Yuzhiran Dev
501 lines
21 KiB
Python
501 lines
21 KiB
Python
import uuid
|
|
import threading
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
|
|
from ..database import get_db
|
|
from ..models import ContentTask, Topic
|
|
from ..schemas import ContentTaskCreate, ContentTaskResponse
|
|
from .auth import get_current_user, org_filter
|
|
|
|
_creator_semaphore = threading.Semaphore(3)
|
|
|
|
router = APIRouter(prefix="/api/tasks", tags=["tasks"])
|
|
|
|
|
|
@router.get("", response_model=List[ContentTaskResponse])
|
|
def list_tasks(
|
|
status: Optional[str] = None,
|
|
topic_id: Optional[str] = None,
|
|
stage: Optional[str] = None,
|
|
limit: int = Query(50, le=200),
|
|
offset: int = Query(0, ge=0),
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
query = db.query(ContentTask, Topic.title.label("topic_title")).join(Topic, ContentTask.topic_id == Topic.id, isouter=True)
|
|
of = org_filter(current_user, Topic)
|
|
if of is not True:
|
|
query = query.filter((ContentTask.topic_id.is_(None)) | (Topic.org_id == current_user.org_id))
|
|
if status:
|
|
query = query.filter(ContentTask.status == status)
|
|
if topic_id:
|
|
query = query.filter(ContentTask.topic_id == topic_id)
|
|
if stage:
|
|
query = query.filter(ContentTask.stage == stage)
|
|
rows = query.order_by(ContentTask.created_at.desc()).offset(offset).limit(limit).all()
|
|
result = []
|
|
for r in rows:
|
|
task = r.ContentTask
|
|
task.topic_title = r.topic_title
|
|
result.append(task)
|
|
return result
|
|
|
|
|
|
@router.get("/active")
|
|
def get_active_tasks(
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
q = db.query(ContentTask).join(Topic, ContentTask.topic_id == Topic.id, isouter=True)
|
|
of = org_filter(current_user, Topic)
|
|
if of is not True:
|
|
q = q.filter((ContentTask.topic_id.is_(None)) | (Topic.org_id == current_user.org_id))
|
|
return q.filter(
|
|
ContentTask.status == "running"
|
|
).order_by(ContentTask.started_at.desc()).all()
|
|
|
|
|
|
@router.post("", response_model=ContentTaskResponse)
|
|
def create_task(
|
|
data: ContentTaskCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
if data.topic_id:
|
|
topic = db.query(Topic).filter(Topic.id == data.topic_id).first()
|
|
if not topic:
|
|
raise HTTPException(status_code=404, detail="选题不存在")
|
|
if current_user.role != "admin" and topic.org_id != current_user.org_id:
|
|
raise HTTPException(status_code=404, detail="选题不存在")
|
|
|
|
task_id = f"task_{uuid.uuid4().hex[:16]}"
|
|
|
|
task = ContentTask(
|
|
task_id=task_id,
|
|
topic_id=data.topic_id,
|
|
stage=data.stage,
|
|
status="pending",
|
|
created_by=data.created_by or current_user.username
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
@router.get("/{task_id}", response_model=ContentTaskResponse)
|
|
def get_task(
|
|
task_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
task = db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="任务不存在")
|
|
return task
|
|
|
|
|
|
@router.put("/{task_id}/start")
|
|
def start_task(
|
|
task_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
task = db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="任务不存在")
|
|
|
|
from datetime import datetime, timezone
|
|
task.status = "running"
|
|
task.started_at = datetime.now(timezone.utc)
|
|
task.message = "任务已启动"
|
|
task.progress = 0
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
@router.put("/{task_id}/progress")
|
|
def update_progress(
|
|
task_id: str,
|
|
progress: int = Query(..., ge=0, le=100),
|
|
message: Optional[str] = None,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
task = db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="任务不存在")
|
|
|
|
task.progress = progress
|
|
if message:
|
|
task.message = message
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
@router.put("/{task_id}/complete")
|
|
def complete_task(
|
|
task_id: str,
|
|
result_data: dict = None,
|
|
message: Optional[str] = None,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
task = db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="任务不存在")
|
|
|
|
from datetime import datetime, timezone
|
|
finished = datetime.now(timezone.utc)
|
|
task.status = "completed"
|
|
task.finished_at = finished
|
|
task.progress = 100
|
|
if message:
|
|
task.message = message
|
|
if result_data:
|
|
task.result_data = result_data
|
|
if task.started_at:
|
|
task.duration = int((finished - task.started_at).total_seconds())
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
@router.put("/{task_id}/fail")
|
|
def fail_task(
|
|
task_id: str,
|
|
error_msg: str = Query(...),
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
task = db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="任务不存在")
|
|
|
|
from datetime import datetime, timezone
|
|
finished = datetime.now(timezone.utc)
|
|
task.status = "failed"
|
|
task.finished_at = finished
|
|
task.error_msg = error_msg
|
|
if task.started_at:
|
|
task.duration = int((finished - task.started_at).total_seconds())
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
@router.delete("/{task_id}")
|
|
def cancel_task(
|
|
task_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
task = db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="任务不存在")
|
|
|
|
task.status = "cancelled"
|
|
db.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/modules/{module_id}/detail")
|
|
def get_module_detail(module_id: str, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
|
|
"""获取定时任务模块的详情：输入参数、产出结果、运行历史"""
|
|
from datetime import datetime as dt_mod, date as date_mod
|
|
from pathlib import Path as PathMod
|
|
import json as json_mod
|
|
import re as re_mod
|
|
|
|
ROOT = PathMod(__file__).resolve().parents[4]
|
|
DATA_DIR = ROOT / "automation" / "data"
|
|
LOGS_DIR = ROOT / "automation" / "logs"
|
|
today_str = date_mod.today().isoformat()
|
|
|
|
try:
|
|
return _get_module_detail_data(module_id, db, ROOT, DATA_DIR, LOGS_DIR, today_str)
|
|
except Exception as e:
|
|
return {
|
|
"module_id": module_id,
|
|
"title": module_id,
|
|
"description": "",
|
|
"status": "stopped",
|
|
"inputs": {},
|
|
"outputs": {"error": str(e)},
|
|
"history": [],
|
|
"log_excerpt": "",
|
|
}
|
|
|
|
def _get_module_detail_data(module_id: str, db, ROOT, DATA_DIR, LOGS_DIR, today_str):
|
|
from datetime import datetime as dt_mod, date as date_mod
|
|
from pathlib import Path as PathMod
|
|
import json as json_mod
|
|
import re as re_mod
|
|
MODULE_META = {
|
|
"scheduled_refresh_search_cache": {"name": "🔍 搜索缓存", "description": "通过 opencode webfetch 联网搜索，刷新 8 个分类的搜索缓存，供内容采集器使用"},
|
|
"scheduled_fetch_trends": {"name": "🔥 热点趋势", "description": "从百度、微博、知乎实时热搜 API 抓取当天热点，LLM 补充，存入 trends.json"},
|
|
"scheduled_collect": {"name": "📡 内容采集", "description": "读取搜索缓存 + 热点趋势 + 历史表现 + AI 建议，经 LLM 分析后生成选题"},
|
|
"scheduled_generate": {"name": "🤖 内容创作", "description": "基于选题，LLM 生成三平台文章（知乎、微信、小红书），存入 articles 表"},
|
|
"scheduled_optimize": {"name": "🔍 合规审查", "description": "LLM 审查已创作文章，检查合规、打分数、优化建议"},
|
|
"scheduled_optimize_sources": {"name": "📡 信息源优化", "description": "AI 分析当前类别和信息源的市场匹配度，给出调整建议"},
|
|
"scheduled_metrics_sync": {"name": "📊 指标同步", "description": "从各平台公开 API 获取已发布文章的互动数据（点赞、阅读、评论等）"},
|
|
"scheduled_task_monitor": {"name": "⏰ 任务监控", "description": "每小时自动检查卡死/中断任务，标记为失败以便重新执行"},
|
|
}
|
|
|
|
meta = MODULE_META.get(module_id, {"name": module_id, "description": ""})
|
|
|
|
# Inputs
|
|
inputs = {}
|
|
outputs = {}
|
|
history = []
|
|
|
|
if module_id == "scheduled_refresh_search_cache":
|
|
cache_file = DATA_DIR / "search_cache.json"
|
|
if cache_file.exists():
|
|
try:
|
|
cache = json_mod.loads(cache_file.read_text(encoding="utf-8"))
|
|
meta_ = cache.pop("_metadata", {})
|
|
for q, results in cache.items():
|
|
inputs.setdefault("搜索词", []).append(q)
|
|
outputs.setdefault("各分类结果", []).append({
|
|
"query": q, "count": len(results),
|
|
"samples": [r.get("title","")[:50] for r in results[:3]]
|
|
})
|
|
outputs["更新时间"] = meta_.get("updated_at", "")
|
|
outputs["结果总数"] = sum(len(v) for v in cache.values())
|
|
except Exception:
|
|
pass
|
|
# try reading queries from yaml
|
|
try:
|
|
import yaml
|
|
cfg_path = ROOT / "config" / "sources.yaml"
|
|
if cfg_path.exists():
|
|
cfg = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
|
|
ws = cfg.get("sustainability_sources", {}).get("web_search", [])
|
|
if ws:
|
|
inputs["采集来源"] = f"config/sources.yaml ({len(ws)} 个搜索词)"
|
|
except Exception:
|
|
pass
|
|
|
|
elif module_id == "scheduled_fetch_trends":
|
|
trends_file = DATA_DIR / "trends.json"
|
|
if trends_file.exists():
|
|
try:
|
|
data = json_mod.loads(trends_file.read_text(encoding="utf-8"))
|
|
trends = data.get("trends", [])
|
|
outputs["采集日期"] = data.get("date", "")
|
|
outputs["更新时间"] = data.get("updated_at", "")
|
|
outputs["热点总数"] = len(trends)
|
|
by_source_clean = {}
|
|
for t in trends:
|
|
s = t.get("source", "")
|
|
label = {"weibo":"微博","zhihu":"知乎","baidu":"百度","llm":"LLM生成"}.get(s, s)
|
|
if label:
|
|
by_source_clean[label] = by_source_clean.get(label, 0) + 1
|
|
outputs["来源分布"] = by_source_clean
|
|
outputs["热点列表"] = [{"topic": t.get("topic",""), "domain": t.get("domain",""), "platform": t.get("platform",""), "source": {"weibo":"微博","zhihu":"知乎","baidu":"百度","llm":"LLM"}.get(t.get("source",""),"")} for t in trends[:20]]
|
|
except Exception:
|
|
pass
|
|
|
|
elif module_id == "scheduled_collect":
|
|
from ..models import CollectorCategory
|
|
pending = db.query(Topic).filter(Topic.status.in_(["pending", "待处理"])).count()
|
|
total_topics = db.query(Topic).count()
|
|
cats = db.query(CollectorCategory).filter(CollectorCategory.is_active == True).all()
|
|
inputs["采集类别"] = [c.name for c in cats]
|
|
inputs["采集数量"] = f"{len(cats)} 个类别"
|
|
outputs["选题总数"] = total_topics
|
|
outputs["待处理"] = pending
|
|
# recent topics
|
|
recent = db.query(Topic).order_by(Topic.created_at.desc()).limit(5).all()
|
|
outputs["最新选题"] = [{"id": t.id, "title": t.title, "field": t.field, "status": t.status, "created": t.created_at.isoformat() if t.created_at else ""} for t in recent]
|
|
|
|
elif module_id == "scheduled_generate":
|
|
review = db.query(Topic).filter(Topic.status.in_(["review", "待审查"])).count()
|
|
pending_t = db.query(Topic).filter(Topic.status.in_(["pending", "待处理"])).count()
|
|
inputs["待创作选题"] = pending_t
|
|
outputs["待审查"] = review
|
|
try:
|
|
from ..models import Article
|
|
recent_articles = db.query(Article, Topic.title.label("topic_title")).join(Topic, Article.topic_id == Topic.id, isouter=True).order_by(Article.created_at.desc()).limit(5).all()
|
|
outputs["最新文章"] = []
|
|
seen_articles = set()
|
|
for a in recent_articles:
|
|
art = a[0]
|
|
tid = a[1] if len(a) > 1 else ""
|
|
if art.id not in seen_articles:
|
|
seen_articles.add(art.id)
|
|
outputs["最新文章"].append({"id": art.id, "platform": art.platform, "topic": tid, "status": art.status, "created": art.created_at.isoformat() if art.created_at else ""})
|
|
except Exception as e:
|
|
outputs["最新文章_错误"] = str(e)
|
|
|
|
elif module_id == "scheduled_optimize":
|
|
outputs["待审查选题"] = db.query(Topic).filter(Topic.status.in_(["review", "待审查"])).count()
|
|
outputs["已审查"] = db.query(Topic).filter(Topic.status.in_(["ready", "待发布", "published", "已发布"])).count()
|
|
# read latest report
|
|
drafts_dir = DATA_DIR / "drafts"
|
|
if drafts_dir.exists():
|
|
dates = sorted([d for d in drafts_dir.iterdir() if d.is_dir() and d.name[:4].isdigit()], reverse=True)
|
|
if dates:
|
|
report_file = dates[0] / "optimization_report.json"
|
|
if report_file.exists():
|
|
try:
|
|
report = json_mod.loads(report_file.read_text(encoding="utf-8"))
|
|
summary = report.get("summary", {})
|
|
outputs["最新报告"] = {"total": summary.get("total_articles", 0), "avg_score": summary.get("average_score", 0), "date": dates[0].name}
|
|
if summary.get("articles"):
|
|
outputs["文章评分"] = [{"title": a.get("title","")[:30], "score": a.get("score","")} for a in summary["articles"][:5]]
|
|
except Exception:
|
|
pass
|
|
|
|
elif module_id == "scheduled_optimize_sources":
|
|
from ..models import SystemConfig, CollectorCategory as Cat2, CollectorSource as Src2
|
|
cats = db.query(Cat2).filter(Cat2.is_active == True).all()
|
|
srcs = db.query(Src2).filter(Src2.is_active == True).all()
|
|
inputs["当前类别"] = [c.name for c in cats]
|
|
inputs["当前信息源"] = [f"{s.name}({s.source_type})" for s in srcs]
|
|
sc = db.query(SystemConfig).filter(SystemConfig.key == "collector_ai_advice").first()
|
|
if sc and sc.value:
|
|
try:
|
|
advice = json_mod.loads(sc.value)
|
|
outputs["AI建议摘要"] = advice.get("summary", "")
|
|
outputs["建议新增类别"] = advice.get("suggested_new_categories", [])
|
|
outputs["建议新增源"] = advice.get("suggested_new_sources", [])
|
|
outputs["类别评估"] = advice.get("category_assessment", [])
|
|
except Exception:
|
|
pass
|
|
|
|
elif module_id == "scheduled_metrics_sync":
|
|
from ..models import ContentMetrics as CM
|
|
total = db.query(CM).count()
|
|
outputs["已同步文章"] = total
|
|
feedback_file = DATA_DIR / "metrics_feedback.json"
|
|
if feedback_file.exists():
|
|
try:
|
|
fb = json_mod.loads(feedback_file.read_text(encoding="utf-8"))
|
|
outputs["高互动领域"] = fb.get("top_domains", [])
|
|
outputs["详情"] = fb.get("detail", [])
|
|
except Exception:
|
|
pass
|
|
latest = db.query(CM, Topic.title.label("t")).join(Topic, CM.topic_id == Topic.id, isouter=True).order_by(CM.last_fetched.desc()).limit(5).all()
|
|
outputs["最新指标"] = []
|
|
for row in latest:
|
|
m = row.CM if hasattr(row, 'CM') else row[0]
|
|
ttl = row.t if hasattr(row, 't') else (row[1] if len(row) > 1 else "")
|
|
outputs["最新指标"].append({"topic": ttl, "views": m.views, "likes": m.likes, "comments": m.comments, "favorites": m.favorites, "fetched": m.last_fetched.isoformat() if m.last_fetched else ""})
|
|
|
|
# History from log files
|
|
log_map = {
|
|
"scheduled_refresh_search_cache": LOGS_DIR / f"opencode_search_{today_str}.log",
|
|
"scheduled_fetch_trends": LOGS_DIR / f"trends_{today_str}.log",
|
|
"scheduled_collect": LOGS_DIR / f"collector_{today_str}.log",
|
|
"scheduled_generate": LOGS_DIR / f"creator_{today_str}.log",
|
|
"scheduled_optimize": LOGS_DIR / f"optimizer_{today_str}.log",
|
|
"scheduled_optimize_sources": LOGS_DIR / f"optimizer_sources_{today_str}.log",
|
|
"scheduled_metrics_sync": LOGS_DIR / f"metrics_sync_{today_str}.log",
|
|
}
|
|
log_file = log_map.get(module_id)
|
|
log_lines = []
|
|
if log_file and log_file.exists():
|
|
content = log_file.read_text(encoding="utf-8", errors="ignore")
|
|
log_lines = content.splitlines()[-30:]
|
|
# Build history from log entries
|
|
for line in log_lines:
|
|
m = re_mod.match(r'^(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}).*', line)
|
|
if m:
|
|
ts = m.group(1)
|
|
is_success = "完成" in line or "success" in line.lower() or "SUCCESS" in line
|
|
is_error = "失败" in line or "failed" in line.lower() or "ERROR" in line
|
|
if is_success or is_error:
|
|
history.append({"time": ts, "status": "success" if is_success else "error", "msg": line.strip()[-60:]})
|
|
history = history[-10:]
|
|
|
|
# Log excerpt
|
|
log_excerpt = "\n".join(log_lines[-15:]) if log_lines else ""
|
|
|
|
return {
|
|
"module_id": module_id,
|
|
"title": meta["name"],
|
|
"description": meta["description"],
|
|
"status": "running", # client will override from modules/status
|
|
"inputs": inputs,
|
|
"outputs": outputs,
|
|
"history": history,
|
|
"log_excerpt": log_excerpt,
|
|
}
|
|
|
|
@router.post("/run-creator")
|
|
def run_creator_task(
|
|
topic_id: Optional[str] = None,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(get_current_user)
|
|
):
|
|
from datetime import datetime, timezone
|
|
|
|
now = datetime.now(timezone.utc)
|
|
task_id = f"task_{uuid.uuid4().hex[:16]}"
|
|
|
|
task = ContentTask(
|
|
task_id=task_id,
|
|
topic_id=topic_id,
|
|
stage="creator",
|
|
status="running",
|
|
started_at=now,
|
|
created_by=current_user.username
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
|
|
def _run():
|
|
from ..database import SessionLocal
|
|
from ..core.generator import run_creator_blocking
|
|
from datetime import datetime, timezone
|
|
import traceback
|
|
_creator_semaphore.acquire()
|
|
try:
|
|
new_db = SessionLocal()
|
|
try:
|
|
new_task = new_db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if new_task:
|
|
new_task.message = "创作脚本运行中..."
|
|
new_task.progress = 30
|
|
new_db.commit()
|
|
|
|
result = run_creator_blocking(topic_id)
|
|
|
|
new_task = new_db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if new_task:
|
|
finished = datetime.now(timezone.utc)
|
|
new_task.status = "completed"
|
|
new_task.finished_at = finished
|
|
new_task.progress = 100
|
|
new_task.message = "创作完成"
|
|
new_task.result_data = {"stdout": (result or {}).get("stdout", "")[:2000]} if isinstance(result, dict) else {"raw": str(result)[:2000]}
|
|
if new_task.started_at:
|
|
new_task.duration = int((finished - new_task.started_at).total_seconds())
|
|
new_db.commit()
|
|
except Exception as e:
|
|
new_db.rollback()
|
|
new_task = new_db.query(ContentTask).filter(ContentTask.task_id == task_id).first()
|
|
if new_task:
|
|
finished = datetime.now(timezone.utc)
|
|
new_task.status = "failed"
|
|
new_task.finished_at = finished
|
|
new_task.error_msg = f"{type(e).__name__}: {e}\n{traceback.format_exc()}"
|
|
if new_task.started_at:
|
|
new_task.duration = int((finished - new_task.started_at).total_seconds())
|
|
new_db.commit()
|
|
finally:
|
|
new_db.close()
|
|
finally:
|
|
_creator_semaphore.release()
|
|
|
|
thread = threading.Thread(target=_run, daemon=True)
|
|
thread.start()
|
|
|
|
return task |