diff --git a/platform/backend/app/api/system.py b/platform/backend/app/api/system.py index ce30a7b..aaa4f8c 100644 --- a/platform/backend/app/api/system.py +++ b/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) diff --git a/platform/backend/app/api/task_configs.py b/platform/backend/app/api/task_configs.py index bfeb005..94b1499 100644 --- a/platform/backend/app/api/task_configs.py +++ b/platform/backend/app/api/task_configs.py @@ -1,5 +1,6 @@ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session +from sqlalchemy import func as sa_func from typing import List, Optional from ..database import get_db @@ -17,15 +18,44 @@ DEFAULT_CONFIGS = { "scheduled_optimize": {"name": "🔍 合规审查", "cron": "03:00", "params": {"auto_pass_threshold": 80}}, "scheduled_optimize_sources": {"name": "📡 信息源优化", "cron": "05:00", "params": {}}, "scheduled_metrics_sync": {"name": "📊 指标同步", "cron": "06:00", "params": {}}, + "scheduled_task_monitor": {"name": "⏰ 任务监控", "cron": "*", "params": {}}, } +def _attach_last_log(resp: TaskConfigResponse, db: Session, module_id: str) -> TaskConfigResponse: + """从最近一次 TaskLog 中附加 last_run_at 和 result_data""" + last_log = db.query(TaskLog).filter( + TaskLog.module_id == module_id, TaskLog.status.in_(["success", "failed"]) + ).order_by(TaskLog.started_at.desc()).first() + if last_log: + resp.last_run_at = last_log.started_at + resp.result_data = last_log.result_data or {} + return resp + @router.get("", response_model=List[TaskConfigResponse]) def list_configs(db: Session = Depends(get_db), admin_user=Depends(get_current_admin)): configs = db.query(TaskConfig).order_by(TaskConfig.id).all() if not configs: _ensure_defaults(db) configs = db.query(TaskConfig).order_by(TaskConfig.id).all() - return [TaskConfigResponse.model_validate(c) for c in configs] + # 批量查询各模块最近一次运行记录 + latest_ids = db.query( + sa_func.max(TaskLog.id).label('max_id') + ).filter( + TaskLog.status.in_(["success", "failed"]) + ).group_by(TaskLog.module_id).subquery() + latest_logs = db.query(TaskLog).filter( + TaskLog.id.in_(db.query(latest_ids.c.max_id)) + ).all() + log_by_module = {log.module_id: log for log in latest_logs} + result = [] + for cfg in configs: + resp = TaskConfigResponse.model_validate(cfg) + log = log_by_module.get(cfg.module_id) + if log: + resp.last_run_at = log.started_at + resp.result_data = log.result_data or {} + result.append(resp) + return result @router.get("/{module_id}", response_model=TaskConfigResponse) def get_config(module_id: str, db: Session = Depends(get_db), admin_user=Depends(get_current_admin)): @@ -33,7 +63,14 @@ def get_config(module_id: str, db: Session = Depends(get_db), admin_user=Depends if not cfg: _ensure_defaults(db) cfg = db.query(TaskConfig).filter(TaskConfig.module_id == module_id).first() - return cfg + if not cfg: + default = DEFAULT_CONFIGS.get(module_id, {}) + cfg = TaskConfig(module_id=module_id, enabled=True, params=default.get("params", {}), schedule=default.get("cron", "")) + db.add(cfg) + db.commit() + db.refresh(cfg) + resp = TaskConfigResponse.model_validate(cfg) + return _attach_last_log(resp, db, module_id) @router.put("/{module_id}", response_model=TaskConfigResponse) def update_config(module_id: str, data: TaskConfigUpdate, db: Session = Depends(get_db), admin_user=Depends(get_current_admin)): diff --git a/platform/backend/app/api/task_logs.py b/platform/backend/app/api/task_logs.py index 5ef9b7f..de8987b 100644 --- a/platform/backend/app/api/task_logs.py +++ b/platform/backend/app/api/task_logs.py @@ -20,6 +20,7 @@ MODULES = { "scheduled_optimize_sources": "📡 信息源优化", "scheduled_metrics_sync": "📊 指标同步", "scheduled_task_monitor": "⏰ 任务监控", + "scheduled_reset_search_usage": "🔁 搜索用量重置", } @router.get("", response_model=List[TaskLogResponse]) @@ -83,8 +84,10 @@ def list_log_types(db: Session = Depends(get_db), admin_user=Depends(get_current "scheduled_collect": "collector", "scheduled_generate": "creator", "scheduled_optimize": "optimizer", - "scheduled_optimize_sources": "collector", - "scheduled_metrics_sync": "sync", + "scheduled_optimize_sources": "optimizer_sources", + "scheduled_metrics_sync": "metrics_sync", + "scheduled_reset_search_usage": "reset_search_usage", + "scheduled_task_monitor": "task_monitor", } result.append({"module_id": mid, "name": name, "log_file": log_file_map.get(mid, mid)}) for mid in used_ids: diff --git a/platform/backend/app/core/collector.py b/platform/backend/app/core/collector.py index c2e7bd6..2d09446 100644 --- a/platform/backend/app/core/collector.py +++ b/platform/backend/app/core/collector.py @@ -8,6 +8,7 @@ from pathlib import Path import logging import os import time +import threading from typing import Optional, Dict logger = logging.getLogger(__name__) @@ -17,6 +18,7 @@ if os.getenv('PROJECT_ROOT'): PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT')) _running_processes: Dict[str, dict] = {} +_lock = threading.Lock() def _get_cmd(): script_path = PROJECT_ROOT / "scripts" / "collector.py" @@ -29,21 +31,26 @@ def _get_cmd(): def run_collector(): """运行选题收集脚本（非阻塞，后台运行）""" - cmd = _get_cmd() - proc = subprocess.Popen( - cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - cwd=PROJECT_ROOT - ) - _running_processes["collector"] = { - "pid": proc.pid, - "started_at": time.time(), - "process": proc - } + with _lock: + existing = _running_processes.get("collector", {}) + proc = existing.get("process") + if proc and proc.poll() is None: + raise RuntimeError("已有内容采集任务正在运行") + cmd = _get_cmd() + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + cwd=PROJECT_ROOT + ) + _running_processes["collector"] = { + "pid": proc.pid, + "started_at": time.time(), + "process": proc, + } logger.info("Collector started in background (pid=%s)", proc.pid) - return {"ok": True, "pid": proc.pid} + return {"ok": True, "pid": proc.pid, "_proc": proc} def run_collector_blocking(timeout: int = 300): """运行选题收集脚本（阻塞，带超时，给定时任务用）""" @@ -60,22 +67,27 @@ def run_collector_blocking(timeout: int = 300): return {"ok": True, "output": result.stdout} def get_collector_status() -> Optional[Dict]: - """获取当前采集任务状态""" + """获取当前采集任务状态（不消费子进程 stdout，留给 _monitor_subprocess）""" info = _running_processes.get("collector") if not info: return None proc: subprocess.Popen = info["process"] + if "finished" in info: + return { + "status": "completed" if info.get("returncode") == 0 else "failed", + "pid": info["pid"], + "elapsed": round(time.time() - info["started_at"], 1), + "returncode": info.get("returncode"), + } if proc.poll() is not None: - stdout, stderr = proc.communicate() + info["returncode"] = proc.returncode + info["finished"] = True elapsed = time.time() - info["started_at"] - del _running_processes["collector"] return { "status": "completed" if proc.returncode == 0 else "failed", "pid": info["pid"], "elapsed": round(elapsed, 1), "returncode": proc.returncode, - "stdout": stdout.strip()[-500:], - "stderr": stderr.strip()[-500:], } return { "status": "running", diff --git a/platform/backend/app/core/generator.py b/platform/backend/app/core/generator.py index 5d9fb20..138f9e2 100644 --- a/platform/backend/app/core/generator.py +++ b/platform/backend/app/core/generator.py @@ -2,8 +2,10 @@ import subprocess from pathlib import Path import logging import os +import json import time -from typing import Optional, Dict +import threading +from typing import Optional, Dict, List logger = logging.getLogger(__name__) @@ -12,6 +14,7 @@ if os.getenv('PROJECT_ROOT'): PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT')) _running_processes: Dict[str, dict] = {} +_lock = threading.Lock() def _get_cmd(topic_id: str = None): script_path = PROJECT_ROOT / "scripts" / "creator.py" @@ -21,15 +24,22 @@ def _get_cmd(topic_id: str = None): cmd = [str(venv_python), str(script_path)] if venv_python.exists() else ["python3", str(script_path)] if topic_id: cmd.extend(["--topic-id", topic_id]) + else: + cmd.append("--today-only") return cmd def run_creator(topic_id: str = None): """非阻塞：后台启动创作脚本，立即返回""" - cmd = _get_cmd(topic_id) - proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(PROJECT_ROOT)) - _running_processes["generator"] = {"pid": proc.pid, "started_at": time.time(), "process": proc, "topic_id": topic_id} + with _lock: + existing = _running_processes.get("generator", {}) + proc = existing.get("process") + if proc and proc.poll() is None: + raise RuntimeError("已有创作任务正在运行") + cmd = _get_cmd(topic_id) + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(PROJECT_ROOT)) + _running_processes["generator"] = {"pid": proc.pid, "started_at": time.time(), "process": proc, "topic_id": topic_id} logger.info("Creator started in background (pid=%s, topic_id=%s)", proc.pid, topic_id) - return {"ok": True, "pid": proc.pid, "topic_id": topic_id} + return {"ok": True, "pid": proc.pid, "topic_id": topic_id, "_proc": proc} def run_creator_blocking(topic_id: str = None, timeout: int = 1800): """阻塞版：带超时，给定时任务使用""" @@ -93,9 +103,11 @@ def get_generator_status() -> Optional[Dict]: if not info: return None proc = info["process"] + if "finished" in info: + return {"status": "completed" if info.get("returncode") == 0 else "failed", "pid": info["pid"], "elapsed": round(time.time() - info["started_at"], 1), "returncode": info.get("returncode"), "topic_id": info.get("topic_id")} if proc.poll() is not None: - stdout, stderr = proc.communicate() + info["returncode"] = proc.returncode + info["finished"] = True elapsed = time.time() - info["started_at"] - del _running_processes["generator"] return {"status": "completed" if proc.returncode == 0 else "failed", "pid": info["pid"], "elapsed": round(elapsed, 1), "returncode": proc.returncode, "topic_id": info["topic_id"]} return {"status": "running", "pid": info["pid"], "elapsed": round(time.time() - info["started_at"], 1)} diff --git a/platform/backend/app/core/optimizer.py b/platform/backend/app/core/optimizer.py index 17b26ea..673ca82 100644 --- a/platform/backend/app/core/optimizer.py +++ b/platform/backend/app/core/optimizer.py @@ -4,6 +4,7 @@ import logging import os import json import time +import threading from datetime import datetime from typing import List, Optional, Dict @@ -14,6 +15,7 @@ if os.getenv('PROJECT_ROOT'): PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT')) _running_processes: Dict[str, dict] = {} +_lock = threading.Lock() def _get_cmd(topic_ids: List[str] = None): script_path = PROJECT_ROOT / "scripts" / "compliance_optimizer.py" @@ -23,15 +25,22 @@ def _get_cmd(topic_ids: List[str] = None): cmd = [str(venv_python), str(script_path)] if venv_python.exists() else ["python3", str(script_path)] if topic_ids: cmd.extend(["--topic-ids", ','.join(topic_ids)]) + else: + cmd.append("--today-only") return cmd def run_optimizer(topic_ids: List[str] = None): """非阻塞：后台启动合规审查脚本，立即返回""" - cmd = _get_cmd(topic_ids) - proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(PROJECT_ROOT)) - _running_processes["optimizer"] = {"pid": proc.pid, "started_at": time.time(), "process": proc, "topic_ids": topic_ids} + with _lock: + existing = _running_processes.get("optimizer", {}) + proc = existing.get("process") + if proc and proc.poll() is None: + raise RuntimeError("已有合规审查任务正在运行，请等待完成后再试") + cmd = _get_cmd(topic_ids) + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(PROJECT_ROOT)) + _running_processes["optimizer"] = {"pid": proc.pid, "started_at": time.time(), "process": proc, "topic_ids": topic_ids} logger.info("Optimizer started in background (pid=%s, topic_ids=%s)", proc.pid, topic_ids) - return {"ok": True, "pid": proc.pid, "topic_ids": topic_ids} + return {"ok": True, "pid": proc.pid, "topic_ids": topic_ids, "_proc": proc} def run_optimizer_blocking(topic_ids: List[str] = None, timeout: int = 600): """阻塞版：带超时，给定时任务使用""" @@ -54,9 +63,11 @@ def get_optimizer_status() -> Optional[Dict]: if not info: return None proc = info["process"] + if "finished" in info: + return {"status": "completed" if info.get("returncode") == 0 else "failed", "pid": info["pid"], "elapsed": round(time.time() - info["started_at"], 1), "returncode": info.get("returncode")} if proc.poll() is not None: - stdout, stderr = proc.communicate() + info["returncode"] = proc.returncode + info["finished"] = True elapsed = time.time() - info["started_at"] - del _running_processes["optimizer"] - return {"status": "completed" if proc.returncode == 0 else "failed", "pid": info["pid"], "elapsed": round(elapsed, 1), "returncode": proc.returncode, "stdout": stdout.strip()[-300:], "stderr": stderr.strip()[-300:]} + return {"status": "completed" if proc.returncode == 0 else "failed", "pid": info["pid"], "elapsed": round(elapsed, 1), "returncode": proc.returncode} return {"status": "running", "pid": info["pid"], "elapsed": round(time.time() - info["started_at"], 1)} diff --git a/platform/backend/app/core/scheduler.py b/platform/backend/app/core/scheduler.py index 37fcc3f..99446b2 100644 --- a/platform/backend/app/core/scheduler.py +++ b/platform/backend/app/core/scheduler.py @@ -31,10 +31,38 @@ MODULES = { "scheduled_task_monitor": {"name": "⏰ 任务监控", "cron": "*"}, } +LOG_FILE_MAP = { + "scheduled_refresh_search_cache": "opencode_search", + "scheduled_fetch_trends": "trends", + "scheduled_collect": "collector", + "scheduled_generate": "creator", + "scheduled_optimize": "optimizer", + "scheduled_optimize_sources": "optimizer_sources", + "scheduled_metrics_sync": "metrics_sync", + "scheduled_reset_search_usage": "reset_search_usage", + "scheduled_task_monitor": "task_monitor", +} + +def _log_to_file(module_id: str, status: str, message: str = None, error_trace: str = None): + log_name = LOG_FILE_MAP.get(module_id, module_id) + log_dir = PROJECT_ROOT / "automation" / "logs" + log_dir.mkdir(parents=True, exist_ok=True) + log_file = log_dir / f"{log_name}_{datetime.now().strftime('%Y-%m-%d')}.log" + ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + line = f"[{ts}] [{status.upper()}] {message or ''}" + if error_trace: + line += f"\n{error_trace[:500]}" + try: + with open(log_file, 'a', encoding='utf-8') as f: + f.write(line + "\n") + except Exception: + pass + def _log_task(module_id: str, status: str, message: str = None, error_trace: str = None, result_data: dict = None, started_at: datetime = None, finished_at: datetime = None, triggered_by: str = "scheduler", next_run_time: datetime = None): + _log_to_file(module_id, status, message, error_trace) try: from ..database import SessionLocal from ..models import TaskLog @@ -159,7 +187,7 @@ class TaskScheduler: logger.info("[Scheduled] Fetching hot trends...") import subprocess result = subprocess.run( - [sys.executable, str(Path(__file__).parent.parent.parent.parent / "scripts" / "trends.py")], + [sys.executable, str(PROJECT_ROOT / "scripts" / "trends.py")], capture_output=True, text=True, timeout=120 ) if result.returncode == 0: @@ -190,7 +218,7 @@ class TaskScheduler: logger.info("[Scheduled] Refreshing search cache via opencode...") import subprocess result = subprocess.run( - [sys.executable, str(Path(__file__).parent.parent.parent.parent / "scripts" / "opencode_search.py"), "--refresh-cache"], + [sys.executable, str(PROJECT_ROOT / "scripts" / "opencode_search.py"), "--refresh-cache"], capture_output=True, text=True, timeout=600 ) for line in result.stdout.strip().split("\n"): @@ -285,10 +313,10 @@ class TaskScheduler: started_at=started, finished_at=datetime.now(timezone.utc)) logger.exception("[Scheduled] Collection failed: %s", e) - def _run_optimize_sources(self): + def _run_optimize_sources(self, triggered_by="scheduler"): """AI自动优化采集类别与信息源：对比市场热点和当前配置，给出调整建议""" started = datetime.now(timezone.utc) - _log_task("scheduled_optimize_sources", "running", started_at=started) + _log_task("scheduled_optimize_sources", "running", started_at=started, triggered_by=triggered_by) try: logger.info("[Scheduled] Starting source optimization with AI...") from .nvidia_client import call_llm @@ -337,19 +365,19 @@ class TaskScheduler: "sources_assessed": len(result.get("source_assessment", [])), "suggested_cats": len(result.get("suggested_new_categories", [])), "suggested_srcs": len(result.get("suggested_new_sources", []))}, - started_at=started, finished_at=datetime.now(timezone.utc)) + started_at=started, finished_at=datetime.now(timezone.utc), triggered_by=triggered_by) db.close() except Exception as e: _log_task("scheduled_optimize_sources", "failed", message=str(e), error_trace=traceback.format_exc(), - started_at=started, finished_at=datetime.now(timezone.utc)) + started_at=started, finished_at=datetime.now(timezone.utc), triggered_by=triggered_by) logger.exception("[Scheduled] Source AI optimization failed: %s", e) - def _run_metrics_sync(self): + def _run_metrics_sync(self, triggered_by="scheduler"): """定时从各平台公开API获取发布文章的效果数据（当前仅支持知乎）""" started = datetime.now(timezone.utc) - _log_task("scheduled_metrics_sync", "running", started_at=started) + _log_task("scheduled_metrics_sync", "running", started_at=started, triggered_by=triggered_by) try: logger.info("[Scheduled] Starting metrics sync (zhihu auto-fetch)...") from ..database import SessionLocal @@ -411,7 +439,7 @@ class TaskScheduler: _log_task("scheduled_metrics_sync", "success", message=f"同步完成，{count} 篇知乎文章", result_data={"articles_synced": count}, - started_at=started, finished_at=datetime.now(timezone.utc)) + started_at=started, finished_at=datetime.now(timezone.utc), triggered_by=triggered_by) # 生成指标反馈：按 field 聚合表现，写入 metrics_feedback.json 供 collector 读取 try: import json as json_mod @@ -436,7 +464,7 @@ class TaskScheduler: "top_domains": [(f, s) for f, s, _ in scored[:5]], "detail": [{"field": f, "score": s, "articles": c} for f, s, c in scored], } - feedback_file = Path(__file__).parent.parent.parent.parent / "automation" / "data" / "metrics_feedback.json" + feedback_file = PROJECT_ROOT / "automation" / "data" / "metrics_feedback.json" feedback_file.parent.mkdir(parents=True, exist_ok=True) feedback_file.write_text(json_mod.dumps(feedback_data, ensure_ascii=False, indent=2), encoding='utf-8') logger.info("[Scheduled] Metrics feedback written: top domain %s (score %.1f)", scored[0][0], scored[0][1]) @@ -445,14 +473,14 @@ class TaskScheduler: else: _log_task("scheduled_metrics_sync", "success", message="无已发布的知乎文章", - started_at=started, finished_at=datetime.now(timezone.utc)) + started_at=started, finished_at=datetime.now(timezone.utc), triggered_by=triggered_by) logger.info("[Scheduled] Metrics sync: no zhihu articles to sync") db.close() except Exception as e: _log_task("scheduled_metrics_sync", "failed", message=str(e), error_trace=traceback.format_exc(), - started_at=started, finished_at=datetime.now(timezone.utc)) + started_at=started, finished_at=datetime.now(timezone.utc), triggered_by=triggered_by) logger.exception("[Scheduled] Metrics sync failed: %s", e) def _run_reset_search_usage(self): diff --git a/platform/backend/app/database.py b/platform/backend/app/database.py index b67ea58..1ab8470 100644 --- a/platform/backend/app/database.py +++ b/platform/backend/app/database.py @@ -3,6 +3,7 @@ from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import os from pathlib import Path +from datetime import datetime, timezone, timedelta PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent.parent @@ -140,6 +141,20 @@ def init_db(): except Exception: pass # SQLite 不支持 IF NOT EXISTS，但 create_all 对 SQLite 够用，这里仅为 PostgreSQL 迁移 + # 清理卡死的 running 任务（超过 30 分钟无更新的标记为失败） + try: + with SessionLocal() as sess: + from sqlalchemy import text as sa_text + cutoff = datetime.now(timezone.utc) - timedelta(minutes=30) + sess.execute( + sa_text("UPDATE task_logs SET status='failed', message='系统启动时清理卡死任务', finished_at=now() " + "WHERE status='running' AND started_at < :cutoff AND module_id != 'scheduled_task_monitor'"), + {"cutoff": cutoff} + ) + sess.commit() + except Exception: + pass + def get_db(): db = SessionLocal() try: diff --git a/platform/backend/app/schemas.py b/platform/backend/app/schemas.py index 532238b..9294157 100644 --- a/platform/backend/app/schemas.py +++ b/platform/backend/app/schemas.py @@ -503,6 +503,8 @@ class TaskConfigUpdate(BaseModel): class TaskConfigResponse(TaskConfigBase): id: int last_modified_by: Optional[str] = None + last_run_at: Optional[datetime] = None + result_data: Optional[Dict[str, Any]] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None diff --git a/platform/frontend/tasks.html b/platform/frontend/tasks.html index 6bed584..ab716f6 100644 --- a/platform/frontend/tasks.html +++ b/platform/frontend/tasks.html @@ -363,78 +363,13 @@ -
-
最近运行结果
-
+
+
{{ key }}: {{ val }}
- -
-
暂无产出数据
-
-
-
{{ key }}
-
- - - - - - -
-
- - - - - - - -
-
- - - -
-
-
-
{{ item.query }} {{ item.count }}条
-
● {{ s }}
-
-
-
-
- {{ item.title }} - -
-
-
{{ val }}
-
-
-
{{ item.name }}
-
{{ item.reason }}
-
-
-
-
- {{ item.status }} - {{ item.name }} — {{ item.reason }} -
-
-
-
- {{ src }}: {{ cnt }}条 -
-
-
-
- {{ item[0] }}: {{ item[1] }}分 -
-
-
{{ val }}
-
+
暂无产出数据
diff --git a/platform/frontend/topics.html b/platform/frontend/topics.html index fdc8129..66c3428 100644 --- a/platform/frontend/topics.html +++ b/platform/frontend/topics.html @@ -459,17 +459,30 @@ const TopicsApp = { } if (platform === 'xiaohongshu') { - const lines = []; + const container = doc.createElement('div'); + if (titleEl) { + const strong = doc.createElement('strong'); + strong.textContent = title; + container.appendChild(strong); + container.appendChild(doc.createElement('br')); + container.appendChild(doc.createElement('br')); + } body.querySelectorAll('h1,h2,h3,h4,p,li').forEach(el => { const text = el.textContent.trim(); if (!text || text.length < 2) return; - const tag = el.tagName.toLowerCase(); - const prefix = tag.startsWith('h') ? '\n### ' : '- '; - const clean = text.replace(/\n/g, ' ').replace(/^[\s#]+|[\s#]+$/g, ''); - if (clean) lines.push(prefix + clean); + const clone = el.cloneNode(true); + container.appendChild(clone); + container.appendChild(doc.createElement('br')); + }); + const cleanHtml = container.innerHTML; + const blob = new Blob([cleanHtml], { type: 'text/html' }); + const richText = new Blob([cleanHtml.replace(/<[^>]+>/g, '').replace(/\n{3,}/g, '\n\n')], { type: 'text/plain' }); + const item = new ClipboardItem({ 'text/html': blob, 'text/plain': richText }); + navigator.clipboard.write([item]).then(() => { + this.$message.success('✅ 已复制（含格式），Ctrl+V 粘贴到小红书'); + }).catch(() => { + navigator.clipboard.writeText(cleanHtml.replace(/<[^>]+>/g, '').replace(/\n{3,}/g, '\n\n')).then(() => this.$message.success('✅ 已复制')).catch(() => this.$message.error('❌ 复制失败')); }); - const md = `**${title}**\n\n${lines.join('\n')}`; - navigator.clipboard.writeText(md).then(() => this.$message.success('✅ 已复制 Markdown（小红书格式）')).catch(() => this.$message.error('❌ 复制失败')); } else { const container = doc.createElement('div'); if (titleEl) { @@ -477,7 +490,7 @@ const TopicsApp = { h1.textContent = title; container.appendChild(h1); } - body.querySelectorAll('h2,h3,h4,p,li,blockquote,img,pre,code,table,hr').forEach(el => { + body.querySelectorAll('h2,h3,h4,p,li,blockquote,pre,code,table,hr').forEach(el => { const clone = el.cloneNode(true); container.appendChild(clone); }); diff --git a/scripts/compliance_optimizer.py b/scripts/compliance_optimizer.py index 018054e..54854c8 100644 --- a/scripts/compliance_optimizer.py +++ b/scripts/compliance_optimizer.py @@ -8,7 +8,7 @@ from pathlib import Path from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, asdict -PROJECT_ROOT = Path('/root/openclaw-workspace/projects/yu-zhi-ran') +PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend")) @@ -92,9 +92,9 @@ def load_topic_map(): topics = export_topics_to_json() return {t['id']: t for t in topics} -def get_articles_from_db(topic_ids: Optional[List[str]] = None) -> List[Tuple[str, str, str]]: +def get_articles_from_db(topic_ids: Optional[List[str]] = None, today_only: bool = False) -> List[Tuple[str, str, str]]: """从 articles 表读取 HTML 内容 - + Returns: [(html_content, platform, topic_id), ...] """ from db_helper import get_articles_by_topic @@ -110,9 +110,14 @@ def get_articles_from_db(topic_ids: Optional[List[str]] = None) -> List[Tuple[st else: from app.database import SessionLocal from app.models import Article + from sqlalchemy import func db = SessionLocal() try: - all_articles = db.query(Article).filter(Article.html_content.isnot(None)).all() + query = db.query(Article).filter(Article.html_content.isnot(None)) + if today_only: + cutoff = datetime.datetime.now() - datetime.timedelta(hours=24) + query = query.filter(Article.created_at >= cutoff) + all_articles = query.all() for a in all_articles: results.append((a.html_content, a.platform, a.topic_id)) finally: @@ -222,14 +227,17 @@ def _load_platform_configs() -> Dict[str, Dict]: finally: db.close() -def main(topic_ids: List[str] = None): +def main(topic_ids: List[str] = None, today_only: bool = False): logger.info("=== 合规审查与优化开始 ===") logger.info("LLM 配置: opencode-go (model=deepseek-v4-flash) — 固定用于合规审查") platform_configs = _load_platform_configs() logger.info(f"已加载 {len(platform_configs)} 个平台配置") - articles = get_articles_from_db(topic_ids) + if today_only: + logger.info("仅处理当天创建的选题文章") + + articles = get_articles_from_db(topic_ids, today_only) if not articles: logger.warning("未找到任何文章（可能尚未创作或同步到 DB）") report_file = DRAFTS_DIR / TODAY / "optimization_report.json" @@ -349,6 +357,7 @@ if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description='合规审查与优化任务') parser.add_argument('--topic-ids', help='逗号分隔的选题ID列表，例如: A01,B02') + parser.add_argument('--today-only', action='store_true', help='仅处理当天创建的选题文章') args = parser.parse_args() topic_ids = args.topic_ids.split(',') if args.topic_ids else None - main(topic_ids) + main(topic_ids, today_only=args.today_only) diff --git a/scripts/creator.py b/scripts/creator.py index 5886fb8..3214d77 100755 --- a/scripts/creator.py +++ b/scripts/creator.py @@ -28,7 +28,7 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) -def select_next_topic(topic_id: str = None) -> Dict: +def select_next_topic(topic_id: str = None, today_only: bool = False) -> Dict: """选择并锁定要创作的选题（趋势引擎匹配 → 回退优先级）""" if topic_id: topic = get_topic_by_id(topic_id) @@ -50,7 +50,7 @@ def select_next_topic(topic_id: str = None) -> Dict: except Exception as e: logger.warning(f"选题引擎失效，回退简单策略: {e}") - topic = get_next_topic(priority='高') or get_next_topic() + topic = get_next_topic(priority='高', today_only=today_only) or get_next_topic(today_only=today_only) if not topic: raise ValueError("No available topics to create (all locked or wrong status)") update_topic_status(topic['id'], 'review') @@ -80,11 +80,11 @@ def run_optimizer_step(topic_id: str) -> bool: logger.info(f"compliance_optimizer 完成: {result.stdout.strip()}") return True -def run_pipeline(topic_id: str = None) -> Dict: +def run_pipeline(topic_id: str = None, today_only: bool = False) -> Dict: """运行完整流水线：研究 → 大纲 → 撰写 → 合规优化""" tid = None try: - topic = select_next_topic(topic_id) + topic = select_next_topic(topic_id, today_only) tid = topic['id'] logger.info(f"开始创作流水线: topic_id={tid}, title={topic.get('title')}") @@ -127,9 +127,10 @@ def main(): import argparse parser = argparse.ArgumentParser(description='内容创作流水线（研究→大纲→撰写→合规优化）') parser.add_argument('--topic-id', help='指定选题ID，不指定则自动选择待处理选题') + parser.add_argument('--today-only', action='store_true', help='仅处理当天创建的选题') args = parser.parse_args() - result = run_pipeline(args.topic_id) + result = run_pipeline(args.topic_id, today_only=args.today_only) print(json.dumps(result, ensure_ascii=False)) sys.exit(0 if result['ok'] else 1) diff --git a/scripts/db_helper.py b/scripts/db_helper.py index 9d15c57..ffc4b1a 100644 --- a/scripts/db_helper.py +++ b/scripts/db_helper.py @@ -6,7 +6,7 @@ import sys import os from pathlib import Path -from datetime import datetime, date +from datetime import datetime, date, timedelta from typing import Optional, Dict, List # 加载 .env（在 scripts/ 目录下运行时需要） @@ -53,7 +53,7 @@ def get_topics_by_status(status: str, db: Optional[Session] = None) -> List[Dict if close_db: db.close() -def get_next_topic(priority: Optional[str] = None, db: Optional[Session] = None) -> Optional[Dict]: +def get_next_topic(priority: Optional[str] = None, db: Optional[Session] = None, today_only: bool = False) -> Optional[Dict]: """获取下一个待处理的选题（状态为 pending/待处理）""" close_db = False if db is None: @@ -65,6 +65,9 @@ def get_next_topic(priority: Optional[str] = None, db: Optional[Session] = None) query = db.query(Topic).filter(Topic.status.in_(status_filter)) if priority: query = query.filter(Topic.priority == priority) + if today_only: + cutoff = datetime.now() - timedelta(hours=24) + query = query.filter(Topic.created_at >= cutoff) topic = query.order_by(Topic.priority_score.desc().nullslast(), Topic.created_at.asc()).first() return topic_to_dict(topic) if topic else None finally: