From d6f8368493241a5935f645538cc0ea8b6b903ce0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Tue, 15 Sep 2026 12:20:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E7=AB=99=E5=86=85=E9=80=9A=E7=9F=A5):=20?= =?UTF-8?q?=E9=BA=A6=E8=B1=A1=E5=BC=82=E5=B8=B8=E6=89=AB=E6=8F=8F=E4=B8=8A?= =?UTF-8?q?=E7=BA=BF=E2=80=94=E2=80=94=E4=BB=BB=E5=8A=A1=E5=81=9C=E6=BB=9E?= =?UTF-8?q?/=E5=A4=B1=E8=B4=A5/=E9=98=9F=E5=88=97=E7=A7=AF=E5=8E=8B/?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E7=A9=BA=E7=BB=93=E6=9E=9C=E6=8E=A8=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E5=91=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户要求「maixiang异常也要发通知」，确认范围（卡住/失败/队列堆积/服务不可用）与 受众（只发管理员）后实现： - MaixiangConsoleClient：18960 后台只读接口客户端（batch/tasks、tasks、queue/status、 batch/detail），token 鉴权、失败只记日志。URL 必须用 URI 对象提交——字符串形态会被 RestClient 二次编码（%20→%2520），end_time 参数实测报 Incorrect DATETIME value， 检查会静默失效；已加回归单测 buildUrlEncodesTimeParameterExactlyOnce - MaixiangAnomalyScanner 五类检查：①批量任务停滞（status 0/1 超时无更新，默认 60 分钟） ②单任务滞留（创建超 30 分钟未完成）③近 30 分钟失败达最小条数 ④批量任务空结果 （worker 报错时批次仍被标"已完成"，是唯一可抓的批量失败信号）⑤队列积压（pending≥300 / processing≥100）；去重按天/小时/任务，全部为管理员全局事件（subjectUserId=null） - 接入 NotificationScanScheduler（分布式锁内同跑）；scene=maixiang_anomaly； AIIMAGE_NOTIFICATION_MAIXIANG_* 环境变量可调，令牌留空=跳过 - 前端 NotificationScene 类型补 maixiang_anomaly（铃铛不按 scene 渲染，无运行时改动） - 测试：客户端解析（真实抓包样例）+ 扫描器 14 例 + 手工联调探针 MaixiangLiveProbeTest （-Dmaixiang.live=true 开启，只读不写通知） 已部署双节点（.env 加 console token、JAR cd3e06c4）并线上验证：首扫推 2×15 条管理员通知 （3 个变体任务停滞 / 39 个跟价任务滞留），二次扫描落库=0 去重生效。 --- .../config/NotificationProperties.java | 24 ++ .../client/MaixiangConsoleClient.java | 240 ++++++++++++++ .../service/MaixiangAnomalyScanner.java | 312 ++++++++++++++++++ .../service/NotificationScanScheduler.java | 6 + .../service/NotificationService.java | 2 + .../src/main/resources/application.yml | 9 + .../client/MaixiangConsoleClientTest.java | 152 +++++++++ .../service/MaixiangAnomalyScannerTest.java | 284 ++++++++++++++++ .../service/MaixiangLiveProbeTest.java | 81 +++++ .../NotificationScanSchedulerTest.java | 3 +- .../shared/api/types/modules/notification.ts | 10 +- 11 files changed, 1120 insertions(+), 3 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClient.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScanner.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClientTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScannerTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangLiveProbeTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java index 363f6492..fb032b3d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java @@ -41,6 +41,30 @@ public class NotificationProperties { /** jikip 代理接口探测开关（复用余额查询接口判活，使用 user-secret 的 jikip 配置）。 */ private boolean jikipProbeEnabled = true; + /** 麦象（18960 任务调度）异常扫描开关：任务停滞/失败/队列积压 → 管理员通知。 */ + private boolean maixiangScanEnabled = true; + + /** + * 麦象后台接口令牌（18960 console token）。留空=跳过麦象异常扫描—— + * 该令牌与「跟价任务 API 地址」(priceTrackApiUrl) 一起构成后台只读接口的访问凭据。 + */ + private String maixiangConsoleToken = ""; + + /** 麦象批量任务停滞阈值（分钟）：status=0/1 且超过该时长无更新视为卡住。 */ + private int maixiangStuckMinutes = 60; + + /** 麦象单任务滞留阈值（分钟）：创建超时仍未完成（status=0/1）视为滞留/无人消费。 */ + private int maixiangSingleStuckMinutes = 30; + + /** 麦象任务失败告警阈值（条）：近 30 分钟窗口内失败数达到该值才告警。 */ + private int maixiangFailMinCount = 1; + + /** 麦象队列积压阈值（条）：task:queue 待处理数达到该值告警。 */ + private int maixiangQueuePendingThreshold = 300; + + /** 麦象队列积压阈值（条）：task:processing 处理中数达到该值告警。 */ + private int maixiangQueueProcessingThreshold = 100; + /** 已读通知保留天数（超期自动清理），默认 90 天。 */ private int readRetentionDays = 90; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClient.java new file mode 100644 index 00000000..0196a90e --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClient.java @@ -0,0 +1,240 @@ +package com.nanri.aiimage.modules.notification.client; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.HttpClientPool; +import com.nanri.aiimage.config.NotificationProperties; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.springframework.web.client.RestClient; +import org.springframework.web.util.UriComponentsBuilder; + +import java.net.URI; +import java.time.LocalDateTime; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.List; + +/** + * 麦象（18960 任务调度）后台只读接口客户端。 + * + *

站内通知的「麦象异常扫描」所需数据全部来自这几个 console 接口（token 鉴权）： + * 批量任务列表（all_task）/ 单任务列表（task_record）/ 队列长度 / 批量任务结果行数。 + * + *

只读、无副作用；调用失败只记中文日志并返回空结果（不抛异常）， + * 避免扫描任务被单次网络抖动打断——服务级不可用由现有 18960 探测告警覆盖。 + * 注意日志里不输出完整 URL（含后台令牌）。 + */ +@Component +@Slf4j +@RequiredArgsConstructor +public class MaixiangConsoleClient { + + private static final int READ_TIMEOUT_MILLIS = 8_000; + private static final DateTimeFormatter TIME_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + /** 麦象后台接口未注册结果表时返回的哨兵值。 */ + public static final long RESULT_TOTAL_UNSUPPORTED = -1L; + + private final NotificationProperties properties; + private final ObjectMapper objectMapper; + + /** 是否已配置（地址 + 后台令牌）；未配置时调用方应跳过扫描。 */ + public boolean configured() { + return hasText(properties.getPriceTrackApiUrl()) && hasText(properties.getMaixiangConsoleToken()); + } + + /** 批量任务列表（all_task）。createdBefore 非空时只取创建时间不晚于该时刻的任务（SQL 字符串比较）。 */ + public BatchTaskPage batchTasks(int status, LocalDateTime createdBefore, int pageSize) { + String url = buildUrl("/api/console/batch/tasks", + "status=" + status, + "page=1", + "page_size=" + pageSize, + createdBefore == null ? null : "end_time=" + TIME_FORMAT.format(createdBefore)); + return parseBatchTaskPage(fetch(url, "批量任务列表 status=" + status)); + } + + /** 单任务列表（task_record）。 */ + public SingleTaskPage singleTasks(int status, LocalDateTime createdBefore, int pageSize) { + String url = buildUrl("/api/console/tasks", + "status=" + status, + "page=1", + "page_size=" + pageSize, + createdBefore == null ? null : "end_time=" + TIME_FORMAT.format(createdBefore)); + return parseSingleTaskPage(fetch(url, "单任务列表 status=" + status)); + } + + /** 单任务队列状态（task:queue:N / task:processing:N，仅返回非零级别）。 */ + public List queueStatus() { + return parseQueueStates(fetch(buildUrl("/api/console/queue/status"), "队列状态")); + } + + /** 批量任务结果行数：未注册结果表（table=null）返回 {@link #RESULT_TOTAL_UNSUPPORTED}。 */ + public long batchResultTotal(String taskId) { + String url = buildUrl("/api/console/batch/detail", "task_id=" + taskId, "page=1", "page_size=1"); + return parseBatchResultTotal(fetch(url, "批量任务结果 task_id=" + taskId)); + } + + // ======================== HTTP ======================== + + /** 拉取并展开统一信封 {code,mes,data}；失败返回 null（已记日志）。 */ + private JsonNode fetch(String url, String scene) { + String body; + try { + // 用 URI 对象提交：URL 已由 buildUrl 编码过一次，不能再经 RestClient 模板再次编码 + // （字符串形态会被二次编码，%20 变 %2520，服务端 DATETIME 参数直接报错） + body = RestClient.builder() + .requestFactory(HttpClientPool.requestFactory(READ_TIMEOUT_MILLIS)) + .build() + .get() + .uri(URI.create(url)) + .retrieve() + .body(String.class); + } catch (Exception ex) { + log.warn("[maixiang-console] 请求失败 scene={} err={}", scene, ex.getMessage()); + return null; + } + return parseEnvelope(body, scene); + } + + /** 组装后台接口 URL（含 token 与查询参数，统一编码一次；包内可见供单测校验编码行为）。 */ + String buildUrl(String path, String... queryPairs) { + UriComponentsBuilder builder = UriComponentsBuilder.fromHttpUrl(baseUrl()) + .path(path) + .queryParam("token", properties.getMaixiangConsoleToken().trim()); + for (String pair : queryPairs) { + if (pair == null) { + continue; + } + int idx = pair.indexOf('='); + builder.queryParam(pair.substring(0, idx), pair.substring(idx + 1)); + } + return builder.build().encode().toUriString(); + } + + private String baseUrl() { + String base = properties.getPriceTrackApiUrl() == null ? "" : properties.getPriceTrackApiUrl().trim(); + return base.endsWith("/") ? base.substring(0, base.length() - 1) : base; + } + + // ======================== 解析（包内可见，供单测用真实样例直接校验） ======================== + + /** 展开信封：code=1 返回 data 节点，否则记日志返回 null。 */ + JsonNode parseEnvelope(String body, String scene) { + if (!hasText(body)) { + log.warn("[maixiang-console] 响应为空 scene={}", scene); + return null; + } + try { + JsonNode root = objectMapper.readTree(body); + if (root.path("code").asInt(-1) != 1) { + log.warn("[maixiang-console] 接口返回失败 scene={} mes={}", scene, root.path("mes").asText("")); + return null; + } + return root.path("data"); + } catch (Exception ex) { + log.warn("[maixiang-console] 响应解析失败 scene={} err={}", scene, ex.getMessage()); + return null; + } + } + + BatchTaskPage parseBatchTaskPage(JsonNode data) { + if (data == null || data.isMissingNode()) { + return BatchTaskPage.EMPTY; + } + List items = new ArrayList<>(); + for (JsonNode node : data.path("items")) { + items.add(new BatchTaskItem( + node.path("task_id").asText(""), + node.path("task_type").asText(""), + node.path("uid").asText(""), + node.path("status").asInt(-1), + parseTime(node.path("create_time").asText(null)), + parseTime(node.path("update_time").asText(null)))); + } + return new BatchTaskPage(data.path("total").asLong(items.size()), items); + } + + SingleTaskPage parseSingleTaskPage(JsonNode data) { + if (data == null || data.isMissingNode()) { + return SingleTaskPage.EMPTY; + } + List items = new ArrayList<>(); + for (JsonNode node : data.path("items")) { + items.add(new SingleTaskItem( + node.path("task_id").asText(""), + node.path("task_type").asText(""), + node.path("task_status").asInt(-1), + parseTime(node.path("create_time").asText(null)), + parseTime(node.path("update_time").asText(null)), + node.path("task_result").asText(""))); + } + return new SingleTaskPage(data.path("total").asLong(items.size()), items); + } + + List parseQueueStates(JsonNode data) { + List items = new ArrayList<>(); + if (data == null || data.isMissingNode()) { + return items; + } + for (JsonNode node : data.path("queues")) { + items.add(new QueueStateItem( + node.path("level").asInt(0), + node.path("queue_pending").asLong(0), + node.path("queue_processing").asLong(0))); + } + return items; + } + + long parseBatchResultTotal(JsonNode data) { + if (data == null || data.isMissingNode()) { + return RESULT_TOTAL_UNSUPPORTED; + } + JsonNode table = data.path("table"); + if (table.isMissingNode() || table.isNull() || !hasText(table.asText(""))) { + // 未注册结果表（或表名非法）：该类型不支持按结果行数判断，跳过 + return RESULT_TOTAL_UNSUPPORTED; + } + return data.path("total").asLong(0); + } + + private LocalDateTime parseTime(String text) { + if (!hasText(text)) { + return null; + } + try { + return LocalDateTime.parse(text.trim(), TIME_FORMAT); + } catch (Exception ex) { + log.warn("[maixiang-console] 时间解析失败 text={}", text); + return null; + } + } + + private boolean hasText(String value) { + return value != null && !value.trim().isEmpty(); + } + + // ======================== 数据模型 ======================== + + /** 批量任务（all_task）单条。 */ + public record BatchTaskItem(String taskId, String taskType, String uid, int status, + LocalDateTime createTime, LocalDateTime updateTime) { + } + + public record BatchTaskPage(long total, List items) { + public static final BatchTaskPage EMPTY = new BatchTaskPage(0, List.of()); + } + + /** 单任务（task_record）单条。 */ + public record SingleTaskItem(String taskId, String taskType, int status, + LocalDateTime createTime, LocalDateTime updateTime, String resultRaw) { + } + + public record SingleTaskPage(long total, List items) { + public static final SingleTaskPage EMPTY = new SingleTaskPage(0, List.of()); + } + + /** 队列状态一行。 */ + public record QueueStateItem(int level, long pending, long processing) { + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScanner.java b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScanner.java new file mode 100644 index 00000000..655e3647 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScanner.java @@ -0,0 +1,312 @@ +package com.nanri.aiimage.modules.notification.service; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.NotificationProperties; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.BatchTaskItem; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.BatchTaskPage; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.QueueStateItem; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.SingleTaskItem; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.SingleTaskPage; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.time.LocalDateTime; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * 麦象（18960 任务调度）异常扫描：把「采集端出了事但没人知道」的情况推给管理员。 + * + *

五类检查（全部只读，通过 {@link MaixiangConsoleClient} 的后台接口取数）： + *

    + *
  1. 批量任务停滞：status=0/1 且超过阈值无更新（如消费端 worker 掉线中断采集）；
    2. + *
  2. 单任务滞留：创建超过阈值仍未完成（跟价等单任务无人消费）；
    3. + *
  3. 任务失败：近窗口内 status=3 的单任务达到最小条数；
    4. + *
  4. 批量任务空结果：已完成但结果 0 条（worker 中途报错时批次仍会被标"已完成"）；
    5. + *
  5. 队列积压：task:queue / task:processing 超过阈值。
    6. + *
+ * + *

全部为管理员全局事件（与「服务探测」一致，不按数据权限过滤）；去重键： + * 停滞类按天、失败/积压按小时、空结果按任务——同一异常窗口内只提醒一次， + * 情况变化（新增停滞任务/失败数增长）时刷新内容并重新提醒。 + * + *

由 {@link NotificationScanScheduler} 在分布式锁内调用；单项检查异常只记日志不中断其余检查。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class MaixiangAnomalyScanner { + + private static final String SCENE = NotificationService.SCENE_MAIXIANG_ANOMALY; + + /** 失败聚合窗口（分钟）：与扫描间隔解耦，窗口内失败数达到阈值才告警。 */ + private static final int FAIL_WINDOW_MINUTES = 30; + /** 空结果检查窗口（分钟）：只看最近完成的批量任务（覆盖数个扫描周期）。 */ + private static final int BATCH_EMPTY_WINDOW_MINUTES = 15; + /** 单类接口单轮拉取上限（停滞类任务量级很小，100 足够覆盖）。 */ + private static final int PAGE_SIZE = 100; + private static final int FAIL_PAGE_SIZE = 50; + /** 通知内容里的任务号预览条数。 */ + private static final int MAX_PREVIEW = 5; + private static final int REASON_MAX_LENGTH = 120; + + private static final DateTimeFormatter DAY_FORMAT = DateTimeFormatter.ofPattern("yyyyMMdd"); + private static final DateTimeFormatter HOUR_FORMAT = DateTimeFormatter.ofPattern("yyyyMMddHH"); + private static final DateTimeFormatter TIME_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + + /** 麦象任务类型 → 中文名（未登记的类型回退原始值）。 */ + private static final Map TYPE_LABELS = Map.ofEntries( + Map.entry("price-track-run", "跟价"), + Map.entry("variant-collection", "变体采集"), + Map.entry("collect-data-run", "采集数据"), + Map.entry("collect-data-run-pagels", "采集数据(搜索)"), + Map.entry("collect-data-run-detail", "采集数据(详情)"), + Map.entry("search-collection", "搜索采集"), + Map.entry("shop-data-crawl-run", "店铺数据采集"), + Map.entry("similar-asin-run", "货源查询"), + Map.entry("appearance-patent-run", "外观专利")); + + private final NotificationProperties properties; + private final MaixiangConsoleClient consoleClient; + private final NotificationDispatchService notificationDispatchService; + private final ObjectMapper objectMapper; + + /** 扫描入口：配置缺失时跳过；各检查互相隔离，单项失败不影响其余。 */ + public void scan() { + if (!properties.isMaixiangScanEnabled()) { + log.info("[maixiang-scan] 麦象异常扫描已关闭（maixiang-scan-enabled=false），跳过"); + return; + } + if (!consoleClient.configured()) { + log.info("[maixiang-scan] 麦象扫描跳过：未配置 price-track-api-url / maixiang-console-token"); + return; + } + NotificationDispatchService.AdminAudience audience = notificationDispatchService.prepareAdminAudience(); + LocalDateTime now = LocalDateTime.now(); + runCheck("批量任务停滞", () -> checkStuckBatchTasks(audience, now)); + runCheck("单任务滞留", () -> checkStuckSingleTasks(audience, now)); + runCheck("任务失败", () -> checkFailedSingleTasks(audience, now)); + runCheck("批量任务空结果", () -> checkEmptyBatchResults(audience, now)); + runCheck("队列积压", () -> checkQueueBacklog(audience, now)); + } + + private void runCheck(String name, Runnable check) { + try { + check.run(); + } catch (Exception ex) { + log.warn("[maixiang-scan] {} 检查异常（不影响其余检查）err={}", name, ex.getMessage(), ex); + } + } + + /** 批量任务停滞：创建早于阈值、仍处 status=0/1 且 update_time 不再推进。 */ + void checkStuckBatchTasks(NotificationDispatchService.AdminAudience audience, LocalDateTime now) { + int thresholdMinutes = Math.max(1, properties.getMaixiangStuckMinutes()); + LocalDateTime cutoff = now.minusMinutes(thresholdMinutes); + List candidates = new ArrayList<>(); + for (int status : new int[]{0, 1}) { + candidates.addAll(consoleClient.batchTasks(status, cutoff, PAGE_SIZE).items()); + } + // 创建时间早于 cutoff 但仍在正常推进的大任务（update_time 晚于 cutoff）不算停滞 + List stuck = new ArrayList<>(); + for (BatchTaskItem task : candidates) { + if (task.updateTime() != null && !task.updateTime().isAfter(cutoff)) { + stuck.add(task); + } + } + if (stuck.isEmpty()) { + log.info("[maixiang-scan] 无停滞的批量任务（阈值 {} 分钟）", thresholdMinutes); + return; + } + LocalDateTime earliest = stuck.stream() + .map(BatchTaskItem::updateTime) + .min(LocalDateTime::compareTo) + .orElse(cutoff); + String content = "麦象有 " + stuck.size() + " 个批量任务超过 " + thresholdMinutes + + " 分钟无进展（" + typeSummary(stuck.stream().map(BatchTaskItem::taskType).toList()) + + "；任务 " + taskPreview(stuck) + + "；最早停滞于 " + TIME_FORMAT.format(earliest) + + "）。消费端可能已掉线，请检查采集机 worker 是否在运行。"; + push(audience, "麦象批量任务停滞", content, "maixiang_stuck_batch:" + now.format(DAY_FORMAT)); + } + + /** 单任务滞留：创建早于阈值、仍处 status=0/1（跟价等单任务应秒级完成）。 */ + void checkStuckSingleTasks(NotificationDispatchService.AdminAudience audience, LocalDateTime now) { + int thresholdMinutes = Math.max(1, properties.getMaixiangSingleStuckMinutes()); + LocalDateTime cutoff = now.minusMinutes(thresholdMinutes); + List stuck = new ArrayList<>(); + for (int status : new int[]{0, 1}) { + SingleTaskPage page = consoleClient.singleTasks(status, cutoff, PAGE_SIZE); + for (SingleTaskItem task : page.items()) { + if (task.updateTime() != null && !task.updateTime().isAfter(cutoff)) { + stuck.add(task); + } + } + } + if (stuck.isEmpty()) { + log.info("[maixiang-scan] 无滞留的单任务（阈值 {} 分钟）", thresholdMinutes); + return; + } + String content = "麦象有 " + stuck.size() + " 个单任务超过 " + thresholdMinutes + + " 分钟未完成（" + typeSummary(stuck.stream().map(SingleTaskItem::taskType).toList()) + + "；任务 " + singleTaskPreview(stuck) + + "）。可能存在任务无人消费或消费端卡住，请检查采集机 worker 与队列消费情况。"; + push(audience, "麦象任务滞留未消费", content, "maixiang_stuck_single:" + now.format(DAY_FORMAT)); + } + + /** 任务失败：近窗口内 status=3 的单任务达到最小条数（按小时桶聚合，原因取最新一条）。 */ + void checkFailedSingleTasks(NotificationDispatchService.AdminAudience audience, LocalDateTime now) { + SingleTaskPage page = consoleClient.singleTasks(3, null, FAIL_PAGE_SIZE); + LocalDateTime windowStart = now.minusMinutes(FAIL_WINDOW_MINUTES); + List failed = new ArrayList<>(); + for (SingleTaskItem task : page.items()) { + if (task.updateTime() != null && !task.updateTime().isBefore(windowStart)) { + failed.add(task); + } + } + int minCount = Math.max(1, properties.getMaixiangFailMinCount()); + if (failed.size() < minCount) { + log.info("[maixiang-scan] 近 {} 分钟麦象失败 {} 条（未达阈值 {}）", FAIL_WINDOW_MINUTES, failed.size(), minCount); + return; + } + String reason = latestReason(failed); + String content = "近 " + FAIL_WINDOW_MINUTES + " 分钟麦象有 " + failed.size() + " 条任务失败（" + + typeSummary(failed.stream().map(SingleTaskItem::taskType).toList()) + "）" + + (reason.isEmpty() ? "" : "；最近失败原因：" + reason) + + "。请排查采集端配置（如代理余额、任务类型注册）。"; + push(audience, "麦象任务失败", content, "maixiang_fail:" + now.format(HOUR_FORMAT)); + } + + /** 批量任务空结果：最近完成但结果 0 条（worker 报错时批次仍会被标为"已完成"）。 */ + void checkEmptyBatchResults(NotificationDispatchService.AdminAudience audience, LocalDateTime now) { + BatchTaskPage page = consoleClient.batchTasks(2, null, PAGE_SIZE); + LocalDateTime windowStart = now.minusMinutes(BATCH_EMPTY_WINDOW_MINUTES); + for (BatchTaskItem task : page.items()) { + if (task.updateTime() == null || task.updateTime().isBefore(windowStart)) { + continue; + } + long total = consoleClient.batchResultTotal(task.taskId()); + if (total != 0) { + continue; + } + String content = "批量任务 " + task.taskId() + "（" + labelOf(task.taskType()) + "，uid=" + task.uid() + + "）已完成但结果 0 条，可能采集中途异常（worker 报错时任务仍会被标为已完成），请核查。"; + push(audience, "麦象批量任务无结果", content, "maixiang_batch_empty:" + task.taskId()); + } + } + + /** 队列积压：task:queue 待处理 / task:processing 处理中超过阈值。 */ + void checkQueueBacklog(NotificationDispatchService.AdminAudience audience, LocalDateTime now) { + int pendingThreshold = Math.max(1, properties.getMaixiangQueuePendingThreshold()); + int processingThreshold = Math.max(1, properties.getMaixiangQueueProcessingThreshold()); + StringBuilder detail = new StringBuilder(); + for (QueueStateItem queue : consoleClient.queueStatus()) { + boolean pendingHit = queue.pending() >= pendingThreshold; + boolean processingHit = queue.processing() >= processingThreshold; + if (!pendingHit && !processingHit) { + continue; + } + if (detail.length() > 0) { + detail.append("；"); + } + detail.append("L").append(queue.level()) + .append(" 待处理 ").append(queue.pending()) + .append(" / 处理中 ").append(queue.processing()); + } + if (detail.length() == 0) { + log.info("[maixiang-scan] 无队列积压（阈值 待处理{} / 处理中{}）", pendingThreshold, processingThreshold); + return; + } + String content = "麦象单任务队列积压：" + detail + "（阈值 待处理 " + pendingThreshold + + " / 处理中 " + processingThreshold + "）。消费端可能掉线或处理变慢，请尽快排查。"; + push(audience, "麦象队列积压", content, "maixiang_queue_backlog:" + now.format(HOUR_FORMAT)); + } + + // ======================== 文案辅助 ======================== + + private void push(NotificationDispatchService.AdminAudience audience, String title, String content, String dedupeKeyBase) { + int pushed = notificationDispatchService.pushToAdmins(audience, SCENE, NotificationService.LEVEL_WARNING, + "麦象异常：" + title, content, dedupeKeyBase, null); + log.info("[maixiang-scan] 已推送管理员通知 title={} 落库={} 条 dedupe={}", title, pushed, dedupeKeyBase); + } + + /** 类型分布摘要，例如「变体采集 3 个、跟价 1 个」。 */ + private String typeSummary(List taskTypes) { + Map counts = new LinkedHashMap<>(); + for (String type : taskTypes) { + counts.merge(labelOf(type), 1, Integer::sum); + } + StringBuilder summary = new StringBuilder(); + for (Map.Entry entry : counts.entrySet()) { + if (summary.length() > 0) { + summary.append("、"); + } + summary.append(entry.getKey()).append(" ").append(entry.getValue()).append(" 个"); + } + return summary.toString(); + } + + private String taskPreview(List tasks) { + return preview(tasks.stream().map(BatchTaskItem::taskId).toList(), tasks.size()); + } + + private String singleTaskPreview(List tasks) { + return preview(tasks.stream().map(SingleTaskItem::taskId).toList(), tasks.size()); + } + + private String preview(List ids, int totalCount) { + List shown = new ArrayList<>(MAX_PREVIEW); + for (String id : ids) { + if (id == null || id.isBlank()) { + continue; + } + shown.add(id); + if (shown.size() >= MAX_PREVIEW) { + break; + } + } + if (shown.isEmpty()) { + return "无"; + } + String text = String.join("、", shown); + return totalCount > shown.size() ? text + " 等" : text; + } + + /** 取最新一条失败原因：task_result 为 JSON（如 {"error": "..."}），解析失败退化为原文截断。 */ + private String latestReason(List failed) { + for (SingleTaskItem task : failed) { + String raw = task.resultRaw(); + if (raw == null || raw.isBlank()) { + continue; + } + String reason = raw; + try { + JsonNode node = objectMapper.readTree(raw); + if (node.path("error").isTextual()) { + reason = node.path("error").asText(); + } + } catch (Exception ignore) { + // 非 JSON：直接用原文 + } + reason = reason.replaceAll("\\s+", " ").trim(); + if (reason.isEmpty()) { + continue; + } + return reason.length() <= REASON_MAX_LENGTH ? reason : reason.substring(0, REASON_MAX_LENGTH) + "…"; + } + return ""; + } + + private String labelOf(String taskType) { + String type = taskType == null ? "" : taskType.trim(); + if (type.isEmpty()) { + return "未知类型"; + } + return TYPE_LABELS.getOrDefault(type, type); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java index 5ecd2d08..bfe7c67c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java @@ -35,6 +35,9 @@ import java.util.Map; *

服务探测：品牌检测服务（15126） / 跟价任务 API（18960） / jikip 代理接口， * 失败立即重试一次（过滤瞬抖），两次都失败才告警；同服务每小时最多一条。 * + *

麦象异常扫描：见 {@link MaixiangAnomalyScanner}（批量/单任务停滞、任务失败、 + * 批量任务空结果、队列积压 → 管理员通知），与任务失败扫描共用本调度与分布式锁。 + * *

双实例经 Redis 分布式锁互斥；所有分支留中文日志便于线上排查。 */ @Slf4j @@ -67,6 +70,7 @@ public class NotificationScanScheduler { private final DistributedJobLockService distributedJobLockService; private final NotificationProperties properties; private final ProxyBalancePort proxyBalancePort; + private final MaixiangAnomalyScanner maixiangAnomalyScanner; /** 上次清理日期（每天最多清理一次；双节点由分布式锁保证只有一个实例执行）。 */ private volatile LocalDate lastCleanupDate; @@ -90,6 +94,8 @@ public class NotificationScanScheduler { if (properties.isServiceProbeEnabled()) { probeServices(); } + // 麦象异常扫描（任务停滞/失败/队列积压）；开关与凭据在扫描器内部判定 + maixiangAnomalyScanner.scan(); cleanupExpiredIfNeeded(); } catch (Exception ex) { log.warn("[notification-scan] 扫描异常终止 err={}", ex.getMessage(), ex); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java index 5b62f46a..353c7f44 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java @@ -36,6 +36,8 @@ public class NotificationService { public static final String SCENE_SECRET_INVALID = "secret_invalid"; public static final String SCENE_TASK_FAILED = "task_failed"; public static final String SCENE_SERVICE_DOWN = "service_down"; + /** 麦象（18960 任务调度）异常：任务停滞/失败/队列积压。 */ + public static final String SCENE_MAIXIANG_ANOMALY = "maixiang_anomaly"; public static final String SCENE_SYSTEM = "system"; private static final long MAX_PAGE_SIZE = 100L; diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 684bacc9..156c0a9d 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -354,6 +354,15 @@ aiimage: brand-service-url: ${AIIMAGE_NOTIFICATION_BRAND_SERVICE_URL:} price-track-api-url: ${AIIMAGE_NOTIFICATION_PRICE_TRACK_API_URL:} jikip-probe-enabled: ${AIIMAGE_NOTIFICATION_JIKIP_PROBE_ENABLED:true} + # 麦象（18960 任务调度）异常扫描：批量/单任务停滞、任务失败、队列积压 → 管理员通知。 + # console-token 留空=跳过麦象扫描；接口地址复用上面的 price-track-api-url。 + maixiang-scan-enabled: ${AIIMAGE_NOTIFICATION_MAIXIANG_SCAN_ENABLED:true} + maixiang-console-token: ${AIIMAGE_NOTIFICATION_MAIXIANG_CONSOLE_TOKEN:} + maixiang-stuck-minutes: ${AIIMAGE_NOTIFICATION_MAIXIANG_STUCK_MINUTES:60} + maixiang-single-stuck-minutes: ${AIIMAGE_NOTIFICATION_MAIXIANG_SINGLE_STUCK_MINUTES:30} + maixiang-fail-min-count: ${AIIMAGE_NOTIFICATION_MAIXIANG_FAIL_MIN_COUNT:1} + maixiang-queue-pending-threshold: ${AIIMAGE_NOTIFICATION_MAIXIANG_QUEUE_PENDING_THRESHOLD:300} + maixiang-queue-processing-threshold: ${AIIMAGE_NOTIFICATION_MAIXIANG_QUEUE_PROCESSING_THRESHOLD:100} read-retention-days: ${AIIMAGE_NOTIFICATION_READ_RETENTION_DAYS:90} security: shop-credential-key: ${AIIMAGE_SHOP_CREDENTIAL_KEY:change-me-shop-credential-key} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClientTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClientTest.java new file mode 100644 index 00000000..98a27651 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/client/MaixiangConsoleClientTest.java @@ -0,0 +1,152 @@ +package com.nanri.aiimage.modules.notification.client; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.NotificationProperties; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.BatchTaskPage; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.QueueStateItem; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.SingleTaskPage; +import org.junit.jupiter.api.Test; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * 解析层单测：样例 JSON 全部来自线上 18960 后台接口真实返回（2026-09-15 抓取）， + * 接口契约变化时这里会第一时间失败。 + */ +class MaixiangConsoleClientTest { + + private final MaixiangConsoleClient client = + new MaixiangConsoleClient(new NotificationProperties(), new ObjectMapper()); + + /** /api/console/batch/tasks 真实返回（截断 task_data）。 */ + private static final String BATCH_TASKS_JSON = """ + {"code":1,"mes":"成功","data":{"total":3,"page":1,"page_size":5,"items":[ + {"id":75,"task_id":"3011bc05-145d-4dfe-accd-5e02d75a4f47","task_type":"variant-collection", + "task_data":"{\\"file_url\\": \\"https://oss.aishufu.top/xx.xlsx\\", \\"execute_country\\": [\\"UK\\"]}", + "status":1,"uid":"1048","create_time":"2026-09-14 14:28:10","update_time":"2026-09-14 22:51:05"}, + {"id":74,"task_id":"a0535546-3c99-4c9f-9b9b-a7c516b18726","task_type":"variant-collection", + "task_data":"{}","status":1,"uid":"1053","create_time":"2026-09-14 10:58:40","update_time":"2026-09-14 22:50:49"} + ]}} + """; + + /** /api/console/tasks 真实返回（截断 task_data）。 */ + private static final String SINGLE_TASKS_JSON = """ + {"code":1,"mes":"成功","data":{"total":39,"page":1,"page_size":3,"items":[ + {"id":1696319,"task_id":"task:price-track-run:d449150a-cbcb-4640-82e6-6388f6d3a4a4", + "task_type":"price-track-run","task_data":"{}","task_status":0,"task_result":null, + "create_time":"2026-09-13 21:08:07","update_time":"2026-09-13 21:08:07"}, + {"id":1740034,"task_id":"task:price-track-run:f5048c8b-e05d-4820-9cf8-e91fc86424d5", + "task_type":"price-track-run","task_data":"{}", + "task_result":"{\\"error\\": \\"未注册的任务类型: price-track-run\\"}","task_status":3, + "create_time":"2026-09-15 03:20:30","update_time":"2026-09-15 03:20:32"} + ]}} + """; + + @Test + void parseBatchTasksWithRealPayload() { + BatchTaskPage page = client.parseBatchTaskPage(client.parseEnvelope(BATCH_TASKS_JSON, "test")); + + assertThat(page.total()).isEqualTo(3); + assertThat(page.items()).hasSize(2); + var first = page.items().get(0); + assertThat(first.taskId()).isEqualTo("3011bc05-145d-4dfe-accd-5e02d75a4f47"); + assertThat(first.taskType()).isEqualTo("variant-collection"); + assertThat(first.uid()).isEqualTo("1048"); + assertThat(first.status()).isEqualTo(1); + assertThat(first.updateTime()).isEqualTo(LocalDateTime.of(2026, 9, 14, 22, 51, 5)); + } + + @Test + void parseSingleTasksWithRealPayload() { + SingleTaskPage page = client.parseSingleTaskPage(client.parseEnvelope(SINGLE_TASKS_JSON, "test")); + + assertThat(page.total()).isEqualTo(39); + assertThat(page.items()).hasSize(2); + var failed = page.items().get(1); + assertThat(failed.status()).isEqualTo(3); + assertThat(failed.resultRaw()).contains("未注册的任务类型"); + assertThat(failed.updateTime()).isEqualTo(LocalDateTime.of(2026, 9, 15, 3, 20, 32)); + } + + @Test + void parseQueueStatesWithRealPayload() { + List empty = client.parseQueueStates( + client.parseEnvelope("{\"code\":1,\"mes\":\"成功\",\"data\":{\"queues\":[]}}", "test")); + assertThat(empty).isEmpty(); + + List queues = client.parseQueueStates(client.parseEnvelope( + "{\"code\":1,\"mes\":\"成功\",\"data\":{\"queues\":[{\"level\":1,\"queue_pending\":500,\"queue_processing\":3}]}}", + "test")); + assertThat(queues).hasSize(1); + assertThat(queues.get(0).level()).isEqualTo(1); + assertThat(queues.get(0).pending()).isEqualTo(500); + assertThat(queues.get(0).processing()).isEqualTo(3); + } + + @Test + void parseBatchResultTotalDistinguishesUnsupportedTable() { + // 已注册结果表：返回真实行数 + long total = client.parseBatchResultTotal(client.parseEnvelope( + "{\"code\":1,\"mes\":\"成功\",\"data\":{\"task\":{\"task_id\":\"t1\"},\"table\":\"variant_collection\",\"columns\":[],\"items\":[],\"total\":0}}", + "test")); + assertThat(total).isZero(); + + // 未注册结果表（table=null，如 search-collection）：哨兵值，不参与空结果判断 + long unsupported = client.parseBatchResultTotal(client.parseEnvelope( + "{\"code\":1,\"mes\":\"成功\",\"data\":{\"task\":{\"task_id\":\"t2\"},\"table\":null,\"columns\":[],\"items\":[],\"total\":0}}", + "test")); + assertThat(unsupported).isEqualTo(MaixiangConsoleClient.RESULT_TOTAL_UNSUPPORTED); + } + + @Test + void parseEnvelopeRejectsFailureResponses() { + assertThat(client.parseEnvelope("{\"code\":-1,\"mes\":\"无权限\",\"data\":null}", "test")).isNull(); + assertThat(client.parseEnvelope("", "test")).isNull(); + assertThat(client.parseEnvelope("not-json", "test")).isNull(); + } + + @Test + void parseToleratesNullUidAndBadTime() { + BatchTaskPage page = client.parseBatchTaskPage(client.parseEnvelope( + "{\"code\":1,\"mes\":\"成功\",\"data\":{\"total\":1,\"items\":[{\"task_id\":\"t3\",\"task_type\":\"search-collection\",\"uid\":\"collect-27475\",\"status\":2,\"create_time\":\"bad\",\"update_time\":\"\"}]}}", + "test")); + assertThat(page.items().get(0).uid()).isEqualTo("collect-27475"); + assertThat(page.items().get(0).updateTime()).isNull(); + } + + @Test + void configuredRequiresBothUrlAndToken() { + NotificationProperties properties = new NotificationProperties(); + MaixiangConsoleClient underTest = new MaixiangConsoleClient(properties, new ObjectMapper()); + assertThat(underTest.configured()).isFalse(); + + properties.setPriceTrackApiUrl("http://192.168.0.170:18960"); + assertThat(underTest.configured()).isFalse(); + + properties.setMaixiangConsoleToken("test-token"); + assertThat(underTest.configured()).isTrue(); + } + + @Test + void buildUrlEncodesTimeParameterExactlyOnce() { + // 回归：end_time 里的空格必须编码成 %20（单次），否则提交时会被二次编码成 %2520， + // 服务端拿到字面 "%20" 直接报 Incorrect DATETIME value（联调时实际踩到过） + NotificationProperties properties = new NotificationProperties(); + properties.setPriceTrackApiUrl("http://192.168.0.170:18960"); + properties.setMaixiangConsoleToken("tok en+特殊"); + MaixiangConsoleClient underTest = new MaixiangConsoleClient(properties, new ObjectMapper()); + + String url = underTest.buildUrl("/api/console/tasks", + "status=0", "page=1", "page_size=100", "end_time=2026-09-15 10:58:39"); + + assertThat(url).startsWith("http://192.168.0.170:18960/api/console/tasks?"); + assertThat(url).contains("end_time=2026-09-15%2010:58:39"); + assertThat(url).doesNotContain("%2520"); + assertThat(url).contains("token=tok%20en+%E7%89%B9%E6%AE%8A"); + // 组装结果必须是可直接提交的合法 URI + assertThat(java.net.URI.create(url).getRawQuery()).contains("end_time=2026-09-15%2010:58:39"); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScannerTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScannerTest.java new file mode 100644 index 00000000..577c6ab9 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangAnomalyScannerTest.java @@ -0,0 +1,284 @@ +package com.nanri.aiimage.modules.notification.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.NotificationProperties; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.BatchTaskItem; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.BatchTaskPage; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.QueueStateItem; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.SingleTaskItem; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient.SingleTaskPage; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class MaixiangAnomalyScannerTest { + + private static final LocalDateTime NOW = LocalDateTime.of(2026, 9, 15, 12, 0, 0); + + private final NotificationProperties properties = new NotificationProperties(); + private final MaixiangConsoleClient consoleClient = mock(MaixiangConsoleClient.class); + private final NotificationDispatchService dispatch = mock(NotificationDispatchService.class); + + private final NotificationDispatchService.AdminAudience audience = + new NotificationDispatchService.AdminAudience(List.of(), Set.of(), Map.of()); + + private MaixiangAnomalyScanner newScanner() { + return new MaixiangAnomalyScanner(properties, consoleClient, dispatch, new ObjectMapper()); + } + + // ======================== 扫描入口 ======================== + + @Test + void scanSkipsWhenDisabled() { + properties.setMaixiangScanEnabled(false); + + newScanner().scan(); + + verifyNoInteractions(consoleClient); + verifyNoInteractions(dispatch); + } + + @Test + void scanSkipsWhenNotConfigured() { + // 未配置 18960 地址/令牌：只做 configured() 探测，不查库、不拉数据、不告警 + newScanner().scan(); + + verifyNoInteractions(dispatch); + verify(consoleClient, never()).batchTasks(anyInt(), any(), anyInt()); + verify(consoleClient, never()).singleTasks(anyInt(), any(), anyInt()); + verify(consoleClient, never()).queueStatus(); + } + + // ======================== 批量任务停滞 ======================== + + @Test + void stuckBatchTasksPushAdminNotificationWithStableContent() { + when(consoleClient.batchTasks(eq(0), any(LocalDateTime.class), anyInt())).thenReturn(BatchTaskPage.EMPTY); + when(consoleClient.batchTasks(eq(1), any(LocalDateTime.class), anyInt())).thenReturn(new BatchTaskPage(2, List.of( + batchTask("3011bc05", "variant-collection", "1048", 1, NOW.minusHours(13)), + batchTask("a0535546", "variant-collection", "1053", 1, NOW.minusHours(13))))); + when(dispatch.pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any())) + .thenReturn(1); + + newScanner().checkStuckBatchTasks(audience, NOW); + + ArgumentCaptor content = ArgumentCaptor.forClass(String.class); + ArgumentCaptor key = ArgumentCaptor.forClass(String.class); + verify(dispatch).pushToAdmins(eq(audience), eq(NotificationService.SCENE_MAIXIANG_ANOMALY), + eq(NotificationService.LEVEL_WARNING), eq("麦象异常：麦象批量任务停滞"), + content.capture(), key.capture(), isNull()); + assertThat(content.getValue()).contains("2 个批量任务超过 60 分钟无进展"); + assertThat(content.getValue()).contains("变体采集 2 个"); + assertThat(content.getValue()).contains("3011bc05"); + assertThat(content.getValue()).contains("最早停滞于 2026-09-14 23:00:00"); + assertThat(key.getValue()).isEqualTo("maixiang_stuck_batch:20260915"); + } + + @Test + void freshBatchTasksDoNotTriggerAlert() { + // 创建时间早于阈值但 update_time 仍在推进（如正常跑着的大任务）不算停滞 + when(consoleClient.batchTasks(eq(0), any(LocalDateTime.class), anyInt())).thenReturn(BatchTaskPage.EMPTY); + when(consoleClient.batchTasks(eq(1), any(LocalDateTime.class), anyInt())).thenReturn(new BatchTaskPage(1, List.of( + batchTask("fresh-task", "variant-collection", "1048", 1, NOW.minusMinutes(5))))); + + newScanner().checkStuckBatchTasks(audience, NOW); + + verify(dispatch, never()).pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any()); + } + + // ======================== 单任务滞留 ======================== + + @Test + void stuckSingleTasksPushAdminNotification() { + when(consoleClient.singleTasks(eq(0), any(LocalDateTime.class), anyInt())) + .thenReturn(new SingleTaskPage(39, List.of( + singleTask("task:price-track-run:d449150a", "price-track-run", 0, NOW.minusDays(2))))); + when(consoleClient.singleTasks(eq(1), any(LocalDateTime.class), anyInt())).thenReturn(SingleTaskPage.EMPTY); + when(dispatch.pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any())) + .thenReturn(1); + + newScanner().checkStuckSingleTasks(audience, NOW); + + ArgumentCaptor content = ArgumentCaptor.forClass(String.class); + ArgumentCaptor key = ArgumentCaptor.forClass(String.class); + verify(dispatch).pushToAdmins(eq(audience), eq(NotificationService.SCENE_MAIXIANG_ANOMALY), + eq(NotificationService.LEVEL_WARNING), eq("麦象异常：麦象任务滞留未消费"), + content.capture(), key.capture(), isNull()); + assertThat(content.getValue()).contains("1 个单任务超过 30 分钟未完成"); + assertThat(content.getValue()).contains("跟价 1 个"); + assertThat(key.getValue()).isEqualTo("maixiang_stuck_single:20260915"); + } + + @Test + void freshSingleTasksDoNotTriggerAlert() { + when(consoleClient.singleTasks(eq(0), any(LocalDateTime.class), anyInt())) + .thenReturn(new SingleTaskPage(1, List.of( + singleTask("task:price-track-run:fresh", "price-track-run", 0, NOW.minusMinutes(1))))); + when(consoleClient.singleTasks(eq(1), any(LocalDateTime.class), anyInt())).thenReturn(SingleTaskPage.EMPTY); + + newScanner().checkStuckSingleTasks(audience, NOW); + + verify(dispatch, never()).pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any()); + } + + // ======================== 任务失败 ======================== + + @Test + void failedTasksPushWithLatestReason() { + when(consoleClient.singleTasks(eq(3), isNull(), anyInt())).thenReturn(new SingleTaskPage(3, List.of( + singleTask("task:price-track-run:f5048c8b", "price-track-run", 3, NOW.minusMinutes(10))))); + when(dispatch.pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any())) + .thenReturn(1); + + newScanner().checkFailedSingleTasks(audience, NOW); + + ArgumentCaptor content = ArgumentCaptor.forClass(String.class); + ArgumentCaptor key = ArgumentCaptor.forClass(String.class); + verify(dispatch).pushToAdmins(eq(audience), eq(NotificationService.SCENE_MAIXIANG_ANOMALY), + eq(NotificationService.LEVEL_WARNING), eq("麦象异常：麦象任务失败"), + content.capture(), key.capture(), isNull()); + assertThat(content.getValue()).contains("1 条任务失败"); + assertThat(content.getValue()).contains("未注册的任务类型: price-track-run"); + assertThat(key.getValue()).isEqualTo("maixiang_fail:2026091512"); + } + + @Test + void failedTasksBelowThresholdDoNotAlert() { + properties.setMaixiangFailMinCount(3); + when(consoleClient.singleTasks(eq(3), isNull(), anyInt())).thenReturn(new SingleTaskPage(2, List.of( + singleTask("t1", "price-track-run", 3, NOW.minusMinutes(5)), + singleTask("t2", "price-track-run", 3, NOW.minusMinutes(6))))); + + newScanner().checkFailedSingleTasks(audience, NOW); + + verify(dispatch, never()).pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any()); + } + + @Test + void failedTasksOutsideWindowDoNotAlert() { + // 更新时间在 30 分钟窗口之外的历史失败不参与聚合 + when(consoleClient.singleTasks(eq(3), isNull(), anyInt())).thenReturn(new SingleTaskPage(1, List.of( + singleTask("t-old", "price-track-run", 3, NOW.minusHours(3))))); + + newScanner().checkFailedSingleTasks(audience, NOW); + + verify(dispatch, never()).pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any()); + } + + // ======================== 批量任务空结果 ======================== + + @Test + void emptyBatchResultPushesOncePerTask() { + BatchTaskItem finished = batchTask("empty-task-1", "variant-collection", "1095", 2, NOW.minusMinutes(3)); + when(consoleClient.batchTasks(eq(2), isNull(), anyInt())) + .thenReturn(new BatchTaskPage(1, List.of(finished))); + when(consoleClient.batchResultTotal("empty-task-1")).thenReturn(0L); + when(dispatch.pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any())) + .thenReturn(1); + + newScanner().checkEmptyBatchResults(audience, NOW); + + ArgumentCaptor content = ArgumentCaptor.forClass(String.class); + ArgumentCaptor key = ArgumentCaptor.forClass(String.class); + verify(dispatch).pushToAdmins(eq(audience), eq(NotificationService.SCENE_MAIXIANG_ANOMALY), + eq(NotificationService.LEVEL_WARNING), eq("麦象异常：麦象批量任务无结果"), + content.capture(), key.capture(), isNull()); + assertThat(content.getValue()).contains("empty-task-1").contains("变体采集").contains("uid=1095"); + assertThat(key.getValue()).isEqualTo("maixiang_batch_empty:empty-task-1"); + } + + @Test + void batchResultWithRowsOrUnsupportedTableDoNotAlert() { + when(consoleClient.batchTasks(eq(2), isNull(), anyInt())).thenReturn(new BatchTaskPage(2, List.of( + batchTask("normal-task", "variant-collection", "1095", 2, NOW.minusMinutes(3)), + batchTask("unsupported-task", "search-collection", "collect-27475", 2, NOW.minusMinutes(3))))); + when(consoleClient.batchResultTotal("normal-task")).thenReturn(12L); + when(consoleClient.batchResultTotal("unsupported-task")) + .thenReturn(MaixiangConsoleClient.RESULT_TOTAL_UNSUPPORTED); + + newScanner().checkEmptyBatchResults(audience, NOW); + + verify(dispatch, never()).pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any()); + } + + // ======================== 队列积压 ======================== + + @Test + void queueBacklogPushesWhenOverThreshold() { + when(consoleClient.queueStatus()).thenReturn(List.of(new QueueStateItem(1, 500, 3))); + when(dispatch.pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any())) + .thenReturn(1); + + newScanner().checkQueueBacklog(audience, NOW); + + ArgumentCaptor content = ArgumentCaptor.forClass(String.class); + verify(dispatch).pushToAdmins(eq(audience), eq(NotificationService.SCENE_MAIXIANG_ANOMALY), + eq(NotificationService.LEVEL_WARNING), eq("麦象异常：麦象队列积压"), + content.capture(), eq("maixiang_queue_backlog:2026091512"), isNull()); + assertThat(content.getValue()).contains("L1 待处理 500 / 处理中 3"); + } + + @Test + void queueUnderThresholdDoNotAlert() { + when(consoleClient.queueStatus()).thenReturn(List.of(new QueueStateItem(1, 10, 2))); + + newScanner().checkQueueBacklog(audience, NOW); + + verify(dispatch, never()).pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any()); + } + + // ======================== 容错 ======================== + + @Test + void scanSwallowsSingleCheckFailure() { + // 配置齐全时，某项检查抛异常不影响其余检查（也不向外抛） + properties.setPriceTrackApiUrl("http://192.168.0.170:18960"); + properties.setMaixiangConsoleToken("token"); + when(consoleClient.configured()).thenReturn(true); + when(dispatch.prepareAdminAudience()).thenReturn(audience); + // 仅批量任务停滞检查（status=0 拉取）抛异常 + doThrow(new RuntimeException("麦象接口超时")).when(consoleClient).batchTasks(eq(0), any(), anyInt()); + doReturn(BatchTaskPage.EMPTY).when(consoleClient).batchTasks(eq(1), any(), anyInt()); + doReturn(BatchTaskPage.EMPTY).when(consoleClient).batchTasks(eq(2), any(), anyInt()); + doReturn(SingleTaskPage.EMPTY).when(consoleClient).singleTasks(anyInt(), any(), anyInt()); + doReturn(List.of()).when(consoleClient).queueStatus(); + + newScanner().scan(); + + // 第一项检查抛异常被吞掉后，末尾的队列检查仍然执行到 + verify(consoleClient, times(1)).queueStatus(); + verify(dispatch, never()).pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any()); + } + + // ======================== 构造辅助 ======================== + + private BatchTaskItem batchTask(String taskId, String type, String uid, int status, LocalDateTime updateTime) { + return new BatchTaskItem(taskId, type, uid, status, updateTime, updateTime); + } + + private SingleTaskItem singleTask(String taskId, String type, int status, LocalDateTime updateTime) { + return new SingleTaskItem(taskId, type, status, updateTime, updateTime, + "{\"error\": \"未注册的任务类型: price-track-run\"}"); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangLiveProbeTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangLiveProbeTest.java new file mode 100644 index 00000000..14c15871 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/MaixiangLiveProbeTest.java @@ -0,0 +1,81 @@ +package com.nanri.aiimage.modules.notification.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.NotificationProperties; +import com.nanri.aiimage.modules.notification.client.MaixiangConsoleClient; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * 手工联调：直连线上 18960 后台接口跑一遍麦象异常扫描，打印各检查的实际判定与文案（只读，不写通知）。 + * + *

默认不执行；需要时显式开启（在 backend-java 目录）： + * 

+ *   mvn test -Dtest=MaixiangLiveProbeTest -Dmaixiang.live=true \
+ *       -Dmaixiang.url=http://192.168.0.170:18960 -Dmaixiang.token=xxx
+ *
+ * 参数可省略，缺省读环境变量 AIIMAGE_NOTIFICATION_PRICE_TRACK_API_URL / AIIMAGE_NOTIFICATION_MAIXIANG_CONSOLE_TOKEN。 + */ +@EnabledIfSystemProperty(named = "maixiang.live", matches = "true") +class MaixiangLiveProbeTest { + + @Test + void probeLiveMaixiangAnomalies() { + NotificationProperties properties = new NotificationProperties(); + properties.setPriceTrackApiUrl(firstNonBlank( + System.getProperty("maixiang.url"), System.getenv("AIIMAGE_NOTIFICATION_PRICE_TRACK_API_URL"))); + properties.setMaixiangConsoleToken(firstNonBlank( + System.getProperty("maixiang.token"), System.getenv("AIIMAGE_NOTIFICATION_MAIXIANG_CONSOLE_TOKEN"))); + System.out.println("[live-probe] 目标 " + properties.getPriceTrackApiUrl() + + "，token " + (properties.getMaixiangConsoleToken().isEmpty() ? "缺失" : "已配置")); + + MaixiangConsoleClient client = new MaixiangConsoleClient(properties, new ObjectMapper()); + NotificationDispatchService dispatch = mock(NotificationDispatchService.class); + List captured = new ArrayList<>(); + when(dispatch.prepareAdminAudience()).thenReturn( + new NotificationDispatchService.AdminAudience(List.of(), Set.of(), Map.of())); + when(dispatch.pushToAdmins(any(), anyString(), anyString(), anyString(), anyString(), anyString(), any())) + .thenAnswer(inv -> { + captured.add("【" + inv.getArgument(3) + "】" + inv.getArgument(4) + + " (dedupe=" + inv.getArgument(5) + ")"); + return 1; + }); + + new MaixiangAnomalyScanner(properties, client, dispatch, new ObjectMapper()).scan(); + + System.out.println("[live-probe] ===== 实际会推送的麦象异常通知（" + captured.size() + " 条）====="); + if (captured.isEmpty()) { + System.out.println("[live-probe] （无——当前线上无麦象异常）"); + } + for (String line : captured) { + System.out.println("[live-probe] " + line); + } + // 控制台在中文 Windows 下编码不可靠，落一份 UTF-8 报告便于人工核对 + try { + java.nio.file.Files.writeString( + java.nio.file.Path.of("target", "maixiang-live-probe.txt"), + String.join(System.lineSeparator(), captured), java.nio.charset.StandardCharsets.UTF_8); + } catch (Exception ignore) { + // 落盘失败不影响探测本身 + } + } + + private String firstNonBlank(String... values) { + for (String value : values) { + if (value != null && !value.trim().isEmpty()) { + return value.trim(); + } + } + return ""; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/NotificationScanSchedulerTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/NotificationScanSchedulerTest.java index 83c9819c..0323a807 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/NotificationScanSchedulerTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/notification/service/NotificationScanSchedulerTest.java @@ -46,10 +46,11 @@ class NotificationScanSchedulerTest { mock(com.nanri.aiimage.common.service.DistributedJobLockService.class); private final NotificationProperties properties = new NotificationProperties(); private final ProxyBalancePort proxyBalancePort = mock(ProxyBalancePort.class); + private final MaixiangAnomalyScanner maixiangAnomalyScanner = mock(MaixiangAnomalyScanner.class); private NotificationScanScheduler newScheduler() { return new NotificationScanScheduler(fileTaskMapper, brandCrawlTaskMapper, notificationService, - dispatch, lockService, properties, proxyBalancePort); + dispatch, lockService, properties, proxyBalancePort, maixiangAnomalyScanner); } @Test diff --git a/frontend-vue/src/shared/api/types/modules/notification.ts b/frontend-vue/src/shared/api/types/modules/notification.ts index 859dac50..0bb30656 100644 --- a/frontend-vue/src/shared/api/types/modules/notification.ts +++ b/frontend-vue/src/shared/api/types/modules/notification.ts @@ -2,8 +2,14 @@ import { get, post, type JavaApiResponse, unwrapJavaResponse } from '../../http. import { buildJavaUrl } from '../../url.ts' import { API_ENDPOINTS } from '../../endpoints.ts' -/** 通知场景：密钥欠费 / 密钥失效 / 任务失败 / 服务异常 / 系统。 */ -export type NotificationScene = 'secret_balance' | 'secret_invalid' | 'task_failed' | 'service_down' | 'system' +/** 通知场景：密钥欠费 / 密钥失效 / 任务失败 / 服务异常 / 麦象异常 / 系统。 */ +export type NotificationScene = + | 'secret_balance' + | 'secret_invalid' + | 'task_failed' + | 'service_down' + | 'maixiang_anomaly' + | 'system' export type NotificationLevel = 'info' | 'warning' | 'error'