优化异步任务架构

applications/tasks/task_config.py

@@ -0,0 +1,198 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class TaskConfig:
+    """任务配置"""
+
+    timeout: int  # 超时时间（秒）
+    max_concurrent: int = 5  # 最大并发数
+    retry_times: int = 0  # 重试次数
+    retryable: bool = True  # 是否可重试
+    alert_on_failure: bool = True  # 失败时是否告警
+
+
+class TaskStatus:
+    """任务状态常量"""
+
+    INIT = 0
+    PROCESSING = 1
+    SUCCESS = 2
+    FAILED = 99
+
+
+class TaskConstants:
+    """任务系统常量"""
+
+    # 默认配置
+    DEFAULT_TIMEOUT = 1800
+    DEFAULT_MAX_CONCURRENT = 5
+    DEFAULT_RETRY_TIMES = 0
+
+    # 数据库表名
+    TASK_TABLE = "long_articles_task_manager"
+
+
+# 所有任务的配置映射
+TASK_CONFIGS = {
+    # 监控类任务
+    "check_kimi_balance": TaskConfig(
+        timeout=20,
+        max_concurrent=1,
+        retry_times=3,
+    ),
+    "get_off_videos": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
+    "check_publish_video_audit_status": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
+    "outside_article_monitor": TaskConfig(
+        timeout=3 * 3600,
+        max_concurrent=2,
+    ),
+    "inner_article_monitor": TaskConfig(
+        timeout=3600,
+        max_concurrent=3,
+    ),
+    "task_processing_monitor": TaskConfig(
+        timeout=300,
+        max_concurrent=1,
+    ),
+    # 爬虫类任务
+    "crawler_toutiao": TaskConfig(
+        timeout=5 * 3600,
+        max_concurrent=3,
+        retry_times=2,
+    ),
+    "crawler_gzh_articles": TaskConfig(
+        timeout=4 * 3600,
+        max_concurrent=3,
+        retry_times=2,
+    ),
+    "crawler_account_manager": TaskConfig(
+        timeout=1800,
+        max_concurrent=5,
+    ),
+    "crawler_detail_analysis": TaskConfig(
+        timeout=3600,
+        max_concurrent=3,
+    ),
+    # 数据处理类任务
+    "daily_publish_articles_recycle": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    "update_root_source_id": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    "recycle_outside_account_articles": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    "update_outside_account_article_root_source_id": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    "fwh_daily_recycle": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    # 算法类任务
+    "article_pool_cold_start": TaskConfig(
+        timeout=4 * 3600,
+        max_concurrent=2,
+        retry_times=1,
+    ),
+    "candidate_account_quality_analysis": TaskConfig(
+        timeout=3600,
+        max_concurrent=3,
+    ),
+    "article_pool_category_generation": TaskConfig(
+        timeout=3600,
+        max_concurrent=3,
+    ),
+    "account_category_analysis": TaskConfig(
+        timeout=3600,
+        max_concurrent=3,
+    ),
+    "update_limited_account_info": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    # LLM 类任务
+    "title_rewrite": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+        retry_times=2,
+    ),
+    "extract_title_features": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+        retry_times=2,
+    ),
+    # 统计分析类任务
+    "update_account_read_rate_avg": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
+    "update_account_read_avg": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
+    "update_account_open_rate_avg": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
+    # 自动化任务
+    "auto_follow_account": TaskConfig(
+        timeout=1800,
+        max_concurrent=2,
+    ),
+    "get_follow_result": TaskConfig(
+        timeout=1800,
+        max_concurrent=2,
+    ),
+    "extract_reply_result": TaskConfig(
+        timeout=1800,
+        max_concurrent=2,
+    ),
+    # 合作账号任务
+    "cooperate_accounts_monitor": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    "cooperate_accounts_detail": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    # 其他任务
+    "mini_program_detail_process": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
+}
+
+
+def get_task_config(task_name: str) -> TaskConfig:
+    """获取任务配置，如果不存在则返回默认配置"""
+    return TASK_CONFIGS.get(
+        task_name,
+        TaskConfig(
+            timeout=TaskConstants.DEFAULT_TIMEOUT,
+            max_concurrent=TaskConstants.DEFAULT_MAX_CONCURRENT,
+            retry_times=TaskConstants.DEFAULT_RETRY_TIMES,
+        ),
+    )
+
+
+__all__ = [
+    "TaskConfig",
+    "TaskStatus",
+    "TaskConstants",
+    "TASK_CONFIGS",
+    "get_task_config",
+]

applications/tasks/task_handler.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Callable, Awaitable, Dict, Any, Optional
 
 from applications.tasks.analysis_task import CrawlerDetailDeal
 from applications.tasks.analysis_task import AccountPositionReadRateAvg
@@ -40,71 +41,119 @@ from applications.tasks.monitor_tasks import OutsideGzhArticlesCollector
 from applications.tasks.monitor_tasks import TaskProcessingMonitor
 from applications.tasks.monitor_tasks import LimitedAccountAnalysisTask
 
-from applications.tasks.task_mapper import TaskMapper
+from applications.tasks.task_config import TaskStatus
+from applications.tasks.task_utils import TaskValidationError
 
 
-class TaskHandler(TaskMapper):
-    def __init__(self, data, log_service, db_client, trace_id):
+_TASK_HANDLER_REGISTRY: Dict[str, Callable] = {}
+
+
+def register(task_name: str):
+    def decorator(func):
+        _TASK_HANDLER_REGISTRY[task_name] = func
+        return func
+
+    return decorator
+
+
+class TaskHandler:
+    """任务处理器基类 - 使用装饰器模式自动注册任务"""
+
+    # 任务注册表
+    _handlers = _TASK_HANDLER_REGISTRY
+
+    def __init__(self, data: dict, log_service, db_client, trace_id: str):
         self.data = data
         self.log_client = log_service
         self.db_client = db_client
         self.trace_id = trace_id
 
-    # ---------- 下面是若干复合任务的局部实现 ----------
+    @classmethod
+    def get_handler(cls, task_name: str) -> Optional[Callable]:
+        """获取任务处理器"""
+        return cls._handlers.get(task_name)
+
+    @classmethod
+    def list_registered_tasks(cls) -> list:
+        """列出所有已注册的任务"""
+        return list(cls._handlers.keys())
+
+    async def _log_task_event(self, event_type: str, **kwargs):
+        """统一的任务日志记录"""
+        log_data = {
+            "timestamp": datetime.now().isoformat(),
+            "trace_id": self.trace_id,
+            "event_type": event_type,
+            "task": self.data.get("task_name"),
+            **kwargs,
+        }
+        await self.log_client.log(contents=log_data)
+
+    # ==================== 监控类任务 ====================
+
+    @register("check_kimi_balance")
     async def _check_kimi_balance_handler(self) -> int:
+        """检查 Kimi 余额"""
         response = await check_kimi_balance()
-        await self.log_client.log(
-            contents={
-                "trace_id": self.trace_id,
-                "task": "check_kimi_balance",
-                "data": response,
-            }
-        )
-        return self.TASK_SUCCESS_STATUS
+        await self._log_task_event("kimi_balance_checked", data=response)
+        return TaskStatus.SUCCESS
 
+    @register("get_off_videos")
     async def _get_off_videos_task_handler(self) -> int:
+        """视频下架任务"""
         sub_task = GetOffVideos(self.db_client, self.log_client, self.trace_id)
         return await sub_task.deal()
 
+    @register("check_publish_video_audit_status")
     async def _check_video_audit_status_handler(self) -> int:
+        """检查视频审核状态"""
         sub_task = CheckVideoAuditStatus(self.db_client, self.log_client, self.trace_id)
         return await sub_task.deal()
 
+    @register("task_processing_monitor")
     async def _task_processing_monitor_handler(self) -> int:
+        """任务处理监控"""
         sub_task = TaskProcessingMonitor(self.db_client)
         await sub_task.deal()
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
+    @register("inner_article_monitor")
     async def _inner_gzh_articles_monitor_handler(self) -> int:
+        """内部公众号文章监控"""
         sub_task = InnerGzhArticlesMonitor(self.db_client)
         return await sub_task.deal()
 
-    async def _title_rewrite_handler(self):
-        sub_task = TitleRewrite(self.db_client, self.log_client, self.trace_id)
-        return await sub_task.deal()
-
-    async def _update_root_source_id_handler(self) -> int:
-        sub_task = UpdateRootSourceIdAndUpdateTimeTask(self.db_client, self.log_client)
-        await sub_task.deal()
-        return self.TASK_SUCCESS_STATUS
-
+    @register("outside_article_monitor")
     async def _outside_monitor_handler(self) -> int:
+        """外部文章监控"""
         collector = OutsideGzhArticlesCollector(self.db_client)
         await collector.deal()
         monitor = OutsideGzhArticlesMonitor(self.db_client)
-        return await monitor.deal()  # 应返回 SUCCESS / FAILED
+        return await monitor.deal()
 
-    async def _recycle_article_data_handler(self) -> int:
-        date_str = self.data.get("date_string") or datetime.now().strftime("%Y-%m-%d")
-        recycle = RecycleDailyPublishArticlesTask(
-            self.db_client, self.log_client, date_str
+    @register("cooperate_accounts_monitor")
+    async def _cooperate_accounts_monitor_handler(self) -> int:
+        """合作账号文章监测"""
+        task = CooperateAccountsMonitorTask(
+            pool=self.db_client, log_client=self.log_client
         )
-        await recycle.deal()
-        check = CheckDailyPublishArticlesTask(self.db_client, self.log_client, date_str)
-        await check.deal()
-        return self.TASK_SUCCESS_STATUS
+        await task.deal(task_name="save_articles")
+        return TaskStatus.SUCCESS
+
+    @register("cooperate_accounts_detail")
+    async def _cooperate_accounts_detail_handler(self) -> int:
+        """合作账号文章详情更新"""
+        task = CooperateAccountsMonitorTask(
+            pool=self.db_client, log_client=self.log_client
+        )
+        await task.deal(task_name="get_detail")
+        return TaskStatus.SUCCESS
 
+    # ==================== 爬虫类任务 ====================
+
+    @register("crawler_toutiao")
     async def _crawler_toutiao_handler(self) -> int:
+        """头条文章/视频抓取"""
         sub_task = CrawlerToutiao(self.db_client, self.log_client, self.trace_id)
         method = self.data.get("method", "account")
         media_type = self.data.get("media_type", "article")
@@ -118,60 +167,17 @@ class TaskHandler(TaskMapper):
             case "search":
                 await sub_task.search_candidate_accounts()
             case _:
-                raise ValueError(f"Unsupported method {method}")
-        return self.TASK_SUCCESS_STATUS
+                raise TaskValidationError(f"Unsupported method: {method}")
 
-    async def _article_pool_cold_start_handler(self) -> int:
-        cold_start = ArticlePoolColdStart(
-            self.db_client, self.log_client, self.trace_id
-        )
-        platform = self.data.get("platform", "weixin")
-        crawler_methods = self.data.get("crawler_methods", [])
-        category_list = self.data.get("category_list", [])
-        strategy = self.data.get("strategy", "strategy_v1")
-        await cold_start.deal(
-            platform=platform,
-            crawl_methods=crawler_methods,
-            category_list=category_list,
-            strategy=strategy,
-        )
-        return self.TASK_SUCCESS_STATUS
-
-    async def _candidate_account_quality_score_handler(self) -> int:
-        task = CandidateAccountQualityScoreRecognizer(
-            self.db_client, self.log_client, self.trace_id
-        )
-        await task.deal()
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    async def _article_pool_category_generation_handler(self) -> int:
-        task = ArticlePoolCategoryGeneration(
-            self.db_client, self.log_client, self.trace_id
-        )
-        limit_num = self.data.get("limit")
-        await task.deal(limit=limit_num)
-        return self.TASK_SUCCESS_STATUS
-
-    async def _crawler_account_manager_handler(self) -> int:
-        platform = self.data.get("platform", "weixin")
-        account_id_list = self.data.get("account_id_list")
-
-        match platform:
-            case "weixin":
-                task = WeixinAccountManager(
-                    self.db_client, self.log_client, self.trace_id
-                )
-            case _:
-                raise ValueError(f"Unsupported platform {platform}")
-
-        await task.deal(platform=platform, account_id_list=account_id_list)
-        return self.TASK_SUCCESS_STATUS
-
-    # 抓取公众号文章
+    @register("crawler_gzh_articles")
     async def _crawler_gzh_article_handler(self) -> int:
+        """抓取公众号文章"""
         account_method = self.data.get("account_method")
         crawl_mode = self.data.get("crawl_mode")
         strategy = self.data.get("strategy")
+
         match crawl_mode:
             case "account":
                 task = CrawlerGzhAccountArticles(
@@ -184,131 +190,220 @@ class TaskHandler(TaskMapper):
                 )
                 await task.deal(strategy)
             case _:
-                raise ValueError(f"Unsupported crawl mode {crawl_mode}")
-        return self.TASK_SUCCESS_STATUS
+                raise TaskValidationError(f"Unsupported crawl mode: {crawl_mode}")
 
-    # 回收服务号文章
-    async def _recycle_fwh_article_handler(self) -> int:
-        task = RecycleFwhDailyPublishArticlesTask(self.db_client, self.log_client)
-        await task.deal()
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 账号品类处理任务
-    async def _account_category_analysis_handler(self) -> int:
-        task = AccountCategoryAnalysis(
-            pool=self.db_client,
-            log_client=self.log_client,
-            trace_id=self.trace_id,
-            data=self.data,
-            date_string=None,
-        )
-        await task.deal()
-        return self.TASK_SUCCESS_STATUS
+    @register("crawler_account_manager")
+    async def _crawler_account_manager_handler(self) -> int:
+        """账号管理任务"""
+        platform = self.data.get("platform", "weixin")
+        account_id_list = self.data.get("account_id_list")
 
-    # 抓取视频/文章详情分析统计
+        match platform:
+            case "weixin":
+                task = WeixinAccountManager(
+                    self.db_client, self.log_client, self.trace_id
+                )
+                await task.deal(platform=platform, account_id_list=account_id_list)
+            case _:
+                raise TaskValidationError(f"Unsupported platform: {platform}")
+
+        return TaskStatus.SUCCESS
+
+    @register("crawler_detail_analysis")
     async def _crawler_article_analysis_handler(self) -> int:
+        """抓取视频/文章详情分析统计"""
         task = CrawlerDetailDeal(pool=self.db_client, trace_id=self.trace_id)
         await task.deal(params=self.data)
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 更新小程序裂变信息
-    async def _mini_program_detail_handler(self) -> int:
-        task = RecycleMiniProgramDetailTask(
-            pool=self.db_client, log_client=self.log_client, trace_id=self.trace_id
+    # ==================== 数据回收类任务 ====================
+
+    @register("daily_publish_articles_recycle")
+    async def _recycle_article_data_handler(self) -> int:
+        """每日发文数据回收"""
+        date_str = self.data.get("date_string") or datetime.now().strftime("%Y-%m-%d")
+        recycle = RecycleDailyPublishArticlesTask(
+            self.db_client, self.log_client, date_str
         )
-        await task.deal(params=self.data)
-        return self.TASK_SUCCESS_STATUS
+        await recycle.deal()
+        check = CheckDailyPublishArticlesTask(self.db_client, self.log_client, date_str)
+        await check.deal()
+        return TaskStatus.SUCCESS
 
-    # 提取标题特征
-    async def _extract_title_features_handler(self) -> int:
-        task = ExtractTitleFeatures(self.db_client, self.log_client, self.trace_id)
-        await task.deal(data=self.data)
-        return self.TASK_SUCCESS_STATUS
+    @register("update_root_source_id")
+    async def _update_root_source_id_handler(self) -> int:
+        """更新 root_source_id"""
+        sub_task = UpdateRootSourceIdAndUpdateTimeTask(self.db_client, self.log_client)
+        await sub_task.deal()
+        return TaskStatus.SUCCESS
 
-    # 回收外部账号文章
+    @register("recycle_outside_account_articles")
     async def _recycle_outside_account_article_handler(self) -> int:
+        """回收外部账号文章"""
         date_str = self.data.get("date_string") or datetime.now().strftime("%Y-%m-%d")
         task = RecycleOutsideAccountArticlesTask(
             pool=self.db_client, log_client=self.log_client, date_string=date_str
         )
         await task.deal()
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 更新外部账号文章的root_source_id和update_time
-    async def _update_outside_account_article_root_source_id_and_update_time_handler(
-        self,
-    ) -> int:
+    @register("update_outside_account_article_root_source_id")
+    async def _update_outside_account_article_root_source_id_handler(self) -> int:
+        """更新外部账号文章的 root_source_id"""
         task = UpdateOutsideRootSourceIdAndUpdateTimeTask(
             pool=self.db_client, log_client=self.log_client
         )
         await task.deal()
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
+
+    @register("fwh_daily_recycle")
+    async def _recycle_fwh_article_handler(self) -> int:
+        """回收服务号文章"""
+        task = RecycleFwhDailyPublishArticlesTask(self.db_client, self.log_client)
+        await task.deal()
+        return TaskStatus.SUCCESS
+
+    # ==================== 算法类任务 ====================
+
+    @register("article_pool_cold_start")
+    async def _article_pool_cold_start_handler(self) -> int:
+        """文章池冷启动"""
+        cold_start = ArticlePoolColdStart(
+            self.db_client, self.log_client, self.trace_id
+        )
+        platform = self.data.get("platform", "weixin")
+        crawler_methods = self.data.get("crawler_methods", [])
+        category_list = self.data.get("category_list", [])
+        strategy = self.data.get("strategy", "strategy_v1")
+
+        await cold_start.deal(
+            platform=platform,
+            crawl_methods=crawler_methods,
+            category_list=category_list,
+            strategy=strategy,
+        )
+        return TaskStatus.SUCCESS
+
+    @register("candidate_account_quality_analysis")
+    async def _candidate_account_quality_score_handler(self) -> int:
+        """候选账号质量分析"""
+        task = CandidateAccountQualityScoreRecognizer(
+            self.db_client, self.log_client, self.trace_id
+        )
+        await task.deal()
+        return TaskStatus.SUCCESS
+
+    @register("article_pool_category_generation")
+    async def _article_pool_category_generation_handler(self) -> int:
+        """文章池品类生成"""
+        task = ArticlePoolCategoryGeneration(
+            self.db_client, self.log_client, self.trace_id
+        )
+        limit_num = self.data.get("limit")
+        await task.deal(limit=limit_num)
+        return TaskStatus.SUCCESS
 
-    # 更新限流账号信息
+    @register("account_category_analysis")
+    async def _account_category_analysis_handler(self) -> int:
+        """账号品类分析"""
+        task = AccountCategoryAnalysis(
+            pool=self.db_client,
+            log_client=self.log_client,
+            trace_id=self.trace_id,
+            data=self.data,
+            date_string=None,
+        )
+        await task.deal()
+        return TaskStatus.SUCCESS
+
+    @register("update_limited_account_info")
     async def _update_limited_account_info_handler(self) -> int:
+        """更新限流账号信息"""
         task = LimitedAccountAnalysisTask(
             pool=self.db_client, log_client=self.log_client
         )
         await task.deal(date_string=self.data.get("date_string"))
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
+
+    # ==================== LLM 类任务 ====================
 
-    # 更新账号阅读率均值
+    @register("title_rewrite")
+    async def _title_rewrite_handler(self) -> int:
+        """标题重写"""
+        sub_task = TitleRewrite(self.db_client, self.log_client, self.trace_id)
+        return await sub_task.deal()
+
+    @register("extract_title_features")
+    async def _extract_title_features_handler(self) -> int:
+        """提取标题特征"""
+        task = ExtractTitleFeatures(self.db_client, self.log_client, self.trace_id)
+        await task.deal(data=self.data)
+        return TaskStatus.SUCCESS
+
+    # ==================== 统计分析类任务 ====================
+
+    @register("update_account_read_rate_avg")
     async def _update_account_read_rate_avg_handler(self) -> int:
+        """更新账号阅读率均值"""
         task = AccountPositionReadRateAvg(
             pool=self.db_client, log_client=self.log_client, trace_id=self.trace_id
         )
         await task.deal(end_date=self.data.get("end_date"))
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 更新账号阅读均值
+    @register("update_account_read_avg")
     async def _update_account_read_avg_handler(self) -> int:
+        """更新账号阅读均值"""
         task = AccountPositionReadAvg(
             pool=self.db_client, log_client=self.log_client, trace_id=self.trace_id
         )
         await task.deal(end_date=self.data.get("end_date"))
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 更新账号打开率均值
+    @register("update_account_open_rate_avg")
     async def _update_account_open_rate_avg_handler(self) -> int:
+        """更新账号打开率均值"""
         task = AccountPositionOpenRateAvg(
             pool=self.db_client, log_client=self.log_client, trace_id=self.trace_id
         )
         await task.deal(date_string=self.data.get("date_string"))
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
+
+    # ==================== 自动化任务 ====================
 
-    # 自动关注公众号账号
+    @register("auto_follow_account")
     async def _auto_follow_account_handler(self) -> int:
+        """自动关注公众号"""
         task = AutoReplyCardsMonitor(pool=self.db_client, log_client=self.log_client)
         await task.deal(task_name="follow_gzh_task")
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 获取自动关注回复
+    @register("get_follow_result")
     async def _get_follow_result_handler(self) -> int:
+        """获取自动关注回复"""
         task = AutoReplyCardsMonitor(pool=self.db_client, log_client=self.log_client)
         await task.deal(task_name="get_auto_reply_task")
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 解析自动回复结果
+    @register("extract_reply_result")
     async def _extract_reply_result_handler(self) -> int:
+        """解析自动回复结果"""
         task = AutoReplyCardsMonitor(pool=self.db_client, log_client=self.log_client)
         await task.deal(task_name="extract_task")
-        return self.TASK_SUCCESS_STATUS
+        return TaskStatus.SUCCESS
 
-    # 定时获取外部文章
-    async def _cooperate_accounts_monitor_handler(self) -> int:
-        task = CooperateAccountsMonitorTask(
-            pool=self.db_client, log_client=self.log_client
-        )
-        await task.deal(task_name="save_articles")
-        return self.TASK_SUCCESS_STATUS
+    # ==================== 其他任务 ====================
 
-    # 定时更新详情
-    async def _cooperate_accounts_detail_handler(self) -> int:
-        task = CooperateAccountsMonitorTask(
-            pool=self.db_client, log_client=self.log_client
+    @register("mini_program_detail_process")
+    async def _mini_program_detail_handler(self) -> int:
+        """更新小程序裂变信息"""
+        task = RecycleMiniProgramDetailTask(
+            pool=self.db_client, log_client=self.log_client, trace_id=self.trace_id
         )
-        await task.deal(task_name="get_detail")
-        return self.TASK_SUCCESS_STATUS
+        await task.deal(params=self.data)
+        return TaskStatus.SUCCESS
 
 
 __all__ = ["TaskHandler"]

applications/tasks/task_scheduler.py

@@ -1,231 +1,448 @@
 import asyncio
 import json
 import time
-import traceback
 from datetime import datetime, timedelta
-from typing import Awaitable, Callable, Dict
+from typing import Optional, Dict, Any, List
 
 from applications.api import feishu_robot
 from applications.utils import task_schedule_response
 from applications.tasks.task_handler import TaskHandler
+from applications.tasks.task_config import (
+    TaskStatus,
+    TaskConstants,
+    get_task_config,
+)
+from applications.tasks.task_utils import (
+    TaskError,
+    TaskValidationError,
+    TaskTimeoutError,
+    TaskConcurrencyError,
+    TaskLockError,
+    TaskUtils,
+)
 
 
 class TaskScheduler(TaskHandler):
-    """统一调度入口：外部只需调用 `await TaskScheduler(data, log_cli, db_cli).deal()`"""
+    """
+    统一任务调度器
 
-    # ---------- 初始化 ----------
-    def __init__(self, data, log_service, db_client, trace_id):
+    使用方法：
+        scheduler = TaskScheduler(data, log_service, db_client, trace_id)
+        result = await scheduler.deal()
+    """
+
+    def __init__(self, data: dict, log_service, db_client, trace_id: str):
         super().__init__(data, log_service, db_client, trace_id)
-        self.data = data
-        self.log_client = log_service
-        self.db_client = db_client
-        self.table = "long_articles_task_manager"
-        self.trace_id = trace_id
+        self.table = TaskUtils.validate_table_name(TaskConstants.TASK_TABLE)
+
+    # ==================== 数据库操作 ====================
 
-    # ---------- 公共数据库工具 ----------
     async def _insert_or_ignore_task(self, task_name: str, date_str: str) -> None:
-        """新建记录（若同键已存在则忽略）"""
-        query = (
-            f"insert ignore into {self.table} "
-            "(date_string, task_name, start_timestamp, task_status, trace_id, data) "
-            "values (%s, %s, %s, %s, %s, %s);"
-        )
+        """新建任务记录（若同键已存在则忽略）"""
+        query = f"""
+            INSERT IGNORE INTO {self.table}
+            (date_string, task_name, start_timestamp, task_status, trace_id, data)
+            VALUES (%s, %s, %s, %s, %s, %s)
+        """
         await self.db_client.async_save(
             query=query,
             params=(
                 date_str,
                 task_name,
                 int(time.time()),
-                self.TASK_INIT_STATUS,
+                TaskStatus.INIT,
                 self.trace_id,
                 json.dumps(self.data, ensure_ascii=False),
             ),
         )
 
     async def _try_lock_task(self) -> bool:
-        """一次 UPDATE 抢锁；返回 True 表示成功上锁"""
-        query = (
-            f"update {self.table} "
-            "set task_status = %s "
-            "where trace_id = %s  and task_status = %s;"
-        )
-        res = await self.db_client.async_save(
+        """
+        尝试获取任务锁（CAS 操作）
+        返回 True 表示成功获取锁
+        """
+        query = f"""
+            UPDATE {self.table}
+            SET task_status = %s
+            WHERE trace_id = %s AND task_status = %s
+        """
+        result = await self.db_client.async_save(
             query=query,
-            params=(
-                self.TASK_PROCESSING_STATUS,
-                self.trace_id,
-                self.TASK_INIT_STATUS,
-            ),
+            params=(TaskStatus.PROCESSING, self.trace_id, TaskStatus.INIT),
         )
-        return True if res else False
+        return bool(result)
 
     async def _release_task(self, status: int) -> None:
-        query = (
-            f"update {self.table} set task_status=%s, finish_timestamp=%s "
-            "where trace_id=%s and task_status=%s;"
-        )
+        """释放任务锁并更新状态"""
+        query = f"""
+            UPDATE {self.table}
+            SET task_status = %s, finish_timestamp = %s
+            WHERE trace_id = %s AND task_status = %s
+        """
         await self.db_client.async_save(
             query=query,
             params=(
                 status,
                 int(time.time()),
                 self.trace_id,
-                self.TASK_PROCESSING_STATUS,
+                TaskStatus.PROCESSING,
             ),
         )
 
-    async def _is_processing_overtime(self, task_name) -> bool:
-        """检测在处理任务是否超时，或者超过最大并行数，若超时会发飞书告警"""
-        query = f"select trace_id from {self.table} where task_status = %s and task_name = %s;"
+    async def _get_processing_tasks(self, task_name: str) -> List[Dict[str, Any]]:
+        """获取正在处理中的任务列表"""
+        query = f"""
+            SELECT trace_id, start_timestamp, data
+            FROM {self.table}
+            WHERE task_status = %s AND task_name = %s
+        """
         rows = await self.db_client.async_fetch(
-            query=query, params=(self.TASK_PROCESSING_STATUS, task_name)
+            query=query,
+            params=(TaskStatus.PROCESSING, task_name),
         )
-        if not rows:
-            return False
+        return rows or []
+
+    # ==================== 任务检查 ====================
+
+    async def _check_task_concurrency_and_timeout(self, task_name: str) -> None:
+        """
+        检查任务并发数和超时情况
+
+        优化点：
+        1. 真正检查任务是否超时（基于时间）
+        2. 分别处理超时和并发限制
+        3. 可选择自动释放超时任务
+
+        Raises:
+            TaskTimeoutError: 发现超时任务
+            TaskConcurrencyError: 超过并发限制
+        """
+        processing_tasks = await self._get_processing_tasks(task_name)
+
+        if not processing_tasks:
+            return
+
+        config = get_task_config(task_name)
+        current_time = int(time.time())
+
+        # 检查超时任务
+        timeout_tasks = [
+            task
+            for task in processing_tasks
+            if current_time - task["start_timestamp"] > config.timeout
+        ]
+
+        if timeout_tasks:
+            await self._log_task_event(
+                "task_timeout_detected",
+                task_name=task_name,
+                timeout_count=len(timeout_tasks),
+                timeout_tasks=[t["trace_id"] for t in timeout_tasks],
+            )
 
-        processing_task_num = len(rows)
-        if processing_task_num >= self.get_task_config(task_name).get(
-            "task_max_num", self.TASK_MAX_NUM
-        ):
             await feishu_robot.bot(
-                title=f"multi {task_name} is processing ",
-                detail={"detail": rows},
+                title=f"Task Timeout Alert: {task_name}",
+                detail={
+                    "task_name": task_name,
+                    "timeout_count": len(timeout_tasks),
+                    "timeout_threshold": config.timeout,
+                    "timeout_tasks": [
+                        {
+                            "trace_id": t["trace_id"],
+                            "running_time": current_time - t["start_timestamp"],
+                        }
+                        for t in timeout_tasks
+                    ],
+                },
             )
-            return True
 
-        return False
+            # 可选：自动释放超时任务（需要谨慎使用）
+            for task in timeout_tasks:
+                await self._force_release_task(task["trace_id"], TaskStatus.FAILED)
 
-    async def _run_with_guard(
-        self, task_name: str, date_str: str, task_coro: Callable[[], Awaitable[int]]
-    ):
-        """公共：检查、建记录、抢锁、后台运行"""
-        # 1. 超时检测
-        if await self._is_processing_overtime(task_name):
-            return await task_schedule_response.fail_response(
-                "5005", "muti tasks with same task_name is processing"
+        # 检查并发限制（排除超时任务）
+        active_tasks = [
+            task
+            for task in processing_tasks
+            if current_time - task["start_timestamp"] <= config.timeout
+        ]
+
+        if len(active_tasks) >= config.max_concurrent:
+            await self._log_task_event(
+                "task_concurrency_limit",
+                task_name=task_name,
+                current_count=len(active_tasks),
+                max_concurrent=config.max_concurrent,
+            )
+
+            await feishu_robot.bot(
+                title=f"Task Concurrency Limit: {task_name}",
+                detail={
+                    "task_name": task_name,
+                    "current_count": len(active_tasks),
+                    "max_concurrent": config.max_concurrent,
+                    "active_tasks": [t["trace_id"] for t in active_tasks],
+                },
             )
 
-        # 2. 记录并尝试抢锁
+            raise TaskConcurrencyError(
+                f"Task {task_name} has reached max concurrency limit "
+                f"({len(active_tasks)}/{config.max_concurrent})",
+                task_name=task_name,
+            )
+
+    # ==================== 任务执行 ====================
+
+    async def _run_with_guard(
+        self,
+        task_name: str,
+        date_str: str,
+        task_handler,
+    ) -> dict:
+        """
+        带保护的任务执行
+
+        优化点：
+        1. 更好的错误处理和重试机制
+        2. 统一的日志记录
+        3. 详细的错误信息
+        """
+        # 1. 检查并发和超时
+        try:
+            await self._check_task_concurrency_and_timeout(task_name)
+        except TaskConcurrencyError as e:
+            return await task_schedule_response.fail_response("5005", str(e))
+
+        # 2. 创建任务记录并尝试获取锁
         await self._insert_or_ignore_task(task_name, date_str)
+
         if not await self._try_lock_task():
             return await task_schedule_response.fail_response(
-                "5001", "task is processing"
+                "5001", "Task is already processing"
             )
 
-        # 3. 真正执行任务 —— 使用后台协程保证不阻塞调度入口
-        async def _wrapper():
-            status = self.TASK_FAILED_STATUS
+        # 3. 后台执行任务
+        async def _task_wrapper():
+            """任务执行包装器 - 处理错误和重试"""
+            status = TaskStatus.FAILED
+            retry_count = 0
+            config = get_task_config(task_name)
+            start_time = time.time()
+
             try:
-                status = await task_coro()
+                await self._log_task_event("task_started", task_name=task_name)
+
+                # 执行任务
+                status = await task_handler()
+
+                duration = time.time() - start_time
+                await self._log_task_event(
+                    "task_completed",
+                    task_name=task_name,
+                    status=status,
+                    duration=duration,
+                )
+
+            except TaskError as e:
+                # 已知的任务错误
+                duration = time.time() - start_time
+                error_detail = TaskUtils.format_error_detail(e)
+
+                await self._log_task_event(
+                    "task_failed",
+                    task_name=task_name,
+                    error=error_detail,
+                    duration=duration,
+                    retry_count=retry_count,
+                )
+
+                # 根据错误类型决定是否告警
+                if config.alert_on_failure:
+                    await feishu_robot.bot(
+                        title=f"Task Failed: {task_name}",
+                        detail={
+                            "task_name": task_name,
+                            "trace_id": self.trace_id,
+                            "error": error_detail,
+                            "duration": duration,
+                            "retryable": e.retryable,
+                        },
+                    )
+
+                # TODO: 实现重试逻辑
+                # if e.retryable and retry_count < config.retry_times:
+                #     await self._schedule_retry(task_name, retry_count + 1)
+
             except Exception as e:
-                await self.log_client.log(
-                    contents={
-                        "trace_id": self.trace_id,
-                        "function": "cor_wrapper",
-                        "task": task_name,
-                        "error": str(e),
-                    }
+                # 未知错误
+                duration = time.time() - start_time
+                error_detail = TaskUtils.format_error_detail(e)
+
+                await self._log_task_event(
+                    "task_error",
+                    task_name=task_name,
+                    error=error_detail,
+                    duration=duration,
                 )
+
                 await feishu_robot.bot(
-                    title=f"{task_name} is failed",
+                    title=f"Task Error: {task_name}",
                     detail={
-                        "task": task_name,
-                        "err": str(e),
-                        "traceback": traceback.format_exc(),
+                        "task_name": task_name,
+                        "trace_id": self.trace_id,
+                        "error": error_detail,
+                        "duration": duration,
                     },
                 )
+
             finally:
                 await self._release_task(status)
 
-        asyncio.create_task(_wrapper(), name=task_name)
+        # 创建后台任务
+        asyncio.create_task(_task_wrapper(), name=f"{task_name}_{self.trace_id}")
+
         return await task_schedule_response.success_response(
             task_name=task_name,
-            data={"code": 0, "message": "task started", "trace_id": self.trace_id},
+            data={
+                "code": 0,
+                "message": "Task started successfully",
+                "trace_id": self.trace_id,
+            },
         )
 
-    # ---------- 主入口 ----------
-    async def deal(self):
-        task_name: str | None = self.data.get("task_name")
+    # ==================== 任务管理接口 ====================
+
+    async def get_task_status(
+        self, trace_id: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
+        """
+        查询任务状态
+
+        Args:
+            trace_id: 任务追踪 ID，默认使用当前实例的 trace_id
+
+        Returns:
+            任务信息字典，如果不存在返回 None
+        """
+        trace_id = trace_id or self.trace_id
+        query = f"SELECT * FROM {self.table} WHERE trace_id = %s"
+        result = await self.db_client.async_fetch_one(query, (trace_id,))
+        return result
+
+    async def cancel_task(self, trace_id: Optional[str] = None) -> bool:
+        """
+        取消任务（将状态设置为失败）
+
+        Args:
+            trace_id: 任务追踪 ID，默认使用当前实例的 trace_id
+
+        Returns:
+            是否成功取消
+        """
+        trace_id = trace_id or self.trace_id
+        query = f"""
+            UPDATE {self.table}
+            SET task_status = %s, finish_timestamp = %s
+            WHERE trace_id = %s AND task_status IN (%s, %s)
+        """
+        result = await self.db_client.async_save(
+            query,
+            (
+                TaskStatus.FAILED,
+                int(time.time()),
+                trace_id,
+                TaskStatus.INIT,
+                TaskStatus.PROCESSING,
+            ),
+        )
+
+        if result:
+            await self._log_task_event("task_cancelled", trace_id=trace_id)
+
+        return bool(result)
+
+    async def retry_task(self, trace_id: Optional[str] = None) -> bool:
+        """
+        重试任务（将状态重置为初始化）
+
+        Args:
+            trace_id: 任务追踪 ID，默认使用当前实例的 trace_id
+
+        Returns:
+            是否成功重置
+        """
+        trace_id = trace_id or self.trace_id
+        query = f"""
+            UPDATE {self.table}
+            SET task_status = %s, start_timestamp = %s, finish_timestamp = NULL
+            WHERE trace_id = %s
+        """
+        result = await self.db_client.async_save(
+            query,
+            (TaskStatus.INIT, int(time.time()), trace_id),
+        )
+
+        if result:
+            await self._log_task_event("task_retried", trace_id=trace_id)
+
+        return bool(result)
+
+    async def _force_release_task(self, trace_id: str, status: int) -> None:
+        """强制释放任务（用于超时任务清理）"""
+        query = f"""
+            UPDATE {self.table}
+            SET task_status = %s, finish_timestamp = %s
+            WHERE trace_id = %s
+        """
+        await self.db_client.async_save(
+            query,
+            (status, int(time.time()), trace_id),
+        )
+        await self._log_task_event(
+            "task_force_released", trace_id=trace_id, status=status
+        )
+
+    # ==================== 主入口 ====================
+
+    async def deal(self) -> dict:
+        """
+        任务调度主入口
+
+        Returns:
+            调度结果字典
+        """
+        # 验证任务名
+        task_name = self.data.get("task_name")
         if not task_name:
             return await task_schedule_response.fail_response(
-                "4003", "task_name must be input"
+                "4003", "task_name is required"
             )
 
+        try:
+            task_name = TaskUtils.validate_task_name(task_name)
+        except TaskValidationError as e:
+            return await task_schedule_response.fail_response("4003", str(e))
+
+        # 获取日期
         date_str = self.data.get("date_string") or (
             datetime.utcnow() + timedelta(hours=8)
         ).strftime("%Y-%m-%d")
 
-        # === 所有任务在此注册：映射到一个返回 int 状态码的异步函数 ===
-        handlers: Dict[str, Callable[[], Awaitable[int]]] = {
-            # 校验kimi余额
-            "check_kimi_balance": self._check_kimi_balance_handler,
-            # 长文视频发布之后，三天后下架
-            "get_off_videos": self._get_off_videos_task_handler,
-            # 长文视频发布之后，三天内保持视频可见状态
-            "check_publish_video_audit_status": self._check_video_audit_status_handler,
-            # 外部服务号发文监测
-            "outside_article_monitor": self._outside_monitor_handler,
-            # 站内发文监测
-            "inner_article_monitor": self._inner_gzh_articles_monitor_handler,
-            # 标题重写（代测试）
-            "title_rewrite": self._title_rewrite_handler,
-            # 每日发文数据回收
-            "daily_publish_articles_recycle": self._recycle_article_data_handler,
-            # 每日发文更新root_source_id
-            "update_root_source_id": self._update_root_source_id_handler,
-            # 头条文章，视频抓取
-            "crawler_toutiao": self._crawler_toutiao_handler,
-            # 文章池冷启动发布
-            "article_pool_cold_start": self._article_pool_cold_start_handler,
-            # 任务超时监控
-            "task_processing_monitor": self._task_processing_monitor_handler,
-            # 候选账号质量分析
-            "candidate_account_quality_analysis": self._candidate_account_quality_score_handler,
-            # 文章内容池--标题品类处理
-            "article_pool_category_generation": self._article_pool_category_generation_handler,
-            # 抓取账号管理
-            "crawler_account_manager": self._crawler_account_manager_handler,
-            # 微信公众号文章抓取
-            "crawler_gzh_articles": self._crawler_gzh_article_handler,
-            # 服务号发文回收
-            "fwh_daily_recycle": self._recycle_fwh_article_handler,
-            # 发文账号品类分析
-            "account_category_analysis": self._account_category_analysis_handler,
-            # 抓取 文章/视频 数量分析
-            "crawler_detail_analysis": self._crawler_article_analysis_handler,
-            # 小程序裂变信息处理
-            "mini_program_detail_process": self._mini_program_detail_handler,
-            # 提取标题特征
-            "extract_title_features": self._extract_title_features_handler,
-            # 回收外部文章
-            "recycle_outside_account_articles": self._recycle_outside_account_article_handler,
-            # 更新外部账号文章的root_source_id和update_time
-            "update_outside_account_article_root_source_id": self._update_outside_account_article_root_source_id_and_update_time_handler,
-            # 更新限流账号信息
-            "update_limited_account_info": self._update_limited_account_info_handler,
-            # 更新账号阅读率均值
-            "update_account_read_rate_avg": self._update_account_read_rate_avg_handler,
-            # 更新账号阅读均值
-            "update_account_read_avg": self._update_account_read_avg_handler,
-            # 更新账号打开率均值
-            "update_account_open_rate_avg": self._update_account_open_rate_avg_handler,
-            # 自动关注公众号账号
-            "auto_follow_account": self._auto_follow_account_handler,
-            # 获取自动关注回复
-            "get_follow_result": self._get_follow_result_handler,
-            # 解析自动回复结果
-            "extract_reply_result": self._extract_reply_result_handler,
-            # 合作账号文章监测
-            "cooperate_accounts_monitor": self._cooperate_accounts_monitor_handler,
-            # 合作账号文章详情更新
-            "cooperate_accounts_detail": self._cooperate_accounts_detail_handler,
-        }
-
-        if task_name not in handlers:
+        # 获取任务处理器
+        handler = self.get_handler(task_name)
+        if not handler:
             return await task_schedule_response.fail_response(
-                "4001", "wrong task name input"
+                "4001",
+                f"Unknown task: {task_name}. "
+                f"Available tasks: {', '.join(self.list_registered_tasks())}",
             )
-        return await self._run_with_guard(task_name, date_str, handlers[task_name])
+
+        # 执行任务
+        return await self._run_with_guard(
+            task_name,
+            date_str,
+            lambda: handler(self),
+        )
 
 
 __all__ = ["TaskScheduler"]

applications/tasks/task_utils.py

@@ -0,0 +1,88 @@
+"""
+任务系统异常定义和工具类
+"""
+
+import re
+from typing import Optional
+
+
+class TaskError(Exception):
+    """任务错误基类"""
+
+    def __init__(
+        self, message: str, retryable: bool = True, task_name: Optional[str] = None
+    ):
+        self.message = message
+        self.retryable = retryable
+        self.task_name = task_name
+        super().__init__(message)
+
+
+class TaskValidationError(TaskError):
+    """任务验证错误（不可重试）"""
+
+    def __init__(self, message: str, task_name: Optional[str] = None):
+        super().__init__(message, retryable=False, task_name=task_name)
+
+
+class TaskTimeoutError(TaskError):
+    """任务超时错误（可重试）"""
+
+    def __init__(self, message: str, task_name: Optional[str] = None):
+        super().__init__(message, retryable=True, task_name=task_name)
+
+
+class TaskConcurrencyError(TaskError):
+    """任务并发限制错误（不可重试）"""
+
+    def __init__(self, message: str, task_name: Optional[str] = None):
+        super().__init__(message, retryable=False, task_name=task_name)
+
+
+class TaskLockError(TaskError):
+    """任务锁获取失败（不可重试）"""
+
+    def __init__(self, message: str, task_name: Optional[str] = None):
+        super().__init__(message, retryable=False, task_name=task_name)
+
+
+class TaskUtils:
+    """任务工具类"""
+
+    @staticmethod
+    def validate_table_name(table: str) -> str:
+        """验证表名只包含安全字符"""
+        if not re.match(r"^[a-zA-Z0-9_]+$", table):
+            raise ValueError(f"Invalid table name: {table}")
+        return table
+
+    @staticmethod
+    def validate_task_name(task_name: str) -> str:
+        """验证任务名"""
+        if not task_name or not isinstance(task_name, str):
+            raise TaskValidationError("task_name must be a non-empty string")
+        if not re.match(r"^[a-zA-Z0-9_]+$", task_name):
+            raise TaskValidationError(f"Invalid task_name format: {task_name}")
+        return task_name
+
+    @staticmethod
+    def format_error_detail(error: Exception) -> dict:
+        """格式化错误详情"""
+        import traceback
+
+        return {
+            "error_type": type(error).__name__,
+            "error_message": str(error),
+            "traceback": traceback.format_exc(),
+            "retryable": getattr(error, "retryable", False),
+        }
+
+
+__all__ = [
+    "TaskError",
+    "TaskValidationError",
+    "TaskTimeoutError",
+    "TaskConcurrencyError",
+    "TaskLockError",
+    "TaskUtils",
+]

ui/src/api/task.ts

@@ -1,48 +1,47 @@
-// src/api/task.ts
-import axios from "axios";
+import axios from 'axios'
+import { API_CONFIG, API_ENDPOINTS } from '../config/api'
 
 const api = axios.create({
-  baseURL: import.meta.env.VITE_API_BASE || "/", // 例如 http://localhost:6060
-  timeout: 15000,
-});
+  baseURL: API_CONFIG.BASE_URL,
+  timeout: API_CONFIG.TIMEOUT,
+})
 
 export interface TaskItem {
-  id: number;
-  date_string: string | null;
-  task_name: string | null;
-  task_status: number;
-  start_timestamp: number | null;
-  finish_timestamp: number | null;
-  trace_id: string | null;
-  data: string | null;
-  status_text?: string;
-  data_json?: any;
+  id: number
+  date_string: string | null
+  task_name: string | null
+  task_status: number
+  start_timestamp: number | null
+  finish_timestamp: number | null
+  trace_id: string | null
+  data: string | null
+  status_text?: string
+  data_json?: any
 }
 
 export interface TaskListResp {
-  items: TaskItem[];
-  total: number;
-  page: number;
-  page_size: number;
+  items: TaskItem[]
+  total: number
+  page: number
+  page_size: number
 }
 
-export async function fetchTasks(params: Record<string, any>) {
-  const { data } = await api.post<TaskListResp>("/api/tasks", params);
-  return data;
+export async function fetchTasks(params: Record<string, any>): Promise<TaskListResp> {
+  const { data } = await api.post<TaskListResp>(API_ENDPOINTS.TASKS, params)
+  return data
 }
 
-
-export async function fetchTaskDetail(id: number) {
-  const { data } = await api.get<TaskItem>(`/api/tasks/${id}`);
-  return data;
+export async function fetchTaskDetail(id: number): Promise<TaskItem> {
+  const { data } = await api.get<TaskItem>(API_ENDPOINTS.TASK_DETAIL(id))
+  return data
 }
 
-export async function retryTask(id: number) {
-  const { data } = await api.post(`/api/tasks/${id}/retry`);
-  return data;
+export async function retryTask(id: number): Promise<any> {
+  const { data } = await api.post(API_ENDPOINTS.TASK_RETRY(id))
+  return data
 }
 
-export async function cancelTask(id: number) {
-  const { data } = await api.post(`/api/tasks/${id}/cancel`);
-  return data;
+export async function cancelTask(id: number): Promise<any> {
+  const { data } = await api.post(API_ENDPOINTS.TASK_CANCEL(id))
+  return data
 }

ui/src/views/TaskManager.vue

@@ -1,22 +1,26 @@
 <template>
   <div class="task-page">
-    <!-- 顶部导航栏 -->
     <div class="top-bar">
       <div class="left">
-        <el-button type="info" plain @click="goBack">🏠 返回首页</el-button>
+        <el-button type="info" plain @click="goBack">返回首页</el-button>
       </div>
-      <div class="title">📋 任务列表</div>
+      <div class="title">任务列表</div>
     </div>
 
     <el-card class="task-card">
-      <!-- 工具栏 -->
       <div class="toolbar" ref="toolbarRef">
         <el-form :inline="true" :model="filters" label-width="90px" @keyup.enter.native="onSearch">
           <el-form-item label="ID">
             <el-input v-model.number="filters.id" placeholder="精确 ID" clearable />
           </el-form-item>
           <el-form-item label="日期">
-            <el-date-picker v-model="filters.date_string" type="date" value-format="YYYY-MM-DD" placeholder="YYYY-MM-DD" clearable />
+            <el-date-picker
+              v-model="filters.date_string"
+              type="date"
+              value-format="YYYY-MM-DD"
+              placeholder="YYYY-MM-DD"
+              clearable
+            />
           </el-form-item>
           <el-form-item label="Trace ID">
             <el-input v-model="filters.trace_id" placeholder="模糊匹配" clearable />
@@ -37,7 +41,6 @@
         </el-form>
       </div>
 
-      <!-- 表格 -->
       <div class="table-wrapper">
         <el-table
           :data="rows"
@@ -54,58 +57,72 @@
           <el-table-column prop="task_name" label="任务名称" min-width="200" :show-overflow-tooltip="true" />
           <el-table-column prop="task_status" label="状态" sortable="custom" width="140">
             <template #default="{ row }">
-              <el-tag :type="statusType(row.task_status)">
+              <el-tag :type="getStatusType(row.task_status)">
                 {{ row.status_text || row.task_status }}
               </el-tag>
             </template>
           </el-table-column>
           <el-table-column prop="start_timestamp" label="开始时间" sortable="custom" width="180">
-            <template #default="{ row }">{{ formatTs(row.start_timestamp) }}</template>
+            <template #default="{ row }">{{ formatTimestamp(row.start_timestamp) }}</template>
           </el-table-column>
           <el-table-column prop="finish_timestamp" label="结束时间" sortable="custom" width="180">
-            <template #default="{ row }">{{ formatTs(row.finish_timestamp) }}</template>
+            <template #default="{ row }">{{ formatTimestamp(row.finish_timestamp) }}</template>
           </el-table-column>
           <el-table-column prop="trace_id" label="Trace ID" min-width="220" :show-overflow-tooltip="true" />
           <el-table-column label="操作" fixed="right" width="240">
             <template #default="{ row }">
               <el-button size="small" @click="openDetail(row)">详情</el-button>
-              <el-button size="small" type="warning" @click="onRetry(row)" :disabled="row.task_status === 1">重试</el-button>
-              <el-button size="small" type="danger" @click="onCancel(row)" :disabled="row.task_status === 2">取消</el-button>
+              <el-button
+                size="small"
+                type="warning"
+                @click="onRetry(row)"
+                :disabled="row.task_status === TASK_STATUS.PROCESSING"
+              >
+                重试
+              </el-button>
+              <el-button
+                size="small"
+                type="danger"
+                @click="onCancel(row)"
+                :disabled="row.task_status === TASK_STATUS.COMPLETED"
+              >
+                取消
+              </el-button>
             </template>
           </el-table-column>
         </el-table>
       </div>
 
-      <!-- 分页 -->
       <div class="pager" ref="pagerRef">
         <el-pagination
           background
           layout="total, sizes, prev, pager, next, jumper"
-          :page-sizes="[10, 20, 50, 100]"
+          :page-sizes="PAGE_SIZES"
           :page-size="query.page_size"
           :total="total"
           :current-page="query.page"
-          @current-change="(p:number)=>{query.page=p; load()}"
-          @size-change="(s:number)=>{query.page_size=s; query.page=1; load()}"
+          @current-change="handlePageChange"
+          @size-change="handleSizeChange"
         />
       </div>
     </el-card>
 
-    <!-- 详情 -->
     <el-drawer v-model="detailOpen" size="52%" title="任务详情">
       <div class="detail">
         <el-descriptions :column="2" border>
           <el-descriptions-item label="ID">{{ detail?.id }}</el-descriptions-item>
           <el-descriptions-item label="日期">{{ detail?.date_string }}</el-descriptions-item>
-          <el-descriptions-item label="状态">{{ detail?.task_status }}（{{ detail?.status_text }}）</el-descriptions-item>
+          <el-descriptions-item label="状态">
+            {{ detail?.task_status }}（{{ detail?.status_text }}）
+          </el-descriptions-item>
           <el-descriptions-item label="Trace ID">{{ detail?.trace_id }}</el-descriptions-item>
-          <el-descriptions-item label="开始">{{ formatTs(detail?.start_timestamp) }}</el-descriptions-item>
-          <el-descriptions-item label="结束">{{ formatTs(detail?.finish_timestamp) }}</el-descriptions-item>
+          <el-descriptions-item label="开始">{{ formatTimestamp(detail?.start_timestamp) }}</el-descriptions-item>
+          <el-descriptions-item label="结束">{{ formatTimestamp(detail?.finish_timestamp) }}</el-descriptions-item>
           <el-descriptions-item label="任务名" :span="2">{{ detail?.task_name }}</el-descriptions-item>
         </el-descriptions>
 
         <h4 class="block-title">请求参数（data）</h4>
-        <pre class="code-block">{{ pretty(detail?.data_json || detail?.data) }}</pre>
+        <pre class="code-block">{{ formatJson(detail?.data_json || detail?.data) }}</pre>
       </div>
     </el-drawer>
   </div>
@@ -116,48 +133,70 @@ import { onMounted, onBeforeUnmount, reactive, ref, nextTick } from 'vue'
 import { useRouter } from 'vue-router'
 import { ElMessage, ElMessageBox } from 'element-plus'
 import { fetchTasks, fetchTaskDetail, retryTask, cancelTask, type TaskItem } from '../api/task'
+import {
+  TASK_STATUS_OPTIONS,
+  TASK_STATUS_TYPE_MAP,
+  PAGE_SIZES,
+  DEFAULT_PAGE_SIZE,
+  AUTO_REFRESH_INTERVAL,
+  MIN_TABLE_HEIGHT,
+  TASK_STATUS,
+} from '../config/constants'
 
 const router = useRouter()
 
-const goBack = () => {
-  router.push('/welcome')
-}
+const goBack = () => router.push('/welcome')
+
+const statusOptions = TASK_STATUS_OPTIONS
 
-const statusOptions = [
-  { label: '初始化(0)', value: 0 },
-  { label: '处理中(1)', value: 1 },
-  { label: '完成(2)', value: 2 },
-  { label: '失败(99)', value: 99 },
-]
+interface Filters {
+  id: number | null
+  date_string: string | null
+  trace_id: string
+  task_status: number | null
+}
 
-const filters = reactive<{ id: number | null; date_string: string | null; trace_id: string; task_status: number | null }>({
+const filters = reactive<Filters>({
   id: null,
   date_string: null,
   trace_id: '',
   task_status: null,
 })
 
-const query = reactive({ page: 1, page_size: 20, sort_by: 'id', sort_dir: 'desc' as 'asc' | 'desc' })
+interface Query {
+  page: number
+  page_size: number
+  sort_by: string
+  sort_dir: 'asc' | 'desc'
+}
+
+const query = reactive<Query>({
+  page: 1,
+  page_size: DEFAULT_PAGE_SIZE,
+  sort_by: 'id',
+  sort_dir: 'desc',
+})
+
 const rows = ref<TaskItem[]>([])
 const total = ref(0)
 const loading = ref(false)
 
-/* 自动刷新 */
 let timer: ReturnType<typeof setInterval> | null = null
 const autoRefresh = ref(false)
+
 const toggleAutoRefresh = () => {
   if (autoRefresh.value) {
-    timer = setInterval(load, 5000)
+    timer = setInterval(load, AUTO_REFRESH_INTERVAL)
   } else if (timer) {
     clearInterval(timer)
     timer = null
   }
 }
 
-/* 精准表格高度计算 */
 const toolbarRef = ref<HTMLElement | null>(null)
 const pagerRef = ref<HTMLElement | null>(null)
 const tableHeight = ref(400)
+
 const calcTableHeight = () => {
   const winH = window.innerHeight
   const toolbarH = toolbarRef.value?.offsetHeight ?? 0
@@ -165,10 +204,9 @@ const calcTableHeight = () => {
   const outerPadding = 16 * 2
   const cardPadding = 20 * 2
   const gap = 12 + 12
-  tableHeight.value = Math.max(260, winH - (outerPadding + cardPadding + toolbarH + pagerH + gap))
+  tableHeight.value = Math.max(MIN_TABLE_HEIGHT, winH - (outerPadding + cardPadding + toolbarH + pagerH + gap))
 }
 
-/* 加载列表 */
 const load = async () => {
   loading.value = true
   try {
@@ -176,7 +214,9 @@ const load = async () => {
     if (filters.id) params.id = filters.id
     if (filters.date_string) params.date_string = filters.date_string
     if (filters.trace_id) params.trace_id = filters.trace_id
-    if (filters.task_status !== null && filters.task_status !== undefined) params.task_status = filters.task_status
+    if (filters.task_status !== null && filters.task_status !== undefined) {
+      params.task_status = filters.task_status
+    }
 
     const resp = await fetchTasks(params)
     rows.value = resp.items
@@ -188,40 +228,67 @@ const load = async () => {
   }
 }
 
-/* 交互 */
-const onSearch = () => { query.page = 1; load() }
+const onSearch = () => {
+  query.page = 1
+  load()
+}
+
 const onReset = () => {
   Object.assign(filters, { id: null, date_string: null, trace_id: '', task_status: null })
   query.page = 1
   load()
 }
+
 const onSortChange = (e: any) => {
   query.sort_by = e.prop || 'id'
   query.sort_dir = e.order === 'ascending' ? 'asc' : 'desc'
   load()
 }
 
-/* 工具函数 */
-const statusType = (s: number) => (s === 0 ? 'info' : s === 1 ? 'warning' : s === 2 ? 'success' : s === 99 ? 'danger' : '')
-const formatTs = (ts?: number | null) => (!ts ? '-' : new Date(ts * 1000).toLocaleString())
-const pretty = (v: any) => { try { return typeof v === 'string' ? v : JSON.stringify(v, null, 2) } catch { return String(v) } }
+const handlePageChange = (page: number) => {
+  query.page = page
+  load()
+}
+
+const handleSizeChange = (size: number) => {
+  query.page_size = size
+  query.page = 1
+  load()
+}
+
+const getStatusType = (status: number): string => {
+  return TASK_STATUS_TYPE_MAP[status] || ''
+}
+
+const formatTimestamp = (ts?: number | null): string => {
+  if (!ts) return '-'
+  return new Date(ts * 1000).toLocaleString()
+}
+
+const formatJson = (value: any): string => {
+  try {
+    return typeof value === 'string' ? value : JSON.stringify(value, null, 2)
+  } catch {
+    return String(value)
+  }
+}
 
-/* 详情 */
 const detailOpen = ref(false)
 const detail = ref<TaskItem | null>(null)
+
 const openDetail = async (row: TaskItem) => {
   const d = await fetchTaskDetail(row.id)
   detail.value = { ...row, ...d }
   detailOpen.value = true
 }
 
-/* 行操作 */
 const onRetry = async (row: TaskItem) => {
   await ElMessageBox.confirm(`确认将任务 #${row.id} 置为初始化并重试？`, '重试确认', { type: 'warning' })
   await retryTask(row.id)
   ElMessage.success('已触发重试')
   load()
 }
+
 const onCancel = async (row: TaskItem) => {
   await ElMessageBox.confirm(`确认取消任务 #${row.id}（置为失败）？`, '取消确认', { type: 'warning' })
   await cancelTask(row.id)
@@ -229,11 +296,11 @@ const onCancel = async (row: TaskItem) => {
   load()
 }
 
-/* 生命周期 */
 onMounted(() => {
   load()
   window.addEventListener('resize', calcTableHeight)
 })
+
 onBeforeUnmount(() => {
   window.removeEventListener('resize', calcTableHeight)
   if (timer) clearInterval(timer)

ui/src/views/TokenManager.vue

@@ -1,27 +1,19 @@
 <template>
   <div class="token-page">
     <div class="top-bar">
-      <el-button type="info" plain @click="goBack">🏠 返回首页</el-button>
+      <el-button type="info" plain @click="goBack">返回首页</el-button>
     </div>
 
     <el-card class="token-card">
-      <h2>🔑 更新 Token / Cookie</h2>
+      <h2>更新 Token / Cookie</h2>
 
       <el-form :model="form" label-width="100px">
         <el-form-item label="gh_id">
-          <el-input
-            v-model="form.gzh_id"
-            placeholder="请输入公众号 ID"
-            clearable
-          />
+          <el-input v-model="form.gzh_id" placeholder="请输入公众号 ID" clearable />
         </el-form-item>
 
         <el-form-item label="Token">
-          <el-input
-            v-model="form.token"
-            placeholder="请输入 Token"
-            clearable
-          />
+          <el-input v-model="form.token" placeholder="请输入 Token" clearable />
         </el-form-item>
 
         <el-form-item label="Cookie">
@@ -35,16 +27,11 @@
         </el-form-item>
 
         <el-form-item>
-          <el-button
-            type="primary"
-            @click="onSave"
-            :loading="loading"
-            style="width: 120px"
-          >
-            💾 保存
+          <el-button type="primary" @click="onSave" :loading="loading" style="width: 120px">
+            保存
           </el-button>
           <el-button @click="onReset" :disabled="loading" style="width: 120px">
-            🔄 重置
+            重置
           </el-button>
         </el-form-item>
       </el-form>
@@ -57,10 +44,17 @@ import { ref } from 'vue'
 import { useRouter } from 'vue-router'
 import { ElMessage } from 'element-plus'
 import axios from 'axios'
+import { API_CONFIG, API_ENDPOINTS } from '../config/api'
 
 const router = useRouter()
 
-const form = ref({
+interface TokenForm {
+  gzh_id: string
+  token: string
+  cookie: string
+}
+
+const form = ref<TokenForm>({
   gzh_id: '',
   token: '',
   cookie: '',
@@ -76,17 +70,17 @@ const onSave = async () => {
 
   loading.value = true
   try {
-    const res = await axios.post('http://192.168.142.66:6060/api/save_token', {
+    const res = await axios.post(`${API_CONFIG.BASE_URL}${API_ENDPOINTS.SAVE_TOKEN}`, {
       gzh_id: form.value.gzh_id,
       token: form.value.token,
       cookie: form.value.cookie,
     })
 
-    if (res.data && res.data.success) {
-      ElMessage.success('更新成功 ✅')
+    if (res.data?.success) {
+      ElMessage.success('更新成功')
       form.value = { gzh_id: '', token: '', cookie: '' }
     } else {
-      ElMessage.error(res.data.error || '更新失败 ❌')
+      ElMessage.error(res.data?.error || '更新失败')
     }
   } catch (err) {
     console.error('保存请求出错:', err)

ui/src/views/Welcome.vue

@@ -5,12 +5,12 @@
       <p class="subtitle">高效、简洁地管理与执行任务。</p>
 
       <div class="btn-group">
-        <el-button type="primary" size="large" @click="goTasks">📋 查看任务列表</el-button>
-        <el-button type="success" size="large" @click="goToken">🔑 更新公众号 Token</el-button>
+        <el-button type="primary" size="large" @click="goTasks">查看任务列表</el-button>
+        <el-button type="success" size="large" @click="goToken">更新公众号 Token</el-button>
       </div>
 
       <div class="task-section">
-        <h2> 定时任务管理器</h2>
+        <h2>定时任务管理器</h2>
         <p class="desc">点击下方按钮可手动触发任务执行（需密码验证）</p>
 
         <el-table :data="tasks" border stripe>
@@ -20,7 +20,7 @@
           <el-table-column label="操作" width="150" align="center">
             <template #default="{ row }">
               <el-button size="small" type="primary" @click="runTask(row)">
-                ▶️ 执行任务
+                执行任务
               </el-button>
             </template>
           </el-table-column>
@@ -28,7 +28,7 @@
       </div>
 
       <div class="footer">
-        <p>🎯 让每个任务都更清晰可控，助你轻松掌握全局。</p>
+        <p>让每个任务都更清晰可控，助你轻松掌握全局。</p>
       </div>
     </el-card>
   </div>
@@ -38,45 +38,15 @@
 import { useRouter } from 'vue-router'
 import { ElMessage, ElMessageBox } from 'element-plus'
 import axios from 'axios'
+import { TASK_LIST, EXEC_PASSWORD, type TaskConfig } from '../config/tasks'
 
 const router = useRouter()
+const tasks = TASK_LIST
+
 const goTasks = () => router.push('/tasks')
 const goToken = () => router.push('/token')
 
-// ✅ 执行密码（改成你自己的）
-const EXEC_PASSWORD = 'curry'
-
-// ✅ 所有 curl 任务
-const tasks = [
-  { name: '合作方账号 daily发文抓取', desc: '抓取合作方的 daily 发文', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'cooperate_accounts_monitor' } },
-  { name: '合作方发文详情获取', desc: '获取合作方发文的详情', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'cooperate_accounts_detail' } },
-  { name: '公众号抓取-手动挑号', desc: '公众文章抓取-手动挑号', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'crawler_gzh_articles', account_method: '1030-手动挑号', crawl_mode: 'account', strategy: 'V1' } },
-  { name: '公众号抓取-合作账号', desc: '公众文章抓取-合作挑号', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'crawler_gzh_articles', account_method: 'cooperate_account', crawl_mode: 'account', strategy: 'V1' } },
-  { name: '头条文章抓取', desc: '头条推荐流文章专区', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'crawler_toutiao' } },
-  { name: '公众号文章冷启动(v1)', desc: '文章路冷启动-策略 1', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'article_pool_cold_start', strategy: 'strategy_v1' } },
-  { name: '公众号文章冷启动(v3)', desc: '文章路冷启动-策略 3', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'article_pool_cold_start', strategy: 'strategy_v3' } },
-  { name: '头条文章冷启动', desc: '文章路冷启动-头条平台', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'article_pool_cold_start', platform: 'toutiao', crawler_methods: ['toutiao_account_association'] } },
-  { name: '每日文章回收任务', desc: 'daily_publish_articles_recycle', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'daily_publish_articles_recycle' } },
-  { name: 'root_source_id 更新', desc: 'update_root_source_id', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'update_root_source_id' } },
-  { name: '视频下架任务', desc: 'get_off_videos', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'get_off_videos' } },
-  { name: '内部文章监测', desc: 'inner_article_monitor', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'inner_article_monitor' } },
-  { name: '关注公众号任务', desc: 'follow_gzh_task', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'auto_follow_account' } },
-  { name: '获取自动回复结果', desc: 'get_auto_reply_task', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'get_follow_result' } },
-  { name: '解析自动回复 xml', desc: 'extract_task', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'extract_reply_result' } },
-  { name: '阅读率均值计算', desc: 'update_account_read_rate_avg', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'update_account_read_rate_avg' } },
-  { name: '阅读均值计算', desc: 'update_account_read_avg', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'update_account_read_avg' } },
-  { name: '打开率均值计算', desc: 'update_account_open_rate_avg', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'update_account_open_rate_avg' } },
-  { name: '视频审核状态校验', desc: 'check_publish_video_audit_status', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'check_publish_video_audit_status' } },
-  { name: 'kimi 余额校验', desc: 'check_kimi_balance', url: 'http://192.168.100.31:6060/api/run_task', data: { task_name: 'check_kimi_balance' } },
-  { name: '小程序信息更新', desc: 'mini_program_detail_process', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'mini_program_detail_process' } },
-  { name: '外部账号文章回收', desc: 'recycle_outside_account_articles', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'recycle_outside_account_articles' } },
-  { name: '外部文章 RootID 更新', desc: 'update_outside_account_article_root_source_id', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'update_outside_account_article_root_source_id' } },
-  { name: '限流文章分析', desc: 'update_limited_account_info', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'update_limited_account_info' } },
-  { name: '候选账号质量分析', desc: 'candidate_account_quality_analysis', url: 'http://192.168.142.66:6060/api/run_task', data: { task_name: 'candidate_account_quality_analysis' } },
-]
-
-// ✅ 执行任务（带密码验证）
-const runTask = async (task: any) => {
+const runTask = async (task: TaskConfig) => {
   try {
     const input = await ElMessageBox.prompt('请输入执行密码以确认操作', '安全验证', {
       confirmButtonText: '执行',
@@ -86,13 +56,13 @@ const runTask = async (task: any) => {
     })
 
     if (input.value !== EXEC_PASSWORD) {
-      ElMessage.error('密码错误 ❌')
+      ElMessage.error('密码错误')
       return
     }
 
     const res = await axios.post(task.url, task.data)
     if (res.data?.success) {
-      ElMessage.success(`✅ ${task.name} 执行成功`)
+      ElMessage.success(`${task.name} 执行成功`)
     } else {
       ElMessage.success(`${task.name} 已触发`)
     }