feat(schedule): 定时任务 8 项 + 用量快照/压缩重试/到期提醒（V17）
PRD 附录 E。每个任务都是两段式：先以平台身份（tenantId=0）扫候选，再逐个租户摆上下文 干活——写数据必须在第二段，否则 tenant_id 被写成 0，通知进平台租户且不报错。 V17 新建 sys_tenant_usage_snapshot（唯一索引保证幂等；tenant_id 必须显式写）并给 collect_file 补 process_retry_count / process_error / last_retry_time + 只索引 「还没处理出规范件」的部分索引；压缩失败重试必须封顶，否则原图损坏的文件会被每 10 分钟重试一次而永远不成功。 跨租户的候选查询全部集中在 ScheduleMapper 一个文件、一律 @InterceptorIgnore + 只读； Map 结果的别名一律 AS "camelCase"（F-83：PG 会把未加引号的别名折成小写， 表现是任务照跑、日志一行 warn、什么都没做）。 验证：e2e-m7-schedule 38/38（含快照幂等与一致性、重试累加与封顶、到期提醒窗口）。
This commit is contained in:
+53
@@ -83,10 +83,63 @@ public class CollectFile extends TenantEntity {
|
||||
|
||||
private String uploadIp;
|
||||
|
||||
/**
|
||||
* 材料来源。**口径以 DDL 的列注释为准**（PRD 第 1894 行）：
|
||||
* {@code 1} 本人上传 / {@code 2} 班主任代传 / {@code 3} 管理员代传 / {@code 4} 整理器导入。
|
||||
*
|
||||
* <p>⚠️ 接口设计 §3.6 的字段表把同一列写成
|
||||
* "1 聊天记录 / 2 相册 / 3 手机文件 / 4 代传" —— 与 DDL 冲突。
|
||||
* 以 DDL 为准的理由：它是"这一列表示什么"的定义，已有数据（全为 1）也与它一致，
|
||||
* 且 PC 端的 {@code FILE_SOURCE} 常量已经按 DDL 写了。差异记在 F-38。
|
||||
*
|
||||
* <p>⚠️ 这一列在 V12 之前**从来没有被代码显式写过** ——
|
||||
* 库里那 10 行 {@code source=1} 全部来自列默认值，
|
||||
* 也就是"学生自己传的"与"老师代传的"在数据上完全区分不出来。V12 起写入侧显式赋值。
|
||||
*/
|
||||
private Integer source;
|
||||
|
||||
/** 代传人 user_id（{@code source=2/3} 时有值）—— 审核时间轴据此显示「由某某老师代传」。 */
|
||||
private Long proxyBy;
|
||||
|
||||
/** 代传原因（选填，会挂到审核时间轴与材料列表上给学生看到）。 */
|
||||
private String proxyReason;
|
||||
|
||||
public static final int NEED_MANUAL_NO = 0;
|
||||
public static final int VALIDATE_PENDING = 0;
|
||||
public static final int CURRENT = 1;
|
||||
public static final int NOT_CURRENT = 0;
|
||||
|
||||
// ── 来源（契约 §10.6：代传必须留痕）────────────
|
||||
|
||||
/** 本人上传（学生自己在填报页传的）。 */
|
||||
public static final int SOURCE_SELF = 1;
|
||||
/** 班主任代传。 */
|
||||
public static final int SOURCE_PROXY_TEACHER = 2;
|
||||
/** 管理员代传。 */
|
||||
public static final int SOURCE_PROXY_ADMIN = 3;
|
||||
/** 历史整理器导入。 */
|
||||
public static final int SOURCE_CLEANUP_IMPORT = 4;
|
||||
|
||||
// ── 处理失败与重试（V17；PRD 附录 E「压缩失败重试」/ FR-611）────
|
||||
|
||||
/** 处理失败后的重试次数（上限 {@link #MAX_PROCESS_RETRY}）。 */
|
||||
private Integer processRetryCount;
|
||||
|
||||
/** 最近一次处理失败的原因（**要给人看**：只留原图而不知道为什么失败，运维只能靠猜）。 */
|
||||
private String processError;
|
||||
|
||||
private java.time.LocalDateTime lastRetryTime;
|
||||
|
||||
/**
|
||||
* 重试上限。
|
||||
*
|
||||
* <p>⚠️ 必须有上限：一个永远失败的图（原图本身损坏）如果不封顶，
|
||||
* 会被每 10 分钟重试一次，把磁盘 IO 与 CPU 白烧掉，而**它永远不会成功**。
|
||||
*/
|
||||
public static final int MAX_PROCESS_RETRY = 3;
|
||||
|
||||
/** 是不是"别人替他传的"（时间轴与材料列表据此显示代传说明）。 */
|
||||
public boolean isProxied() {
|
||||
return source != null && (source == SOURCE_PROXY_TEACHER || source == SOURCE_PROXY_ADMIN);
|
||||
}
|
||||
}
|
||||
|
||||
+54
@@ -5,6 +5,7 @@ import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
import org.apache.ibatis.annotations.Update;
|
||||
|
||||
public interface FileMapper extends BaseMapper<CollectFile> {
|
||||
|
||||
@@ -92,4 +93,57 @@ public interface FileMapper extends BaseMapper<CollectFile> {
|
||||
WHERE id = #{id} AND tenant_id = #{tenantId} AND deleted = 0
|
||||
""")
|
||||
CollectFile selectForAudit(@Param("id") Long id, @Param("tenantId") Long tenantId);
|
||||
|
||||
// ── 处理失败与重试（V17；PRD 附录 E「压缩失败重试」）──────────
|
||||
|
||||
/**
|
||||
* 处理失败、还可以重试的文件。
|
||||
*
|
||||
* <p>候选条件的三段各有理由：
|
||||
* <ul>
|
||||
* <li>{@code std_path IS NULL} —— **还没有规范件**。规范件是唯一产物，
|
||||
* 有它就说明处理成功过，不该再压一遍；</li>
|
||||
* <li>{@code raw_path IS NOT NULL} —— **还留着原始件**。没有原图就无从重试
|
||||
* （PRD FR-611 的"保留原图"正是为这一步）；</li>
|
||||
* <li>{@code process_retry_count < #{maxRetry}} —— **封顶**。
|
||||
* 原图本身损坏的文件重试一万次也不会成功，而每 10 分钟一次会白烧 CPU。</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>⚠️ 别名加双引号：返回 `Map`，未加引号的别名会被 PostgreSQL 折成小写（F-83）。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT id AS "id", tenant_id AS "tenantId", raw_path AS "rawPath",
|
||||
process_retry_count AS "processRetryCount"
|
||||
FROM collect_file
|
||||
WHERE deleted = 0
|
||||
AND is_current = 1
|
||||
AND (std_path IS NULL OR std_path = '')
|
||||
AND raw_path IS NOT NULL
|
||||
AND process_retry_count < #{maxRetry}
|
||||
ORDER BY last_retry_time NULLS FIRST, id
|
||||
LIMIT #{limit}
|
||||
""")
|
||||
java.util.List<java.util.Map<String, Object>> listProcessFailed(
|
||||
@Param("maxRetry") int maxRetry, @Param("limit") int limit);
|
||||
|
||||
/** 重试成功：清掉失败原因与计数（它已经好了，计数留着会误导）。 */
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Update("""
|
||||
UPDATE collect_file
|
||||
SET process_retry_count = 0, process_error = NULL, last_retry_time = NOW(),
|
||||
update_time = NOW()
|
||||
WHERE id = #{id}
|
||||
""")
|
||||
int markRetryOk(@Param("id") Long id);
|
||||
|
||||
/** 重试失败：累加计数并记下原因（PRD FR-611 的「标记失败原因」）。 */
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Update("""
|
||||
UPDATE collect_file
|
||||
SET process_retry_count = process_retry_count + 1,
|
||||
process_error = #{error}, last_retry_time = NOW(), update_time = NOW()
|
||||
WHERE id = #{id}
|
||||
""")
|
||||
int markRetryFailed(@Param("id") Long id, @Param("error") String error);
|
||||
}
|
||||
|
||||
+54
@@ -0,0 +1,54 @@
|
||||
package cn.tianbaotong.fill.module.schedule.controller;
|
||||
|
||||
import cn.tianbaotong.fill.common.api.ApiResult;
|
||||
import cn.tianbaotong.fill.module.schedule.service.ScheduleJobs;
|
||||
import cn.tianbaotong.fill.security.PermissionDef;
|
||||
import cn.tianbaotong.fill.security.RequirePerm;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 定时任务的**手工触发**入口（PRD 附录 E 的 9 项）。
|
||||
*
|
||||
* <h3>为什么必须有这个入口</h3>
|
||||
*
|
||||
* <p>定时任务的"等不起"是测试与运维的**共同**问题：验证"任务截止后会怎样"
|
||||
* 总不能在测试里等到明天凌晨（F-37 已经为导出包清理踩过一次，当时加的是
|
||||
* {@code POST /exports/cleanup}）。同一个理由、同一类做法，这里集中成一个入口。
|
||||
*
|
||||
* <p>运维也需要它：某天凌晨的清理没跑成（服务在重启），
|
||||
* 上午要知道"能不能手工补一次"而不是"等明天"。
|
||||
*
|
||||
* <h3>⚠️ 权限：它是**运维动作**，不是业务动作</h3>
|
||||
*
|
||||
* <p>用 `log:view:all`（运营侧排障权限）而不是任务/审核权限 ——
|
||||
* 手工触发会**跨租户**跑（平台身份），普通管理员不该能触发它。
|
||||
*/
|
||||
@Tag(name = "定时任务")
|
||||
@RestController
|
||||
@RequestMapping("/api/v1/schedules")
|
||||
@RequiredArgsConstructor
|
||||
public class ScheduleController {
|
||||
|
||||
private final ScheduleJobs jobs;
|
||||
|
||||
@PostMapping("/run")
|
||||
@RequirePerm(PermissionDef.LOG_VIEW_ALL)
|
||||
@Operation(summary = "手动触发一轮定时任务",
|
||||
description = "`job` 取 `close`（任务自动截止）/ `urge`（自动催报）/ "
|
||||
+ "`timeout`（审核超时提醒）/ `raw`（原始件清理）。"
|
||||
+ "⚠️ 它是**运维动作**：以平台身份跨租户跑，所以用运营侧的排障权限，"
|
||||
+ "而不是任务或审核权限")
|
||||
public ApiResult<Map<String, Object>> run(@RequestBody(required = false) Map<String, Object> body) {
|
||||
String job = body == null || body.get("job") == null ? null : String.valueOf(body.get("job"));
|
||||
String msg = jobs.runNow(job);
|
||||
return ApiResult.ok(Map.of("job", job == null ? "" : job.trim(), "message", msg));
|
||||
}
|
||||
}
|
||||
+202
@@ -0,0 +1,202 @@
|
||||
package cn.tianbaotong.fill.module.schedule.mapper;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
import org.apache.ibatis.annotations.Update;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 定时任务的**候选查询**（PRD 附录 E 的 9 项）。
|
||||
*
|
||||
* <h3>为什么这些查询要集中放、且全部 {@@code @InterceptorIgnore}</h3>
|
||||
*
|
||||
* <p>定时任务的处境与匿名接口一样：**线程里没有登录用户，租户拦截器会直接拒绝**
|
||||
* （{@code 40101}）。但它又必须**先看到所有租户的候选行**才知道要为哪些租户干活 ——
|
||||
* 这与"匿名接口只能看一个租户"正好相反。
|
||||
*
|
||||
* <p>所以这个类的规矩是（与 {@code module/mp/mapper/MpSceneMapper} 同一条）：
|
||||
* <ol>
|
||||
* <li>关掉拦截器；</li>
|
||||
* <li>**显式带 {@code tenant_id}** —— 要么是查询条件的一部分（跨租户扫描时不必带），
|
||||
* 要么把 {@code tenant_id} 作为结果列返回，让调用方拿着它去摆上下文；</li>
|
||||
* <li>**只做"找出候选"，不在这里改数据** —— 改数据一律回到各自模块的 Service，
|
||||
* 在正确的租户上下文里做。这样"关掉框架保护"的范围就被压到最小：只有 SELECT。</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>⚠️ 只读这件事是刻意的。一个能跨租户 UPDATE 的方法，写错一个条件就是全校数据被改。
|
||||
*
|
||||
* <h3>⚠️ 别名必须加双引号：{@code AS "tenantId"}</h3>
|
||||
*
|
||||
* <p>这些方法返回 {@code Map<String, Object>}，而 **PostgreSQL 会把未加引号的别名折成小写**
|
||||
* —— {@code AS tenantId} 在结果里是 {@code tenantid}，于是 {@code row.get("tenantId")}
|
||||
* 拿到 {@code null}。返回 POJO 的 mapper（如 {@code UrgeMapper}）不受影响
|
||||
* （MyBatis 按属性名忽略大小写匹配），**所以这个坑只在 Map 结果上出现**，
|
||||
* 很容易"照着别的 mapper 抄一遍"就中招。
|
||||
*
|
||||
* <p>它的表现极其安静：定时任务照跑，日志里只有一行 warn，**什么都没做**。
|
||||
* 这与 F-54 是同一个坑（当时在 {@code ValidationIssueMapper}），
|
||||
* 所以这里的规矩是：**这个文件里的别名一律带引号**。
|
||||
*/
|
||||
public interface ScheduleMapper {
|
||||
|
||||
/**
|
||||
* 已到截止时间但还在"收集中"的任务。
|
||||
*
|
||||
* <p>⚠️ 用 `end_time < now()` 而不是 `<=`：`end_time` 在库里是**日期**（当天 00:00），
|
||||
* 所以"今天截止"的任务不应该在凌晨就被关掉 —— 学生今天还能交。
|
||||
* 判定写成"截止日**已经过去**"（严格小于今天），语义是"从明天起不能再交"。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT id, tenant_id AS "tenantId", task_name AS "taskName", end_time AS "endTime"
|
||||
FROM collect_task
|
||||
WHERE deleted = 0
|
||||
AND status = 1
|
||||
AND end_time IS NOT NULL
|
||||
AND end_time < CURRENT_DATE
|
||||
ORDER BY id
|
||||
""")
|
||||
List<Map<String, Object>> tasksToClose();
|
||||
|
||||
/**
|
||||
* 截止日正好在某一天的任务（给"提前 7/3/1 天催报"用）。
|
||||
*
|
||||
* <p>为什么按"正好那一天"而不是"小于等于"：催报是**每天一次**的定时任务，
|
||||
* 用 `<=` 会让"还剩 2 天"的任务在 7 天、3 天、1 天各被催一次（因为 2 < 3 也成立）
|
||||
* —— 学生一天收到三条，然后关掉通知。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT id, tenant_id AS "tenantId", task_name AS "taskName", end_time AS "endTime"
|
||||
FROM collect_task
|
||||
WHERE deleted = 0
|
||||
AND status = 1
|
||||
AND end_time = #{date}::date
|
||||
ORDER BY id
|
||||
""")
|
||||
List<Map<String, Object>> tasksEndingOn(@Param("date") String date);
|
||||
|
||||
/**
|
||||
* 可以清掉原始件的行（`raw_expire_time` 到期）。
|
||||
*
|
||||
* <p>只取"确实有原始件"的（`raw_path` 非空）—— 契约默认 {@code keep-raw=false}，
|
||||
* 所以绝大多数行根本没有原始件，扫它们只是白跑。
|
||||
*
|
||||
* <p>带 {@code limit}：一次清一大批会把磁盘 IO 打满（而且删除不可回滚），
|
||||
* 分几轮清完更安全。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT id, tenant_id AS "tenantId", raw_path AS "rawPath"
|
||||
FROM collect_file
|
||||
WHERE deleted = 0
|
||||
AND raw_path IS NOT NULL
|
||||
AND raw_expire_time IS NOT NULL
|
||||
AND raw_expire_time < NOW()
|
||||
ORDER BY raw_expire_time
|
||||
LIMIT #{limit}
|
||||
""")
|
||||
List<Map<String, Object>> rawFilesToClean(@Param("limit") int limit);
|
||||
|
||||
/** 清掉一行的原始件引用（**只改列，不删行** —— 规范件还在，行必须留）。 */
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Update("""
|
||||
UPDATE collect_file
|
||||
SET raw_path = NULL, update_time = NOW()
|
||||
WHERE id = #{id} AND deleted = 0
|
||||
""")
|
||||
int clearRawPath(@Param("id") Long id);
|
||||
|
||||
/**
|
||||
* 超时未审的填报单（审核超时提醒）。
|
||||
*
|
||||
* <p>口径与 §11.6 的审核统计**必须一致**（班级 48 小时 / 后台 72 小时），
|
||||
* 否则会出现"统计里说超时 3 件、提醒只发了 1 条"这种对不上的情况。
|
||||
* 两个小时的阈值由调用方传进来，但**取的是同一份配置**（`collect_task.audit_config`）。
|
||||
*
|
||||
* <p>只取"当前卡在哪一级"对应的那条：`status=1` 是待初审（班级级），
|
||||
* `status=2` 是待终审（后台级）。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT s.id AS "submissionId",
|
||||
s.tenant_id AS "tenantId",
|
||||
s.task_id AS "taskId",
|
||||
s.status AS status,
|
||||
s.submitter_id AS "submitterId",
|
||||
s.submit_time AS "submitTime",
|
||||
t.task_name AS "taskName",
|
||||
r.real_name AS "realName"
|
||||
FROM collect_submission s
|
||||
JOIN collect_task t ON t.id = s.task_id AND t.deleted = 0
|
||||
LEFT JOIN sys_user u ON u.id = s.submitter_id AND u.deleted = 0
|
||||
LEFT JOIN collect_roster r
|
||||
ON r.task_id = s.task_id AND r.bind_user_id = s.submitter_id AND r.deleted = 0
|
||||
WHERE s.deleted = 0
|
||||
AND s.status IN (1, 2)
|
||||
AND s.submit_time IS NOT NULL
|
||||
AND s.submit_time < #{before}::timestamp
|
||||
ORDER BY s.submit_time
|
||||
LIMIT #{limit}
|
||||
""")
|
||||
List<Map<String, Object>> auditTimeouts(@Param("before") java.time.LocalDateTime before,
|
||||
@Param("limit") int limit);
|
||||
|
||||
/**
|
||||
* 某个任务下"该审批的人"（初审=班级负责人，终审=学校管理员）。
|
||||
*
|
||||
* <p>找人的口径沿用审核台的范围查询（`collect_task_org.head_user_id` /
|
||||
* `deputy_user_ids` / 组织的 `leader_user_id`）—— 提醒要发给**真的要动手的那个人**，
|
||||
* 发给所有人等于没人负责。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT DISTINCT u.id AS "userId", u.real_name AS "realName"
|
||||
FROM collect_task_org cto
|
||||
JOIN sys_user u ON u.deleted = 0 AND u.status = 1
|
||||
AND (u.id = cto.head_user_id
|
||||
OR cto.deputy_user_ids LIKE ('%,' || u.id || ',%'))
|
||||
WHERE cto.task_id = #{taskId} AND cto.status = 1 AND cto.tenant_id = #{tenantId}
|
||||
""")
|
||||
List<Map<String, Object>> classAuditors(@Param("tenantId") Long tenantId,
|
||||
@Param("taskId") Long taskId);
|
||||
|
||||
/**
|
||||
* 到期日正好在某一天的租户（给"提前 30/7/1 天提醒"用）。
|
||||
*
|
||||
* <p>与催报同理：按"正好那一天"而不是"小于等于" —— 用 `<=` 会连报三次，
|
||||
* 而客户看到的第一反应是"我知道了"，第二次是"又来了"，第三次是关掉通知。
|
||||
*
|
||||
* <p>只取启用中的租户：已经停用的不用再提醒续费（那已经是另一个话题）。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT id AS "tenantId", tenant_name AS "tenantName", expire_time AS "expireTime"
|
||||
FROM sys_tenant
|
||||
WHERE deleted = 0 AND status = 1
|
||||
AND expire_time IS NOT NULL
|
||||
AND expire_time::date = #{date}::date
|
||||
ORDER BY id
|
||||
""")
|
||||
List<Map<String, Object>> tenantsExpiringOn(@Param("date") String date);
|
||||
|
||||
/**
|
||||
* 学校管理员（终审超时提醒的收件人）。
|
||||
*
|
||||
* <p>按角色找而不是按某个固定账号 —— 学校换管理员时不该改代码。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT DISTINCT u.id AS "userId", u.real_name AS "realName"
|
||||
FROM sys_user u
|
||||
JOIN sys_user_role ur ON ur.user_id = u.id AND ur.deleted = 0
|
||||
JOIN sys_role ro ON ro.id = ur.role_id AND ro.deleted = 0
|
||||
WHERE u.tenant_id = #{tenantId} AND u.deleted = 0 AND u.status = 1
|
||||
AND ro.role_code = #{roleCode}
|
||||
""")
|
||||
List<Map<String, Object>> usersOfRole(@Param("tenantId") Long tenantId,
|
||||
@Param("roleCode") String roleCode);
|
||||
}
|
||||
+647
@@ -0,0 +1,647 @@
|
||||
package cn.tianbaotong.fill.module.schedule.service;
|
||||
|
||||
import cn.hutool.json.JSONUtil;
|
||||
import cn.tianbaotong.fill.common.api.ErrorCode;
|
||||
import cn.tianbaotong.fill.common.exception.BizException;
|
||||
import cn.tianbaotong.fill.module.notify.service.NotificationService;
|
||||
import cn.tianbaotong.fill.module.notify.service.UrgeService;
|
||||
import cn.tianbaotong.fill.module.schedule.mapper.ScheduleMapper;
|
||||
import cn.tianbaotong.fill.module.task.mapper.TaskMapper;
|
||||
import cn.tianbaotong.fill.module.task.entity.CollectTask;
|
||||
import cn.tianbaotong.fill.module.task.service.TaskService;
|
||||
import cn.tianbaotong.fill.security.BuiltinRole;
|
||||
import cn.tianbaotong.fill.support.TenantScope;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* 定时任务（PRD 附录 E 的 9 项）。
|
||||
*
|
||||
* <h3>这个类存在的意义：把"没人点的那部分系统"集中到一处</h3>
|
||||
*
|
||||
* <p>9 项里已经有的两件不在这个类里（各归各的模块）：
|
||||
* {@code ExportCleanupService.cleanupExpired}（导出包清理）与
|
||||
* {@code OpenPushService.retryDue}（开放接口推送重试）。剩下的在这里。
|
||||
*
|
||||
* <h3>⚠️ 定时任务最大的坑：租户上下文</h3>
|
||||
*
|
||||
* <p>定时线程里**没有登录用户**，而租户拦截器默认拒绝（{@code 40101}）。
|
||||
* 所以每个任务都是同一个两段式：
|
||||
*
|
||||
* <ol>
|
||||
* <li><b>先以平台身份扫候选</b>（{@code TenantScope.runAs(0L, ...)}）——
|
||||
* 平台租户是唯一允许跨租户的身份；候选行里带着 {@code tenantId}；</li>
|
||||
* <li><b>再逐个租户摆好上下文干活</b>（{@code TenantScope.runAs(候选.tenantId, ...)}）——
|
||||
* 写数据、发通知都必须在这一步，否则 {@code tenant_id} 会被写成 0
|
||||
* （通知写进平台租户，那个学校的人**永远看不到**，而且不报错）。</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>反过来（先摆租户上下文再扫）就只能扫到一个租户 —— 而"漏掉某个学校的催报"
|
||||
* 没有任何地方会报错。
|
||||
*
|
||||
* <h3>为什么每个任务都 try/catch 住整个身体</h3>
|
||||
*
|
||||
* <p>{@code @Scheduled} 抛异常只会打一行日志，下一轮照跑；但如果异常发生在
|
||||
* "遍历到第 3 个租户"时，**前面 2 个已经干了、后面 7 个这轮不干了**。
|
||||
* 所以这里的取舍是：**一个租户失败不影响其它租户**（循环体内各自 try），
|
||||
* 整体再包一层（避免把调度线程搞挂）。
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class ScheduleJobs {
|
||||
|
||||
private final ScheduleMapper scheduleMapper;
|
||||
private final TaskService taskService;
|
||||
private final UrgeService urgeService;
|
||||
private final NotificationService notificationService;
|
||||
private final TaskMapper taskMapper;
|
||||
private final cn.tianbaotong.fill.module.tenant.mapper.SnapshotMapper snapshotMapper;
|
||||
private final cn.tianbaotong.fill.module.tenant.service.TenantUsageService usageService;
|
||||
private final cn.tianbaotong.fill.module.file.service.MaterialFileService fileService;
|
||||
private final cn.tianbaotong.fill.module.file.mapper.FileMapper fileMapper;
|
||||
|
||||
/** 原始件一次最多清多少行（分轮清完，别一次把磁盘 IO 打满）。 */
|
||||
private static final int RAW_CLEAN_BATCH = 200;
|
||||
|
||||
/** 审核超时提醒一轮最多处理多少条（防止某天积压上万条时把接口打爆）。 */
|
||||
private static final int TIMEOUT_BATCH = 500;
|
||||
|
||||
/** 提前几天催报（PRD 附录 E 的「7/3/1 天规则」）。 */
|
||||
private static final int[] URGE_DAYS_BEFORE = {7, 3, 1};
|
||||
|
||||
/** 租户到期前提醒的天数（PRD 附录 E 的「30/7/1 天」）。 */
|
||||
private static final int[] EXPIRE_REMIND_DAYS = {30, 7, 1};
|
||||
|
||||
/** 一轮最多重试多少个文件（分轮清完，别一次把 CPU 打满）。 */
|
||||
private static final int RETRY_BATCH = 50;
|
||||
|
||||
/** 平台身份（跨租户扫描时用）。 */
|
||||
private static final long PLATFORM = 0L;
|
||||
|
||||
// ══════════════ ① 任务自动截止（每小时）══════════════
|
||||
|
||||
/**
|
||||
* 到期的任务自动 {@code 收集中 → 已截止}。
|
||||
*
|
||||
* <p>⚠️ 这件事**必须自动做**：截止时间到了而任务还是"收集中"，学生就能继续提交
|
||||
* —— 而"截止后提交"会在导出与上报时变成一个说不清的问题（这批到底算不算）。
|
||||
* 靠管理员记着手动点"截止"，一定会漏。
|
||||
*/
|
||||
@Scheduled(cron = "0 5 * * * *")
|
||||
public void autoCloseExpiredTasks() {
|
||||
try {
|
||||
List<Map<String, Object>> tasks = TenantScope.callAs(PLATFORM, PLATFORM, Set.of(),
|
||||
scheduleMapper::tasksToClose);
|
||||
assertTenantPresent(tasks, "close");
|
||||
int closed = 0;
|
||||
for (Map<String, Object> row : tasks) {
|
||||
Long tenantId = longOf(row, "tenantId");
|
||||
Long taskId = longOf(row, "id");
|
||||
try {
|
||||
TenantScope.runAs(tenantId, PLATFORM, Set.of(SYSTEM_ROLE),
|
||||
() -> taskService.autoClose(taskId));
|
||||
closed++;
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("自动截止失败：tenant={} task={} 原因={}", tenantId, taskId, e.toString());
|
||||
}
|
||||
}
|
||||
if (closed > 0) {
|
||||
log.info("自动截止：扫到 {} 个到期任务，关闭 {} 个", tasks.size(), closed);
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("自动截止任务失败（下一轮会重试）", e);
|
||||
}
|
||||
}
|
||||
|
||||
// ══════════════ ② 自动催报（每日 09:00）══════════════
|
||||
|
||||
/**
|
||||
* 按截止日 7/3/1 天规则自动催报**未提交者**。
|
||||
*
|
||||
* <p>为什么是 09:00：催报要落在学生**当天会看手机**的时间。凌晨发的话，
|
||||
* 通知会沉在消息列表底下，等于没发。
|
||||
*
|
||||
* <p>为什么只催"还没提交"：已经交了的人收到"请尽快完成填报"会以为自己没交成功
|
||||
* —— 这比不催更糟。
|
||||
*/
|
||||
@Scheduled(cron = "0 0 9 * * *")
|
||||
public void autoUrgeUnsubmitted() {
|
||||
try {
|
||||
LocalDate today = LocalDate.now();
|
||||
for (int days : URGE_DAYS_BEFORE) {
|
||||
LocalDate target = today.plusDays(days);
|
||||
List<Map<String, Object>> tasks = TenantScope.callAs(PLATFORM, PLATFORM, Set.of(),
|
||||
() -> scheduleMapper.tasksEndingOn(target.toString()));
|
||||
assertTenantPresent(tasks, "urge");
|
||||
int sent = 0;
|
||||
for (Map<String, Object> row : tasks) {
|
||||
Long tenantId = longOf(row, "tenantId");
|
||||
Long taskId = longOf(row, "id");
|
||||
try {
|
||||
sent += TenantScope.callAs(tenantId, PLATFORM, Set.of(SYSTEM_ROLE),
|
||||
() -> urgeService.autoUrge(tenantId, taskId, days));
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("自动催报失败：tenant={} task={} 原因={}", tenantId, taskId, e.toString());
|
||||
}
|
||||
}
|
||||
if (!tasks.isEmpty()) {
|
||||
log.info("自动催报：截止日 {}（还有 {} 天）有 {} 个任务，发出 {} 条",
|
||||
target, days, tasks.size(), sent);
|
||||
}
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("自动催报任务失败（下一轮会重试）", e);
|
||||
}
|
||||
}
|
||||
|
||||
// ══════════════ ③ 审核超时提醒（每小时）══════════════
|
||||
|
||||
/**
|
||||
* 超过时效还没审的，提醒**该动手的那个人**。
|
||||
*
|
||||
* <p>时效口径与 §11.6 的审核统计一致（班级 48h / 后台 72h）。这里用**默认值**，
|
||||
* 任务自己配了 `audit_config` 时以任务为准 —— 两者不一致会让
|
||||
* "统计里说超时 3 件、提醒只发了 1 条"，而那种不一致没人查得出来。
|
||||
*
|
||||
* <p>提醒发给谁：
|
||||
* <ul>
|
||||
* <li>{@code status=1}（待初审）→ 这个班的班主任 / 副班主任；</li>
|
||||
* <li>{@code status=2}（待终审）→ 学校管理员。</li>
|
||||
* </ul>
|
||||
* 发给"真的要动手的那个人"，发给所有人等于没人负责。
|
||||
*/
|
||||
@Scheduled(cron = "0 25 * * * *")
|
||||
public void remindAuditTimeout() {
|
||||
try {
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
// 取两个阈值里**更宽松**的那个作为扫描下界：具体每个任务用哪个阈值，
|
||||
// 在下面按 status（初审/终审）分别判 —— 先粗筛再精筛，避免全表扫
|
||||
int maxHours = Math.max(DEFAULT_CLASS_TIMEOUT_HOURS, DEFAULT_ADMIN_TIMEOUT_HOURS);
|
||||
LocalDateTime before = now.minusHours(maxHours);
|
||||
List<Map<String, Object>> rows = TenantScope.callAs(PLATFORM, PLATFORM, Set.of(),
|
||||
() -> scheduleMapper.auditTimeouts(before, TIMEOUT_BATCH));
|
||||
assertTenantPresent(rows, "timeout");
|
||||
int notified = 0;
|
||||
for (Map<String, Object> row : rows) {
|
||||
Long tenantId = longOf(row, "tenantId");
|
||||
Long taskId = longOf(row, "taskId");
|
||||
Integer status = intOf(row, "status");
|
||||
LocalDateTime submitTime = timeOf(row, "submitTime");
|
||||
if (tenantId == null || taskId == null || status == null || submitTime == null) {
|
||||
continue;
|
||||
}
|
||||
int limitHours = status == 1 ? classTimeoutHours(tenantId, taskId)
|
||||
: adminTimeoutHours(tenantId, taskId);
|
||||
long waited = java.time.Duration.between(submitTime, now).toHours();
|
||||
if (waited < limitHours) {
|
||||
continue; // 还没到它那一级的时效
|
||||
}
|
||||
try {
|
||||
notified += TenantScope.callAs(tenantId, PLATFORM, Set.of(SYSTEM_ROLE),
|
||||
() -> remindOne(tenantId, row, status, waited, limitHours));
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("审核超时提醒失败：submission={} 原因={}", row.get("submissionId"),
|
||||
e.toString());
|
||||
}
|
||||
}
|
||||
if (notified > 0) {
|
||||
log.info("审核超时提醒：扫到 {} 条超时，发出 {} 条提醒", rows.size(), notified);
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("审核超时提醒任务失败（下一轮会重试）", e);
|
||||
}
|
||||
}
|
||||
|
||||
private int remindOne(Long tenantId, Map<String, Object> row, Integer status,
|
||||
long waited, int limitHours) {
|
||||
Long taskId = longOf(row, "taskId");
|
||||
String taskName = strOf(row, "taskName");
|
||||
String realName = strOf(row, "realName");
|
||||
String who = realName == null ? "有 1 份填报单" : realName + " 的填报单";
|
||||
String levelText = status == 1 ? "班级初审" : "后台终审";
|
||||
List<Map<String, Object>> receivers = status == 1
|
||||
? scheduleMapper.classAuditors(tenantId, taskId)
|
||||
: scheduleMapper.usersOfRole(tenantId, BuiltinRole.SCHOOL_ADMIN.getCode());
|
||||
boolean escalated = false;
|
||||
if (receivers.isEmpty()) {
|
||||
// ⚠️ 「找不到班主任」不是"跳过这一条"，而是**必须升级**。
|
||||
//
|
||||
// 那些单子不是"晚点会有人审"，而是**永远不会有人审**（没人负责这个班），
|
||||
// 而学生那边只会看到"一直没通过"。所以按 PRD 的原话「提醒并升级」：
|
||||
// 找不到班级负责人时，升级给学校管理员，并在文案里说清原因 ——
|
||||
// 管理员看到才知道要去配班主任。
|
||||
receivers = scheduleMapper.usersOfRole(tenantId, BuiltinRole.SCHOOL_ADMIN.getCode());
|
||||
escalated = !receivers.isEmpty();
|
||||
if (!escalated) {
|
||||
// 连管理员都没有 → 这个租户的数据不完整，留痕（这是配置问题，不是提醒问题）
|
||||
log.warn("审核超时但找不到任何提醒对象：tenant={} task={}（没有管理员？）",
|
||||
tenantId, taskId);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
String title = escalated ? "有填报单审核超时（该班没有负责人）" : "有填报单审核超时";
|
||||
String content = who + "在《" + nz(taskName) + "》里等你" + levelText
|
||||
+ "，已经等了 " + waited + " 小时（时效 " + limitHours + " 小时）。"
|
||||
+ (escalated ? "⚠️ 这个班没有配负责人，请先指定班主任/初审人。" : "");
|
||||
int n = 0;
|
||||
for (Map<String, Object> u : receivers) {
|
||||
Long uid = longOf(u, "userId");
|
||||
if (uid == null) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
notificationService.notify(new NotificationService.NotifyRequest(
|
||||
tenantId, uid, // 契约 §6 的 notifyBizType 里没有"审核超时"，用 2「临期提醒」——
|
||||
// 对审核员来说"这份单子等你 48 小时了"就是一次临期提醒。
|
||||
// **不新造第 8 个码**：契约里没有的码前端与客服都不认识
|
||||
cn.tianbaotong.fill.module.notify.entity.CollectMessage.BIZ_DEADLINE,
|
||||
title, content,
|
||||
cn.tianbaotong.fill.module.notify.entity.CollectMessage.LINK_TASK, taskId,
|
||||
NotificationService.TPL_AUDIT_RESULT, Map.of()));
|
||||
n++;
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("审核超时提醒发送失败：uid={} 原因={}", uid, e.toString());
|
||||
}
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
// ══════════════ ④ 原始件清理（每日 02:00）══════════════
|
||||
|
||||
/**
|
||||
* 清掉 `raw_expire_time` 到期的**原始件**（默认保留 30 天）。
|
||||
*
|
||||
* <p>⚠️ 语义是"**清掉原始件、保留规范件**"，不是删文件记录：
|
||||
* 规范件才是交付物，行必须留着（学生要能看到自己传过什么）。
|
||||
*
|
||||
* <p>先改库（把 `raw_path` 置空）再删磁盘文件 —— 与导出包清理同一个顺序理由：
|
||||
* 最坏结果是"文件还在但库里不再引用它"（浪费磁盘），
|
||||
* 而不是"库里有引用但文件没了"（用户点开报错）。
|
||||
*/
|
||||
@Scheduled(cron = "0 0 2 * * *")
|
||||
public void cleanExpiredRawFiles() {
|
||||
try {
|
||||
List<Map<String, Object>> rows = TenantScope.callAs(PLATFORM, PLATFORM, Set.of(),
|
||||
() -> scheduleMapper.rawFilesToClean(RAW_CLEAN_BATCH));
|
||||
assertTenantPresent(rows, "raw");
|
||||
int cleaned = 0;
|
||||
for (Map<String, Object> row : rows) {
|
||||
Long id = longOf(row, "id");
|
||||
String rawPath = strOf(row, "rawPath");
|
||||
if (id == null) {
|
||||
continue;
|
||||
}
|
||||
scheduleMapper.clearRawPath(id);
|
||||
if (rawPath != null) {
|
||||
try {
|
||||
Path p = Paths.get(rawPath);
|
||||
if (Files.deleteIfExists(p)) {
|
||||
cleaned++;
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.warn("删除原始件失败（引用已清，下轮不再重试）：{} 原因={}",
|
||||
rawPath, e.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!rows.isEmpty()) {
|
||||
log.info("原始件清理：清掉 {} 行的引用，删除 {} 个文件", rows.size(), cleaned);
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("原始件清理任务失败（下一轮会重试）", e);
|
||||
}
|
||||
}
|
||||
|
||||
// ══════════════ ⑤ 用量快照（每日 01:00）══════════════
|
||||
|
||||
/**
|
||||
* 把每个租户的用量写一行快照（PRD 附录 E「用量统计：每日 01:00」）。
|
||||
*
|
||||
* <p>为什么要快照而不是每次实时算 —— 见 `sys_tenant_usage_snapshot` 的表注释。
|
||||
* 一句话：实时统计只能回答"现在多少"，而计费问的是"**上个月**用了多少"。
|
||||
*
|
||||
* <p>⚠️ 幂等：一个租户一个周期一行（唯一索引 + 先查后写）。
|
||||
* 定时任务重复跑、或运维手工补跑，都不该造出两行 —— 两行的话
|
||||
* "这个月用了多少"就有两个答案，而报表会随机取其中一个。
|
||||
*/
|
||||
@Scheduled(cron = "0 0 1 * * *")
|
||||
public void snapshotUsage() {
|
||||
try {
|
||||
List<Map<String, Object>> tenants = TenantScope.callAs(PLATFORM, PLATFORM, Set.of(),
|
||||
snapshotMapper::activeTenants);
|
||||
assertTenantPresent(tenants, "usage");
|
||||
int written = 0;
|
||||
for (Map<String, Object> row : tenants) {
|
||||
Long tenantId = longOf(row, "tenantId");
|
||||
if (tenantId == null) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
written += TenantScope.callAs(tenantId, PLATFORM, Set.of(SYSTEM_ROLE),
|
||||
() -> snapshotOne(tenantId));
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("用量快照失败：tenant={} 原因={}", tenantId, e.toString());
|
||||
}
|
||||
}
|
||||
if (written > 0) {
|
||||
log.info("用量快照：{} 个租户，写入 {} 行", tenants.size(), written);
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("用量快照任务失败（下一轮会重试）", e);
|
||||
}
|
||||
}
|
||||
|
||||
/** 一个租户的快照（也供手工触发与测试调用）。返回写入行数（0 或 1）。 */
|
||||
public int snapshotOne(Long tenantId) {
|
||||
String period = java.time.LocalDate.now().format(
|
||||
java.time.format.DateTimeFormatter.ofPattern("yyyy-MM"));
|
||||
if (snapshotMapper.findByPeriod(tenantId, period) != null) {
|
||||
return 0; // 这个周期已经有快照了，不覆盖
|
||||
}
|
||||
cn.tianbaotong.fill.module.tenant.service.TenantUsageService.UsageVO u =
|
||||
usageService.usage(tenantId, period);
|
||||
cn.tianbaotong.fill.module.tenant.entity.SysTenantUsageSnapshot s =
|
||||
new cn.tianbaotong.fill.module.tenant.entity.SysTenantUsageSnapshot();
|
||||
// ⚠️ 必须显式写 tenantId：平台上下文里拦截器不生效，不会自动填
|
||||
s.setTenantId(tenantId);
|
||||
s.setPeriod(period);
|
||||
s.setStudentCount(u.getStudentCount());
|
||||
s.setStorageMb(u.getStorageMb());
|
||||
s.setOcrCount(u.getOcrCount());
|
||||
s.setSmsCount(u.getSmsCount());
|
||||
snapshotMapper.insert(s);
|
||||
return 1;
|
||||
}
|
||||
|
||||
// ══════════════ ⑥ 压缩失败重试（每 10 分钟）══════════════
|
||||
|
||||
/**
|
||||
* 重试"处理失败"的材料（PRD 附录 E「压缩失败重试：重试 ≤3 次」）。
|
||||
*
|
||||
* <p>候选是**还没有规范件、但保留着原始件**的行 ——
|
||||
* 这正是 PRD 那句「处理失败 → 保留原图，允许一键重试」对应的状态。
|
||||
*
|
||||
* <p>⚠️ 两条边界：
|
||||
* <ol>
|
||||
* <li><b>必须封顶 3 次</b>：原图本身损坏的文件重试一万次也不会成功，
|
||||
* 而每 10 分钟一次的重试会白烧磁盘 IO 与 CPU；</li>
|
||||
* <li><b>没有原始件的行不重试</b>：规范件是唯一的产物，原图又没了 ——
|
||||
* 重试无从下手。这类行只记一次日志（它们需要人工重新上传，
|
||||
* 而不是系统反复尝试）。</li>
|
||||
* </ol>
|
||||
*/
|
||||
@Scheduled(cron = "0 */10 * * * *")
|
||||
public void retryFailedProcess() {
|
||||
try {
|
||||
List<Map<String, Object>> rows = TenantScope.callAs(PLATFORM, PLATFORM, Set.of(),
|
||||
() -> fileMapper.listProcessFailed(cn.tianbaotong.fill.module.file.entity
|
||||
.CollectFile.MAX_PROCESS_RETRY, RETRY_BATCH));
|
||||
assertTenantPresent(rows, "retry");
|
||||
int ok = 0;
|
||||
int failed = 0;
|
||||
for (Map<String, Object> row : rows) {
|
||||
Long tenantId = longOf(row, "tenantId");
|
||||
Long fileId = longOf(row, "id");
|
||||
if (tenantId == null || fileId == null) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
TenantScope.runAs(tenantId, PLATFORM, Set.of(SYSTEM_ROLE),
|
||||
() -> fileService.process(fileId));
|
||||
fileMapper.markRetryOk(fileId);
|
||||
ok++;
|
||||
} catch (RuntimeException e) {
|
||||
failed++;
|
||||
fileMapper.markRetryFailed(fileId, clip(e.getMessage()));
|
||||
log.warn("材料处理重试失败：file={} 第 {} 次 原因={}", fileId,
|
||||
longOf(row, "processRetryCount"), e.toString());
|
||||
}
|
||||
}
|
||||
if (ok + failed > 0) {
|
||||
log.info("材料处理重试：成功 {}，失败 {}（失败会累加计数，到 {} 次停止）",
|
||||
ok, failed, cn.tianbaotong.fill.module.file.entity.CollectFile.MAX_PROCESS_RETRY);
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("材料处理重试任务失败（下一轮会重试）", e);
|
||||
}
|
||||
}
|
||||
|
||||
// ══════════════ ⑦ 租户到期提醒（每日 08:00）══════════════
|
||||
|
||||
/**
|
||||
* 租户到期前 30/7/1 天提醒（PRD 附录 E「租户到期提醒：每日 08:00」）。
|
||||
*
|
||||
* <p>⚠️ 提醒发给**这个租户自己的管理员**，不是发给我们 ——
|
||||
* 因为"要不要续费、走什么流程"是客户内部要安排的事，
|
||||
* 提前 30 天告诉他们才有时间走流程；提前 1 天说等于没说。
|
||||
*
|
||||
* <p>平台运营侧另外有一条（同一批数据，语气不同）：
|
||||
* 运营要的是"这家该催了"，而不是"你们快到期了"。
|
||||
*/
|
||||
@Scheduled(cron = "0 0 8 * * *")
|
||||
public void remindTenantExpiring() {
|
||||
try {
|
||||
int sent = 0;
|
||||
for (int days : EXPIRE_REMIND_DAYS) {
|
||||
java.time.LocalDate target = java.time.LocalDate.now().plusDays(days);
|
||||
List<Map<String, Object>> rows = TenantScope.callAs(PLATFORM, PLATFORM, Set.of(),
|
||||
() -> scheduleMapper.tenantsExpiringOn(target.toString()));
|
||||
assertTenantPresent(rows, "expire");
|
||||
for (Map<String, Object> row : rows) {
|
||||
Long tenantId = longOf(row, "tenantId");
|
||||
String name = strOf(row, "tenantName");
|
||||
if (tenantId == null) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
sent += TenantScope.callAs(tenantId, PLATFORM, Set.of(SYSTEM_ROLE),
|
||||
() -> remindExpire(tenantId, name, days));
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("租户到期提醒失败：tenant={} 原因={}", tenantId, e.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
if (sent > 0) {
|
||||
log.info("租户到期提醒：发出 {} 条", sent);
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("租户到期提醒任务失败（下一轮会重试）", e);
|
||||
}
|
||||
}
|
||||
|
||||
private int remindExpire(Long tenantId, String tenantName, int daysLeft) {
|
||||
var receivers = scheduleMapper.usersOfRole(tenantId,
|
||||
cn.tianbaotong.fill.security.BuiltinRole.SCHOOL_ADMIN.getCode());
|
||||
if (receivers.isEmpty()) {
|
||||
log.warn("租户到期提醒但该租户没有管理员：tenant={}", tenantId);
|
||||
return 0;
|
||||
}
|
||||
String title = daysLeft <= 1 ? "你的单位账号明天到期" : "你的单位账号还有 " + daysLeft + " 天到期";
|
||||
String content = "「" + nz(tenantName) + "」的服务将在 " + daysLeft
|
||||
+ " 天后到期。到期后学生与老师将无法登录、无法继续填报。"
|
||||
+ "请及时联系服务方安排续费 —— 已提交的数据不会丢失，但到期期间不能收新材料。";
|
||||
int n = 0;
|
||||
for (Map<String, Object> u : receivers) {
|
||||
Long uid = longOf(u, "userId");
|
||||
if (uid == null) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
notificationService.notify(new NotificationService.NotifyRequest(
|
||||
tenantId, uid,
|
||||
cn.tianbaotong.fill.module.notify.entity.CollectMessage.BIZ_DEADLINE,
|
||||
title, content,
|
||||
cn.tianbaotong.fill.module.notify.entity.CollectMessage.LINK_NONE, null,
|
||||
null, Map.of()));
|
||||
n++;
|
||||
} catch (RuntimeException e) {
|
||||
log.warn("租户到期提醒发送失败：uid={} 原因={}", uid, e.toString());
|
||||
}
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
// ══════════════ 供手工触发 / E2E 调用 ══════════════
|
||||
|
||||
/**
|
||||
* 立刻跑一轮（给 {@code POST /schedules/run} 用）。
|
||||
*
|
||||
* <p>为什么要这个入口：**定时任务的"等不起"是测试与运维的共同问题** ——
|
||||
* 验证"截止后会怎样"总不能在测试里等到明天 09:00（F-37 已经踩过一次）。
|
||||
*
|
||||
* @param job 任务名（{@code close} / {@code urge} / {@code timeout} / {@code raw}）
|
||||
* @return 这一轮做了什么（给人看的说明）
|
||||
*/
|
||||
public String runNow(String job) {
|
||||
if (job == null || job.isBlank()) {
|
||||
throw new BizException(ErrorCode.VALIDATION_BLOCKED,
|
||||
"请指定要跑的任务：close / urge / timeout / raw / usage / retry / expire");
|
||||
}
|
||||
switch (job.trim().toLowerCase()) {
|
||||
case "close" -> autoCloseExpiredTasks();
|
||||
case "urge" -> autoUrgeUnsubmitted();
|
||||
case "timeout" -> remindAuditTimeout();
|
||||
case "raw" -> cleanExpiredRawFiles();
|
||||
case "usage" -> snapshotUsage();
|
||||
case "retry" -> retryFailedProcess();
|
||||
case "expire" -> remindTenantExpiring();
|
||||
default -> throw new BizException(ErrorCode.VALIDATION_BLOCKED,
|
||||
"不认识的任务名「" + job + "」：只有 close / urge / timeout / raw / usage / retry / expire");
|
||||
}
|
||||
return "已手动触发一轮：" + job;
|
||||
}
|
||||
|
||||
// ══════════════ 内部工具 ══════════════
|
||||
|
||||
/** 系统身份的角色码（定时任务的写操作要能过权限校验）。 */
|
||||
private static final String SYSTEM_ROLE = BuiltinRole.PLATFORM_ADMIN.getCode();
|
||||
|
||||
/** 默认时效（小时），与 §11.6 的审核统计同口径。 */
|
||||
private static final int DEFAULT_CLASS_TIMEOUT_HOURS = 48;
|
||||
private static final int DEFAULT_ADMIN_TIMEOUT_HOURS = 72;
|
||||
|
||||
private int classTimeoutHours(Long tenantId, Long taskId) {
|
||||
return timeoutFromConfig(tenantId, taskId, "classTimeoutHours", DEFAULT_CLASS_TIMEOUT_HOURS);
|
||||
}
|
||||
|
||||
private int adminTimeoutHours(Long tenantId, Long taskId) {
|
||||
return timeoutFromConfig(tenantId, taskId, "adminTimeoutHours", DEFAULT_ADMIN_TIMEOUT_HOURS);
|
||||
}
|
||||
|
||||
/**
|
||||
* 从任务的 {@code audit_config} 里取时效；没有就回默认值。
|
||||
*
|
||||
* <p>⚠️ 读的是**任务自己配的那个值** —— 统计接口也是这么算的。
|
||||
* 两边各用一套默认值，就会出现"统计说超时 3 件、提醒只发了 1 条"。
|
||||
*/
|
||||
private int timeoutFromConfig(Long tenantId, Long taskId, String key, int fallback) {
|
||||
try {
|
||||
CollectTask t = taskMapper.selectById(taskId);
|
||||
if (t == null || t.getAuditConfig() == null || t.getAuditConfig().isBlank()) {
|
||||
return fallback;
|
||||
}
|
||||
Object v = JSONUtil.parseObj(t.getAuditConfig()).get(key);
|
||||
return v instanceof Number n ? n.intValue() : fallback;
|
||||
} catch (RuntimeException e) {
|
||||
return fallback;
|
||||
}
|
||||
}
|
||||
|
||||
private static Long longOf(Map<String, Object> row, String key) {
|
||||
Object v = row.get(key);
|
||||
return v instanceof Number n ? n.longValue() : null;
|
||||
}
|
||||
|
||||
private static Integer intOf(Map<String, Object> row, String key) {
|
||||
Object v = row.get(key);
|
||||
return v instanceof Number n ? n.intValue() : null;
|
||||
}
|
||||
|
||||
private static String strOf(Map<String, Object> row, String key) {
|
||||
Object v = row.get(key);
|
||||
return v == null ? null : String.valueOf(v);
|
||||
}
|
||||
|
||||
private static LocalDateTime timeOf(Map<String, Object> row, String key) {
|
||||
Object v = row.get(key);
|
||||
if (v instanceof LocalDateTime t) {
|
||||
return t;
|
||||
}
|
||||
if (v instanceof java.sql.Timestamp ts) {
|
||||
return ts.toLocalDateTime();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private static String nz(String s) {
|
||||
return s == null ? "本次任务" : s;
|
||||
}
|
||||
|
||||
/** 失败原因入库前截断（列是 varchar(256)，异常消息可能很长）。 */
|
||||
private static String clip(String s) {
|
||||
if (s == null) {
|
||||
return "处理失败（异常没有消息）";
|
||||
}
|
||||
return s.length() <= 250 ? s : s.substring(0, 250);
|
||||
}
|
||||
|
||||
/**
|
||||
* 候选行自检：缺 `tenantId` 就明确报出来。
|
||||
*
|
||||
* <p>这个检查的存在理由是一次真实的失败：`AS tenantId`（没加双引号）会被
|
||||
* PostgreSQL 折成小写，于是 `row.get("tenantId")` 是 null →
|
||||
* `TenantScope.runAs(null, ...)` 抛 → 被 catch 吞掉 →
|
||||
* **定时任务看起来跑过了，其实什么都没做**，日志里只有一行 warn。
|
||||
*
|
||||
* <p>所以宁可在这里吵一次：把候选行里**实际的键**打出来，
|
||||
* 让人一眼看到"哦，是 tenantid 不是 tenantId"。
|
||||
*/
|
||||
private void assertTenantPresent(List<Map<String, Object>> rows, String job) {
|
||||
if (rows.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
if (longOf(rows.get(0), "tenantId") == null) {
|
||||
throw new BizException(ErrorCode.INTERNAL_ERROR,
|
||||
"定时任务「" + job + "」的候选行里没有 tenantId（实际键："
|
||||
+ rows.get(0).keySet() + "）。"
|
||||
+ "最常见的原因：SQL 别名没加双引号，被 PostgreSQL 折成了小写");
|
||||
}
|
||||
}
|
||||
|
||||
/** 供测试断言用：这一轮"应该"处理哪些任务（不改数据）。 */
|
||||
public List<Map<String, Object>> previewClose() {
|
||||
return TenantScope.callAs(PLATFORM, PLATFORM, Set.of(), scheduleMapper::tasksToClose);
|
||||
}
|
||||
}
|
||||
+61
@@ -0,0 +1,61 @@
|
||||
package cn.tianbaotong.fill.module.tenant.entity;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Getter;
|
||||
import lombok.Setter;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* 租户用量快照（V17；PRD 附录 E「用量统计：每日 01:00」）。
|
||||
*
|
||||
* <h3>为什么必须落库，而不是每次实时算</h3>
|
||||
*
|
||||
* <ol>
|
||||
* <li>实时统计只能回答"**现在**多少"，而计费与超额告警问的是"**上个月**用了多少"；</li>
|
||||
* <li>历史快照能回答"用量是什么时候涨上去的" —— 招生季与毕业季的形状完全不同；</li>
|
||||
* <li>存储量尤其重要：原始件 30 天后被清理，**清理前的占用在实时统计里再也看不到**。</li>
|
||||
* </ol>
|
||||
*
|
||||
* <h3>⚠️ 它不继承 {@code BaseEntity} / {@code TenantEntity}</h3>
|
||||
*
|
||||
* <p>那张表**没有** {@code create_by / update_by / update_time / deleted} ——
|
||||
* 快照是"只增不改的事实"，不需要软删，也不需要"谁改的"。
|
||||
* 继承基类会让 MyBatis-Plus 去写不存在的列，表现是每次 insert 都报
|
||||
* {@code column "update_time" does not exist}。
|
||||
* （这类不一致由 {@code scripts/check-entity-columns.py} 拦。）
|
||||
*
|
||||
* <p>⚠️ 另一个坑：{@code tenant_id} 在基类之外**不会**被自动填 ——
|
||||
* 定时任务是平台上下文（拦截器对平台身份不生效），所以写的时候
|
||||
* **必须显式 setTenantId**，否则所有租户的快照会挤到 tenant_id=0 上，
|
||||
* 而那看起来像"没有数据"。
|
||||
*/
|
||||
@Getter
|
||||
@Setter
|
||||
@TableName("sys_tenant_usage_snapshot")
|
||||
public class SysTenantUsageSnapshot implements Serializable {
|
||||
|
||||
@TableId(type = IdType.AUTO)
|
||||
private Long id;
|
||||
|
||||
private Long tenantId;
|
||||
|
||||
/** 统计周期：`2026-09`（月）或 `2026Q3`（季），与 §6.7 的 period 同口径。 */
|
||||
private String period;
|
||||
|
||||
/** 按证件号去重的学生数。 */
|
||||
private Long studentCount;
|
||||
|
||||
/** 规范件体积合计（MB）。**不含原始件**。 */
|
||||
private Long storageMb;
|
||||
|
||||
/** 识别通道用量（一期恒为 0：M4 未接）。 */
|
||||
private Integer ocrCount;
|
||||
|
||||
private Integer smsCount;
|
||||
|
||||
private LocalDateTime createTime;
|
||||
}
|
||||
+55
@@ -0,0 +1,55 @@
|
||||
package cn.tianbaotong.fill.module.tenant.mapper;
|
||||
|
||||
import cn.tianbaotong.fill.module.tenant.entity.SysTenantUsageSnapshot;
|
||||
import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 租户用量快照（V17）。
|
||||
*
|
||||
* <p>与 {@code ScheduleMapper} 同一条规矩：定时任务在**平台上下文**里先看到所有租户，
|
||||
* 所以这里的跨租户查询显式关掉租户拦截器，并把 {@code tenant_id} 作为结果列带出来。
|
||||
*/
|
||||
public interface SnapshotMapper extends BaseMapper<SysTenantUsageSnapshot> {
|
||||
|
||||
/** 所有启用中的租户（快照要逐个租户算）。 */
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT id AS "tenantId", tenant_name AS "tenantName"
|
||||
FROM sys_tenant
|
||||
WHERE deleted = 0 AND status = 1
|
||||
ORDER BY id
|
||||
""")
|
||||
List<Map<String, Object>> activeTenants();
|
||||
|
||||
/**
|
||||
* 某个租户、某个周期的**已有快照**（幂等：同一天重复跑不会造出两行）。
|
||||
*
|
||||
* <p>用 `selectOne` 而不是"先查再插"两处各写一遍 —— 唯一索引是最后一道保险，
|
||||
* 但不该靠它来发现重复。
|
||||
*/
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT * FROM sys_tenant_usage_snapshot
|
||||
WHERE tenant_id = #{tenantId} AND period = #{period}
|
||||
LIMIT 1
|
||||
""")
|
||||
SysTenantUsageSnapshot findByPeriod(@Param("tenantId") Long tenantId,
|
||||
@Param("period") String period);
|
||||
|
||||
/** 运营视角：某周期所有租户的快照。 */
|
||||
@InterceptorIgnore(tenantLine = "true")
|
||||
@Select("""
|
||||
SELECT s.*, t.tenant_name AS tenantName
|
||||
FROM sys_tenant_usage_snapshot s
|
||||
LEFT JOIN sys_tenant t ON t.id = s.tenant_id AND t.deleted = 0
|
||||
WHERE s.period = #{period}
|
||||
ORDER BY s.student_count DESC
|
||||
""")
|
||||
List<Map<String, Object>> listByPeriod(@Param("period") String period);
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
-- ═══════════════════════════════════════════════════════════
|
||||
-- V17 · 定时任务收尾所需的 schema（PRD 附录 E）
|
||||
--
|
||||
-- 这一版只加三样东西，都是"不做就实现不了剩下三项定时任务"的：
|
||||
-- ① 文件处理失败的重试计数与失败原因（"压缩失败重试"要用）
|
||||
-- ② 租户用量快照表（"用量统计"要用 —— 快照必须落库，
|
||||
-- 否则"上个月的用量"永远算不出来：实时统计只能回答"现在多少"）
|
||||
--
|
||||
-- ⚠️ 只加列/加表，**不改已有列** —— 滚动发布期间新旧两版同时在跑，
|
||||
-- 改列类型会让旧实例在新库上出错（见 docs/09-部署与运维.md §2.3）。
|
||||
-- ═══════════════════════════════════════════════════════════
|
||||
|
||||
-- ── ① 文件处理失败：重试计数 + 失败原因 ──────────────────────
|
||||
|
||||
ALTER TABLE collect_file ADD COLUMN IF NOT EXISTS process_retry_count SMALLINT NOT NULL DEFAULT 0;
|
||||
ALTER TABLE collect_file ADD COLUMN IF NOT EXISTS process_error VARCHAR(256);
|
||||
ALTER TABLE collect_file ADD COLUMN IF NOT EXISTS last_retry_time TIMESTAMP;
|
||||
|
||||
COMMENT ON COLUMN collect_file.process_retry_count IS
|
||||
'处理失败后的重试次数。上限 3（PRD 附录 E「压缩失败重试：重试 ≤3 次」）——
|
||||
到上限就停，不然一个永远失败的图会被每 10 分钟重试一次，把磁盘与 CPU 白烧掉';
|
||||
COMMENT ON COLUMN collect_file.process_error IS
|
||||
'最近一次处理失败的原因（PRD FR-611「标记失败原因」）。
|
||||
⚠️ 与"保留原图"配套：只留原图而不知道为什么失败，运维只能靠猜';
|
||||
COMMENT ON COLUMN collect_file.last_retry_time IS '最近一次重试时间（观察重试是否在推进）';
|
||||
|
||||
-- 部分索引：只索引"还没处理出规范件"的行。
|
||||
-- 失败是**少数**（正常上传一步就压好了），全表索引会白白变大。
|
||||
CREATE INDEX IF NOT EXISTS idx_file_retry
|
||||
ON collect_file (process_retry_count, last_retry_time)
|
||||
WHERE deleted = 0 AND std_path IS NULL;
|
||||
|
||||
|
||||
-- ── ② 租户用量快照 ──────────────────────────────────────────
|
||||
|
||||
CREATE TABLE IF NOT EXISTS sys_tenant_usage_snapshot (
|
||||
id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
|
||||
tenant_id BIGINT NOT NULL,
|
||||
period VARCHAR(16) NOT NULL,
|
||||
student_count BIGINT NOT NULL DEFAULT 0,
|
||||
storage_mb BIGINT NOT NULL DEFAULT 0,
|
||||
ocr_count INTEGER NOT NULL DEFAULT 0,
|
||||
sms_count INTEGER NOT NULL DEFAULT 0,
|
||||
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
COMMENT ON TABLE sys_tenant_usage_snapshot IS
|
||||
'租户用量快照（PRD 附录 E「用量统计：每日 01:00」）。
|
||||
⚠️ 为什么必须落库而不是每次实时算：
|
||||
① 实时统计只能回答"现在多少"，而计费与超额告警问的是"**上个月**用了多少"；
|
||||
② 历史快照能回答"用量是什么时候涨上去的" —— 招生季与毕业季的形状完全不同；
|
||||
③ 存储量尤其重要：原始件 30 天后被清理，**清理前的占用在实时统计里再也看不到**';
|
||||
COMMENT ON COLUMN sys_tenant_usage_snapshot.period IS
|
||||
'统计周期，`2026-09`（月）或 `2026Q3`（季）—— 与 §6.7 的 period 同口径';
|
||||
COMMENT ON COLUMN sys_tenant_usage_snapshot.student_count IS
|
||||
'按证件号去重的学生数（同一个人多个任务只算一次）';
|
||||
COMMENT ON COLUMN sys_tenant_usage_snapshot.storage_mb IS
|
||||
'规范件体积合计（MB）。**不含原始件** —— 原始件 30 天后会被清掉，算进去会让曲线自己往下掉';
|
||||
COMMENT ON COLUMN sys_tenant_usage_snapshot.ocr_count IS
|
||||
'识别通道用量。一期恒为 0（M4 未接，等 OCR 选型），不编数字';
|
||||
|
||||
-- 一个租户一个周期一行：定时任务重复跑（或手工补跑）不会造出两行
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uk_usage_snapshot
|
||||
ON sys_tenant_usage_snapshot (tenant_id, period);
|
||||
-- 运营侧"某个月所有租户用了多少"的查询
|
||||
CREATE INDEX IF NOT EXISTS idx_usage_snapshot_period
|
||||
ON sys_tenant_usage_snapshot (period, tenant_id);
|
||||
Reference in New Issue
Block a user