Skip to content

谢明辉 / recv_data_longhua

Go to a project
Commit ab87334c by xmh
Options

视频推送

1 parent 880a79a0
Inline Side-by-side
Showing 10 changed files with 170 additions and 46 deletions
src/main/java/com/viontech/Application.java
......@@ -2,6 +2,7 @@ package com.viontech;
import com.viontech.model.KeepAlive;
import com.viontech.netty.ChannelGroup;
import com.viontech.utils.VideoUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
......@@ -15,6 +16,7 @@ public class Application {
public final static long REAL_TIME = 0x00000001L;
public final static long NOT_REAL_TIME = 0x00000002L;
public static void main(String[] args) {
try {
SpringApplication.run(Application.class, args);
......@@ -27,4 +29,9 @@ public class Application {
public void keepAlive() {
ChannelGroup.broadcast(new KeepAlive());
}
@Scheduled(fixedRate = 1000 * 60 * 10)
public void cleanVideo() {
VideoUtil.clean();
}
}
src/main/java/com/viontech/controller/DataController.java
......@@ -3,12 +3,17 @@ package com.viontech.controller;
import com.viontech.model.BaseModel;
import com.viontech.netty.ChannelGroup;
import com.viontech.process.ProcessService;
import com.viontech.utils.VideoUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.multipart.MultipartHttpServletRequest;
import javax.annotation.Resource;
import java.io.IOException;
/**
* .
......@@ -32,5 +37,12 @@ public class DataController {
return "success";
}
@PostMapping("/video")
public Object receiveVideo(@RequestParam(name = "file") MultipartFile file, MultipartHttpServletRequest request) throws IOException {
String refid = request.getParameter("refid");
VideoUtil.addVideo(refid, file.getBytes());
return "success";
}
}
src/main/java/com/viontech/ftp/FTPClientHelper.java
......@@ -211,9 +211,10 @@ public class FTPClientHelper {
} catch (InterruptedException e) {
}
}
log.error("通过ftp存储文件" + remote + "时发生异常 ，开始尝试存储到本地路径" + tempPath + "下", throwException);
return storeTemp(remote, content);
// log.error("通过ftp存储文件" + remote + "时发生异常 ，开始尝试存储到本地路径" + tempPath + "下", throwException);
log.error("通过ftp存储文件" + remote + "时发生异常");
// return storeTemp(remote, content);
return false;
}
private boolean makeDirectorys(FTPClient ftpclient, String remote) throws Exception {
......
src/main/java/com/viontech/process/BehaviorProcess.java
......@@ -2,18 +2,20 @@ package com.viontech.process;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.viontech.ftp.FTPClientHelper;
import com.viontech.model.BaseModel;
import com.viontech.model.BehaviorModel;
import com.viontech.utils.DateUtil;
import com.viontech.utils.IntUtils;
import com.viontech.utils.VideoUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.util.Base64;
import java.util.Date;
import java.util.Optional;
import java.util.Properties;
......@@ -25,6 +27,8 @@ public class BehaviorProcess implements Process {
private Properties illegalTypeProp;
@Resource
private Properties eventProp;
@Resource
private FTPClientHelper ftpClientHelper;
@Override
public BaseModel process(JSONObject jsonObject) throws ParseException {
......@@ -34,26 +38,13 @@ public class BehaviorProcess implements Process {
JSONObject eventData = jsonObject.getJSONObject("event_data");
JSONArray videos = jsonObject.getJSONArray("video");
JSONArray pics = jsonObject.getJSONArray("pics");
if (videos != null && videos.size() > 0) {
JSONObject video = videos.getJSONObject(0);
String startStr = video.getString("start_dt");
String endStr = video.getString("end_dt");
if (startStr != null && !"".equals(startStr.trim())) {
long start = DateUtil.parse(DateUtil.FORMAT_FULL, startStr,8).getTime();
model.setRecordingStartSecond((int) (start / 1000));
model.setRecordingStartMillisecond((int) (start % 1000));
}
if (endStr != null && !"".equals(endStr.trim())) {
long end = DateUtil.parse(DateUtil.FORMAT_FULL, endStr,8).getTime();
model.setRecordingEndSecond((int) (end / 1000));
model.setRecordingEndMillisecond((int) (end % 1000));
}
}
String deviceSerialnum = jsonObject.getString("vchan_refid");
String eventDtStr = jsonObject.getString("event_dt");
Date eventTime = new Date();
if (eventDtStr != null) {
long time = DateUtil.parse(DateUtil.FORMAT_FULL, eventDtStr,8).getTime();
eventTime = DateUtil.parse(DateUtil.FORMAT_FULL, eventDtStr, 8);
long time = eventTime.getTime();
model.setEventStartSecond((int) (time / 1000));
model.setEventStartMillisecond((int) (time % 1000));
//结束时间暂定和开始时间一致
......@@ -65,12 +56,42 @@ public class BehaviorProcess implements Process {
if (eventType != null) {
// eventType
String eventStr = eventProp.getProperty(eventType, "0x00000000");
int event = Integer.parseInt(eventStr, 16);
int event = Integer.getInteger(eventStr, 16);
model.setEventCode(event);
}
if (videos != null && videos.size() > 0) {
JSONObject video = videos.getJSONObject(0);
String startStr = video.getString("start_dt");
String endStr = video.getString("end_dt");
if (startStr != null && !"".equals(startStr.trim())) {
Date startTime = DateUtil.parse(DateUtil.FORMAT_FULL, startStr, 8);
long start = startTime.getTime();
model.setRecordingStartSecond((int) (start / 1000));
model.setRecordingStartMillisecond((int) (start % 1000));
}
if (endStr != null && !"".equals(endStr.trim())) {
long end = DateUtil.parse(DateUtil.FORMAT_FULL, endStr, 8).getTime();
model.setRecordingEndSecond((int) (end / 1000));
model.setRecordingEndMillisecond((int) (end % 1000));
}
String filename = video.getString("ofilename");
String refid = video.getString("unid");
if (filename != null && !"".equals(filename)) {
String filePath = "";
try {
filePath = Process.getFileFTPPath(eventTime, deviceSerialnum, filename);
// 提交推送任务
VideoUtil.submitTask(refid, filePath, ftpClientHelper);
} catch (Exception ignore) {
}
byte[] videoPathBytes = filePath.getBytes(Charset.forName("GBK"));
System.arraycopy(videoPathBytes, 0, model.getEventSnapshotPath(), 0, videoPathBytes.length);
}
}
if (pics != null && pics.size() > 0) {
JSONObject pic1 = pics.getJSONObject(1);
JSONObject pic1 = pics.getJSONObject(0);
String picBase64 = pic1.getString("pic_base64");
if (picBase64 != null) {
byte[] picBytes = Base64.getDecoder().decode(picBase64);
......@@ -102,9 +123,6 @@ public class BehaviorProcess implements Process {
byte[] locationNameBytes = locationName.getBytes(Charset.forName("GBK"));
System.arraycopy(locationNameBytes, 0, model.getEventLocation(), 0, Math.min(locationNameBytes.length, 32));
}
// todo 录像路径
byte[] videoPathBytes = "testVideoPath".getBytes(Charset.forName("GBK"));
System.arraycopy(videoPathBytes, 0, model.getEventSnapshotPath(), 0, videoPathBytes.length);
}
if (speed != null) {
......@@ -120,7 +138,7 @@ public class BehaviorProcess implements Process {
JSONObject body = vehicle.getJSONObject("body");
if (plate != null) {
String text = plate.getString("text");
if(StringUtils.isBlank(text) || StringUtils.equals(text,"无牌")){
if (StringUtils.isBlank(text) || StringUtils.equals(text, "无牌")) {
text = "11111111";
}
byte[] bytes = text.getBytes(Charset.forName("GBK"));
......
src/main/java/com/viontech/process/Process.java
......@@ -27,7 +27,7 @@ public interface Process {
ip = "null";
LOGGER.error("", e);
}
return "video/" + yyyyMMdd + "/" + ip + "/" + deviceId + "/" + fileName;
return "12401/video/" + yyyyMMdd + "/" + ip + "/" + deviceId + "/" + fileName;
}
static byte[] downloadFile(String url) {
......
src/main/java/com/viontech/process/ProcessService.java
......@@ -43,21 +43,17 @@ public class ProcessService {
//违法
case "traffic":
if ("vehicle".equals(eventType)) {
log.info("收到繁星发来的一条违法记录");
log.warn("收到繁星发来的一条违法记录");
return trafficProcess.process(jsonObject);
}
// 流量
else if ("tflow".equals(eventType)) {
log.info("收到繁星发来的一条交通流量记录");
log.warn("收到繁星发来的一条交通流量记录");
return flowProcess.process(jsonObject);
}
return null;
//综治
case "behavior":
log.info("收到繁星发来的一条综治记录");
log.warn("收到繁星发来的一条综治记录");
return behaviorProcess.process(jsonObject);
default:
log.error("收到繁星发来的一条记录，不在博康接收范围内");
......
src/main/java/com/viontech/process/TrafficProcess.java
......@@ -7,12 +7,13 @@ import com.viontech.model.BaseModel;
import com.viontech.model.TrafficModel;
import com.viontech.utils.DateUtil;
import com.viontech.utils.IntUtils;
import com.viontech.utils.VideoUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.util.Base64;
import java.util.Date;
......@@ -20,6 +21,7 @@ import java.util.Optional;
import java.util.Properties;
@Component
@Slf4j
public class TrafficProcess implements Process {
@Resource
......@@ -47,6 +49,7 @@ public class TrafficProcess implements Process {
if (illegal != null && illegal.getInteger("state") == 0) {
return null;
}
log.info("收到繁星发来的一条违法记录");
JSONArray pics = jsonObject.getJSONArray("pics");
model.setSerialNum(IntUtils.next());
String deviceSerialnum = jsonObject.getString("vchan_refid");
......@@ -60,7 +63,7 @@ public class TrafficProcess implements Process {
}
// ------------------------ 车辆信息 车牌信息 --------------
String eventTimeStr = jsonObject.getString("event_dt");
Date eventTime = DateUtil.parse(DateUtil.FORMAT_FULL, eventTimeStr,8);
Date eventTime = DateUtil.parse(DateUtil.FORMAT_FULL, eventTimeStr, 8);
long eventTimeL = eventTime.getTime();
model.setDiscernTime(BaseModel.toInt(eventTimeL / 1000));
model.setDiscernMillisecondTime(BaseModel.toInt(eventTimeL % 1000));
......@@ -84,19 +87,20 @@ public class TrafficProcess implements Process {
private void video(TrafficModel model, JSONArray video, String deviceSerialnum, Date eventTime) {
if (video != null && video.size() > 0) {
JSONObject item = video.getJSONObject(0);
String src_url = item.getString("src_url");
if (src_url == null || "".equals(src_url)) {
String filename = item.getString("ofilename");
// unid 应该就是 refid
String refid = item.getString("unid");
if (filename == null || "".equals(filename)) {
return;
}
// todo 下载录像
byte[] bytes = Process.downloadFile(src_url);
String filename = item.getString("ofilename");
String filePath = "";
String filePath;
try {
filePath = Process.getFileFTPPath(eventTime, deviceSerialnum, filename);
// 提交推送任务
VideoUtil.submitTask(refid, filePath, ftpClientHelper);
} catch (Exception ignore) {
return;
}
// ftpClientHelper.storeFile(filePath, bytes);
byte[] videoPathBytes = filePath.getBytes(Charset.forName("GBK"));
System.arraycopy(videoPathBytes, 0, model.getVideoPath(), 0, videoPathBytes.length);
}
......@@ -133,10 +137,9 @@ public class TrafficProcess implements Process {
String fileFTPPath = "";
try {
fileFTPPath = Process.getFileFTPPath(eventTime, deviceSerialnum, filename);
ftpClientHelper.storeFile(fileFTPPath, pic1Bytes);
} catch (Exception ignore) {
}
ftpClientHelper.storeFile(fileFTPPath, pic1Bytes);
byte[] pic1UrlBytes = fileFTPPath.getBytes();
System.arraycopy(pic1UrlBytes, 0, model.getPicturePath(), 0, Math.min(128, pic1UrlBytes.length));
}
......@@ -148,7 +151,7 @@ public class TrafficProcess implements Process {
// ----------------- 车牌 ------------------
if (plate != null) {
String palteNum = plate.getString("text");
if(StringUtils.isBlank(palteNum) || StringUtils.equals(palteNum,"无牌")){
if (StringUtils.isBlank(palteNum) || StringUtils.equals(palteNum, "无牌")) {
palteNum = "11111111";
}
byte[] text = palteNum.getBytes(Charset.forName("GBK"));
......@@ -230,11 +233,11 @@ public class TrafficProcess implements Process {
long redStart = 0;
long redEnd = 0;
if (redStartStr != null && !"".equals(redStartStr.trim())) {
redStart = DateUtil.parse(DateUtil.FORMAT_FULL, redStartStr,8).getTime();
redStart = DateUtil.parse(DateUtil.FORMAT_FULL, redStartStr, 8).getTime();
}
String redEndStr = detail.getString("end_dt");
if (redEndStr != null && !"".equals(redEndStr.trim())) {
redEnd = DateUtil.parse(DateUtil.FORMAT_FULL, redEndStr,8).getTime();
redEnd = DateUtil.parse(DateUtil.FORMAT_FULL, redEndStr, 8).getTime();
}
model.setRedLightBeginTime((int) (redStart / 1000));
model.setRedLightBeginMillisecondTime((int) (redStart % 1000));
......
src/main/java/com/viontech/utils/DateUtil.java
......@@ -2,10 +2,12 @@ package com.viontech.utils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.context.i18n.LocaleContextHolder;
import sun.util.resources.cldr.es.CalendarData_es_AR;
import java.text.DateFormatSymbols;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.util.*;
public class DateUtil extends DateUtils {
......@@ -1327,6 +1329,7 @@ public class DateUtil extends DateUtils {
}
public static Date parse(String dateFormatStr, String dateStr,int offset) throws ParseException {
Locale locale = LocaleContextHolder.getLocale();
Date parse = parse(dateFormatStr, dateStr, locale);
return DateUtils.addHours(parse, offset);
......
src/main/java/com/viontech/utils/VideoUtil.java 0 → 100644
package com.viontech.utils;
import com.viontech.ftp.FTPClientHelper;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* .
*
* @author 谢明辉
* @date 2020/9/25
*/
public class VideoUtil {
private final static ConcurrentHashMap<String, byte[]> VIDEO_MAP = new ConcurrentHashMap<>();
private final static ConcurrentHashMap<String, Long> VIDEO_EXPIRE_MAP = new ConcurrentHashMap<>();
private final static ThreadPoolExecutor POOL = new ThreadPoolExecutor(20, 30, 1, TimeUnit.HOURS, new LinkedBlockingQueue<>(), new ThreadPoolExecutor.DiscardPolicy());
public synchronized static void addVideo(String refId, byte[] fileBytes) {
VIDEO_MAP.put(refId, fileBytes);
VIDEO_EXPIRE_MAP.put(refId, System.currentTimeMillis());
}
public synchronized static byte[] getVideo(String refId) {
return VIDEO_MAP.remove(refId);
}
public synchronized static void clean() {
for (Map.Entry<String, Long> entry : VIDEO_EXPIRE_MAP.entrySet()) {
if ((System.currentTimeMillis() - entry.getValue()) > TimeUnit.HOURS.toMillis(3)) {
getVideo(entry.getKey());
}
}
}
public static void submitTask(String refId, String filePath, FTPClientHelper ftpClientHelper) {
POOL.submit(new VideoPushRunnable(refId, filePath, ftpClientHelper));
}
@Slf4j
private static class VideoPushRunnable implements Runnable {
private final String refId;
private final String filePath;
private final FTPClientHelper ftpClientHelper;
public VideoPushRunnable(String refId, String filePath, FTPClientHelper ftpClientHelper) {
this.refId = refId;
this.filePath = filePath;
this.ftpClientHelper = ftpClientHelper;
}
@Override
public void run() {
try {
long begin = System.currentTimeMillis();
while (System.currentTimeMillis() - begin < TimeUnit.MINUTES.toMillis(5)) {
byte[] video = VideoUtil.getVideo(refId);
if (video != null) {
ftpClientHelper.storeFile(filePath, video);
log.info("视频推送成功,refId:[{}],path:[{}]", refId, filePath);
return;
}
Long aLong = VideoUtil.VIDEO_EXPIRE_MAP.get(refId);
if (aLong != null) {
log.info("视频已发送,refId:[{}],path:[{}]", refId, filePath);
return;
}
TimeUnit.SECONDS.sleep(10);
log.info("视频未找到:[{}]", refId);
}
} catch (Exception e) {
log.error("推送视频出错,refId:[" + refId + "],path:[" + filePath + "]", e);
}
}
}
}
src/main/resources/application.yml
......@@ -6,6 +6,9 @@ spring:
jackson:
time-zone: GMT+8
date-format: yyyy-MM-dd HH:mm:ss
servlet:
multipart:
max-file-size: 20MB
server:
port: 30000
servlet:
......
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!