From dcaf28afac290fb6da074549ba6bd65db3d5814d Mon Sep 17 00:00:00 2001 From: liuwanwan1 Date: Tue, 14 Jul 2026 10:34:32 +0800 Subject: [PATCH] fix: harden task execution and deployment --- .gitignore | 3 +- README.md | 13 +- pom.xml | 34 +--- sql/migrations/V2__task_integrity.sql | 5 + sql/th_wk_queue.sql | 5 +- sql/th_wk_user.sql | 5 +- sql/tihai.sql | 10 +- .../tihai/config/FreeThreadPoolConfig.java | 4 +- .../com/tihai/config/OperatorAuthFilter.java | 61 +++++++ .../tihai/controller/SuperStarController.java | 29 +-- .../com/tihai/exception/GlobalException.java | 12 +- .../tihai/factory/PriorityRejectPolicy.java | 21 +-- .../tihai/manager/CoursePointNavigator.java | 35 ++++ .../com/tihai/manager/RollBackManager.java | 16 +- .../java/com/tihai/manager/TaskLifecycle.java | 22 +++ .../queue/BoundedPriorityBlockingQueue.java | 165 +++++++++++++++++ .../superstar/SuperStarTaskService.java | 5 + .../impl/SuperStarCookieServiceImpl.java | 17 +- .../impl/SuperStarLoginServiceImpl.java | 35 ++-- .../impl/SuperStarTaskServiceImpl.java | 172 +++++++++++++----- .../impl/SuperStarUserServiceImpl.java | 2 +- src/main/java/com/tihai/utils/CourseUtil.java | 52 +++++- .../com/tihai/utils/CredentialCipher.java | 72 ++++++++ src/main/resources/application-dev.yml | 19 +- src/main/resources/application-prod.yml | 10 +- .../manager/CoursePointNavigatorTest.java | 36 ++++ .../com/tihai/manager/TaskLifecycleTest.java | 20 ++ .../BoundedPriorityBlockingQueueTest.java | 23 +++ .../tihai/utils/CourseUtilSessionTest.java | 45 +++++ .../com/tihai/utils/CredentialCipherTest.java | 23 +++ 30 files changed, 805 insertions(+), 166 deletions(-) create mode 100644 sql/migrations/V2__task_integrity.sql create mode 100644 src/main/java/com/tihai/config/OperatorAuthFilter.java create mode 100644 src/main/java/com/tihai/manager/CoursePointNavigator.java create mode 100644 src/main/java/com/tihai/manager/TaskLifecycle.java create mode 100644 src/main/java/com/tihai/queue/BoundedPriorityBlockingQueue.java create mode 100644 src/main/java/com/tihai/utils/CredentialCipher.java create mode 100644 src/test/java/com/tihai/manager/CoursePointNavigatorTest.java create mode 100644 src/test/java/com/tihai/manager/TaskLifecycleTest.java create mode 100644 src/test/java/com/tihai/queue/BoundedPriorityBlockingQueueTest.java create mode 100644 src/test/java/com/tihai/utils/CourseUtilSessionTest.java create mode 100644 src/test/java/com/tihai/utils/CredentialCipherTest.java diff --git a/.gitignore b/.gitignore index e99d4c0..73d75a7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ src/main/resources/application-dev.yml src/main/resources/application-prod.yml src/main/resources/super-star.yml -src/main/resources/bootstrap.yml \ No newline at end of file +target/ +src/main/resources/bootstrap.yml diff --git a/README.md b/README.md index ce086af..6f4557b 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ SuperAutoStudy 是一个**开源、可自部署**的学习通任务自动化工 - 服务治理：Apache Dubbo - 异步调度：RabbitMQ - 缓存加速：Redis -- **灵活部署**：提供 Docker Compose 一键部署方案（支持 ARM/x86）。 +- **灵活部署**：支持本地或服务器部署；请按下方配置完成数据库和运行密钥设置。 --- @@ -53,8 +53,15 @@ cd SuperAutoStudy ### 2. 部署方式 -- **完整版（含中间件）**：参考 [部署文档](https://doc.xxtmooc.com/pages/793dcb/) -- **轻量版（仅刷课）**：使用 `lightweight` 分支或预编译 JAR（见 [Releases](https://github.com/DuanInnovator/SuperAutoStudy/releases)） +- 创建 MySQL 数据库后导入 `sql/tihai.sql`。该文件会重建表，仅应在新数据库中执行。 +- 复制并填写 `src/main/resources/application-dev.yml` 中的数据库配置。 +- 设置 `SUPER_AUTO_CREDENTIAL_KEY`：值必须为 16、24 或 32 字节的随机字符串。它用于加密数据库中的账号密码和 Cookie，丢失后旧数据无法解密。 +- 对外部署还必须设置 `SUPER_AUTO_API_TOKEN`，并在调用时传入 `Authorization: Bearer ` 或 `X-API-Token: `；未配置令牌时仅允许本机回环地址访问。 +- 执行 `mvn package`，再运行 `java -jar target/SuperAuto-1.0-SNAPSHOT.jar`。 + +首次部署请先调用 `POST /chaoxing/course` 验证账号并获取课程，再调用 `POST /chaoxing` 提交任务。可用 `GET /chaoxing/tasks?loginAccount=...` 查看任务，`DELETE /chaoxing/tasks/{taskId}` 暂停尚未执行的任务。 + +> 数据库从旧版本升级时，请先清理重复的“账号 + 课程”任务，再执行 `sql/migrations/V2__task_integrity.sql`。 > 💡 推荐首次使用者阅读 [快速上手指南](https://doc.xxtmooc.com/pages/793dcb/)。 diff --git a/pom.xml b/pom.xml index c9607b4..c90eacc 100644 --- a/pom.xml +++ b/pom.xml @@ -46,7 +46,7 @@ org.projectlombok lombok - 1.18.22 + 1.18.36 compile @@ -63,12 +63,6 @@ 4.12.0 - - com.fasterxml.jackson.core - jackson-databind - 2.15.4 - - org.jsoup jsoup @@ -226,13 +220,6 @@ - - com.baomidou - mybatis-plus-boot-starter - 3.4.1 - - - javax.servlet javax.servlet-api @@ -272,24 +259,23 @@ org.apache.maven.plugins maven-compiler-plugin - 3.5.1 + 3.11.0 + 8 - -Xlint:deprecation - - - - - ${java.home}\lib\rt.jar;${java.home}\lib\jce.jar - + + org.apache.maven.plugins + maven-surefire-plugin + 3.2.5 + + - \ No newline at end of file + diff --git a/sql/migrations/V2__task_integrity.sql b/sql/migrations/V2__task_integrity.sql new file mode 100644 index 0000000..8bcfc60 --- /dev/null +++ b/sql/migrations/V2__task_integrity.sql @@ -0,0 +1,5 @@ +-- Apply this migration to existing installations after resolving duplicate tasks. +ALTER TABLE th_wk_queue MODIFY COLUMN id varchar(36) NOT NULL COMMENT '任务id'; +ALTER TABLE th_wk_queue ADD UNIQUE KEY uk_th_wk_queue_account_course (login_account, course_id); +ALTER TABLE th_wk_user ADD COLUMN id bigint(20) NOT NULL AUTO_INCREMENT PRIMARY KEY FIRST; +ALTER TABLE th_wk_user ADD UNIQUE KEY uk_th_wk_user_account (account); diff --git a/sql/th_wk_queue.sql b/sql/th_wk_queue.sql index 5377143..30a2835 100644 --- a/sql/th_wk_queue.sql +++ b/sql/th_wk_queue.sql @@ -7,7 +7,7 @@ SET FOREIGN_KEY_CHECKS = 0; -- ---------------------------- DROP TABLE IF EXISTS `th_wk_queue`; CREATE TABLE `th_wk_queue` ( - `id` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '订单id（注意：有可能在总订单表不存在这个id）', + `id` varchar(36) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '任务id', `login_account` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '登陆账号', `password` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '密码', `course_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '课程id', @@ -17,7 +17,8 @@ CREATE TABLE `th_wk_queue` ( `retry_count` int(11) NULL DEFAULT 0 COMMENT '重试次数', `creat_time` datetime NULL DEFAULT NULL COMMENT '创建时间', `machine_num` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '所处机器编号', - PRIMARY KEY (`id`) USING BTREE + PRIMARY KEY (`id`) USING BTREE, + UNIQUE KEY `uk_th_wk_queue_account_course` (`login_account`, `course_id`) ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '网课任务队列表' ROW_FORMAT = Dynamic; SET FOREIGN_KEY_CHECKS = 1; diff --git a/sql/th_wk_user.sql b/sql/th_wk_user.sql index ed031b3..564bc67 100644 --- a/sql/th_wk_user.sql +++ b/sql/th_wk_user.sql @@ -8,12 +8,15 @@ SET FOREIGN_KEY_CHECKS = 0; -- ---------------------------- DROP TABLE IF EXISTS `th_wk_user`; CREATE TABLE `th_wk_user` ( + `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键', `account` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '账号', `password` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '密码', `school_name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '学校名称', `name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '姓名', `fid` bigint(20) NULL DEFAULT NULL COMMENT '学校id', - `cookies` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL COMMENT '用户当前cookie' + `cookies` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL COMMENT '用户当前cookie', + PRIMARY KEY (`id`) USING BTREE, + UNIQUE KEY `uk_th_wk_user_account` (`account`) ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '网课-用户信息表' ROW_FORMAT = Dynamic; SET FOREIGN_KEY_CHECKS = 1; diff --git a/sql/tihai.sql b/sql/tihai.sql index 445a847..e6080ca 100644 --- a/sql/tihai.sql +++ b/sql/tihai.sql @@ -28,7 +28,7 @@ CREATE TABLE `th_wk_log` ( -- ---------------------------- DROP TABLE IF EXISTS `th_wk_queue`; CREATE TABLE `th_wk_queue` ( - `id` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '订单id（注意：有可能在总订单表不存在这个id）', + `id` varchar(36) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '任务id', `login_account` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '登陆账号', `password` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '密码', `course_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '课程id', @@ -38,7 +38,8 @@ CREATE TABLE `th_wk_queue` ( `retry_count` int(11) NULL DEFAULT 0 COMMENT '重试次数', `creat_time` datetime NULL DEFAULT NULL COMMENT '创建时间', `machine_num` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '所处机器编号', - PRIMARY KEY (`id`) USING BTREE + PRIMARY KEY (`id`) USING BTREE, + UNIQUE KEY `uk_th_wk_queue_account_course` (`login_account`, `course_id`) ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '网课任务队列表' ROW_FORMAT = Dynamic; -- ---------------------------- @@ -46,12 +47,15 @@ CREATE TABLE `th_wk_queue` ( -- ---------------------------- DROP TABLE IF EXISTS `th_wk_user`; CREATE TABLE `th_wk_user` ( + `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键', `account` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '账号', `password` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '密码', `school_name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '学校名称', `name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '姓名', `fid` bigint(20) NULL DEFAULT NULL COMMENT '学校id', - `cookies` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL COMMENT '用户当前cookie' + `cookies` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL COMMENT '用户当前cookie', + PRIMARY KEY (`id`) USING BTREE, + UNIQUE KEY `uk_th_wk_user_account` (`account`) ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '网课-用户信息表' ROW_FORMAT = Dynamic; SET FOREIGN_KEY_CHECKS = 1; diff --git a/src/main/java/com/tihai/config/FreeThreadPoolConfig.java b/src/main/java/com/tihai/config/FreeThreadPoolConfig.java index 564b023..11bc362 100644 --- a/src/main/java/com/tihai/config/FreeThreadPoolConfig.java +++ b/src/main/java/com/tihai/config/FreeThreadPoolConfig.java @@ -3,13 +3,13 @@ import com.tihai.factory.CustomThreadFactory; import com.tihai.factory.PriorityRejectPolicy; import com.tihai.properties.ThreadPoolProperties; +import com.tihai.queue.BoundedPriorityBlockingQueue; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.task.TaskExecutor; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -30,7 +30,7 @@ public ThreadPoolExecutor customThreadPool(ThreadPoolProperties config) { log.error("当前核心线程数为:{}", config.getCoreSize()); // 创建优先级阻塞队列 - BlockingQueue queue = new PriorityBlockingQueue<>(config.getQueueCapacity()); + BlockingQueue queue = new BoundedPriorityBlockingQueue<>(config.getQueueCapacity()); // 构建线程池 ThreadPoolExecutor executor = new ThreadPoolExecutor( diff --git a/src/main/java/com/tihai/config/OperatorAuthFilter.java b/src/main/java/com/tihai/config/OperatorAuthFilter.java new file mode 100644 index 0000000..ab8c7d2 --- /dev/null +++ b/src/main/java/com/tihai/config/OperatorAuthFilter.java @@ -0,0 +1,61 @@ +package com.tihai.config; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; +import org.springframework.web.filter.OncePerRequestFilter; + +import javax.servlet.FilterChain; +import javax.servlet.ServletException; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; + +/** + * Protects the operational task endpoints. With no configured token the + * application is intentionally limited to loopback callers. + */ +@Component +public class OperatorAuthFilter extends OncePerRequestFilter { + + private final String apiToken; + + public OperatorAuthFilter(@Value("${security.api-token:}") String apiToken) { + this.apiToken = apiToken; + } + + @Override + protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, + FilterChain filterChain) throws ServletException, IOException { + if (!request.getRequestURI().startsWith("/chaoxing")) { + filterChain.doFilter(request, response); + return; + } + + if (isAuthorized(request)) { + filterChain.doFilter(request, response); + return; + } + + response.setStatus(HttpServletResponse.SC_UNAUTHORIZED); + response.setContentType("application/json;charset=UTF-8"); + response.getWriter().write("{\"code\":201,\"subCode\":401,\"msg\":\"未授权\"}"); + } + + private boolean isAuthorized(HttpServletRequest request) { + if (!StringUtils.hasText(apiToken)) { + String remoteAddress = request.getRemoteAddr(); + return "127.0.0.1".equals(remoteAddress) || "::1".equals(remoteAddress) + || "0:0:0:0:0:0:0:1".equals(remoteAddress); + } + String authorization = request.getHeader("Authorization"); + String suppliedToken = authorization != null && authorization.startsWith("Bearer ") + ? authorization.substring("Bearer ".length()) + : request.getHeader("X-API-Token"); + return suppliedToken != null && MessageDigest.isEqual( + apiToken.getBytes(StandardCharsets.UTF_8), + suppliedToken.getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/src/main/java/com/tihai/controller/SuperStarController.java b/src/main/java/com/tihai/controller/SuperStarController.java index 017fa3c..2355ef0 100644 --- a/src/main/java/com/tihai/controller/SuperStarController.java +++ b/src/main/java/com/tihai/controller/SuperStarController.java @@ -2,12 +2,11 @@ import com.tihai.common.R; import com.tihai.dubbo.dto.CourseSubmitTaskDTO; -import com.tihai.enums.BizCodeEnum; import com.tihai.service.superstar.impl.SuperStarTaskServiceImpl; -import ma.glasnost.orika.MapperFacade; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; import javax.validation.Valid; +import java.util.List; /** * @Copyright : DuanInnovator @@ -24,23 +23,14 @@ public class SuperStarController { @Autowired private SuperStarTaskServiceImpl chaoXingTaskService; - @Autowired - private MapperFacade mapperFacade; - /** * 添加网课代刷任务 * * @return */ @PostMapping("") - public R addChaxingTask(@RequestBody @Valid CourseSubmitTaskDTO courseSubmitTaskDTO) { - - try { - chaoXingTaskService.addChaoXingTask(courseSubmitTaskDTO); - } catch (Exception e) { - - return R.error(BizCodeEnum.TASK_ALREADY_EXIST.getCode(), BizCodeEnum.TASK_ALREADY_EXIST.getMsg()); - } + public R addChaxingTask(@RequestBody @Valid CourseSubmitTaskDTO courseSubmitTaskDTO) throws Exception { + chaoXingTaskService.addChaoXingTask(courseSubmitTaskDTO); return R.ok(); } @@ -51,5 +41,18 @@ public R addChaxingTask(@RequestBody @Valid CourseSubmitTaskDTO courseSubmitTask public void start(){ chaoXingTaskService.startChaoxingTask(); } + + @GetMapping("/tasks") + public R listTasks(@RequestParam(required = false) String loginAccount) { + List tasks = chaoXingTaskService.listTasks(loginAccount); + return R.ok().put(tasks); + } + + @DeleteMapping("/tasks/{taskId}") + public R pauseTask(@PathVariable String taskId) { + return chaoXingTaskService.pauseTask(taskId) + ? R.ok() + : R.error(404, "任务不存在或已完成"); + } } diff --git a/src/main/java/com/tihai/exception/GlobalException.java b/src/main/java/com/tihai/exception/GlobalException.java index 3196927..76fde4c 100644 --- a/src/main/java/com/tihai/exception/GlobalException.java +++ b/src/main/java/com/tihai/exception/GlobalException.java @@ -80,10 +80,14 @@ public R handleBindException(BindException ex) { } + @ExceptionHandler(BusinessException.class) + public R handleBusinessException(BusinessException ex) { + return R.error(ex.getCode(), ex.getMsg(), ex.getData()); + } - @ExceptionHandler(Exception.class) //捕获所有异常 + @ExceptionHandler(Exception.class) public R ex(Exception ex) { - ex.printStackTrace(); - return R.error(404, BizCodeEnum.SYSTEM_ERROR.getMsg(), null); + log.error("未处理的请求异常", ex); + return R.error(BizCodeEnum.SYSTEM_ERROR.getCode(), BizCodeEnum.SYSTEM_ERROR.getMsg(), null); } -} \ No newline at end of file +} diff --git a/src/main/java/com/tihai/factory/PriorityRejectPolicy.java b/src/main/java/com/tihai/factory/PriorityRejectPolicy.java index e777b1e..d4a1443 100644 --- a/src/main/java/com/tihai/factory/PriorityRejectPolicy.java +++ b/src/main/java/com/tihai/factory/PriorityRejectPolicy.java @@ -1,27 +1,16 @@ package com.tihai.factory; -import com.tihai.queue.PriorityTaskWrapper; - import java.util.concurrent.RejectedExecutionHandler; -import java.util.concurrent.ThreadPoolExecutor; // 关键修正：使用JDK标准线程池类 +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; /** - * 自定义拒绝策略（优先级降级重试） + * Rejects explicitly so the task lifecycle can return the task to PENDING. */ public class PriorityRejectPolicy implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { - if (executor.isShutdown()) { - return; - } - - if (r instanceof PriorityTaskWrapper) { - PriorityTaskWrapper task = (PriorityTaskWrapper) r; - if(task.getPriority()>=1){ - task.decreasePriority(1); // 优先级降级 - } - executor.execute(task); // 重新入队 - } + throw new RejectedExecutionException("任务队列已满或线程池已关闭"); } -} \ No newline at end of file +} diff --git a/src/main/java/com/tihai/manager/CoursePointNavigator.java b/src/main/java/com/tihai/manager/CoursePointNavigator.java new file mode 100644 index 0000000..3770425 --- /dev/null +++ b/src/main/java/com/tihai/manager/CoursePointNavigator.java @@ -0,0 +1,35 @@ +package com.tihai.manager; + +import com.tihai.common.ChapterPoint; +import com.tihai.common.CoursePoint; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * Normalizes the upstream course-point response into one ordered chapter list. + */ +public final class CoursePointNavigator { + + private CoursePointNavigator() { + } + + public static List flatten(List coursePoints) { + if (coursePoints == null || coursePoints.isEmpty()) { + return Collections.emptyList(); + } + + List chapters = new ArrayList<>(); + for (CoursePoint coursePoint : coursePoints) { + if (coursePoint != null && coursePoint.getPoints() != null) { + chapters.addAll(coursePoint.getPoints()); + } + } + return chapters; + } + + public static boolean isComplete(int nextChapterIndex, int chapterCount) { + return nextChapterIndex >= chapterCount; + } +} diff --git a/src/main/java/com/tihai/manager/RollBackManager.java b/src/main/java/com/tihai/manager/RollBackManager.java index 95caeae..6f33c81 100644 --- a/src/main/java/com/tihai/manager/RollBackManager.java +++ b/src/main/java/com/tihai/manager/RollBackManager.java @@ -2,6 +2,9 @@ import org.springframework.stereotype.Component; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + /** * @Copyright : DuanInnovator * @Description :回滚管理器 @@ -11,18 +14,13 @@ **/ @Component public class RollBackManager { - private int rollbackTimes = 0; - private String rollbackId = ""; + private final ConcurrentHashMap rollbackCounts = new ConcurrentHashMap<>(); public void addTimes(String id) throws Exception { - if (id.equals(rollbackId) && rollbackTimes >= 3) { + int attempts = rollbackCounts.computeIfAbsent(id, key -> new AtomicInteger()).incrementAndGet(); + if (attempts > 3) { + rollbackCounts.remove(id); throw new RuntimeException("回滚次数已达3次，请手动检查学习通任务点完成情况"); - } else if (!id.equals(rollbackId)) { - // 新任务 - rollbackId = id; - rollbackTimes = 1; - } else { - rollbackTimes++; } } } diff --git a/src/main/java/com/tihai/manager/TaskLifecycle.java b/src/main/java/com/tihai/manager/TaskLifecycle.java new file mode 100644 index 0000000..14043a7 --- /dev/null +++ b/src/main/java/com/tihai/manager/TaskLifecycle.java @@ -0,0 +1,22 @@ +package com.tihai.manager; + +import com.tihai.enums.WkTaskStatusEnum; + +/** + * Defines the persisted task states that can safely be recovered after a + * process restart. + */ +public final class TaskLifecycle { + + private TaskLifecycle() { + } + + public static boolean shouldRecover(Integer status) { + return WkTaskStatusEnum.QUEUE.getCode().equals(status) + || WkTaskStatusEnum.PROCESSING.getCode().equals(status); + } + + public static Integer recoveredStatus() { + return WkTaskStatusEnum.PENDING.getCode(); + } +} diff --git a/src/main/java/com/tihai/queue/BoundedPriorityBlockingQueue.java b/src/main/java/com/tihai/queue/BoundedPriorityBlockingQueue.java new file mode 100644 index 0000000..222bf86 --- /dev/null +++ b/src/main/java/com/tihai/queue/BoundedPriorityBlockingQueue.java @@ -0,0 +1,165 @@ +package com.tihai.queue; + +import java.util.AbstractQueue; +import java.util.Collection; +import java.util.Iterator; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +/** + * A priority queue with a real capacity. {@link PriorityBlockingQueue} alone + * is unbounded, so it cannot trigger a ThreadPoolExecutor rejection policy. + */ +public class BoundedPriorityBlockingQueue extends AbstractQueue implements BlockingQueue { + + private final PriorityBlockingQueue delegate = new PriorityBlockingQueue<>(); + private final Semaphore slots; + + public BoundedPriorityBlockingQueue(int capacity) { + if (capacity <= 0) { + throw new IllegalArgumentException("capacity must be positive"); + } + this.slots = new Semaphore(capacity); + } + + @Override + public boolean offer(E element) { + if (!slots.tryAcquire()) { + return false; + } + boolean added = false; + try { + added = delegate.offer(element); + return added; + } finally { + if (!added) { + slots.release(); + } + } + } + + @Override + public void put(E element) throws InterruptedException { + slots.acquire(); + boolean added = false; + try { + added = delegate.offer(element); + } finally { + if (!added) { + slots.release(); + } + } + } + + @Override + public boolean offer(E element, long timeout, TimeUnit unit) throws InterruptedException { + if (!slots.tryAcquire(timeout, unit)) { + return false; + } + boolean added = false; + try { + added = delegate.offer(element); + return added; + } finally { + if (!added) { + slots.release(); + } + } + } + + @Override + public E poll() { + E element = delegate.poll(); + releaseSlot(element); + return element; + } + + @Override + public E take() throws InterruptedException { + E element = delegate.take(); + releaseSlot(element); + return element; + } + + @Override + public E poll(long timeout, TimeUnit unit) throws InterruptedException { + E element = delegate.poll(timeout, unit); + releaseSlot(element); + return element; + } + + @Override + public E peek() { + return delegate.peek(); + } + + @Override + public int size() { + return delegate.size(); + } + + @Override + public int remainingCapacity() { + return slots.availablePermits(); + } + + @Override + public boolean remove(Object element) { + boolean removed = delegate.remove(element); + if (removed) { + slots.release(); + } + return removed; + } + + @Override + public int drainTo(Collection collection) { + int drained = delegate.drainTo(collection); + slots.release(drained); + return drained; + } + + @Override + public int drainTo(Collection collection, int maxElements) { + int drained = delegate.drainTo(collection, maxElements); + slots.release(drained); + return drained; + } + + @Override + public void clear() { + int count = delegate.size(); + delegate.clear(); + slots.release(count); + } + + @Override + public Iterator iterator() { + final Iterator iterator = delegate.iterator(); + return new Iterator() { + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public E next() { + return iterator.next(); + } + + @Override + public void remove() { + iterator.remove(); + slots.release(); + } + }; + } + + private void releaseSlot(E element) { + if (element != null) { + slots.release(); + } + } +} diff --git a/src/main/java/com/tihai/service/superstar/SuperStarTaskService.java b/src/main/java/com/tihai/service/superstar/SuperStarTaskService.java index 8c790b1..23df751 100644 --- a/src/main/java/com/tihai/service/superstar/SuperStarTaskService.java +++ b/src/main/java/com/tihai/service/superstar/SuperStarTaskService.java @@ -6,6 +6,7 @@ import com.tihai.dubbo.dto.CourseSubmitTaskDTO; import java.io.IOException; +import java.util.List; /** * @Copyright : DuanInnovator @@ -55,5 +56,9 @@ public interface SuperStarTaskService extends IService { * @param task */ void executeCourseTask(SuperStarTask task) throws IOException; + + List listTasks(String loginAccount); + + boolean pauseTask(String taskId); } diff --git a/src/main/java/com/tihai/service/superstar/impl/SuperStarCookieServiceImpl.java b/src/main/java/com/tihai/service/superstar/impl/SuperStarCookieServiceImpl.java index af7cd46..6767997 100644 --- a/src/main/java/com/tihai/service/superstar/impl/SuperStarCookieServiceImpl.java +++ b/src/main/java/com/tihai/service/superstar/impl/SuperStarCookieServiceImpl.java @@ -5,6 +5,8 @@ import com.tihai.domain.chaoxing.WkUser; import com.tihai.mapper.SuperStarCookieMapper; import com.tihai.service.superstar.SuperStarCookieService; +import com.tihai.utils.CredentialCipher; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; /** @@ -17,6 +19,9 @@ @Service public class SuperStarCookieServiceImpl extends ServiceImpl implements SuperStarCookieService { + @Autowired + private CredentialCipher credentialCipher; + /** * 根据登陆账号获取cookie * @param loginAccount 登录账号 @@ -26,8 +31,8 @@ public String getWkUserCookie(String loginAccount) { LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); wrapper.eq(WkUser::getAccount, loginAccount); wrapper.select(WkUser::getCookies); - WkUser wkUser = this.getOne(wrapper); - return wkUser.getCookies(); + WkUser wkUser = this.getOne(wrapper, false); + return wkUser == null ? null : credentialCipher.decrypt(wkUser.getCookies()); } /** @@ -36,7 +41,13 @@ public String getWkUserCookie(String loginAccount) { */ @Override public void updateWkUserCookies(WkUser wkUser) { - this.updateById(wkUser); + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.eq(WkUser::getAccount, wkUser.getAccount()); + WkUser existingUser = this.getOne(wrapper, false); + if (existingUser != null) { + existingUser.setCookies(credentialCipher.encrypt(wkUser.getCookies())); + this.updateById(existingUser); + } } } diff --git a/src/main/java/com/tihai/service/superstar/impl/SuperStarLoginServiceImpl.java b/src/main/java/com/tihai/service/superstar/impl/SuperStarLoginServiceImpl.java index d0c7c8a..c8acea9 100644 --- a/src/main/java/com/tihai/service/superstar/impl/SuperStarLoginServiceImpl.java +++ b/src/main/java/com/tihai/service/superstar/impl/SuperStarLoginServiceImpl.java @@ -10,6 +10,7 @@ import com.tihai.service.superstar.SuperStarUserService; import com.tihai.utils.AESCipher; import com.tihai.utils.JsonParser; +import com.tihai.utils.CredentialCipher; import ma.glasnost.orika.MapperFacade; import okhttp3.*; import org.springframework.beans.factory.annotation.Autowired; @@ -21,8 +22,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ThreadPoolExecutor; import java.util.stream.Collectors; /** @@ -48,7 +47,7 @@ public class SuperStarLoginServiceImpl extends ServiceImpl loadForRequest(HttpUrl url) { FormBody formBody = new FormBody.Builder() .add("fid", "-1") .add("uname", cipher.encrypt(wkUser.getAccount())) - .add("password", cipher.encrypt(wkUser.getPassword())) + .add("password", cipher.encrypt(password)) .add("refer", "https%3A%2F%2Fi.chaoxing.com") .add("t", "true") .add("forbidotherlogin", "0") @@ -114,26 +114,25 @@ public List loadForRequest(HttpUrl url) { .map(Cookie::toString) .collect(Collectors.joining(", ")); - CompletableFuture.runAsync(() -> { - if (userByAccount != null) { - userByAccount.setCookies(cookieStr); - this.baseMapper.updateById(userByAccount); - } else { - wkUser.setCookies(cookieStr); - this.save(wkUser); - } - startLog.setRemark(GlobalConstant.LOGIN_SUCCESS); - }, threadPoolExecutor); + if (userByAccount != null) { + userByAccount.setCookies(credentialCipher.encrypt(cookieStr)); + userByAccount.setPassword(credentialCipher.encrypt(password)); + this.baseMapper.updateById(userByAccount); + } else { + wkUser.setPassword(credentialCipher.encrypt(password)); + wkUser.setCookies(credentialCipher.encrypt(cookieStr)); + this.save(wkUser); + } + startLog.setRemark(GlobalConstant.LOGIN_SUCCESS); return cookieStr; } else { - startLog.setRemark(result.get("msg2") != null ? result.get("msg2").toString() : GlobalConstant.LOGIN_FAIL); + startLog.setRemark(result != null && result.get("msg2") != null ? result.get("msg2").toString() : GlobalConstant.LOGIN_FAIL); } } catch (IOException e) { - Map ret = new HashMap<>(); startLog.setRemark(GlobalConstant.LOGIN_FAIL + e.getMessage()); } superStarLogService.saveLog(startLog); diff --git a/src/main/java/com/tihai/service/superstar/impl/SuperStarTaskServiceImpl.java b/src/main/java/com/tihai/service/superstar/impl/SuperStarTaskServiceImpl.java index 1c03312..33d48bd 100644 --- a/src/main/java/com/tihai/service/superstar/impl/SuperStarTaskServiceImpl.java +++ b/src/main/java/com/tihai/service/superstar/impl/SuperStarTaskServiceImpl.java @@ -18,16 +18,20 @@ import com.tihai.factory.CustomThreadFactory; import com.tihai.factory.PriorityRejectPolicy; import com.tihai.manager.RollBackManager; +import com.tihai.manager.CoursePointNavigator; +import com.tihai.manager.TaskLifecycle; import com.tihai.mapper.SuperStarMapper; import com.tihai.properties.StudyProperties; import com.tihai.properties.ThreadPoolProperties; import com.tihai.queue.PriorityTaskWrapper; +import com.tihai.queue.BoundedPriorityBlockingQueue; import com.tihai.service.superstar.SuperStarLogService; import com.tihai.service.superstar.SuperStarLoginService; import com.tihai.service.superstar.SuperStarTaskService; import com.tihai.service.superstar.SuperStarUserService; import com.tihai.utils.CourseUtil; import com.tihai.utils.ServerInfoUtil; +import com.tihai.utils.CredentialCipher; import lombok.extern.slf4j.Slf4j; import ma.glasnost.orika.MapperFacade; import org.apache.commons.lang3.StringUtils; @@ -35,9 +39,9 @@ import org.springframework.stereotype.Service; import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; import java.math.BigDecimal; import java.time.LocalDateTime; -import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.*; @@ -83,6 +87,9 @@ public class SuperStarTaskServiceImpl extends ServiceImpl logQueue = new LinkedBlockingQueue<>(); private final ConcurrentHashMap logMap = new ConcurrentHashMap<>(); + private Thread logWriterThread; public SuperStarTaskServiceImpl() throws NacosException { } @@ -173,7 +181,7 @@ public void restartThreadPoolAndTasks() { return; } - BlockingQueue queue = new PriorityBlockingQueue<>(config.getQueueCapacity()); + BlockingQueue queue = new BoundedPriorityBlockingQueue<>(config.getQueueCapacity()); // 重新构建线程池 taskExecutor = new ThreadPoolExecutor( @@ -192,8 +200,32 @@ public void restartThreadPoolAndTasks() { @PostConstruct public void init() { - // 启动一个单独的线程来处理日志的批量保存 - new Thread(this::batchSaveLogs).start(); + recoverInterruptedTasks(); + logWriterThread = new Thread(this::batchSaveLogs, "super-star-log-writer"); + logWriterThread.setDaemon(true); + logWriterThread.start(); + startChaoxingTask(); + } + + @PreDestroy + public void stopLogWriter() { + if (logWriterThread != null) { + logWriterThread.interrupt(); + } + } + + private void recoverInterruptedTasks() { + List interruptedTasks = list(new LambdaQueryWrapper() + .in(SuperStarTask::getStatus, + WkTaskStatusEnum.QUEUE.getCode(), + WkTaskStatusEnum.PROCESSING.getCode())); + interruptedTasks.stream() + .filter(task -> TaskLifecycle.shouldRecover(task.getStatus())) + .forEach(task -> task.setStatus(TaskLifecycle.recoveredStatus())); + if (!interruptedTasks.isEmpty()) { + updateBatchById(interruptedTasks); + log.info("已恢复 {} 个因服务重启中断的任务", interruptedTasks.size()); + } } private void batchSaveLogs() { @@ -245,7 +277,7 @@ public SuperStarTask getSuperStarTask(String loginAccount, String courseId) { LambdaQueryWrapper superStarTaskLambdaQueryWrapper = new LambdaQueryWrapper<>(); superStarTaskLambdaQueryWrapper.eq(SuperStarTask::getLoginAccount, loginAccount); superStarTaskLambdaQueryWrapper.eq(SuperStarTask::getCourseId, courseId); - return this.getOne(superStarTaskLambdaQueryWrapper); + return this.getOne(superStarTaskLambdaQueryWrapper, false); } /** @@ -259,7 +291,7 @@ public SuperStarTask getSuperStarTaskByCourseName(String loginAccount, String co LambdaQueryWrapper superStarTaskLambdaQueryWrapper = new LambdaQueryWrapper<>(); superStarTaskLambdaQueryWrapper.eq(SuperStarTask::getLoginAccount, loginAccount); superStarTaskLambdaQueryWrapper.eq(SuperStarTask::getCourseName, courseName); - return this.getOne(superStarTaskLambdaQueryWrapper); + return this.getOne(superStarTaskLambdaQueryWrapper, false); } /** @@ -277,9 +309,11 @@ public void addChaoXingTask(CourseSubmitTaskDTO courseSubmitTaskDTO) throws Naco } if (task == null) { SuperStarTask chaoXingTask = mapperFacade.map(courseSubmitTaskDTO, SuperStarTask.class); + chaoXingTask.setPassword(credentialCipher.encrypt(chaoXingTask.getPassword())); chaoXingTask.setStatus(WkTaskStatusEnum.PENDING.getCode()); - chaoXingTask.setId(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS"))); //当前version使用UUID + chaoXingTask.setId(UUID.randomUUID().toString()); chaoXingTask.setPriority(1); + chaoXingTask.setRetryCount(0); chaoXingTask.setCreatTime(LocalDateTime.now()); chaoXingTask.setMachineNum(serverInfoUtil.getCurrentServerInstance()); this.baseMapper.insert(chaoXingTask); @@ -310,30 +344,71 @@ public void startChaoxingTask() { throw new RuntimeException(e); } boolean isUpdated = this.updateById(task); + if (!isUpdated) { + log.warn("任务入队状态更新失败，taskId={}", task.getId()); + return; + } Runnable taskWrapper = new PriorityTaskWrapper(() -> { try { executeCourseTask(task); } catch (Exception e) { - if (task.getRetryCount() == RetryConstant.DEFAULT_RETRY_COUNT) { + int retryCount = task.getRetryCount() == null ? 0 : task.getRetryCount(); + if (retryCount >= RetryConstant.DEFAULT_RETRY_COUNT) { task.setStatus(WkTaskStatusEnum.ABNORMAL.getCode()); } else { - task.setRetryCount(task.getRetryCount() + 1); + task.setRetryCount(retryCount + 1); task.setStatus(WkTaskStatusEnum.PENDING.getCode()); } this.updateById(task); + log.warn("任务执行失败，将按状态重试。taskId={}, retryCount={}", + task.getId(), task.getRetryCount(), e); + if (WkTaskStatusEnum.PENDING.getCode().equals(task.getStatus())) { + startChaoxingTask(); + } } }, task.getPriority(), task.getId()); - taskExecutor.execute(taskWrapper); + try { + taskExecutor.execute(taskWrapper); + } catch (RejectedExecutionException e) { + task.setStatus(WkTaskStatusEnum.PENDING.getCode()); + this.updateById(task); + throw e; + } }); } + @Override + public List listTasks(String loginAccount) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.eq(StringUtils.isNotBlank(loginAccount), SuperStarTask::getLoginAccount, loginAccount); + wrapper.orderByDesc(SuperStarTask::getCreatTime); + return list(wrapper); + } + + @Override + public boolean pauseTask(String taskId) { + SuperStarTask task = getById(taskId); + if (task == null || !(WkTaskStatusEnum.PENDING.getCode().equals(task.getStatus()) + || WkTaskStatusEnum.QUEUE.getCode().equals(task.getStatus()))) { + return false; + } + task.setStatus(WkTaskStatusEnum.PAUSED.getCode()); + return updateById(task); + } + /** * 执行超星学习任务 * * @param task 超星学习任务 */ public void executeCourseTask(SuperStarTask task) { + SuperStarTask persistedTask = getById(task.getId()); + if (persistedTask == null || WkTaskStatusEnum.PAUSED.getCode().equals(persistedTask.getStatus())) { + return; + } + task.setStatus(WkTaskStatusEnum.PROCESSING.getCode()); + this.updateById(task); WkUser user = userService.getUserByAccount(task.getLoginAccount()); SuperStarLog log = superStarLogService.getLatestLogByLoginAccount(task.getLoginAccount(), task.getCourseName()); @@ -344,28 +419,22 @@ public void executeCourseTask(SuperStarTask task) { } - if (user == null || user.getCookies() == null) { - if (task.getRetryCount() < RetryConstant.DEFAULT_RETRY_COUNT) { - task.setStatus(WkTaskStatusEnum.PENDING.getCode()); - task.setRetryCount(task.getRetryCount() + 1); - } else { - task.setStatus(WkTaskStatusEnum.ABNORMAL.getCode()); + try { + if (user == null || user.getCookies() == null) { + WkUser wkUser = new WkUser(); + wkUser.setAccount(task.getLoginAccount()); + wkUser.setPassword(task.getPassword()); + String cookies = loginService.login(wkUser, true); + if (StringUtils.isEmpty(cookies)) { + throw new IllegalStateException("用户登录失败"); + } + user = wkUser; + user.setCookies(cookies); } - WkUser wkUser = new WkUser(); - wkUser.setAccount(task.getLoginAccount()); - wkUser.setPassword(task.getPassword()); - loginService.login(wkUser, true); - this.updateById(task); - log.setStatus(WkTaskStatusEnum.ABNORMAL.getCode()); - log.setErrorMessage("用户未登录"); - logQueue.offer(log); - return; - } - try { Course readyCourse = new Course(); courseUtil.setAccount(task.getLoginAccount()); - courseUtil.setCookies(user.getCookies()); + courseUtil.setCookies(credentialCipher.decrypt(user.getCookies())); if (task.getCourseId() != null) { readyCourse = courseUtil.getCourseList().stream() .filter(course -> course.getCourseId().equals(task.getCourseId())) @@ -388,25 +457,28 @@ public void executeCourseTask(SuperStarTask task) { } if (readyCourse == null) { - loginService.login(user, true); - log.setStatus(WkTaskStatusEnum.ABNORMAL.getCode()); - log.setErrorMessage(GlobalConstant.COURSE_INFO_GET_FAIL); - logQueue.offer(log); - return; + throw new IllegalStateException(GlobalConstant.COURSE_INFO_GET_FAIL); } List pointList = courseUtil.getCoursePoint( readyCourse.getCourseId(), readyCourse.getClazzId(), readyCourse.getCpi()); - processCoursePoints(task, log, readyCourse, pointList); + if (!processCoursePoints(task, log, readyCourse, pointList)) { + throw new IllegalStateException("课程章节执行失败"); + } + task.setStatus(WkTaskStatusEnum.FINISHED.getCode()); + this.updateById(task); } catch (Exception e) { log.setStatus(WkTaskStatusEnum.ABNORMAL.getCode()); - log.setErrorMessage("系统异常"); + log.setErrorMessage(e.getMessage()); logQueue.offer(log); + throw new IllegalStateException("课程任务执行失败: " + task.getId(), e); + } finally { + courseUtil.clearSession(); } } - private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course readyCourse, List pointList) { + private boolean processCoursePoints(SuperStarTask task, SuperStarLog log, Course readyCourse, List pointList) { if (log == null) { log = mapperFacade.map(task, SuperStarLog.class); log.setId(Long.valueOf(UUID.randomUUID().toString())); @@ -416,7 +488,7 @@ private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course re int pointIndex = log.getCurrentChapterIndex() != null ? log.getCurrentChapterIndex() : 0; - List chapterPointList = pointList.get(0).getPoints(); + List chapterPointList = CoursePointNavigator.flatten(pointList); while (pointIndex < chapterPointList.size()) { @@ -438,8 +510,7 @@ private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course re continue; } - if (jobs.stream().findFirst().map(job -> job instanceof Map && Boolean.TRUE.equals(((Map) job).get("notOpen"))).orElse(false)) { - pointIndex--; + if (jobInfo != null && jobInfo.stream().anyMatch(info -> Boolean.TRUE.equals(info.getNotOpen()))) { rb.addTimes(chapterPoint.getId()); continue; } @@ -452,13 +523,13 @@ private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course re case JobTypeConstant.VIDEO: boolean isAudio = false; try { - courseUtil.studyVideo(readyCourse, job, jobInfo.get(0), studyProperties.getSpeed(), "Video", log); + courseUtil.studyVideo(readyCourse, job, requiredJobInfo(jobInfo, job), studyProperties.getSpeed(), "Video", log); } catch (Exception e) { isAudio = true; } if (isAudio) { try { - courseUtil.studyVideo(readyCourse, job, jobInfo.get(0), studyProperties.getSpeed(), "Audio", log); + courseUtil.studyVideo(readyCourse, job, requiredJobInfo(jobInfo, job), studyProperties.getSpeed(), "Audio", log); } catch (Exception e) { log.setErrorMessage("异常任务 -> 章节: " + job.getId() + "，已跳过"); } @@ -468,10 +539,10 @@ private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course re courseUtil.studyDocument(readyCourse, job); break; case JobTypeConstant.READ: - courseUtil.studyRead(readyCourse, job, jobInfo.get(0), log); + courseUtil.studyRead(readyCourse, job, requiredJobInfo(jobInfo, job), log); break; case JobTypeConstant.QUESTION: - courseUtil.studyWork(readyCourse, job, jobInfo.get(0), log); + courseUtil.studyWork(readyCourse, job, requiredJobInfo(jobInfo, job), log); default: break; @@ -481,6 +552,7 @@ private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course re log.setEndTime(LocalDateTime.now()); log.setStatus(WkTaskStatusEnum.ABNORMAL.getCode()); log.setErrorMessage("任务异常，任务ID=" + job.getTitle()); + throw new IllegalStateException("任务点执行失败: " + job.getId(), e); } } pointIndex++; @@ -497,11 +569,11 @@ private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course re log.setStatus(WkTaskStatusEnum.ABNORMAL.getCode()); log.setErrorMessage("章节处理异常"); logQueue.offer(log); - break; + return false; } } - if (pointIndex >= pointList.size()) { + if (CoursePointNavigator.isComplete(pointIndex, chapterPointList.size())) { log.setEndTime(LocalDateTime.now()); log.setCurrentChapterIndex(pointIndex - 1); log.setCurrentProgress(BigDecimal.valueOf(100)); @@ -509,5 +581,13 @@ private void processCoursePoints(SuperStarTask task, SuperStarLog log, Course re logQueue.offer(log); } + return CoursePointNavigator.isComplete(pointIndex, chapterPointList.size()); + } + + private JobInfo requiredJobInfo(List jobInfo, Job job) { + if (jobInfo == null || jobInfo.isEmpty()) { + throw new IllegalStateException("任务点缺少执行信息: " + job.getId()); + } + return jobInfo.get(0); } -} \ No newline at end of file +} diff --git a/src/main/java/com/tihai/service/superstar/impl/SuperStarUserServiceImpl.java b/src/main/java/com/tihai/service/superstar/impl/SuperStarUserServiceImpl.java index 16e1cb5..349f67a 100644 --- a/src/main/java/com/tihai/service/superstar/impl/SuperStarUserServiceImpl.java +++ b/src/main/java/com/tihai/service/superstar/impl/SuperStarUserServiceImpl.java @@ -26,7 +26,7 @@ public class SuperStarUserServiceImpl extends ServiceImpl wkUserLambdaQueryWrapper = new LambdaQueryWrapper<>(); wkUserLambdaQueryWrapper.eq(WkUser::getAccount, account); - return this.getOne(wkUserLambdaQueryWrapper); + return this.getOne(wkUserLambdaQueryWrapper, false); } } diff --git a/src/main/java/com/tihai/utils/CourseUtil.java b/src/main/java/com/tihai/utils/CourseUtil.java index 9b92ebf..07d1f1c 100644 --- a/src/main/java/com/tihai/utils/CourseUtil.java +++ b/src/main/java/com/tihai/utils/CourseUtil.java @@ -42,10 +42,8 @@ public class CourseUtil { - public Map cookies; - public List cookieList = new ArrayList<>(); - - private String account; + private final ThreadLocal sessionContext = + ThreadLocal.withInitial(SessionContext::new); @Autowired private SuperStarLogService superStarLogService; @@ -60,13 +58,21 @@ public class CourseUtil { private Query query; public void setAccount(String account) { - this.account = account; + sessionContext.get().account = account; } public void setCookies(String cookies) { - this.cookies = convertCookieStringToMap(cookies); - cookieList = convertCookieStringToList(cookies); + SessionContext session = sessionContext.get(); + session.cookies = convertCookieStringToMap(cookies); + session.cookieList = convertCookieStringToList(cookies); + } + + /** + * Clears request-scoped session state before a pooled worker is reused. + */ + public void clearSession() { + sessionContext.remove(); } /** @@ -101,6 +107,9 @@ public static List convertCookieStringToList(String cookieStr) { } public static Map convertCookieStringToMap(String cookieStr) { + if (cookieStr == null || cookieStr.trim().isEmpty()) { + return new LinkedHashMap<>(); + } // 去掉前后的方括号 if (cookieStr.startsWith("[") && cookieStr.endsWith("]")) { cookieStr = cookieStr.substring(1, cookieStr.length() - 1); @@ -129,7 +138,27 @@ public static Map convertCookieStringToMap(String cookieStr) { } public String getValue(String key) { - return cookies.get(key); + return sessionContext.get().cookies.get(key); + } + + private String getAccount() { + return sessionContext.get().account; + } + + private List getCookieList() { + return sessionContext.get().cookieList; + } + + private String getCookieHeader() { + return getCookieList().stream() + .map(cookie -> cookie.name() + "=" + cookie.value()) + .collect(Collectors.joining("; ")); + } + + private static final class SessionContext { + private String account; + private Map cookies = new LinkedHashMap<>(); + private List cookieList = new ArrayList<>(); } @@ -197,6 +226,10 @@ public X509Certificate[] getAcceptedIssuers() { @Override public void saveFromResponse(HttpUrl url, List cookies) { String domain = url.host(); + String account = getAccount(); + if (account == null || cookieService == null) { + return; + } // 获取现有Cookie（从内存或数据库） List existingCookies = cookieCache.computeIfAbsent(domain, k -> { @@ -331,6 +364,7 @@ private Cookie parseSingleCookie(String cookieStr) { public List loadForRequest(HttpUrl url) { try { + String account = getAccount(); // 1. 检查基础条件 if (url == null || account == null || cookieService == null) { System.out.println("获取Cookie异常: 参数错误 -> " + @@ -468,7 +502,7 @@ public List getCoursePoint(String courseId, String clazzId, String OkHttpClient client = initSession(false, false); String url = String.format("https://mooc2-ans.chaoxing.com/mooc2-ans/mycourse/studentcourse?courseid=%s&clazzid=%s&cpi=%s&ut=s", courseId, clazzId, cpi); - Request request = new Request.Builder().url(url).addHeader("Cookie", cookieList.toString()).build(); + Request request = new Request.Builder().url(url).addHeader("Cookie", getCookieHeader()).build(); Response response = client.newCall(request).execute(); String respText = response.body().string(); CoursePoint stringObjectMap = Decode.decodeCoursePoint(respText); diff --git a/src/main/java/com/tihai/utils/CredentialCipher.java b/src/main/java/com/tihai/utils/CredentialCipher.java new file mode 100644 index 0000000..9018a73 --- /dev/null +++ b/src/main/java/com/tihai/utils/CredentialCipher.java @@ -0,0 +1,72 @@ +package com.tihai.utils; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.security.SecureRandom; +import java.util.Arrays; +import java.util.Base64; + +/** Encrypts locally persisted credentials and session cookies with AES-GCM. */ +@Component +public class CredentialCipher { + + private static final String VERSION = "v1:"; + private static final int IV_LENGTH = 12; + private static final int TAG_LENGTH = 128; + + private final SecretKeySpec key; + private final SecureRandom secureRandom = new SecureRandom(); + + public CredentialCipher(@Value("${security.credential-key:}") String keyMaterial) { + byte[] keyBytes = keyMaterial == null ? new byte[0] : keyMaterial.getBytes(StandardCharsets.UTF_8); + if (keyBytes.length != 16 && keyBytes.length != 24 && keyBytes.length != 32) { + throw new IllegalStateException("security.credential-key 必须是 16、24 或 32 字节，并应通过环境变量提供"); + } + this.key = new SecretKeySpec(keyBytes, "AES"); + } + + public String encrypt(String plainText) { + if (plainText == null || plainText.isEmpty()) { + return plainText; + } + try { + byte[] iv = new byte[IV_LENGTH]; + secureRandom.nextBytes(iv); + Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding"); + cipher.init(Cipher.ENCRYPT_MODE, key, new GCMParameterSpec(TAG_LENGTH, iv)); + byte[] encrypted = cipher.doFinal(plainText.getBytes(StandardCharsets.UTF_8)); + byte[] payload = new byte[iv.length + encrypted.length]; + System.arraycopy(iv, 0, payload, 0, iv.length); + System.arraycopy(encrypted, 0, payload, iv.length, encrypted.length); + return VERSION + Base64.getEncoder().encodeToString(payload); + } catch (Exception e) { + throw new IllegalStateException("敏感数据加密失败", e); + } + } + + /** Supports a gradual migration of existing plaintext rows on their next update. */ + public String decrypt(String value) { + if (value == null || value.isEmpty() || !isEncrypted(value)) { + return value; + } + try { + byte[] payload = Base64.getDecoder().decode(value.substring(VERSION.length())); + byte[] iv = Arrays.copyOfRange(payload, 0, IV_LENGTH); + byte[] encrypted = Arrays.copyOfRange(payload, IV_LENGTH, payload.length); + Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding"); + cipher.init(Cipher.DECRYPT_MODE, key, new GCMParameterSpec(TAG_LENGTH, iv)); + return new String(cipher.doFinal(encrypted), StandardCharsets.UTF_8); + } catch (Exception e) { + throw new IllegalStateException("敏感数据解密失败", e); + } + } + + public boolean isEncrypted(String value) { + return value != null && value.startsWith(VERSION); + } +} diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index fcb5e86..3d26d11 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -55,12 +55,15 @@ large: model: qwen-api-key: #通义千问api-key -tiku: - settings: - endpoints: - - name: #示例local,icodef,buguake,wanneng，不填写默认使用所有免费题库 - token: - - name: - token: - +tiku: + settings: + endpoints: + - name: #示例local,icodef,buguake,wanneng，不填写默认使用所有免费题库 + token: + - name: + token: + +security: + credential-key: ${SUPER_AUTO_CREDENTIAL_KEY:} + api-token: ${SUPER_AUTO_API_TOKEN:} diff --git a/src/main/resources/application-prod.yml b/src/main/resources/application-prod.yml index b3cf30d..1282a42 100644 --- a/src/main/resources/application-prod.yml +++ b/src/main/resources/application-prod.yml @@ -48,6 +48,10 @@ custom: allow-core-thread-timeout: true -study: - chaoxing: - speed: 3 +study: + chaoxing: + speed: 3 + +security: + credential-key: ${SUPER_AUTO_CREDENTIAL_KEY:} + api-token: ${SUPER_AUTO_API_TOKEN:} diff --git a/src/test/java/com/tihai/manager/CoursePointNavigatorTest.java b/src/test/java/com/tihai/manager/CoursePointNavigatorTest.java new file mode 100644 index 0000000..8949bec --- /dev/null +++ b/src/test/java/com/tihai/manager/CoursePointNavigatorTest.java @@ -0,0 +1,36 @@ +package com.tihai.manager; + +import com.tihai.common.ChapterPoint; +import com.tihai.common.CoursePoint; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CoursePointNavigatorTest { + + @Test + void traversesEveryCoursePointAndFinishesOnlyAfterTheLastChapter() { + CoursePoint firstGroup = coursePoint("chapter-1"); + CoursePoint secondGroup = coursePoint("chapter-2"); + + List chapters = CoursePointNavigator.flatten(Arrays.asList(firstGroup, secondGroup)); + + assertEquals(Arrays.asList("chapter-1", "chapter-2"), Arrays.asList( + chapters.get(0).getId(), chapters.get(1).getId())); + assertFalse(CoursePointNavigator.isComplete(1, chapters.size())); + assertTrue(CoursePointNavigator.isComplete(2, chapters.size())); + } + + private CoursePoint coursePoint(String chapterId) { + ChapterPoint chapter = new ChapterPoint(); + chapter.setId(chapterId); + CoursePoint point = new CoursePoint(); + point.setPoints(Arrays.asList(chapter)); + return point; + } +} diff --git a/src/test/java/com/tihai/manager/TaskLifecycleTest.java b/src/test/java/com/tihai/manager/TaskLifecycleTest.java new file mode 100644 index 0000000..b3f59ce --- /dev/null +++ b/src/test/java/com/tihai/manager/TaskLifecycleTest.java @@ -0,0 +1,20 @@ +package com.tihai.manager; + +import com.tihai.enums.WkTaskStatusEnum; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class TaskLifecycleTest { + + @Test + void recoversOnlyInterruptedTasksAndDoesNotRetryTerminalTasks() { + assertTrue(TaskLifecycle.shouldRecover(WkTaskStatusEnum.QUEUE.getCode())); + assertTrue(TaskLifecycle.shouldRecover(WkTaskStatusEnum.PROCESSING.getCode())); + assertFalse(TaskLifecycle.shouldRecover(WkTaskStatusEnum.FINISHED.getCode())); + assertFalse(TaskLifecycle.shouldRecover(WkTaskStatusEnum.ABNORMAL.getCode())); + assertEquals(WkTaskStatusEnum.PENDING.getCode(), TaskLifecycle.recoveredStatus()); + } +} diff --git a/src/test/java/com/tihai/queue/BoundedPriorityBlockingQueueTest.java b/src/test/java/com/tihai/queue/BoundedPriorityBlockingQueueTest.java new file mode 100644 index 0000000..63d7f50 --- /dev/null +++ b/src/test/java/com/tihai/queue/BoundedPriorityBlockingQueueTest.java @@ -0,0 +1,23 @@ +package com.tihai.queue; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class BoundedPriorityBlockingQueueTest { + + @Test + void enforcesCapacityWhileReturningHigherPriorityItemsFirst() { + BoundedPriorityBlockingQueue queue = new BoundedPriorityBlockingQueue<>(2); + + assertTrue(queue.offer(5)); + assertTrue(queue.offer(1)); + assertFalse(queue.offer(3)); + assertEquals(Integer.valueOf(1), queue.poll()); + assertTrue(queue.offer(3)); + assertEquals(Integer.valueOf(3), queue.poll()); + assertEquals(Integer.valueOf(5), queue.poll()); + } +} diff --git a/src/test/java/com/tihai/utils/CourseUtilSessionTest.java b/src/test/java/com/tihai/utils/CourseUtilSessionTest.java new file mode 100644 index 0000000..bad5272 --- /dev/null +++ b/src/test/java/com/tihai/utils/CourseUtilSessionTest.java @@ -0,0 +1,45 @@ +package com.tihai.utils; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CourseUtilSessionTest { + + @Test + void keepsCookieStateIsolatedBetweenConcurrentTasks() throws Exception { + CourseUtil courseUtil = new CourseUtil(); + ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch firstSessionReady = new CountDownLatch(1); + CountDownLatch secondSessionReady = new CountDownLatch(1); + + try { + Future firstTask = executor.submit(() -> { + courseUtil.setAccount("first-account"); + courseUtil.setCookies("_uid=first-user"); + firstSessionReady.countDown(); + assertTrue(secondSessionReady.await(2, TimeUnit.SECONDS)); + return courseUtil.getValue("_uid"); + }); + Future secondTask = executor.submit(() -> { + assertTrue(firstSessionReady.await(2, TimeUnit.SECONDS)); + courseUtil.setAccount("second-account"); + courseUtil.setCookies("_uid=second-user"); + secondSessionReady.countDown(); + return courseUtil.getValue("_uid"); + }); + + assertEquals("first-user", firstTask.get(2, TimeUnit.SECONDS)); + assertEquals("second-user", secondTask.get(2, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + } +} diff --git a/src/test/java/com/tihai/utils/CredentialCipherTest.java b/src/test/java/com/tihai/utils/CredentialCipherTest.java new file mode 100644 index 0000000..b8f5ce3 --- /dev/null +++ b/src/test/java/com/tihai/utils/CredentialCipherTest.java @@ -0,0 +1,23 @@ +package com.tihai.utils; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CredentialCipherTest { + + @Test + void encryptsSensitiveValuesWithANonDeterministicAuthenticatedCiphertext() { + CredentialCipher cipher = new CredentialCipher("0123456789abcdef0123456789abcdef"); + + String first = cipher.encrypt("sensitive-value"); + String second = cipher.encrypt("sensitive-value"); + + assertTrue(cipher.isEncrypted(first)); + assertFalse(first.equals(second)); + assertEquals("sensitive-value", cipher.decrypt(first)); + assertEquals("legacy-value", cipher.decrypt("legacy-value")); + } +}