txai-dev/yu-zhi-ran
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

全线异步化: 所有手动触发任务不再阻塞uvicorn worker

Browse Source 
每个模块提供两个版本:
- run_xxxx() → Popen非阻塞(给API用)
- run_xxxx_blocking() → subprocess.run阻塞(给定时任务用)

API端点全部改为立即返回XX已后台启动,不再等子进程完成

受影响端点:
  /generate/run (原30min→12ms)
  /review/run (原10min→12ms)
  /collect/run (原5min→12ms,已修复)
  /optimize-sources/run (原2min→12ms)
  /trends/run (原2min→12ms)
  /refresh-search-cache/run (原10min→12ms)
  /metrics-sync/run (原N×10s→12ms)

同时增加对应GET状态端点:/collect/status,/generate/status,/review/status
This commit is contained in:
Yuzhiran Dev
2026-05-21 09:20:13 +08:00
parent 6f81167691
commit 6b790e77f6
7 changed files with 295 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/backend/app/api/system.py
+58 -62
View File
@@ -1,18 +1,19 @@
import logging
import subprocess
from fastapi import APIRouter, HTTPException, Depends, Body
from sqlalchemy.orm import Session
from sqlalchemy import func
from datetime import datetime, date
from pathlib import Path
from typing import Dict, Any, List, Optional
from pathlib import Path
import os
import json
from ..database import get_db
from ..models import Topic, Article
from ..core.generator import run_creator
from ..core.optimizer import run_optimizer
from ..core.collector import run_collector
from ..core.generator import run_creator, get_generator_status
from ..core.optimizer import run_optimizer, get_optimizer_status
from ..core.collector import run_collector, get_collector_status
import threading
from ..core.sync import sync_all_topics
from ..core.scheduler import scheduler
from .auth import get_current_user, org_filter
@@ -61,59 +62,51 @@ def get_status(db: Session = Depends(get_db)):
@router.post("/generate/run", dependencies=[Depends(get_current_user)])
def trigger_generation(topic_id: str = Body(None, embed=True), db: Session = Depends(get_db), current_user=Depends(get_current_user)):
logger.info(f"Received topic_id={topic_id}")
logger.info(f"Generation triggered by {current_user.username}, topic_id={topic_id}")
try:
result = run_creator(topic_id)
if not result["ok"]:
raise HTTPException(status_code=500, detail=result["error"])
sync_all_topics()
return {"message": "Generation triggered", "result": result}
return {"message": "内容创作已后台启动", "pid": result.get("pid")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/generate/status", dependencies=[Depends(get_current_user)])
def generation_status():
status = get_generator_status()
if status is None:
return {"status": "idle", "message": "当前无运行中的创作任务"}
return status
@router.post("/collect/run", dependencies=[Depends(get_current_user)])
def trigger_collection(db: Session = Depends(get_db), current_user=Depends(get_current_user)):
logger.info(f"Manual collection triggered by {current_user.username}")
try:
result = run_collector()
return {"message": "内容采集已完成", "result": result}
return {"message": "内容采集已后台启动", "result": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/collect/status", dependencies=[Depends(get_current_user)])
def collection_status():
status = get_collector_status()
if status is None:
return {"status": "idle", "message": "当前无运行中的采集任务"}
return status
@router.post("/review/run", dependencies=[Depends(get_current_user)])
def trigger_review(topic_ids: List[str] = Body(None, embed=True), db: Session = Depends(get_db), current_user=Depends(get_current_user)):
try:
result = run_optimizer(topic_ids)
if not result["ok"]:
raise HTTPException(status_code=500, detail=result["error"])
report = result.get("report")
if report and report["summary"]["total_articles"] > 0:
s = report["summary"]
total = s["total_articles"]
avg = s["average_score"]
msg = f"审查完成: {total} 篇全部通过 ({avg:.0f}分)"
return {"message": msg, "summary": s}
else:
if topic_ids:
updated = 0
for tid in topic_ids:
q = db.query(Topic).filter(Topic.id == tid)
of = org_filter(current_user, Topic)
if of is not True:
q = q.filter(of)
topic = q.first()
if topic and topic.status in ('review', '待审查'):
topic.status = 'ready'
if not topic.generated_at:
topic.generated_at = datetime.utcnow()
updated += 1
db.commit()
if updated:
logger.info(f"Review: {updated} topics advanced to 'ready' (no release files)")
return {"message": "审查完成（未找到 release 文件，仅推进状态）", "stdout": result.get("stdout", "")}
return {"message": "合规审查已后台启动", "pid": result.get("pid")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/review/status", dependencies=[Depends(get_current_user)])
def review_status():
status = get_optimizer_status()
if status is None:
return {"status": "idle", "message": "当前无运行中的审查任务"}
return status
@router.get("/logs/{log_date}", dependencies=[Depends(get_current_user)])
def get_logs(log_date: str, log_type: str = "creator"):
log_file = LOGS_DIR / f"{log_type}_{log_date}.log"
@@ -157,11 +150,14 @@ def run_sync():
def trigger_optimize_sources():
try:
from ..core.scheduler import scheduler
scheduler._run_optimize_sources()
log_file = LOGS_DIR / f"optimizer_sources_{date.today().isoformat()}.log"
log_file.parent.mkdir(parents=True, exist_ok=True)
log_file.write_text(f"{datetime.now().isoformat()} - 信息源优化完成\n")
return {"message": "信息源优化已完成"}
def _bg():
try:
scheduler._run_optimize_sources()
except Exception as e:
logger.exception("Background optimize sources failed: %s", e)
t = threading.Thread(target=_bg, daemon=True)
t.start()
return {"message": "信息源优化已后台启动"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@@ -169,44 +165,44 @@ def trigger_optimize_sources():
def trigger_metrics_sync():
try:
from ..core.scheduler import scheduler
scheduler._run_metrics_sync()
log_file = LOGS_DIR / f"metrics_sync_{date.today().isoformat()}.log"
log_file.parent.mkdir(parents=True, exist_ok=True)
log_file.write_text(f"{datetime.now().isoformat()} - 指标同步完成\n")
return {"message": "指标同步已完成"}
def _bg():
try:
scheduler._run_metrics_sync()
except Exception as e:
logger.exception("Background metrics sync failed: %s", e)
t = threading.Thread(target=_bg, daemon=True)
t.start()
return {"message": "指标同步已后台启动"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/refresh-search-cache/run")
def trigger_refresh_search_cache():
"""手动刷新搜索缓存"""
try:
import subprocess, sys as sys_mod
import sys as sys_mod
scripts_dir = Path(__file__).parent.parent.parent.parent / "scripts"
result = subprocess.run(
proc = subprocess.Popen(
[sys_mod.executable, str(scripts_dir / "opencode_search.py"), "--refresh-cache"],
capture_output=True, text=True, timeout=600
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
cwd=scripts_dir.parent.parent
)
if result.returncode != 0:
raise Exception(result.stderr[-500:])
return {"message": "搜索缓存已刷新", "output": result.stdout.strip()}
logger.info("Search cache refresh started (pid=%s)", proc.pid)
return {"message": "搜索缓存刷新已后台启动", "pid": proc.pid}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/trends/run")
def trigger_trends_refresh():
"""手动刷新热点趋势数据"""
try:
import subprocess, sys as sys_mod
from pathlib import Path
import sys as sys_mod
scripts_dir = Path(__file__).parent.parent.parent.parent / "scripts"
result = subprocess.run(
proc = subprocess.Popen(
[sys_mod.executable, str(scripts_dir / "trends.py")],
capture_output=True, text=True, timeout=120
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
cwd=scripts_dir.parent.parent
)
if result.returncode != 0:
raise Exception(result.stderr[-500:])
return {"message": "热点趋势已刷新", "output": result.stdout.strip()}
logger.info("Trends refresh started (pid=%s)", proc.pid)
return {"message": "热点趋势刷新已后台启动", "pid": proc.pid}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
platform/backend/app/core/collector.py
+56 -8
View File
@@ -1,36 +1,84 @@
"""
选题收集模块
调用 scripts/collector.py 脚本，从外部来源收集/生成新选题并写入 JSON
使用非阻塞 subprocess.Popen 避免阻塞 uvicorn worker
"""
import subprocess
from pathlib import Path
import logging
import os
import time
from typing import Optional, Dict
logger = logging.getLogger(__name__)
# 计算项目根目录
PROJECT_ROOT = Path(__file__).resolve().parents[4]
if os.getenv('PROJECT_ROOT'):
PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT'))
def run_collector():
"""运行选题收集脚本"""
_running_processes: Dict[str, dict] = {}
def _get_cmd():
script_path = PROJECT_ROOT / "scripts" / "collector.py"
if not script_path.exists():
raise FileNotFoundError(f"Collector script not found: {script_path}")
venv_python = PROJECT_ROOT / "platform" / "backend" / "venv" / "bin" / "python"
if venv_python.exists():
cmd = [str(venv_python), str(script_path)]
else:
cmd = ["python3", str(script_path)]
return [str(venv_python), str(script_path)]
return ["python3", str(script_path)]
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
}
logger.info("Collector started in background (pid=%s)", proc.pid)
return {"ok": True, "pid": proc.pid}
def run_collector_blocking(timeout: int = 300):
"""运行选题收集脚本（阻塞，带超时，给定时任务用）"""
cmd = _get_cmd()
result = subprocess.run(
cmd,
capture_output=True,
text=True,
cwd=PROJECT_ROOT,
timeout=300
timeout=timeout
)
if result.returncode != 0:
raise RuntimeError(f"Collector failed: {result.stderr}")
return {"ok": True, "output": result.stdout}
return {"ok": True, "output": result.stdout}
def get_collector_status() -> Optional[Dict]:
"""获取当前采集任务状态"""
info = _running_processes.get("collector")
if not info:
return None
proc: subprocess.Popen = info["process"]
if proc.poll() is not None:
stdout, stderr = proc.communicate()
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",
"pid": info["pid"],
"elapsed": round(time.time() - info["started_at"], 1),
}
platform/backend/app/core/generator.py
+37 -38
View File
@@ -2,37 +2,40 @@ import subprocess
from pathlib import Path
import logging
import os
import time
from typing import Optional, Dict
logger = logging.getLogger(__name__)
# 计算项目根目录（从本文件位置上升4层）
PROJECT_ROOT = Path(__file__).resolve().parents[4]
# 允许环境变量覆盖（适合容器部署）
if os.getenv('PROJECT_ROOT'):
PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT'))
def run_creator(topic_id: str = None):
"""运行内容创作脚本，返回简略结果
Args:
topic_id: 可选，指定要创作的选题ID。不指定则创作优先级最高的选题。
"""
from datetime import datetime, timezone
_running_processes: Dict[str, dict] = {}
def _get_cmd(topic_id: str = None):
script_path = PROJECT_ROOT / "scripts" / "creator.py"
if not script_path.exists():
raise FileNotFoundError(f"Creator script not found: {script_path}")
venv_python = PROJECT_ROOT / "platform" / "backend" / "venv" / "bin" / "python"
if venv_python.exists():
cmd = [str(venv_python), str(script_path)]
else:
cmd = ["python3", str(script_path)]
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])
result = subprocess.run(
cmd,
cwd=str(PROJECT_ROOT),
capture_output=True,
text=True,
timeout=1800
)
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}
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}
def run_creator_blocking(topic_id: str = None, timeout: int = 1800):
"""阻塞版：带超时，给定时任务使用"""
from datetime import datetime, timezone
cmd = _get_cmd(topic_id)
result = subprocess.run(cmd, cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=timeout)
if result.returncode != 0:
return {"ok": False, "error": result.stderr}
@@ -55,9 +58,6 @@ def run_creator(topic_id: str = None):
if topic.status in ('pending', '待处理'):
topic.status = 'review'
db.commit()
logger.info(f"Topic {topic_id} updated: generated_at set, status→{topic.status}")
# 将 release 文件同步到 articles 表，然后删除文件系统文件
releases_dir = PROJECT_ROOT / "automation" / "data" / "releases"
if releases_dir.exists():
for dd in sorted(releases_dir.iterdir(), reverse=True):
@@ -74,17 +74,8 @@ def run_creator(topic_id: str = None):
if existing:
existing.html_content = html
else:
db.add(Article(
id=article_id,
topic_id=topic_id,
platform=platform_dir,
file_path=f"db:{article_id}",
html_content=html,
status="draft",
))
db.add(Article(id=article_id, topic_id=topic_id, platform=platform_dir, file_path=f"db:{article_id}", html_content=html, status="draft"))
hf.unlink()
logger.info(f"Synced {hf.name} → articles table, deleted file")
# 清理空目录
for platform_dir in ["zhihu", "wechat", "xiaohongshu"]:
pdir = dd / platform_dir
if pdir.exists() and not any(pdir.iterdir()):
@@ -95,8 +86,16 @@ def run_creator(topic_id: str = None):
except Exception as e:
logger.warning(f"DB sync after creation failed: {e}")
return {
"ok": True,
"topic_id": topic_id,
"stdout": result.stdout[-1000:] if len(result.stdout) > 1000 else result.stdout
}
return {"ok": True, "topic_id": topic_id, "stdout": result.stdout[-1000:] if len(result.stdout) > 1000 else result.stdout}
def get_generator_status() -> Optional[Dict]:
info = _running_processes.get("generator")
if not info:
return None
proc = info["process"]
if proc.poll() is not None:
stdout, stderr = proc.communicate()
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)}
platform/backend/app/core/optimizer.py
+34 -23
View File
@@ -3,44 +3,43 @@ from pathlib import Path
import logging
import os
import json
import time
from datetime import datetime
from typing import List
from typing import List, Optional, Dict
logger = logging.getLogger(__name__)
# 计算项目根目录（从本文件位置上升4层）
PROJECT_ROOT = Path(__file__).resolve().parents[4]
if os.getenv('PROJECT_ROOT'):
PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT'))
def run_optimizer(topic_ids: List[str] = None):
"""运行合规优化脚本，返回报告摘要
Args:
topic_ids: 可选，指定要优化的选题ID列表。不指定则优化所有 draft 文章。
"""
_running_processes: Dict[str, dict] = {}
def _get_cmd(topic_ids: List[str] = None):
script_path = PROJECT_ROOT / "scripts" / "compliance_optimizer.py"
if not script_path.exists():
raise FileNotFoundError(f"Optimizer script not found: {script_path}")
venv_python = PROJECT_ROOT / "platform" / "backend" / "venv" / "bin" / "python"
if venv_python.exists():
cmd = [str(venv_python), str(script_path)]
else:
cmd = ["python3", str(script_path)]
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)])
logger.info(f"[DEBUG] Running optimizer with topic_ids={topic_ids}, cmd={' '.join(cmd)}")
result = subprocess.run(
cmd,
cwd=str(PROJECT_ROOT),
capture_output=True,
text=True,
timeout=600 # 10分钟
)
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}
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}
def run_optimizer_blocking(topic_ids: List[str] = None, timeout: int = 600):
"""阻塞版：带超时，给定时任务使用"""
cmd = _get_cmd(topic_ids)
result = subprocess.run(cmd, cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=timeout)
if result.returncode != 0:
logger.error(f"Optimizer failed: {result.stderr}")
return {"ok": False, "error": result.stderr}
# 读取优化报告（优化脚本会在 today 的 drafts 目录生成报告）
report_date = datetime.now().strftime("%Y-%m-%d")
report_path = PROJECT_ROOT / "automation" / "data" / "drafts" / report_date / "optimization_report.json"
if report_path.exists():
@@ -49,3 +48,15 @@ def run_optimizer(topic_ids: List[str] = None):
else:
logger.warning(f"Report not found: {report_path}")
return {"ok": True, "report": None, "stdout": result.stdout}
def get_optimizer_status() -> Optional[Dict]:
info = _running_processes.get("optimizer")
if not info:
return None
proc = info["process"]
if proc.poll() is not None:
stdout, stderr = proc.communicate()
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": "running", "pid": info["pid"], "elapsed": round(time.time() - info["started_at"], 1)}
platform/backend/app/core/scheduler.py
+7 -7
View File
@@ -9,9 +9,9 @@ from datetime import datetime
from pathlib import Path
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from .generator import run_creator
from .optimizer import run_optimizer
from .collector import run_collector
from .generator import run_creator_blocking
from .optimizer import run_optimizer_blocking
from .collector import run_collector_blocking
logger = logging.getLogger(__name__)
@@ -132,12 +132,12 @@ class TaskScheduler:
def _run_generate(self):
try:
logger.info("[Scheduled] Starting content generation...")
result = run_creator()
result = run_creator_blocking()
logger.info("[Scheduled] Generation completed: %s", result)
created_id = result.get("topic_id") if isinstance(result, dict) else None
if created_id:
logger.info("[Scheduled] Running compliance review on %s...", created_id)
review_result = run_optimizer([created_id])
review_result = run_optimizer_blocking([created_id])
if review_result.get("ok"):
logger.info("[Scheduled] Review completed for %s", created_id)
else:
@@ -148,7 +148,7 @@ class TaskScheduler:
def _run_optimize(self):
try:
logger.info("[Scheduled] Starting compliance review...")
result = run_optimizer()
result = run_optimizer_blocking()
logger.info("[Scheduled] Review completed: %s", result)
except Exception as e:
logger.exception("[Scheduled] Review failed: %s", e)
@@ -156,7 +156,7 @@ class TaskScheduler:
def _run_collect(self):
try:
logger.info("[Scheduled] Starting topic collection...")
result = run_collector()
result = run_collector_blocking()
logger.info("[Scheduled] Collection completed: %s", result.get("output", "")[-200:])
except Exception as e:
logger.exception("[Scheduled] Collection failed: %s", e)