From 8f69b1e95bb7529f47c25a3409d943e232d68cde Mon Sep 17 00:00:00 2001 From: yinhuaiwei Date: Thu, 15 Oct 2026 17:04:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B1=87=E8=81=9A=E7=AB=AF=20nginx=20?= =?UTF-8?q?=E6=8C=89=20stationCode=20=E5=B0=86=E5=89=8D=E7=AB=AF=E5=AF=B9?= =?UTF-8?q?=E5=9B=BE=E7=89=87=E3=80=81=E6=96=87=E4=BB=B6=E3=80=81=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E6=B5=81=E7=9A=84=E8=AF=B7=E6=B1=82=E5=8F=8D=E4=BB=A3?= =?UTF-8?q?=E5=88=B0=E5=90=84=E7=AB=99=E7=AB=AF=EF=BC=9B=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E6=B5=81=E7=9A=84=E5=90=AF=E5=81=9C=E7=94=B1=E6=B1=87=E8=81=9A?= =?UTF-8?q?=E7=9B=B4=E6=8E=A5=E8=B0=83=E7=94=A8=E7=AB=99=E7=AB=AF=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/inspect/agg/config/AggProperties.java | 14 +- .../agg/service/AggStatisticService.java | 62 ++- .../agg/support/StationFileUrlResolver.java | 174 +++++++++ .../agg/video/RealtimeStreamBinding.java | 33 -- .../agg/video/RealtimeStreamRegistry.java | 65 ---- .../agg/video/RealtimeStreamService.java | 340 ----------------- .../agg/video/RealtimeViewerRegistry.java | 105 ----- .../com/inspect/agg/video/StationStream.java | 18 - .../agg/video/StationStreamRegistry.java | 35 -- .../StationDirectStreamCleanupScheduler.java} | 14 +- .../direct/StationDirectStreamRegistry.java | 100 +++++ .../direct/StationDirectStreamService.java | 314 +++++++++++++++ .../agg/web/AggDefectRecordController.java | 21 + .../web/AggInspectionReportController.java | 16 + .../inspect/agg/web/AggVideoController.java | 14 +- .../agg/web/StationProxyController.java | 196 +++++++++- .../inspect/agg/web/VideoProxyController.java | 358 ------------------ .../src/main/resources/application.yml | 11 +- 18 files changed, 910 insertions(+), 980 deletions(-) create mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/support/StationFileUrlResolver.java delete mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamBinding.java delete mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamRegistry.java delete mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamService.java delete mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeViewerRegistry.java delete mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStream.java delete mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStreamRegistry.java rename inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/{scheduler/RealtimeViewerCleanupScheduler.java => video/direct/StationDirectStreamCleanupScheduler.java} (52%) create mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamRegistry.java create mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamService.java delete mode 100644 inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/VideoProxyController.java diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/config/AggProperties.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/config/AggProperties.java index f19b474..b8293e5 100644 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/config/AggProperties.java +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/config/AggProperties.java @@ -20,6 +20,8 @@ public class AggProperties { private Video video = new Video(); + private File file = new File(); + private Http http = new Http(); @Data @@ -64,6 +66,8 @@ public class AggProperties { private String app = "live"; /** 播放地址对办公网络暴露的地址（默认同 zlmBaseUrl） */ private String publicBaseUrl = ""; + /** 站端 ZLM 直连流的集控 nginx 地址；留空时按当前请求 Host 自动拼接 */ + private String nginxBaseUrl = ""; /** 是否启用 RTSP TCP 传输（0=TCP, 1=UDP） */ private int rtspTransport = 0; /** 拉流重试次数（-1=无限） */ @@ -75,9 +79,17 @@ public class AggProperties { } + @Data + public static class File { + /** 站端图片/文件反代的集控 nginx 地址；留空时按当前请求 Host 自动拼接 */ + private String nginxBaseUrl = ""; + /** 是否在 /agg/proxy 通用代理的 JSON 响应中递归改写图片地址 */ + private boolean rewriteProxyResponse = true; + } + @Data public static class Http { private int connectTimeoutMs = 10_000; private int readTimeoutMs = 30_000; } -} \ No newline at end of file +} diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/service/AggStatisticService.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/service/AggStatisticService.java index 86b5ed7..635ade6 100644 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/service/AggStatisticService.java +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/service/AggStatisticService.java @@ -5,6 +5,8 @@ import java.time.YearMonth; import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.LinkedHashMap; +import java.util.HashSet; +import java.util.Set; import java.util.List; import java.util.Map; @@ -27,6 +29,7 @@ public class AggStatisticService { private static final DateTimeFormatter MONTH_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM"); private static final String MONTH = "month"; private static final String QUARTER = "quarter"; + private static final String YEAR = "year"; private final AggAnalysisDefectMonthMapper defectMonthMapper; private final AggPatrolTaskStatusMapper taskStatusMapper; @@ -34,7 +37,7 @@ public class AggStatisticService { public List> defectStatistic(String stationCode, String algCategory, String dateType) { - AlgCategory category = AlgCategory.fromLabel(algCategory); + List categories = parseCategories(algCategory); String normalizedType = normalizeDateType(dateType); YearMonth current = YearMonth.now(); LocalDate begin; @@ -43,16 +46,32 @@ public class AggStatisticService { YearMonth currentQuarter = quarterStart(current); begin = currentQuarter.minusMonths(9).atDay(1); endExclusive = currentQuarter.plusMonths(3).atDay(1); + } else if (YEAR.equals(normalizedType)) { + begin = LocalDate.of(current.getYear() - 2, 1, 1); + endExclusive = LocalDate.of(current.getYear() + 1, 1, 1); } else { begin = current.minusMonths(11).atDay(1); endExclusive = current.plusMonths(1).atDay(1); } List rows = defectMonthMapper.selectRange(stationCode, begin, endExclusive); - Map monthlyMap = toMonthlyMap(rows, category == null ? "" : category.getCode()); + Map monthlyMap = toMonthlyMap(rows, categories); String stationName = stationDisplayName(stationCode); List> result = new ArrayList<>(); + if (YEAR.equals(normalizedType)) { + int currentYear = current.getYear(); + for (int year = currentYear - 2; year <= currentYear; year++) { + Counts counts = new Counts(); + for (int monthOffset = 0; monthOffset < 12; monthOffset++) { + counts.add(monthlyMap.get(YearMonth.of(year, 1).plusMonths(monthOffset))); + } + result.add(toPeriodResult(counts, String.valueOf(year), + normalizedType, algCategory, stationCode, stationName)); + } + return result; + } + if (QUARTER.equals(normalizedType)) { YearMonth firstQuarter = quarterStart(current).minusMonths(9); for (int i = 0; i < 4; i++) { @@ -108,10 +127,17 @@ public class AggStatisticService { return toInspectionResult(category.getLabel(), taskStats, defectStats, stationCode, stationName); } - private Map toMonthlyMap(List rows, String pointType) { + private Map toMonthlyMap(List rows, + List categories) { + Set codes = new HashSet<>(); + for (AlgCategory category : categories) { + codes.add(category.getCode()); + } Map result = new LinkedHashMap<>(); for (AggAnalysisDefectMonth row : rows) { - if (!pointType.equals(emptyIfNull(row.getPointType()))) { + String type = emptyIfNull(row.getPointType()); + boolean matched = categories.isEmpty() ? "".equals(type) : codes.contains(type); + if (!matched) { continue; } YearMonth month = YearMonth.from(row.getStatMonth()); @@ -172,8 +198,34 @@ public class AggStatisticService { return stationCode; } + private static List parseCategories(String algCategory) { + List categories = new ArrayList<>(); + Set codes = new HashSet<>(); + if (!StringUtils.hasText(algCategory)) { + return categories; + } + for (String part : algCategory.split(",")) { + String label = part.trim(); + if (label.isEmpty()) { + continue; + } + AlgCategory category = AlgCategory.fromLabel(label); + if (codes.add(category.getCode())) { + categories.add(category); + } + } + return categories; + } + private static String normalizeDateType(String dateType) { - return QUARTER.equalsIgnoreCase(dateType == null ? "" : dateType.trim()) ? QUARTER : MONTH; + String value = dateType == null ? "" : dateType.trim(); + if (QUARTER.equalsIgnoreCase(value)) { + return QUARTER; + } + if (YEAR.equalsIgnoreCase(value)) { + return YEAR; + } + return MONTH; } private static YearMonth quarterStart(YearMonth month) { diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/support/StationFileUrlResolver.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/support/StationFileUrlResolver.java new file mode 100644 index 0000000..35d6ef8 --- /dev/null +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/support/StationFileUrlResolver.java @@ -0,0 +1,174 @@ +package com.inspect.agg.support; + +import java.nio.charset.StandardCharsets; +import java.util.regex.Pattern; + +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import com.inspect.agg.config.AggProperties; + + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * 将站端 SFTP 文件相对路径改写为集控 nginx 代理地址。 + * 图片：{nginxBase}/img/{stationCode}/{原始路径}；文档（巡视报告等）：{nginxBase}/file/{stationCode}/{原始路径}。 + * 集控 nginx 按 stationCode 转发到站端 nginx，由站端规则补全/剥除固定前缀后转到文件/图片服务。 + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class StationFileUrlResolver { + + /** 图片扩展名 */ + private static final Pattern IMAGE_EXT = Pattern.compile( + "(?i).+\\.(jpg|jpeg|png|gif|bmp|webp)(\\?.*)?$"); + + /** 可下载的文档/附件扩展名 */ + private static final Pattern DOC_EXT = Pattern.compile( + "(?i).+\\.(doc|docx|xls|xlsx|ppt|pptx|pdf|zip|rar|7z|txt|csv)(\\?.*)?$"); + + /** + * 站端存储内部路径形态，第一段为存储实例编号，如： + * 图片 /1/2026/09/29/1796/CCD/x.jpg，报告 /1/Model/xxx.docx。 + */ + private static final Pattern STORE_PATH = Pattern.compile( + "(?i)^/?\\d{1,4}/.+\\.[0-9a-z]+(\\?.*)?$"); + + /** 字段名命中即视为图片字段 */ + private static final Pattern IMAGE_FIELD = Pattern.compile( + "(?i).*(img|image|photo|pic|picture).*"); + + /** 字段名命中即视为文件字段 */ + private static final Pattern FILE_FIELD = Pattern.compile( + "(?i).*(file|filepath|fileurl|attach|reporturl).*"); + + private final AggProperties props; + + /** 明确是图片字段的出参（缺陷列表的 img 等）。支持逗号分隔多图。 */ + public String resolveImage(String stationCode, String value) { + return rewrite(stationCode, null, value, true, false); + } + + /** 明确是文件字段的出参（巡视报告 filePath 等）。 */ + public String resolveFile(String stationCode, String value) { + return rewrite(stationCode, null, value, false, true); + } + + /** + * 通用代理响应的不确定字段：仅当值看起来是站端存储的图片/文档路径时才改写， + * 避免误伤地图 XML 等无关 filePath。 + */ + public String rewriteIfStationFile(String stationCode, String fieldName, String value) { + return rewrite(stationCode, fieldName, value, false, false); + } + + private String rewrite(String stationCode, String fieldName, String value, + boolean knownImageField, boolean knownFileField) { + if (!StringUtils.hasText(value)) { + return value; + } + String[] tokens = value.split(","); + boolean changed = false; + for (int i = 0; i < tokens.length; i++) { + String rewritten = rewriteOne(stationCode, fieldName, tokens[i], + knownImageField, knownFileField); + if (rewritten != null && !rewritten.equals(tokens[i])) { + tokens[i] = rewritten; + changed = true; + } + } + return changed ? String.join(",", tokens) : value; + } + + private String rewriteOne(String stationCode, String fieldName, String raw, + boolean knownImageField, boolean knownFileField) { + if (raw == null) { + return null; + } + String path = raw.trim(); + if (path.isEmpty() + || path.startsWith("http://") || path.startsWith("https://") + || path.startsWith("data:") || path.startsWith("//")) { + return raw; + } + + boolean image = IMAGE_EXT.matcher(path).matches(); + boolean doc = DOC_EXT.matcher(path).matches(); + boolean storePath = STORE_PATH.matcher(path).matches(); + String field = StringUtils.hasText(fieldName) ? fieldName : ""; + boolean imageField = IMAGE_FIELD.matcher(field).matches(); + boolean fileField = FILE_FIELD.matcher(field).matches(); + + boolean isImage = image && (knownImageField || storePath || imageField); + boolean isDoc = doc && (knownFileField || storePath || fileField); + if (!isImage && !isDoc) { + return raw; + } + + if (!path.startsWith("/")) { + path = "/" + path; + } + String segment = isImage ? "img" : "file"; + log.info("mediaBase: {}", mediaBase()); + return mediaBase() + "/" + segment + "/" + encodeSegment(stationCode) + encodePathSegments(path); + } + + /** + * 对存储路径逐段做百分号编码（保留 /），防止中文、空格（如报告文件名中的空格） + * 导致 nginx 400。nginx 转发到上游时保持编码形式，由文件服务解码为真实文件名。 + * 已包含百分号编码（%XX）的段视为已编码，不重复处理；查询串原样保留。 + */ + private static String encodePathSegments(String pathAndQuery) { + int queryIndex = pathAndQuery.indexOf('?'); + String path = queryIndex >= 0 ? pathAndQuery.substring(0, queryIndex) : pathAndQuery; + String query = queryIndex >= 0 ? pathAndQuery.substring(queryIndex) : ""; + String[] segments = path.split("/", -1); + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < segments.length; i++) { + if (i > 0) { + sb.append('/'); + } + if (!segments[i].isEmpty()) { + sb.append(encodeSegment(segments[i])); + } + } + return sb.append(query).toString(); + } + + /** + * 只编码必须编码的字符：空格和非 ASCII（中文等），其余 RFC 3986 路径允许的字符 + * （含冒号）保持原样，与浏览器地址栏行为一致。 + * 不能用 URLEncoder：它会把冒号编成 %3A，经过 nginx 变量 proxy_pass 原样透传后， + * 上游文件服务会返回 400。已包含 %XX 的段视为已编码，不重复处理。 + */ + private static String encodeSegment(String segment) { + if (segment.contains("%")) { + return segment; + } + StringBuilder sb = new StringBuilder(segment.length() + 16); + for (int i = 0; i < segment.length(); ) { + int cp = segment.codePointAt(i); + if (cp <= 0x20 || cp > 0x7E) { + byte[] bytes = new String(Character.toChars(cp)).getBytes(StandardCharsets.UTF_8); + for (byte b : bytes) { + sb.append('%').append(String.format("%02X", b & 0xFF)); + } + } else { + sb.append((char) cp); + } + i += Character.charCount(cp); + } + return sb.toString(); + } + /** + * 配置为空时返回空串，出参为相对地址 /img、/file，由前端拼接 /app/htjc6/htjcImage 前缀； + * 显式配置 agg.file.nginx-base-url 时才输出绝对地址。 + */ + private String mediaBase() { + String configured = props.getFile().getNginxBaseUrl(); + return StringUtils.hasText(configured) ? configured.trim().replaceAll("/+$", "") : ""; + } +} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamBinding.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamBinding.java deleted file mode 100644 index 4f20379..0000000 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamBinding.java +++ /dev/null @@ -1,33 +0,0 @@ -package com.inspect.agg.video; - -import lombok.Data; - -/** - * 汇聚实时流与站端实时流之间的绑定关系。 - */ -@Data -public class RealtimeStreamBinding { - - /** 站点编码。 */ - private String stationCode; - - /** 原始摄像机编码，保留 cameraCode 中 # 后的附加信息。 */ - private String cameraCode; - - /** 汇聚 ZLMediaKit 中使用的确定性流 ID。 */ - private String aggStreamId; - - /** 站端实时流 ID；直连 RTSP 模式为空。 */ - private String stationStreamId; - - /** 汇聚 ZLMediaKit 拉取的源 RTSP 地址。 */ - private String sourceRtsp; - - /** 流类型，透传给站端 stop 接口。 */ - private Integer streamType; - - /** 是否由站端启动和停止；false 表示调用方直接传入 RTSP。 */ - private boolean stationManaged = true; - - private long createdAt = System.currentTimeMillis(); -} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamRegistry.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamRegistry.java deleted file mode 100644 index 9c42985..0000000 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamRegistry.java +++ /dev/null @@ -1,65 +0,0 @@ -package com.inspect.agg.video; - -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.springframework.stereotype.Component; -import org.springframework.util.StringUtils; - -/** - * 实时流绑定注册表：按站点 + 摄像机维护汇聚流与站端流的关系。 - */ -@Component -public class RealtimeStreamRegistry { - - private final Map byCamera = new ConcurrentHashMap<>(); - private final Map aggStreamIdToCameraKey = new ConcurrentHashMap<>(); - - public static String cameraKey(String stationCode, String cameraCode) { - String base = cameraCode; - if (base != null && base.contains("#")) { - base = base.substring(0, base.indexOf('#')); - } - return stationCode + "|" + base; - } - - public void put(RealtimeStreamBinding binding) { - if (binding == null - || !StringUtils.hasText(binding.getStationCode()) - || !StringUtils.hasText(binding.getCameraCode()) - || !StringUtils.hasText(binding.getAggStreamId())) { - throw new IllegalArgumentException("实时流绑定信息不完整"); - } - String key = cameraKey(binding.getStationCode(), binding.getCameraCode()); - byCamera.put(key, binding); - aggStreamIdToCameraKey.put(binding.getAggStreamId(), key); - } - - public RealtimeStreamBinding getByCamera(String stationCode, String cameraCode) { - if (!StringUtils.hasText(stationCode) || !StringUtils.hasText(cameraCode)) { - return null; - } - return byCamera.get(cameraKey(stationCode, cameraCode)); - } - - public RealtimeStreamBinding getByAggStreamId(String aggStreamId) { - String cameraKey = aggStreamIdToCameraKey.get(aggStreamId); - return cameraKey == null ? null : byCamera.get(cameraKey); - } - - public RealtimeStreamBinding removeByAggStreamId(String aggStreamId) { - String cameraKey = aggStreamIdToCameraKey.remove(aggStreamId); - if (cameraKey == null) { - return null; - } - RealtimeStreamBinding binding = byCamera.remove(cameraKey); - if (binding == null) { - return null; - } - if (!aggStreamId.equals(binding.getAggStreamId())) { - // 正常不会发生；若同摄像机绑定已被替换，恢复反向索引，避免误删新绑定。 - aggStreamIdToCameraKey.put(binding.getAggStreamId(), cameraKey); - } - return binding; - } -} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamService.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamService.java deleted file mode 100644 index 6c68957..0000000 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeStreamService.java +++ /dev/null @@ -1,340 +0,0 @@ -package com.inspect.agg.video; - -import java.net.MalformedURLException; -import java.net.URL; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.springframework.stereotype.Component; -import org.springframework.util.StringUtils; - -import com.fasterxml.jackson.databind.JsonNode; -import com.inspect.agg.dto.VideoRequest; -import com.inspect.agg.entity.AggStation; -import com.inspect.agg.service.StationRegistry; -import com.inspect.agg.station.StationApiClient; - -import lombok.Data; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; - -/** - * 实时流生命周期服务：统一处理站端启动、汇聚代理、观看引用计数和停止清理。 - */ -@Slf4j -@Component -@RequiredArgsConstructor -public class RealtimeStreamService { - - private final StationRegistry stationRegistry; - private final StationApiClient stationApiClient; - private final ZlmApiClient zlmApiClient; - private final RealtimeViewerRegistry viewerRegistry; - private final RealtimeStreamRegistry streamRegistry; - - /** 单实例部署下按相机串行化 start/stop，避免并发重复创建代理。 */ - private final Map cameraLocks = new ConcurrentHashMap<>(); - - public StartResult start(String stationCode, VideoRequest request) { - if (request == null - || !StringUtils.hasText(stationCode) - || !StringUtils.hasText(request.getCameraCode())) { - throw new IllegalArgumentException("stationCode / cameraCode 不能为空"); - } - String cameraCode = request.getCameraCode(); - Integer streamType = request.getStreamType() == null ? 2 : request.getStreamType(); - String directRtsp = request.getRtspUrl(); - AggStation station = stationRegistry.requireByCode(stationCode); - String aggStreamId = aggStreamId(stationCode, cameraCode); - - synchronized (cameraLock(stationCode, cameraCode)) { - boolean mediaActive = zlmApiClient.streamExists(aggStreamId); - JsonNode existingProxy = zlmApiClient.findStreamProxy(aggStreamId); - boolean proxyActive = existingProxy != null; - RealtimeStreamBinding binding = streamRegistry.getByCamera(stationCode, cameraCode); - boolean stationStartAttempted = false; - try { - if (!mediaActive && !proxyActive) { - // 既无在线媒体也无拉流代理：调站端启动并新建汇聚代理。 - if (StringUtils.hasText(directRtsp)) { - binding = directBinding(stationCode, cameraCode, aggStreamId, directRtsp, streamType); - } else { - binding = startStationStream(station, cameraCode, aggStreamId, streamType); - stationStartAttempted = true; - } - // addStreamProxy 已对 already exists 幂等处理，兜底并发/残留同名代理。 - zlmApiClient.addStreamProxy(aggStreamId, binding.getSourceRtsp()); - } else if (binding == null) { - // ZLM 已有在线媒体或拉流代理，但本机绑定丢失（例如服务重启后）。 - // 不再重复调用站端 start（避免站端重复生成一路流），改用代理源地址补登绑定。 - if (StringUtils.hasText(directRtsp)) { - binding = directBinding(stationCode, cameraCode, aggStreamId, directRtsp, streamType); - } else { - String sourceRtsp = existingProxy == null ? null : existingProxy.path("url").asText(null); - binding = recoveredBinding(stationCode, cameraCode, aggStreamId, streamType, sourceRtsp); - } - } else { - binding.setAggStreamId(aggStreamId); - } - - streamRegistry.put(binding); - viewerRegistry.add(aggStreamId); - // 媒体已注册可播放返回 PLAYING；代理已存在但尚未出流返回 STARTING。 - return new StartResult(aggStreamId, mediaActive); - } catch (RuntimeException e) { - if (stationStartAttempted) { - stopStationQuietly(station, binding, cameraCode, streamType, null); - } - throw e; - } - } - } - /** - * 停止一个观看者。 - * - * @return -1 没有观看票据；0 最后一个观看者已退出并停止流；正数表示剩余观看者数量 - */ - public int stop(String stationCode, VideoRequest request) { - if (request == null - || !StringUtils.hasText(stationCode) - || !StringUtils.hasText(request.getCameraCode())) { - throw new IllegalArgumentException("stationCode / cameraCode 不能为空"); - } - String cameraCode = request.getCameraCode(); - Integer streamType = request.getStreamType() == null ? 2 : request.getStreamType(); - AggStation station = stationRegistry.requireByCode(stationCode); - String aggStreamId = aggStreamId(stationCode, cameraCode); - - synchronized (cameraLock(stationCode, cameraCode)) { - int remaining = viewerRegistry.remove(aggStreamId); - if (remaining != 0) { - return remaining; - } - - boolean deleted; - try { - deleted = zlmApiClient.delStreamProxy(aggStreamId); - } catch (RuntimeException e) { - viewerRegistry.add(aggStreamId); - throw e; - } - - if (!deleted) { - boolean stillExists; - try { - stillExists = zlmApiClient.streamExists(aggStreamId); - } catch (RuntimeException e) { - viewerRegistry.add(aggStreamId); - throw e; - } - if (stillExists) { - viewerRegistry.add(aggStreamId); - throw new IllegalStateException("汇聚 ZLMediaKit 流仍存在，停止失败"); - } - } - - RealtimeStreamBinding binding = streamRegistry.removeByAggStreamId(aggStreamId); - viewerRegistry.clear(aggStreamId); - stopStationQuietly(station, binding, cameraCode, streamType, request.getRtspUrl()); - return 0; - } - } - - /** 定时清理过期观看票据；仅清理没有活跃观看者的汇聚流，并尽力停止对应站端流。 */ - public void cleanupExpired() { - long now = System.currentTimeMillis(); - List expiredStreams = viewerRegistry.expire(now); - for (String aggStreamId : expiredStreams) { - RealtimeStreamBinding existing = streamRegistry.getByAggStreamId(aggStreamId); - String lockKey = existing == null - ? "expired:" + aggStreamId - : RealtimeStreamRegistry.cameraKey(existing.getStationCode(), existing.getCameraCode()); - synchronized (cameraLocks.computeIfAbsent(lockKey, ignored -> new Object())) { - if (viewerRegistry.count(aggStreamId) > 0) { - continue; - } - - boolean deleted; - try { - deleted = zlmApiClient.delStreamProxy(aggStreamId); - } catch (Exception e) { - log.warn("delete expired agg stream failed, stream={} err={}", aggStreamId, e.getMessage()); - continue; - } - if (!deleted) { - try { - if (zlmApiClient.streamExists(aggStreamId)) { - continue; - } - } catch (Exception e) { - log.warn("check expired agg stream failed, stream={} err={}", aggStreamId, e.getMessage()); - continue; - } - } - - RealtimeStreamBinding binding = streamRegistry.removeByAggStreamId(aggStreamId); - viewerRegistry.clear(aggStreamId); - RealtimeStreamBinding target = binding != null ? binding : existing; - if (target == null) { - continue; - } - try { - AggStation station = stationRegistry.requireByCode(target.getStationCode()); - stopStationQuietly(station, target, target.getCameraCode(), target.getStreamType(), null); - log.info("realtime viewer tickets expired, close stream={}", aggStreamId); - } catch (Exception e) { - log.warn("stop expired station stream failed, stream={} err={}", aggStreamId, e.getMessage()); - } - } - } - } - - private RealtimeStreamBinding startStationStream(AggStation station, String cameraCode, - String aggStreamId, Integer streamType) { - JsonNode root = stationApiClient.fetchRealtimeStart(station, cameraCode, streamType); - JsonNode payload = successPayload(root); - String stationStreamId = textAnyDeep(payload, "streamId", "StreamID", "streamID", "id"); - String sourceRtsp = textAnyDeep(payload, "RTSP", "rtspURL", "rtspUrl", "rtsp"); - if (!StringUtils.hasText(sourceRtsp)) { - String stationIp = extractIp(station.getBaseUrl()); - if (!StringUtils.hasText(stationStreamId) || !StringUtils.hasText(stationIp)) { - throw new IllegalStateException("站端未返回有效 streamId / RTSP 地址"); - } - sourceRtsp = String.format("rtsp://%s:8554/live/%s", stationIp, stationStreamId); - } - - RealtimeStreamBinding binding = new RealtimeStreamBinding(); - binding.setStationCode(station.getStationCode()); - binding.setCameraCode(cameraCode); - binding.setAggStreamId(aggStreamId); - binding.setStationStreamId(stationStreamId); - binding.setSourceRtsp(sourceRtsp); - binding.setStreamType(streamType); - binding.setStationManaged(true); - return binding; - } - - private RealtimeStreamBinding directBinding(String stationCode, String cameraCode, - String aggStreamId, String rtspUrl, Integer streamType) { - RealtimeStreamBinding binding = new RealtimeStreamBinding(); - binding.setStationCode(stationCode); - binding.setCameraCode(cameraCode); - binding.setAggStreamId(aggStreamId); - binding.setSourceRtsp(rtspUrl); - binding.setStreamType(streamType); - binding.setStationManaged(false); - return binding; - } - - /** - * 服务重启后依据 ZLM 已有代理恢复内存绑定。 - * 此时不掌握站端 streamId，stop 时仅靠 cameraCode 尽力通知站端停止。 - */ - private RealtimeStreamBinding recoveredBinding(String stationCode, String cameraCode, - String aggStreamId, Integer streamType, String sourceRtsp) { - RealtimeStreamBinding binding = new RealtimeStreamBinding(); - binding.setStationCode(stationCode); - binding.setCameraCode(cameraCode); - binding.setAggStreamId(aggStreamId); - binding.setSourceRtsp(sourceRtsp); - binding.setStreamType(streamType); - binding.setStationManaged(true); - return binding; - } - private void stopStationQuietly(AggStation station, RealtimeStreamBinding binding, - String fallbackCameraCode, Integer fallbackStreamType, - String directRtsp) { - if (binding != null && !binding.isStationManaged()) { - return; - } - if (binding == null && StringUtils.hasText(directRtsp)) { - return; - } - - String cameraCode = binding != null && StringUtils.hasText(binding.getCameraCode()) - ? binding.getCameraCode() : fallbackCameraCode; - String stationStreamId = binding == null ? null : binding.getStationStreamId(); - Integer streamType = binding != null && binding.getStreamType() != null - ? binding.getStreamType() : fallbackStreamType; - if (!StringUtils.hasText(cameraCode)) { - return; - } - - try { - JsonNode root = stationApiClient.fetchRealtimeStop(station, stationStreamId, cameraCode, streamType); - if (root.has("code") && root.path("code").asInt(-1) != 200) { - log.warn("stop station realtime stream returned non-success, station={} camera={} resp={}", - station.getStationCode(), cameraCode, root); - } - } catch (Exception e) { - log.warn("stop station realtime stream failed, station={} camera={} stationStreamId={} err={}", - station.getStationCode(), cameraCode, stationStreamId, e.getMessage()); - } - } - - private Object cameraLock(String stationCode, String cameraCode) { - String key = RealtimeStreamRegistry.cameraKey(stationCode, cameraCode); - return cameraLocks.computeIfAbsent(key, ignored -> new Object()); - } - - private static JsonNode successPayload(JsonNode root) { - if (root.has("code") && root.path("code").asInt(-1) != 200) { - String message = root.path("msg").asText(root.path("message").asText("站端接口返回失败")); - throw new IllegalStateException(message); - } - JsonNode payload = root.path("data"); - return payload.isObject() ? payload : root; - } - - private static String textAnyDeep(JsonNode node, String... fields) { - String value = textAny(node, fields); - if (!StringUtils.hasText(value) && node.path("data").isObject()) { - value = textAny(node.path("data"), fields); - } - return value; - } - - private static String textAny(JsonNode node, String... fields) { - if (node == null || fields == null) { - return null; - } - for (String field : fields) { - String value = text(node, field); - if (StringUtils.hasText(value)) { - return value; - } - } - return null; - } - - private static String text(JsonNode node, String field) { - JsonNode value = node == null ? null : node.get(field); - return value == null || value.isNull() ? null : value.asText(); - } - - /** 汇聚实时流 ID：站点 + 摄像机；去除 cameraCode 中 # 后的附属信息。 */ - private static String aggStreamId(String stationCode, String cameraCode) { - String base = cameraCode.contains("#") ? cameraCode.substring(0, cameraCode.indexOf('#')) : cameraCode; - return (stationCode + "_" + base).replaceAll("[^A-Za-z0-9_]", "_"); - } - - private static String extractIp(String urlString) { - try { - return new URL(urlString).getHost(); - } catch (MalformedURLException e) { - return null; - } - } - - @Data - public static class StartResult { - private final String streamId; - private final boolean existing; - - public StartResult(String streamId, boolean existing) { - this.streamId = streamId; - this.existing = existing; - } - } -} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeViewerRegistry.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeViewerRegistry.java deleted file mode 100644 index 4f99904..0000000 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/RealtimeViewerRegistry.java +++ /dev/null @@ -1,105 +0,0 @@ -package com.inspect.agg.video; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.springframework.stereotype.Component; - -/** - * 实时流观看会话注册表：一个 streamId 对应多个观看票据。 - * start 增加一个票据，stop 只减少一个票据，最后一个观看者退出时才允许删除 ZLM 流。 - */ -@Component -public class RealtimeViewerRegistry { - - /** 观看票据过期时间；当前没有心跳接口，暂用 24 小时作为兜底清理。 */ - private static final long TICKET_TTL_MILLIS = 24L * 60 * 60 * 1000; - - private final Map> tickets = new ConcurrentHashMap<>(); - private final Map lastActivity = new ConcurrentHashMap<>(); - - public void add(String streamId) { - tickets.compute(streamId, (key, current) -> { - List list = current == null ? new ArrayList<>() : current; - synchronized (list) { - list.add(System.currentTimeMillis()); - } - return list; - }); - } - - /** - * 移除一个观看票据。 - * - * @return -1 表示没有票据；0 表示本次移除后无人观看；大于 0 表示仍有 N 个观看者 - */ - public int remove(String streamId) { - final int[] remaining = {-2}; - tickets.compute(streamId, (key, current) -> { - if (current == null || current.isEmpty()) { - remaining[0] = -1; - return null; - } - synchronized (current) { - current.remove(current.size() - 1); - if (current.isEmpty()) { - remaining[0] = 0; - lastActivity.remove(streamId); - return null; - } - remaining[0] = current.size(); - return current; - } - }); - return remaining[0]; - } - - public void touch(String streamId) { - if (streamId != null) { - lastActivity.put(streamId, System.currentTimeMillis()); - } - } - - public long lastActivityAge(String streamId, long now) { - Long last = lastActivity.get(streamId); - return last == null ? Long.MAX_VALUE : now - last; - } - - public void clear(String streamId) { - tickets.remove(streamId); - lastActivity.remove(streamId); - } - - public int count(String streamId) { - List current = tickets.get(streamId); - if (current == null) { - return 0; - } - synchronized (current) { - return current.size(); - } - } - - /** - * 清理过期票据，返回因此没有观看者的 streamId 列表。 - */ - public List expire(long now) { - List closed = new ArrayList<>(); - for (String streamId : new ArrayList<>(tickets.keySet())) { - tickets.computeIfPresent(streamId, (key, current) -> { - synchronized (current) { - current.removeIf(ticketTime -> now - ticketTime > TICKET_TTL_MILLIS); - if (current.isEmpty()) { - closed.add(streamId); - lastActivity.remove(streamId); - return null; - } - return current; - } - }); - } - return closed; - } -} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStream.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStream.java deleted file mode 100644 index 29cec40..0000000 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStream.java +++ /dev/null @@ -1,18 +0,0 @@ -package com.inspect.agg.video; - -import lombok.Data; - -/** - * 站端转流信息（来自 /ivs/api/v1/stream/start）。 - */ -@Data -public class StationStream { - - private String stationCode; - /** 站端 StreamID */ - private String stationStreamId; - /** 站端 ZLM 的 HTTP-FLV / HLS 地址 */ - private String flvUrl; - private String hlsUrl; - private long createdAt = System.currentTimeMillis(); -} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStreamRegistry.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStreamRegistry.java deleted file mode 100644 index 64d99da..0000000 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/StationStreamRegistry.java +++ /dev/null @@ -1,35 +0,0 @@ -package com.inspect.agg.video; - -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.springframework.stereotype.Component; -import org.springframework.util.StringUtils; - -/** - * 会话内的站端流信息注册表（内存，进程重启后失效，前端重新调用 /play 即可）。 - */ -@Component -public class StationStreamRegistry { - - private final Map streams = new ConcurrentHashMap<>(); - - public static String key(String stationCode, String stationStreamId) { - return stationCode + "/" + stationStreamId; - } - - public void put(StationStream stream) { - streams.put(key(stream.getStationCode(), stream.getStationStreamId()), stream); - } - - public StationStream get(String stationCode, String streamId) { - if (!StringUtils.hasText(stationCode) || !StringUtils.hasText(streamId)) { - return null; - } - return streams.get(key(stationCode, streamId)); - } - - public void remove(String stationCode, String streamId) { - streams.remove(key(stationCode, streamId)); - } -} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/scheduler/RealtimeViewerCleanupScheduler.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamCleanupScheduler.java similarity index 52% rename from inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/scheduler/RealtimeViewerCleanupScheduler.java rename to inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamCleanupScheduler.java index 2b33b5b..5ee8067 100644 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/scheduler/RealtimeViewerCleanupScheduler.java +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamCleanupScheduler.java @@ -1,31 +1,29 @@ -package com.inspect.agg.scheduler; +package com.inspect.agg.video.direct; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; -import com.inspect.agg.video.RealtimeStreamService; - import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; /** - * 清理过期的实时观看票据；票据全部过期后关闭对应汇聚和站端实时流。 + * 清理过期的站端直连实时观看票据，票据全部过期后通知站端停流。 */ @Slf4j @Component @RequiredArgsConstructor -public class RealtimeViewerCleanupScheduler { +public class StationDirectStreamCleanupScheduler { private static final long CLEAN_INTERVAL_MS = 30L * 60 * 1000; - private final RealtimeStreamService realtimeStreamService; + private final StationDirectStreamService stationDirectStreamService; @Scheduled(fixedDelay = CLEAN_INTERVAL_MS, initialDelay = CLEAN_INTERVAL_MS) public void cleanup() { try { - realtimeStreamService.cleanupExpired(); + stationDirectStreamService.cleanupExpired(); } catch (Exception e) { - log.error("cleanup expired realtime streams failed: {}", e.getMessage(), e); + log.error("cleanup expired station direct realtime streams failed: {}", e.getMessage(), e); } } } \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamRegistry.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamRegistry.java new file mode 100644 index 0000000..230efbe --- /dev/null +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamRegistry.java @@ -0,0 +1,100 @@ +package com.inspect.agg.video.direct; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import lombok.Data; + +/** + * 站端直连实时流注册表：按站点 + 摄像机维护站端 streamId 和观看票据。 + */ +@Component +public class StationDirectStreamRegistry { + + private static final long TICKET_TTL_MILLIS = 24L * 60 * 60 * 1000; + + private final Map sessions = new ConcurrentHashMap<>(); + private final Map locks = new ConcurrentHashMap<>(); + + public static String cameraKey(String stationCode, String cameraCode) { + String base = cameraCode; + if (base != null && base.contains("#")) { + base = base.substring(0, base.indexOf('#')); + } + return stationCode + "|" + base; + } + + public Object lock(String stationCode, String cameraCode) { + return locks.computeIfAbsent(cameraKey(stationCode, cameraCode), ignored -> new Object()); + } + + public Session get(String stationCode, String cameraCode) { + if (!StringUtils.hasText(stationCode) || !StringUtils.hasText(cameraCode)) { + return null; + } + return sessions.get(cameraKey(stationCode, cameraCode)); + } + + public Session create(String stationCode, String cameraCode, String stationStreamId, Integer streamType) { + Session session = new Session(); + session.setStationCode(stationCode); + session.setCameraCode(cameraCode); + session.setStationStreamId(stationStreamId); + session.setStreamType(streamType); + sessions.put(cameraKey(stationCode, cameraCode), session); + return session; + } + + public Session remove(String stationCode, String cameraCode) { + return sessions.remove(cameraKey(stationCode, cameraCode)); + } + + public List expiredSessions(long now) { + List expired = new ArrayList<>(); + for (Session session : new ArrayList<>(sessions.values())) { + if (session.expireViewers(now) == 0) { + expired.add(session); + } + } + return expired; + } + + @Data + public static class Session { + private String stationCode; + private String cameraCode; + private String stationStreamId; + private String hlsPath; + private String flvPath; + private Integer streamType; + private long createdAt = System.currentTimeMillis(); + private final List viewers = new ArrayList<>(); + + public synchronized int addViewer() { + viewers.add(System.currentTimeMillis()); + return viewers.size(); + } + + public synchronized int removeViewer() { + if (viewers.isEmpty()) { + return -1; + } + viewers.remove(viewers.size() - 1); + return viewers.size(); + } + + public synchronized int viewerCount() { + return viewers.size(); + } + + public synchronized int expireViewers(long now) { + viewers.removeIf(ticketTime -> now - ticketTime > TICKET_TTL_MILLIS); + return viewers.size(); + } + } +} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamService.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamService.java new file mode 100644 index 0000000..50b8f34 --- /dev/null +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/video/direct/StationDirectStreamService.java @@ -0,0 +1,314 @@ +package com.inspect.agg.video.direct; + +import java.net.HttpURLConnection; +import java.net.URI; +import java.net.URL; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; +import org.springframework.web.context.request.RequestContextHolder; +import org.springframework.web.context.request.ServletRequestAttributes; + +import javax.servlet.http.HttpServletRequest; + +import com.fasterxml.jackson.databind.JsonNode; +import com.inspect.agg.config.AggProperties; +import com.inspect.agg.dto.VideoRequest; +import com.inspect.agg.entity.AggStation; +import com.inspect.agg.service.StationRegistry; +import com.inspect.agg.station.StationApiClient; +import com.inspect.agg.video.direct.StationDirectStreamRegistry.Session; + +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * 站端直连实时流：启流和停流都调用站端业务接口，媒体流经集控 nginx 反代站端 ZLM。 + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class StationDirectStreamService { + + private final StationRegistry stationRegistry; + private final StationApiClient stationApiClient; + private final StationDirectStreamRegistry streamRegistry; + private final AggProperties props; + + private final Map cleanupLocks = new ConcurrentHashMap<>(); + + private static final int PROBE_CONNECT_TIMEOUT_MS = 2000; + private static final int PROBE_READ_TIMEOUT_MS = 2000; + + public StartResult start(String stationCode, VideoRequest request) { + if (request == null + || !StringUtils.hasText(stationCode) + || !StringUtils.hasText(request.getCameraCode())) { + throw new IllegalArgumentException("stationCode / cameraCode 不能为空"); + } + String cameraCode = request.getCameraCode(); + Integer streamType = request.getStreamType() == null ? 2 : request.getStreamType(); + AggStation station = stationRegistry.requireByCode(stationCode); + + synchronized (streamRegistry.lock(stationCode, cameraCode)) { + Session existing = streamRegistry.get(stationCode, cameraCode); + if (existing != null) { + if (isStreamLive(existing)) { + existing.addViewer(); + return result(existing, true); + } + log.warn("cached station stream offline, drop session and restart: station={} camera={} streamId={}", + stationCode, cameraCode, existing.getStationStreamId()); + streamRegistry.remove(stationCode, cameraCode); + } + + JsonNode root = stationApiClient.fetchRealtimeStart(station, cameraCode, streamType); + log.info("root: {}", root); + JsonNode payload = successPayload(root); + String stationStreamId = textAnyDeep(payload, "streamId", "StreamID", "streamID", "id"); + String stationHlsUrl = textAnyDeep(payload, "hlsUrl", "hlsURL", "hls", "HLS"); + String stationFlvUrl = textAnyDeep(payload, "flvUrl", "flvURL", "flv", "FLV"); + if (!StringUtils.hasText(stationStreamId)) { + stationStreamId = extractStreamId(stationHlsUrl, stationFlvUrl, + textAnyDeep(payload, "rtspUrl", "rtspURL", "RTSP")); + } + if (!StringUtils.hasText(stationStreamId)) { + throw new IllegalStateException("站端未返回有效 streamId"); + } + + Session session = streamRegistry.create(stationCode, cameraCode, stationStreamId, streamType); + session.setHlsPath(mediaPath(stationHlsUrl, "/live/" + stationStreamId + "/hls.m3u8")); + session.setFlvPath(mediaPath(stationFlvUrl, "/live/" + stationStreamId + ".live.flv")); + session.addViewer(); + return result(session, false); + } + } + + /** + * @return -1 没有本地观看票据但已按 cameraCode 尽力停止；0 最后一个观看者已停止；正数为剩余观看者数量 + */ + public int stop(String stationCode, VideoRequest request) { + if (request == null + || !StringUtils.hasText(stationCode) + || !StringUtils.hasText(request.getCameraCode())) { + throw new IllegalArgumentException("stationCode / cameraCode 不能为空"); + } + String cameraCode = request.getCameraCode(); + Integer streamType = request.getStreamType() == null ? 2 : request.getStreamType(); + AggStation station = stationRegistry.requireByCode(stationCode); + + synchronized (streamRegistry.lock(stationCode, cameraCode)) { + Session session = streamRegistry.get(stationCode, cameraCode); + if (session == null) { + stopStation(station, null, cameraCode, streamType); + return -1; + } + + int remaining = session.removeViewer(); + if (remaining > 0) { + return remaining; + } + + stopStation(station, session.getStationStreamId(), cameraCode, session.getStreamType()); + streamRegistry.remove(stationCode, cameraCode); + return 0; + } + } + + /** 定时清理超过 24 小时未刷新的观看票据；无剩余观看者时通知站端停流。 */ + public void cleanupExpired() { + long now = System.currentTimeMillis(); + for (Session session : streamRegistry.expiredSessions(now)) { + String key = StationDirectStreamRegistry.cameraKey(session.getStationCode(), session.getCameraCode()); + synchronized (cleanupLocks.computeIfAbsent(key, ignored -> new Object())) { + try { + if (session.expireViewers(now) > 0) { + continue; + } + AggStation station = stationRegistry.requireByCode(session.getStationCode()); + stopStation(station, session.getStationStreamId(), session.getCameraCode(), session.getStreamType()); + streamRegistry.remove(session.getStationCode(), session.getCameraCode()); + log.info("station direct realtime tickets expired, close stream: station={} camera={} streamId={}", + session.getStationCode(), session.getCameraCode(), session.getStationStreamId()); + } catch (Exception e) { + log.warn("cleanup expired station direct stream failed, station={} camera={} streamId={} err={}", + session.getStationCode(), session.getCameraCode(), session.getStationStreamId(), e.getMessage()); + } + } + } + } + + private void stopStation(AggStation station, String stationStreamId, String cameraCode, Integer streamType) { + JsonNode root = stationApiClient.fetchRealtimeStop(station, stationStreamId, cameraCode, streamType); + successPayload(root); + } + + private StartResult result(Session session, boolean existing) { + String hlsUrl = mediaBase(session.getStationCode()) + session.getHlsPath() + + (session.getHlsPath().contains("?") ? "&" : "?") + "_ts=" + System.currentTimeMillis(); + String flvUrl = mediaBase(session.getStationCode()) + session.getFlvPath(); + return new StartResult(session.getStationStreamId(), existing, hlsUrl, flvUrl); + } + + private String mediaBase(String stationCode) { + String configured = props.getVideo().getNginxBaseUrl(); + String base = StringUtils.hasText(configured) + ? configured.trim().replaceAll("/+$", "") + : currentRequestOrigin(); + return base + "/video/" + stationCode; + } + + /** + * Probe the HLS playlist through the aggregation nginx: ZLM returns 200 while the + * upstream stream is online and 404 when the stream does not exist. Any error is + * treated as offline so the caller falls back to a fresh station start. + */ + private boolean isStreamLive(Session session) { + String probeUrl = mediaBase(session.getStationCode()) + session.getHlsPath(); + HttpURLConnection connection = null; + try { + connection = (HttpURLConnection) new URL(probeUrl).openConnection(); + connection.setRequestMethod("GET"); + connection.setConnectTimeout(PROBE_CONNECT_TIMEOUT_MS); + connection.setReadTimeout(PROBE_READ_TIMEOUT_MS); + connection.setInstanceFollowRedirects(false); + connection.setRequestProperty("Connection", "close"); + int status = connection.getResponseCode(); + boolean live = status >= 200 && status < 300; + log.info("station stream liveness probe: station={} streamId={} status={} live={}", + session.getStationCode(), session.getStationStreamId(), status, live); + return live; + } catch (Exception e) { + log.warn("station stream liveness probe failed, treat as offline: station={} streamId={} error={}", + session.getStationCode(), session.getStationStreamId(), e.getMessage()); + return false; + } finally { + if (connection != null) { + connection.disconnect(); + } + } + } + + private String currentRequestOrigin() { + try { + ServletRequestAttributes attributes = + (ServletRequestAttributes) RequestContextHolder.getRequestAttributes(); + if (attributes == null) { + return ""; + } + HttpServletRequest request = attributes.getRequest(); + String forwardedHost = firstForwardedValue(request.getHeader("X-Forwarded-Host")); + String host = StringUtils.hasText(forwardedHost) + ? forwardedHost + : request.getHeader("Host"); + if (!StringUtils.hasText(host)) { + host = request.getServerName(); + int port = request.getServerPort(); + if (port > 0 && port != 80 && port != 443) { + host = host + ":" + port; + } + } + + String forwardedProto = firstForwardedValue(request.getHeader("X-Forwarded-Proto")); + String scheme = StringUtils.hasText(forwardedProto) + ? forwardedProto + : request.getScheme(); + return scheme + "://" + host; + } catch (Exception e) { + log.warn("resolve current media origin failed, fallback to relative url: {}", e.getMessage()); + return ""; + } + } + + private static String firstForwardedValue(String headerValue) { + if (!StringUtils.hasText(headerValue)) { + return null; + } + return headerValue.split(",")[0].trim(); + } + + private static String mediaPath(String stationMediaUrl, String fallback) { + if (StringUtils.hasText(stationMediaUrl)) { + try { + String path = URI.create(stationMediaUrl.trim()).getPath(); + if (StringUtils.hasText(path) && path.startsWith("/live/")) { + return path; + } + } catch (Exception ignored) { + // 使用站端标准路径兜底 + } + } + return fallback; + } + + private JsonNode successPayload(JsonNode root) { + if (root.path("code").asInt(-1) != 200) { + String message = root.path("msg").asText(root.path("message").asText("站端接口返回失败")); + throw new IllegalStateException(message); + } + JsonNode payload = root.path("data"); + return payload.isObject() ? payload : root; + } + + private static String textAnyDeep(JsonNode node, String... fields) { + String value = textAny(node, fields); + if (!StringUtils.hasText(value) && node.path("data").isObject()) { + value = textAny(node.path("data"), fields); + } + return value; + } + + private static String textAny(JsonNode node, String... fields) { + if (node == null || fields == null) { + return null; + } + for (String field : fields) { + JsonNode value = node.get(field); + if (value != null && !value.isNull() && StringUtils.hasText(value.asText())) { + return value.asText(); + } + } + return null; + } + + private static String extractStreamId(String... mediaUrls) { + for (String mediaUrl : mediaUrls) { + if (!StringUtils.hasText(mediaUrl)) { + continue; + } + try { + String path = URI.create(mediaUrl.trim()).getPath(); + int marker = path.indexOf("/live/"); + if (marker < 0) { + continue; + } + String id = path.substring(marker + "/live/".length()).replaceAll("/+$", ""); + if (id.endsWith("/hls.m3u8")) { + id = id.substring(0, id.length() - "/hls.m3u8".length()); + } else { + id = id.replaceAll("\\.m3u8$", "").replaceAll("\\.live\\.flv$", ""); + } + if (id.contains("/")) { + id = id.substring(0, id.indexOf('/')); + } + if (StringUtils.hasText(id)) { + return id; + } + } catch (Exception ignored) { + // 继续尝试其他地址 + } + } + return null; + } + + @Data + public static class StartResult { + private final String streamId; + private final boolean existing; + private final String hlsUrl; + private final String flvUrl; + } +} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggDefectRecordController.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggDefectRecordController.java index 3f05510..e3b540b 100644 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggDefectRecordController.java +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggDefectRecordController.java @@ -11,6 +11,7 @@ import com.github.pagehelper.PageHelper; import com.github.pagehelper.PageInfo; import com.inspect.agg.entity.AggDefectRecord; import com.inspect.agg.mapper.AggDefectRecordMapper; +import com.inspect.agg.support.StationFileUrlResolver; import com.inspect.base.core.web.domain.AjaxResult; import com.inspect.base.core.web.page.TableDataInfo; @@ -25,6 +26,7 @@ import lombok.RequiredArgsConstructor; public class AggDefectRecordController { private final AggDefectRecordMapper defectRecordMapper; + private final StationFileUrlResolver fileUrlResolver; @GetMapping("/defect/algNameOptions") public AjaxResult algNameOptions(@RequestParam(required = false) String stationCode) { @@ -44,6 +46,7 @@ public class AggDefectRecordController { PageHelper.startPage(safePageNum, safePageSize); List list = defectRecordMapper.selectList( stationCode, pointName, patrolTimeBegin, patrolTimeEnd, algName); + rewriteImageUrls(stationCode, list); PageInfo pageInfo = new PageInfo<>(list); TableDataInfo result = new TableDataInfo(); result.setCode(200); @@ -52,4 +55,22 @@ public class AggDefectRecordController { result.setTotal(pageInfo.getTotal()); return result; } + + /** + * 出参图片字段仍是站端 SFTP 相对路径，按 stationCode 统一拼为集控 nginx 反代地址， + * 前端字段名和用法不变，直接作为 使用。 + */ + private void rewriteImageUrls(String stationCode, List list) { + for (AggDefectRecord record : list) { + String code = (stationCode == null || stationCode.trim().isEmpty()) + ? record.getStationCode() : stationCode; + if (code == null || code.trim().isEmpty()) { + continue; + } + record.setImg(fileUrlResolver.resolveImage(code, record.getImg())); + record.setImgAnalyse(fileUrlResolver.resolveImage(code, record.getImgAnalyse())); + record.setBrightImgAnalyse(fileUrlResolver.resolveImage(code, record.getBrightImgAnalyse())); + record.setImageNormalUrlPath(fileUrlResolver.resolveImage(code, record.getImageNormalUrlPath())); + } + } } diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggInspectionReportController.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggInspectionReportController.java index d8442b0..c9f418b 100644 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggInspectionReportController.java +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggInspectionReportController.java @@ -11,6 +11,7 @@ import com.github.pagehelper.PageHelper; import com.github.pagehelper.PageInfo; import com.inspect.agg.entity.AggInspectionReport; import com.inspect.agg.mapper.AggInspectionReportMapper; +import com.inspect.agg.support.StationFileUrlResolver; import com.inspect.base.core.web.page.TableDataInfo; import lombok.RequiredArgsConstructor; @@ -24,6 +25,7 @@ import lombok.RequiredArgsConstructor; public class AggInspectionReportController { private final AggInspectionReportMapper reportMapper; + private final StationFileUrlResolver fileUrlResolver; @GetMapping("/list") public TableDataInfo list(@RequestParam(required = false) String stationCode, @@ -38,10 +40,24 @@ public class AggInspectionReportController { PageHelper.startPage(safePageNum, safePageSize); List list = reportMapper.selectList( stationCode, cycleStartTime, cycleEndTime, inspectionTaskName); + rewriteFilePaths(stationCode, list); PageInfo pageInfo = new PageInfo<>(list); return dataTable(list, pageInfo.getTotal()); } + /** + * filePath 为站端 SFTP 相对路径（如 /1/Model/xxx.docx），按 stationCode 改写为集控 nginx 反代地址。 + */ + private void rewriteFilePaths(String stationCode, List list) { + for (AggInspectionReport report : list) { + String code = (stationCode == null || stationCode.trim().isEmpty()) + ? report.getStationCode() : stationCode; + if (code == null || code.trim().isEmpty()) { + continue; + } + report.setFilePath(fileUrlResolver.resolveFile(code, report.getFilePath())); + } + } private static TableDataInfo dataTable(List rows, long total) { TableDataInfo result = new TableDataInfo(); result.setCode(200); diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggVideoController.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggVideoController.java index c0c8338..955a501 100644 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggVideoController.java +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/AggVideoController.java @@ -7,7 +7,7 @@ import com.inspect.agg.service.StationRegistry; import com.inspect.agg.station.StationApiClient; import com.inspect.agg.video.PlaybackSession; import com.inspect.agg.video.PlaybackSessionRegistry; -import com.inspect.agg.video.RealtimeStreamService; +import com.inspect.agg.video.direct.StationDirectStreamService; import com.inspect.agg.video.ZlmApiClient; import com.inspect.base.core.web.domain.AjaxResult; import lombok.RequiredArgsConstructor; @@ -34,7 +34,7 @@ public class AggVideoController { private final StationApiClient stationApiClient; private final ZlmApiClient zlmApiClient; private final PlaybackSessionRegistry playbackSessionRegistry; - private final RealtimeStreamService realtimeStreamService; + private final StationDirectStreamService stationDirectStreamService; private static String resolvePlaybackUrl(AggStation station, String url) { if (url.startsWith("http://") || url.startsWith("https://")) { @@ -116,16 +116,16 @@ public class AggVideoController { return AjaxResult.error(400, "stationCode / cameraCode 不能为空"); } try { - RealtimeStreamService.StartResult result = realtimeStreamService.start(stationCode, request); + StationDirectStreamService.StartResult result = stationDirectStreamService.start(stationCode, request); String streamId = result.getStreamId(); - String hlsUrl = cacheBustedHlsUrl(streamId); - String flvUrl = zlmApiClient.flvUrl(streamId); + String hlsUrl = result.getHlsUrl(); + String flvUrl = result.getFlvUrl(); Map data = new LinkedHashMap<>(); data.put("stationCode", stationCode); data.put("cameraCode", cameraCode); data.put("streamId", streamId); - data.put("sourceMode", "RTSP_DIRECT"); + data.put("sourceMode", "STATION_DIRECT"); data.put("status", result.isExisting() ? "PLAYING" : "STARTING"); data.put("hlsUrl", hlsUrl); data.put("flvUrl", flvUrl); @@ -226,7 +226,7 @@ public class AggVideoController { return AjaxResult.error(400, "stationCode / cameraCode 不能为空"); } try { - int remaining = realtimeStreamService.stop(stationCode, request); + int remaining = stationDirectStreamService.stop(stationCode, request); if (remaining == -1) { return AjaxResult.success("未找到可停止的观看会话"); } diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/StationProxyController.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/StationProxyController.java index 4c8b161..0d7b3c8 100644 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/StationProxyController.java +++ b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/StationProxyController.java @@ -1,6 +1,13 @@ package com.inspect.agg.web; +import java.net.URLDecoder; +import java.net.URLEncoder; import java.nio.charset.StandardCharsets; +import java.util.function.UnaryOperator; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.Iterator; +import java.util.Map; import java.util.Arrays; import java.util.stream.Collectors; @@ -19,6 +26,12 @@ import org.springframework.web.bind.annotation.RestController; import org.springframework.web.client.HttpStatusCodeException; import org.springframework.web.client.RestTemplate; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.fasterxml.jackson.databind.node.TextNode; +import com.inspect.agg.support.StationFileUrlResolver; + import com.inspect.agg.config.AggProperties; import com.inspect.agg.entity.AggStation; import com.inspect.agg.service.StationRegistry; @@ -41,10 +54,16 @@ public class StationProxyController { private static final String PREFIX = "/agg/proxy/"; + private static final String IR_DOWNLOAD_CSV_PATH = "simulator/ir/downloadCsv"; + /** /img/{stationCode}/... 前缀，stationCode 为任意非空段 */ + private static final Pattern IMG_PROXY_PREFIX = Pattern.compile("^/img/[^/]+(?/.*)$"); + private final StationRegistry stationRegistry; private final StationApiClient stationApiClient; private final RestTemplate restTemplate; private final AggProperties props; + private final ObjectMapper objectMapper; + private final StationFileUrlResolver fileUrlResolver; @RequestMapping(value = "/agg/proxy/**", method = {RequestMethod.GET, RequestMethod.POST, RequestMethod.PUT, RequestMethod.DELETE}) @@ -68,7 +87,7 @@ public class StationProxyController { return error(404, e.getMessage()); } - String query = filterStationCode(request.getQueryString()); + String query = buildStationQuery(request.getQueryString()); String targetUrl = stripTrailingSlash(station.getBaseUrl()) + "/" + subPath + (StringUtils.hasText(query) ? "?" + query : ""); @@ -85,21 +104,80 @@ public class StationProxyController { } byte[] body = readBody(request); + body = rewriteRequestBody(subPath, body, contentType); HttpMethod method = HttpMethod.valueOf(request.getMethod()); try { ResponseEntity resp = restTemplate.exchange(targetUrl, method, new HttpEntity<>(body, headers), byte[].class); - return relay(resp.getStatusCodeValue(), resp.getHeaders().getContentType(), resp.getBody()); + return relay(resp.getStatusCodeValue(), resp.getHeaders().getContentType(), + rewriteStationFiles(stationCode, resp.getBody(), resp.getHeaders().getContentType())); } catch (HttpStatusCodeException e) { // 站点返回 4xx/5xx 也原样透传 - return relay(e.getRawStatusCode(), e.getResponseHeaders() == null ? null - : e.getResponseHeaders().getContentType(), e.getResponseBodyAsByteArray()); + HttpHeaders errorHeaders = e.getResponseHeaders(); + MediaType errorType = errorHeaders == null ? null : errorHeaders.getContentType(); + return relay(e.getRawStatusCode(), errorType, + rewriteStationFiles(stationCode, e.getResponseBodyAsByteArray(), errorType)); } catch (Exception e) { log.error("proxy failed: station={} path={} err={}", stationCode, subPath, e.getMessage()); return error(502, "代理调用站点失败: " + e.getMessage()); } } + /** + * 通用代理不感知具体接口字段，因此对 JSON 响应做递归扫描： + * 只要字段名像图片/文件字段且值为站端存储相对路径，就改写为 /img/ 或 /file/{stationCode}/ 代理地址。 + * 非 JSON（图片二进制、视频流等）原样透传；解析失败也原样透传。 + */ + private byte[] rewriteStationFiles(String stationCode, byte[] body, MediaType contentType) { + if (body == null || body.length == 0 || !isJson(contentType) + || props.getFile() == null || !props.getFile().isRewriteProxyResponse()) { + return body; + } + try { + JsonNode root = objectMapper.readTree(body); + rewriteNode(stationCode, root, null); + return objectMapper.writeValueAsBytes(root); + } catch (Exception e) { + log.warn("rewrite proxy image urls failed, return original body: station={} err={}", + stationCode, e.getMessage()); + return body; + } + } + + private void rewriteNode(String stationCode, JsonNode node, String fieldName) { + if (node == null) { + return; + } + if (node.isObject()) { + ObjectNode objectNode = (ObjectNode) node; + Iterator> fields = objectNode.fields(); + while (fields.hasNext()) { + Map.Entry entry = fields.next(); + JsonNode child = entry.getValue(); + if (child != null && child.isTextual()) { + String original = child.asText(); + String rewritten = fileUrlResolver.rewriteIfStationFile(stationCode, entry.getKey(), original); + if (rewritten != null && !rewritten.equals(original)) { + objectNode.set(entry.getKey(), TextNode.valueOf(rewritten)); + } + } else { + rewriteNode(stationCode, child, entry.getKey()); + } + } + } else if (node.isArray()) { + for (JsonNode child : node) { + rewriteNode(stationCode, child, fieldName); + } + } + } + + private static boolean isJson(MediaType contentType) { + if (contentType == null) { + return false; + } + String subtype = contentType.getSubtype(); + return subtype != null && (subtype.equals("json") || subtype.endsWith("+json")); + } private boolean allowed(String subPath) { if (props.getProxy() == null || props.getProxy().getAllowedPrefixes() == null || props.getProxy().getAllowedPrefixes().isEmpty()) { @@ -108,6 +186,116 @@ public class StationProxyController { return props.getProxy().getAllowedPrefixes().stream().anyMatch(subPath::startsWith); } + /** 对特定接口的请求体做参数定制：处理 JSON 或表单 body 中的 filePath。 */ + private byte[] rewriteRequestBody(String subPath, byte[] body, String contentType) { + if (!IR_DOWNLOAD_CSV_PATH.equals(subPath) || body == null || body.length == 0) { + return body; + } + try { + MediaType mediaType = StringUtils.hasText(contentType) ? MediaType.parseMediaType(contentType) : null; + if (mediaType != null && MediaType.APPLICATION_JSON.includes(mediaType)) { + JsonNode root = objectMapper.readTree(body); + rewriteBodyFilePath(root); + return objectMapper.writeValueAsBytes(root); + } + if (mediaType != null && MediaType.APPLICATION_FORM_URLENCODED.includes(mediaType)) { + String raw = new String(body, StandardCharsets.UTF_8); + String rewritten = rewriteQueryParam(raw, "filePath", + StationProxyController::stripImageProxyPrefix); + return rewritten.getBytes(StandardCharsets.UTF_8); + } + } catch (Exception e) { + log.warn("rewrite proxy request body failed, return original body: stationPath={} err={}", + subPath, e.getMessage()); + } + return body; + } + + private void rewriteBodyFilePath(JsonNode node) { + if (node == null) { + return; + } + if (node.isObject()) { + ObjectNode objectNode = (ObjectNode) node; + Iterator> fields = objectNode.fields(); + while (fields.hasNext()) { + Map.Entry entry = fields.next(); + JsonNode child = entry.getValue(); + if (child != null && child.isTextual() && "filePath".equals(entry.getKey())) { + objectNode.put(entry.getKey(), stripImageProxyPrefix(child.asText())); + } else { + rewriteBodyFilePath(child); + } + } + } else if (node.isArray()) { + for (JsonNode child : node) { + rewriteBodyFilePath(child); + } + } + } + /** 组装转发查询串：仅剔除 stationCode，其余查询参数原样透传。 */ + private String buildStationQuery(String rawQuery) { + return filterStationCode(rawQuery); + } + + /** 仅对指定查询参数解码、变换后重新编码，其余参数保持原始编码原样透传。 */ + private static String rewriteQueryParam(String rawQuery, String targetName, UnaryOperator mapper) { + StringBuilder sb = new StringBuilder(); + for (String pair : rawQuery.split("&")) { + if (sb.length() > 0) { + sb.append('&'); + } + int idx = pair.indexOf('='); + String encodedName = idx >= 0 ? pair.substring(0, idx) : pair; + String encodedValue = idx >= 0 ? pair.substring(idx + 1) : ""; + String name = urlDecode(encodedName); + if (targetName.equals(name)) { + String value = urlDecode(encodedValue); + value = mapper.apply(value); + sb.append(urlEncode(name)) + .append('=') + .append(urlEncode(value)); + } else { + sb.append(pair); + } + } + return sb.toString(); + } + + private static String urlDecode(String value) { + try { + return URLDecoder.decode(value, StandardCharsets.UTF_8.name()); + } catch (java.io.UnsupportedEncodingException e) { + throw new IllegalStateException(e); + } + } + + private static String urlEncode(String value) { + try { + return URLEncoder.encode(value, StandardCharsets.UTF_8.name()); + } catch (java.io.UnsupportedEncodingException e) { + throw new IllegalStateException(e); + } + } + /** /img/{stationCode}/x -> /x；不符合形态时原样返回。 */ + private static String stripImageProxyPrefix(String value) { + if (!StringUtils.hasText(value)) { + return value; + } + Matcher matcher = IMG_PROXY_PREFIX.matcher(value.trim()); + return matcher.matches() ? matcher.group("rest") : value; + } + + + + public static void main(String[] args) { + String fi = filterStationCode("stationCode=shaoxing"); + System.out.println(fi); +// String filePath = "/img/shaoxing/1/2026/10/08/1810/CCD/1810_20261008174906780_2093864_02160558340721910101_20261008175026.jpg"; +// System.out.println(stripImageProxyPrefix(filePath)); + + } + private static String filterStationCode(String rawQuery) { if (!StringUtils.hasText(rawQuery)) { return null; diff --git a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/VideoProxyController.java b/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/VideoProxyController.java deleted file mode 100644 index 7f7cd12..0000000 --- a/inspect-main/inspect-main-agg/src/main/java/com/inspect/agg/web/VideoProxyController.java +++ /dev/null @@ -1,358 +0,0 @@ -package com.inspect.agg.web; - -import java.io.InputStream; -import java.io.OutputStream; -import java.net.HttpURLConnection; -import java.net.URI; -import java.net.URL; -import java.net.URLEncoder; -import java.nio.charset.StandardCharsets; -import java.util.LinkedHashMap; -import java.util.Map; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -import org.springframework.http.HttpHeaders; -import org.springframework.http.MediaType; -import org.springframework.http.ResponseEntity; -import org.springframework.util.StringUtils; -import org.springframework.web.bind.annotation.GetMapping; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RequestParam; -import org.springframework.web.bind.annotation.RestController; -import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody; - -import com.fasterxml.jackson.databind.JsonNode; -import com.inspect.agg.config.AggProperties; -import com.inspect.agg.entity.AggStation; -import com.inspect.agg.service.StationRegistry; -import com.inspect.agg.station.StationApiClient; -import com.inspect.agg.video.StationStream; -import com.inspect.agg.video.StationStreamRegistry; - -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; - -/** - * 方案 B：集控纯 HTTP 反向代理（不拉流、不转码）。 - * 办公端只访问集控地址；集控把站端 HLS/FLV 流式转发，并重写 m3u8 中的分片地址。 - * - * GET /agg/video/play?stationCode=&cameraCode= 启动站端转流并返回集控播放地址 - * GET /agg/video/flv?stationCode=&streamId= HTTP-FLV 流式代理 - * GET /agg/video/hls?stationCode=&streamId= HLS 清单（重写分片地址） - * GET /agg/video/segment?stationCode=&streamId=&u= HLS 分片流式代理 - * GET /agg/video/proxy/stop?stationCode=&streamId= 停止站端转流 - */ -@Slf4j -@RestController -@RequestMapping("/agg/video/stream") -@RequiredArgsConstructor -public class VideoProxyController { - - private static final Pattern TAG_URI = Pattern.compile("URI=\"([^\"]+)\""); - - private final StationRegistry stationRegistry; - private final StationApiClient stationApiClient; - private final StationStreamRegistry streamRegistry; - private final AggProperties props; - - /** 启动播放：调站端 /v1/stream/start，返回经集控代理的 HLS/FLV 地址 */ - @GetMapping("/play") - public Map play(@RequestParam String stationCode, - @RequestParam String cameraCode) { - if (!StringUtils.hasText(stationCode) || !StringUtils.hasText(cameraCode)) { - return error(400, "stationCode / cameraCode 不能为空"); - } - AggStation station; - try { - station = stationRegistry.requireByCode(stationCode); - } catch (IllegalArgumentException e) { - return error(404, e.getMessage()); - } - try { - JsonNode root = stationApiClient.fetchStreamStart(station, mainCode(cameraCode), 2); - if (!StringUtils.hasText(text(root, "rtspURL")) && !StringUtils.hasText(text(root, "FLV")) - && !StringUtils.hasText(text(root, "HLS"))) { - root = stationApiClient.fetchStreamStart(station, cameraCode, 2); - } - String stationStreamId = text(root, "StreamID"); - String flv = text(root, "FLV"); - String hls = text(root, "HLS"); - if (!StringUtils.hasText(stationStreamId)) { - return error(502, "站端未返回 StreamID: " + root); - } - StationStream stream = new StationStream(); - stream.setStationCode(stationCode); - stream.setStationStreamId(stationStreamId); - stream.setFlvUrl(flv); - stream.setHlsUrl(hls); - streamRegistry.put(stream); - - Map data = new LinkedHashMap<>(); - data.put("stationCode", stationCode); - data.put("cameraCode", cameraCode); - data.put("streamId", stationStreamId); - data.put("mode", "HTTP_PROXY"); - if (StringUtils.hasText(hls)) { - data.put("hls", "/api/agg/video/hls?stationCode=" + enc(stationCode) + "&streamId=" + enc(stationStreamId)); - } - if (StringUtils.hasText(flv)) { - data.put("flv", "/api/agg/video/flv?stationCode=" + enc(stationCode) + "&streamId=" + enc(stationStreamId)); - } - Map m = new LinkedHashMap<>(); - m.put("code", 200); - m.put("data", data); - return m; - } catch (Exception e) { - log.error("video play failed: station={} camera={} err={}", stationCode, cameraCode, e.getMessage()); - return error(502, "启动视频流失败: " + e.getMessage()); - } - } - - /** 停止站端转流 */ - @GetMapping("/stop") - public Map stop(@RequestParam String stationCode, @RequestParam String streamId) { - AggStation station; - try { - station = stationRegistry.requireByCode(stationCode); - } catch (IllegalArgumentException e) { - return error(404, e.getMessage()); - } - try { - stationApiClient.fetchStreamStop(station, streamId); - streamRegistry.remove(stationCode, streamId); - Map m = new LinkedHashMap<>(); - m.put("code", 200); - m.put("data", true); - return m; - } catch (Exception e) { - return error(502, "停止视频流失败: " + e.getMessage()); - } - } - - /** HTTP-FLV 流式代理（不缓冲，边收边发） */ - @GetMapping("/flv") - public ResponseEntity flv(@RequestParam String stationCode, - @RequestParam String streamId) { - StationStream stream = streamRegistry.get(stationCode, streamId); - if (stream == null || !StringUtils.hasText(stream.getFlvUrl())) { - return ResponseEntity.status(404).build(); - } - return streaming(stream.getFlvUrl(), "video/x-flv"); - } - - /** HLS 清单：拉取站端 m3u8 并重写分片地址到集控代理 */ - @GetMapping("/hls") - public ResponseEntity hls(@RequestParam String stationCode, @RequestParam String streamId) { - StationStream stream = streamRegistry.get(stationCode, streamId); - if (stream == null || !StringUtils.hasText(stream.getHlsUrl())) { - return ResponseEntity.status(404).build(); - } - try { - PlaylistResult pr = readHlsWithFallback(stream.getHlsUrl()); - if (!pr.url.equals(stream.getHlsUrl())) { - // 记录实际生效的清单地址，后续分片按它的目录解析 - stream.setHlsUrl(pr.url); - } - String rewritten = rewritePlaylist(pr.content, stationCode, streamId, pr.url); - return ResponseEntity.ok() - .header(HttpHeaders.CONTENT_TYPE, "application/vnd.apple.mpegurl") - .header(HttpHeaders.CACHE_CONTROL, "no-store") - .body(rewritten.getBytes(StandardCharsets.UTF_8)); - } catch (Exception e) { - log.error("hls proxy failed: stream={} err={}", streamId, e.getMessage()); - return ResponseEntity.status(502).build(); - } - } - - /** HLS 分片流式代理 */ - @GetMapping("/segment") - public ResponseEntity segment(@RequestParam String stationCode, - @RequestParam String streamId, - @RequestParam("u") String uri) { - StationStream stream = streamRegistry.get(stationCode, streamId); - if (stream == null) { - return ResponseEntity.status(404).build(); - } - String target = resolve(stream.getHlsUrl(), uri); - if (target == null) { - return ResponseEntity.status(403).build(); - } - return streaming(target, null); - } - - /** 把 m3u8 里所有 URI（分片/密钥/初始化段）改写成集控代理地址 */ - private String rewritePlaylist(String playlist, String stationCode, String streamId, String hlsUrl) { - StringBuilder sb = new StringBuilder(playlist.length() + 256); - for (String line : playlist.split("\n")) { - String trimmed = line.trim(); - if (trimmed.isEmpty()) { - continue; - } - if (trimmed.startsWith("#")) { - Matcher matcher = TAG_URI.matcher(trimmed); - if (matcher.find()) { - String rewritten = proxyUri(stationCode, streamId, matcher.group(1)); - sb.append(trimmed, 0, matcher.start(1)).append(rewritten).append(trimmed.substring(matcher.end(1))); - } else { - sb.append(trimmed); - } - } else { - sb.append(proxyUri(stationCode, streamId, trimmed)); - } - sb.append('\n'); - } - return sb.toString(); - } - - private String proxyUri(String stationCode, String streamId, String uri) { - return "/api/agg/video/segment?stationCode=" + enc(stationCode) - + "&streamId=" + enc(streamId) + "&u=" + enc(uri); - } - - /** 解析分片绝对地址（仅允许与站端 HLS 同 host:port，避免开放代理） */ - private String resolve(String hlsUrl, String uri) { - try { - URI base = URI.create(hlsUrl); - String baseDir = hlsUrl.substring(0, hlsUrl.lastIndexOf('/') + 1); - String target = uri.startsWith("http://") || uri.startsWith("https://") ? uri - : uri.startsWith("/") ? base.getScheme() + "://" + base.getAuthority() + uri - : baseDir + uri; - URI t = URI.create(target); - if (!base.getAuthority().equalsIgnoreCase(t.getAuthority())) { - log.warn("拒绝跨主机分片代理: {}", target); - return null; - } - return target; - } catch (Exception e) { - return null; - } - } - - /** 流式转发：不缓冲，边读边写 */ - private ResponseEntity streaming(String url, String contentType) { - StreamingResponseBody body = outputStream -> { - HttpURLConnection conn = null; - try { - conn = (HttpURLConnection) new URL(url).openConnection(); - conn.setConnectTimeout(props.getHttp().getConnectTimeoutMs()); - conn.setReadTimeout(props.getHttp().getReadTimeoutMs()); - conn.setInstanceFollowRedirects(true); - conn.setRequestProperty("User-Agent", "inspect-aggregation-proxy"); - if (StringUtils.hasText(contentType)) { - conn.setRequestProperty(HttpHeaders.ACCEPT, contentType); - } - try (InputStream in = conn.getInputStream()) { - copy(in, outputStream); - } - } catch (Exception e) { - log.warn("stream proxy interrupted: url={} err={}", url, e.getMessage()); - } finally { - if (conn != null) { - conn.disconnect(); - } - } - }; - ResponseEntity.BodyBuilder builder = ResponseEntity.ok() - .header(HttpHeaders.CACHE_CONTROL, "no-store") - .header("X-Accel-Buffering", "no"); - builder.header(HttpHeaders.CONTENT_TYPE, - StringUtils.hasText(contentType) ? contentType : MediaType.APPLICATION_OCTET_STREAM_VALUE); - return builder.body(body); - } - - /** - * 兼容站端 HLS 路径风格差异：站端 start 接口可能返回 /live/{id}.m3u8（旧）， - * 实际可用为 /live/{id}/hls.m3u8，反之亦然。 - */ - private PlaylistResult readHlsWithFallback(String hlsUrl) throws Exception { - try { - return new PlaylistResult(readString(hlsUrl), hlsUrl); - } catch (Exception first) { - for (String alt : hlsAlternatives(hlsUrl)) { - try { - log.info("HLS 主地址不可用，尝试备用地址: {}", alt); - return new PlaylistResult(readString(alt), alt); - } catch (Exception ignore) { - // 继续尝试 - } - } - throw first; - } - } - - /** 清单内容 + 实际生效的清单地址 */ - private static final class PlaylistResult { - private final String content; - private final String url; - - PlaylistResult(String content, String url) { - this.content = content; - this.url = url; - } - } - - private static java.util.List hlsAlternatives(String hlsUrl) { - java.util.List list = new java.util.ArrayList<>(); - if (hlsUrl == null) { - return list; - } - if (hlsUrl.endsWith("/hls.m3u8")) { - // /live/{id}/hls.m3u8 -> /live/{id}.m3u8 - String base = hlsUrl.substring(0, hlsUrl.length() - "/hls.m3u8".length()); - list.add(base + ".m3u8"); - } else if (hlsUrl.endsWith(".m3u8")) { - // /live/{id}.m3u8 -> /live/{id}/hls.m3u8 - String base = hlsUrl.substring(0, hlsUrl.length() - ".m3u8".length()); - list.add(base + "/hls.m3u8"); - } - return list; - } - - private String readString(String url) throws Exception { - HttpURLConnection conn = (HttpURLConnection) new URL(url).openConnection(); - conn.setConnectTimeout(props.getHttp().getConnectTimeoutMs()); - conn.setReadTimeout(props.getHttp().getReadTimeoutMs()); - conn.setRequestProperty("User-Agent", "inspect-aggregation-proxy"); - try (InputStream in = conn.getInputStream()) { - java.io.ByteArrayOutputStream bos = new java.io.ByteArrayOutputStream(); - copy(in, bos); - return new String(bos.toByteArray(), StandardCharsets.UTF_8); - } finally { - conn.disconnect(); - } - } - - private static void copy(InputStream in, OutputStream out) throws Exception { - byte[] buf = new byte[8192]; - int len; - while ((len = in.read(buf)) != -1) { - out.write(buf, 0, len); - out.flush(); - } - } - - private static String mainCode(String cameraCode) { - return cameraCode.contains("#") ? cameraCode.substring(0, cameraCode.indexOf('#')) : cameraCode; - } - - private static String enc(String s) { - try { - return URLEncoder.encode(s, "UTF-8"); - } catch (Exception e) { - return s; - } - } - - private static String text(JsonNode node, String field) { - JsonNode v = node == null ? null : node.get(field); - return v == null || v.isNull() ? null : v.asText(); - } - - private static Map error(int code, String msg) { - Map m = new LinkedHashMap<>(); - m.put("code", code); - m.put("msg", msg); - return m; - } -} \ No newline at end of file diff --git a/inspect-main/inspect-main-agg/src/main/resources/application.yml b/inspect-main/inspect-main-agg/src/main/resources/application.yml index ebd3ac1..934f631 100644 --- a/inspect-main/inspect-main-agg/src/main/resources/application.yml +++ b/inspect-main/inspect-main-agg/src/main/resources/application.yml @@ -1,4 +1,4 @@ -server: +server: port: 18088 spring: @@ -29,6 +29,15 @@ agg: zlm-secret: Xg@123456 app: live public-base-url: http://192.168.1.116:33380 + # 站端直连流的集控 nginx 地址；留空时按当前请求 Host 自动拼接，也可用 AGG_VIDEO_NGINX_BASE_URL 覆盖 + nginx-base-url: ${AGG_VIDEO_NGINX_BASE_URL:} rtsp-transport: 0 retry-count: -1 + file: + # 留空：出参返回相对地址 /img、/file，由前端拼 /app/htjc6/htjcImage 前缀走集控 nginx + # 仅当前端不再拼前缀、需要绝对地址时，才配置 AGG_FILE_NGINX_BASE_URL=http://集控:18891 + nginx-base-url: ${AGG_FILE_NGINX_BASE_URL:} + # 是否在 /agg/proxy 通用代理的 JSON 响应中递归改写图片/文件地址 + rewrite-proxy-response: true +