优雅的 Ctrl C 退出

ui/backend/main.py

@@ -28,16 +28,9 @@ async def lifespan(app: FastAPI):
     """应用生命周期管理"""
     tasks = await startup()
     yield
+    logger.info(f"quite {tasks}")
     for name,model in celery_worker.workers_model.items():
         await celery_worker.stop_worker(name)
-    # 先取消所有任务再清理进程
-    for task in tasks:
-        task.cancel()
-        try:
-            await task
-        except asyncio.CancelledError:
-            pass
-    await process_manager.cleanup()
 
 # 创建 FastAPI 应用实例
 app = FastAPI(

ui/backend/routers/worker.py

@@ -207,6 +207,7 @@ async def delete_keyword(keyword: str):
         # 检查响应状态码，如果请求失败则抛出异常
         response.raise_for_status()
         ret = response.json()
+        logger.info(f"response {ret}")
     if ret.get("err") == 0:
         return JSONResponse(
             status_code=200,

ui/backend/src/services/celery_worker.py

@@ -52,8 +52,8 @@ class CrawlTaskConfig(BaseModel):
     queue_name: str = 'crawl_queue'  # 新增队列名称参数
 
 class ConvertTaskParams(BaseModel):
-    result_ids: List[str] = Field(..., min_length=1)
-    batch_size: Optional[int] = Field(10, gt=0)
+    result_ids: List[str] = Field(..., min_length=0)
+    batch_size: Optional[int] = Field(0, gt=-1)
 
 class WorkerModel(BaseModel):
     name: str
@@ -115,6 +115,7 @@ class CeleryWorker:
                 continue
 
     async def run(self):
+        logger.info(f"config {config.model_dump_json(indent=2)}")
         python_exe = self.python_exe
         logger.info(f"WORKER_DIR_BASE {WORKER_DIR_BASE}")
         # return
@@ -169,7 +170,7 @@ class CeleryWorker:
         if not worker_model:
             raise ValueError(f"Invalid worker name: {name}")
         # 检查进程是否真实存在
-        if psutil.pid_exists(worker_model.pid):
+        if worker_model.pid and psutil.pid_exists(worker_model.pid):
             proc = await asyncio.to_thread(psutil.Process, worker_model.pid)  # 异步获取进程对象
             await asyncio.to_thread(proc.terminate)              # 异步发送终止信号
             try:
@@ -303,7 +304,7 @@ class CeleryWorker:
         TASK_MAP = {
             'search': 'search_worker.search_all_uncompleted_keywords',
             'crawl': 'crawl_worker.crawl_all_unprocessed_pages',
-            'convert': 'html_convert_tasks.convert_all_unprocessed_results'
+            'convert': 'html_convert_tasks.convert_all_results'
         }
 
         if worker_name not in TASK_MAP:
@@ -313,6 +314,8 @@ class CeleryWorker:
             task_model = pre_task_model.config
         elif worker_name == 'crawl':
             task_model = self._prepare_crawl_task(data, select_proxy)
+        elif worker_name == 'convert':
+            task_model = ConvertTaskParams(**data)
         logger.info(f"{task_model}")
         self.celery_app.send_task(
                 name=TASK_MAP[worker_name],

ui/backend/utils/process_mgr.py

@@ -28,8 +28,8 @@ class ProcessManager:
         if platform.system() == 'Windows':
             self._create_windows_job()
 
-        self._register_signal_handlers()
-        atexit.register(self._sync_cleanup)
+        # self._register_signal_handlers()
+        # atexit.register(self._sync_cleanup)
 
     def _create_windows_job(self):
         """创建Windows作业对象用于进程生命周期管理"""