fix: 合规审查卡死修复 + 小红书复制格式 + today-only过滤

platform/backend/app/api/system.py

@@ -1,5 +1,6 @@
 import logging
 import subprocess
+import re
 from fastapi import APIRouter, HTTPException, Depends, Body
 from sqlalchemy.orm import Session
 from sqlalchemy import func
@@ -7,7 +8,6 @@ from datetime import datetime, date, timezone, timedelta
 from pathlib import Path
 from typing import Dict, Any, List, Optional
 import os
-import os
 import json
 from ..database import get_db
 from ..models import Topic, Article, TaskConfig, TaskLog
@@ -29,14 +29,48 @@ router = APIRouter(prefix="/api/system", tags=["system"])
 
 _active_monitors: Dict[int, dict] = {}
 
+def _parse_stdout_result(stdout: str, module_id: str) -> dict:
+    """从脚本 stdout 中提取关键指标存入 result_data"""
+    result = {}
+    if module_id == "scheduled_collect":
+        m = re.search(r'SUCCESS:\s*Collected\s*(\d+)\s*cases?\s*and\s*(\d+)\s*topics?', stdout)
+        if m:
+            result["cases_saved"] = int(m.group(1))
+            result["topics_saved"] = int(m.group(2))
+        m = re.search(r'WARNING:\s*(.*)', stdout)
+        if m:
+            result["warning"] = m.group(1).strip()
+    elif module_id in ("scheduled_generate", "scheduled_optimize"):
+        # 尝试解析 JSON 输出
+        try:
+            data = json.loads(stdout.strip())
+            if isinstance(data, dict):
+                for k, v in data.items():
+                    if isinstance(v, (str, int, float, bool)):
+                        result[k] = v
+        except (json.JSONDecodeError, ValueError):
+            pass
+        if not result:
+            m = re.search(r'SUCCESS:\s*(.*)', stdout)
+            if m:
+                result["summary"] = m.group(1).strip()
+    return result
+
 def _monitor_subprocess(log_id: int, proc, module_id: str, task_name: str, db_session_factory):
     """阻塞等待子进程退出（最长 1800s），完成后更新 task_logs"""
+    stdout, stderr = "", ""
     try:
-        returncode = proc.wait(timeout=1800)
+        stdout, stderr = proc.communicate(timeout=1800)
+        returncode = proc.returncode
     except subprocess.TimeoutExpired:
         proc.kill()
+        stdout, stderr = proc.communicate(timeout=5)
         returncode = -1
         logger.warning("Subprocess %s (pid=%s) killed after 1800s timeout", module_id, proc.pid)
+    except Exception as e:
+        logger.warning("Subprocess %s monitor error: %s", module_id, e)
+        returncode = -1
+        stdout = stderr = ""
     finished_at = datetime.now(timezone.utc)
     try:
         db = db_session_factory()
@@ -44,8 +78,17 @@ def _monitor_subprocess(log_id: int, proc, module_id: str, task_name: str, db_se
         if log:
             log.status = "success" if returncode == 0 else "failed"
             log.finished_at = finished_at
+            msg = f"{task_name} {'完成' if returncode == 0 else '失败'}"
+            log.message = msg
+            if returncode != 0:
+                log.error_trace = (stderr or stdout or "")[:2000]
             if log.started_at:
                 log.duration = int((finished_at - log.started_at).total_seconds())
+            # 解析 stdout 提取结果指标
+            if returncode == 0 and stdout:
+                parsed = _parse_stdout_result(stdout, module_id)
+                if parsed:
+                    log.result_data = parsed
             db.commit()
         db.close()
     except Exception as e:
@@ -97,12 +140,13 @@ def trigger_generation(topic_id: Optional[str] = None, db: Session = Depends(get
         db.add(log)
         db.commit()
         log_id = log.id
-        proc_info = result.get("proc") or result
-        proc = _generator_running.get("generator", {}).get("process") if "pid" in result else None
+        proc = result.get("_proc")
         if proc:
             t = threading.Thread(target=_monitor_subprocess, args=(log_id, proc, "scheduled_generate", "🤖 内容创作", SessionLocal), daemon=True)
             t.start()
         return {"message": "内容创作已后台启动", "pid": result.get("pid"), "log_id": log_id}
+    except RuntimeError as e:
+        raise HTTPException(status_code=409, detail=str(e))
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
@@ -123,11 +167,13 @@ def trigger_collection(db: Session = Depends(get_db), current_user=Depends(get_c
         db.add(log)
         db.commit()
         log_id = log.id
-        proc = _collector_running.get("collector", {}).get("process") if "pid" in result else None
+        proc = result.get("_proc")
         if proc:
             t = threading.Thread(target=_monitor_subprocess, args=(log_id, proc, "scheduled_collect", "📡 内容采集", SessionLocal), daemon=True)
             t.start()
         return {"message": "内容采集已后台启动", "result": result, "log_id": log_id}
+    except RuntimeError as e:
+        raise HTTPException(status_code=409, detail=str(e))
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
@@ -139,7 +185,8 @@ def collection_status():
     return status
 
 @router.post("/review/run")
-def trigger_review(topic_ids: Optional[List[str]] = None, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
+def trigger_review(data: Dict = Body({}), db: Session = Depends(get_db), current_user=Depends(get_current_user)):
+    topic_ids = data.get("topic_ids") if isinstance(data, dict) else None
     try:
         result = run_optimizer(topic_ids)
         from ..database import SessionLocal
@@ -147,11 +194,13 @@ def trigger_review(topic_ids: Optional[List[str]] = None, db: Session = Depends(
         db.add(log)
         db.commit()
         log_id = log.id
-        proc = _optimizer_running.get("optimizer", {}).get("process") if "pid" in result else None
+        proc = result.get("_proc")
         if proc:
             t = threading.Thread(target=_monitor_subprocess, args=(log_id, proc, "scheduled_optimize", "🔍 合规审查", SessionLocal), daemon=True)
             t.start()
         return {"message": "合规审查已后台启动", "pid": result.get("pid"), "log_id": log_id}
+    except RuntimeError as e:
+        raise HTTPException(status_code=409, detail=str(e))
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
@@ -207,7 +256,7 @@ def trigger_optimize_sources():
         from ..core.scheduler import scheduler
         def _bg():
             try:
-                scheduler._run_optimize_sources()
+                scheduler._run_optimize_sources(triggered_by="manual")
             except Exception as e:
                 logger.exception("Background optimize sources failed: %s", e)
         t = threading.Thread(target=_bg, daemon=True)
@@ -222,7 +271,7 @@ def trigger_metrics_sync():
         from ..core.scheduler import scheduler
         def _bg():
             try:
-                scheduler._run_metrics_sync()
+                scheduler._run_metrics_sync(triggered_by="manual")
             except Exception as e:
                 logger.exception("Background metrics sync failed: %s", e)
         t = threading.Thread(target=_bg, daemon=True)