feat(A1/A3+C8): 客户端令牌链路打通(按人鉴权就绪)+ 快照 JSON 写入节流
A1/A3 客户端令牌链路(服务端守卫已能按 JWT 鉴权,缺口在客户端不带身份) - 前端:新增 user-token-bridge(纯逻辑,值变化才推送)+ user-token-sync(启动安装, pywebviewready/storage/60s 轮询补推),main.ts 接入;桥接口补 set_user_token - 客户端:新增 crawler_core/user_token.py(持有 + 对自家 Java 端点注入 Authorization: Bearer <jwt>,Session.request 包装,第三方域名不注入、已有头不覆盖、 登出清空、SHUFUAI_DISABLE_USER_TOKEN_HOOK 可关);main.py 暴露桥方法并在启动安装钩子 - 服务端:守卫契约测试补 5 个用户态前缀用例(开关关闭放行 / 打开后匿名 401 / 用户令牌通过 / 内部令牌仍放行 / /api/ziniao 第二层) - 翻开关的前置条件(客户端铺开后置 aiimage.security.user-tool-guard-enabled=true)写入报告 C8 快照 JSON 写入节流 - 读端改以 biz_task_result_item 行为准、JSON 仅兜底(历史任务 JSON 仍是唯一副本时可用) - 整档 JSON 改 30s 节流写,终态路径 force 立即写;新增 2 个契约测试钉住语义 验证:mvn test 2888 全绿;npm test 741 全绿;user_token 钩子自测(注入/归一/第三方跳过/不覆盖/登出)通过
This commit is contained in:
+50
-6
@@ -63,7 +63,20 @@ public class QueryAsinTaskService {
|
||||
private static final int RESULT_SUCCESS = 1;
|
||||
private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断";
|
||||
|
||||
/**
|
||||
* 整档快照 JSON 的写入节流间隔(2026-09 审查 C8)。
|
||||
*
|
||||
* <p>分片回传很频繁时逐次重写整份文档是 O(n²) 序列化开销;读端已改为**以
|
||||
* biz_task_result_item 行为准、JSON 仅作兜底**,因此允许 JSON 短时滞后。
|
||||
* 终态(结果文件生成/结束)走 force 立即写,保证历史详情最终一致。
|
||||
*/
|
||||
private static final long SNAPSHOT_JSON_WRITE_THROTTLE_MILLIS = 30_000L;
|
||||
|
||||
private final FileTaskMapper fileTaskMapper;
|
||||
|
||||
/** taskId → 上次整档 JSON 写入时间(仅本进程节流;跨实例由终态强制写兜底) */
|
||||
private final java.util.concurrent.ConcurrentHashMap<Long, Long> snapshotJsonWrittenAt =
|
||||
new java.util.concurrent.ConcurrentHashMap<>();
|
||||
private final FileResultMapper fileResultMapper;
|
||||
private final QueryAsinResolveService queryAsinResolveService;
|
||||
private final QueryAsinExcelAssemblyService excelAssemblyService;
|
||||
@@ -619,9 +632,11 @@ public class QueryAsinTaskService {
|
||||
private Map<Long, Map<Long, QueryAsinResultItemVo>> buildSnapshotMap(Map<Long, FileTaskEntity> taskMap) {
|
||||
Map<Long, Map<Long, QueryAsinResultItemVo>> out = new LinkedHashMap<>();
|
||||
for (Map.Entry<Long, FileTaskEntity> entry : taskMap.entrySet()) {
|
||||
List<QueryAsinResultItemVo> snapshots = parseTaskSnapshots(entry.getValue().getResultJson());
|
||||
// 以行表为准(增量真源),JSON 仅作兜底(历史任务的 JSON 是唯一副本)
|
||||
List<QueryAsinResultItemVo> snapshots =
|
||||
taskResultItemService.listResultSnapshots(entry.getKey(), MODULE_TYPE, QueryAsinResultItemVo.class);
|
||||
if (snapshots.isEmpty()) {
|
||||
snapshots = taskResultItemService.listResultSnapshots(entry.getKey(), MODULE_TYPE, QueryAsinResultItemVo.class);
|
||||
snapshots = parseTaskSnapshots(entry.getValue().getResultJson());
|
||||
}
|
||||
out.put(entry.getKey(), indexSnapshotByResultId(snapshots));
|
||||
}
|
||||
@@ -722,7 +737,12 @@ public class QueryAsinTaskService {
|
||||
}
|
||||
|
||||
private List<QueryAsinResultItemVo> buildSnapshotFromDb(FileTaskEntity task, List<FileResultEntity> rows) {
|
||||
List<QueryAsinResultItemVo> existing = parseTaskSnapshots(task.getResultJson());
|
||||
// 同上:行表优先,JSON 兜底
|
||||
List<QueryAsinResultItemVo> existing =
|
||||
taskResultItemService.listResultSnapshots(task.getId(), MODULE_TYPE, QueryAsinResultItemVo.class);
|
||||
if (existing.isEmpty()) {
|
||||
existing = parseTaskSnapshots(task.getResultJson());
|
||||
}
|
||||
Map<Long, QueryAsinResultItemVo> snapshotByResultId = indexSnapshotByResultId(existing);
|
||||
List<QueryAsinResultItemVo> list = new ArrayList<>();
|
||||
for (FileResultEntity row : rows) {
|
||||
@@ -962,7 +982,8 @@ public class QueryAsinTaskService {
|
||||
}
|
||||
}
|
||||
|
||||
persistSnapshotJson(task, snapshots);
|
||||
// 终态:强制重写整档 JSON,保证历史详情与结果文件口径一致
|
||||
persistSnapshotJson(task, snapshots, true);
|
||||
fileTaskMapper.updateById(task);
|
||||
taskCacheService.deleteTaskCache(task.getId());
|
||||
}
|
||||
@@ -1008,7 +1029,7 @@ public class QueryAsinTaskService {
|
||||
}
|
||||
updateResultFileColumns(successRowIds, filename, objectKey, fileSize, rowCount);
|
||||
updateTaskStatusFromRows(task, rows);
|
||||
persistSnapshotJson(task, buildSnapshotFromDb(task, rows));
|
||||
persistSnapshotJson(task, buildSnapshotFromDb(task, rows), true);
|
||||
fileTaskMapper.updateById(task);
|
||||
} finally {
|
||||
FileUtil.del(xlsx);
|
||||
@@ -1090,14 +1111,37 @@ public class QueryAsinTaskService {
|
||||
}
|
||||
|
||||
private void persistSnapshotJson(FileTaskEntity task, List<QueryAsinResultItemVo> snapshots) {
|
||||
persistSnapshotJson(task, snapshots, false);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param force true=立即重写整档 JSON(终态路径用);false=按节流窗口决定是否重写
|
||||
*/
|
||||
private void persistSnapshotJson(FileTaskEntity task, List<QueryAsinResultItemVo> snapshots, boolean force) {
|
||||
try {
|
||||
task.setResultJson(objectMapper.writeValueAsString(snapshots == null ? List.of() : snapshots));
|
||||
// 行表始终同步(增量 upsert,内容未变时自身会跳过)
|
||||
syncSnapshotTables(task, snapshots);
|
||||
if (!force && !snapshotJsonWriteDue(task)) {
|
||||
return;
|
||||
}
|
||||
task.setResultJson(objectMapper.writeValueAsString(snapshots == null ? List.of() : snapshots));
|
||||
if (task.getId() != null) {
|
||||
snapshotJsonWrittenAt.put(task.getId(), System.currentTimeMillis());
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
throw new BusinessException("查询ASIN任务快照保存失败");
|
||||
}
|
||||
}
|
||||
|
||||
private boolean snapshotJsonWriteDue(FileTaskEntity task) {
|
||||
if (task == null || task.getId() == null) {
|
||||
return true;
|
||||
}
|
||||
Long lastWrittenAt = snapshotJsonWrittenAt.get(task.getId());
|
||||
return lastWrittenAt == null
|
||||
|| System.currentTimeMillis() - lastWrittenAt >= SNAPSHOT_JSON_WRITE_THROTTLE_MILLIS;
|
||||
}
|
||||
|
||||
private void syncSnapshotTables(FileTaskEntity task, List<QueryAsinResultItemVo> snapshots) {
|
||||
List<QueryAsinResultItemVo> safe = snapshots == null ? List.of() : snapshots;
|
||||
taskResultItemService.replaceTaskSnapshots(task.getId(), MODULE_TYPE, safe, new TaskResultItemService.SnapshotKeyResolver() {
|
||||
|
||||
Reference in New Issue
Block a user