Yuzhiran Dev
c6628b22f6
- 任务抽屉产出结果 tab 优先显示 last_message（日志消息）， 过滤 result_data 中的 None/空值 - 修复 tasks.py 因 opencode 块删除导致的 SyntaxError - 修复 scheduled_generate 存储 None 值问题 - 清理 DB 中旧 scheduled_refresh_search_cache 的 56 条 TaskLog + 1 条 TaskConfig - 更新 PROGRESS.md
590 lines
29 KiB
Python
590 lines
29 KiB
Python
"""
|
|
定时任务调度器
|
|
基于 APScheduler，支持在 FastAPI 生命周期内运行定时任务
|
|
"""
|
|
import os, sys, logging, json
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent.parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT / 'scripts'))
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
from prompt_loader import get_prompt, get_prompt_params
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
from .generator import run_creator_blocking
|
|
from .optimizer import run_optimizer_blocking
|
|
from .collector import run_collector_blocking
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def _set_task_llm_provider(module_id: str):
|
|
"""从 TaskConfig 读取 llm_provider 并设为环境变量，供子进程和 call_llm 读取"""
|
|
try:
|
|
from ..database import SessionLocal
|
|
from ..models import TaskConfig
|
|
db = SessionLocal()
|
|
cfg = db.query(TaskConfig).filter(TaskConfig.module_id == module_id).first()
|
|
db.close()
|
|
if cfg and cfg.params:
|
|
provider = cfg.params.get("llm_provider")
|
|
if provider:
|
|
os.environ["LLM_TASK_PROVIDER"] = provider
|
|
logger.debug("[%s] LLM provider set to %s", module_id, provider)
|
|
return
|
|
except Exception:
|
|
pass
|
|
os.environ.pop("LLM_TASK_PROVIDER", None)
|
|
|
|
MODULES = {
|
|
"scheduled_fetch_trends": {"name": "🔥 热点趋势", "cron": "01:10"},
|
|
"scheduled_collect": {"name": "📡 内容采集", "cron": "01:30"},
|
|
"scheduled_generate": {"name": "🤖 内容创作", "cron": "02:00"},
|
|
"scheduled_optimize": {"name": "🔍 合规审查", "cron": "03:00"},
|
|
"scheduled_optimize_sources": {"name": "📡 信息源优化", "cron": "05:00"},
|
|
"scheduled_metrics_sync": {"name": "📊 指标同步", "cron": "06:00"},
|
|
"scheduled_reset_search_usage": {"name": "🔁 搜索用量重置", "cron": "00:05"},
|
|
"scheduled_task_monitor": {"name": "⏰ 任务监控", "cron": "*"},
|
|
}
|
|
|
|
LOG_FILE_MAP = {
|
|
"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_id: int = None):
|
|
"""写入/更新 TaskLog。如果 log_id 不为 None 则 UPDATE 已有行，否则 INSERT。返回 log_id。"""
|
|
_log_to_file(module_id, status, message, error_trace)
|
|
try:
|
|
from ..database import SessionLocal
|
|
from ..models import TaskLog
|
|
db = SessionLocal()
|
|
try:
|
|
duration = None
|
|
if started_at and finished_at:
|
|
duration = int((finished_at - started_at).total_seconds())
|
|
if log_id:
|
|
log = db.query(TaskLog).filter(TaskLog.id == log_id).first()
|
|
if log:
|
|
log.status = status
|
|
log.message = message
|
|
if error_trace:
|
|
log.error_trace = error_trace
|
|
if result_data is not None:
|
|
log.result_data = result_data
|
|
if finished_at:
|
|
log.finished_at = finished_at
|
|
if duration is not None:
|
|
log.duration = duration
|
|
db.commit()
|
|
return log_id
|
|
log = TaskLog(
|
|
module_id=module_id,
|
|
task_name=MODULES.get(module_id, {}).get("name", module_id),
|
|
status=status,
|
|
message=message,
|
|
error_trace=error_trace,
|
|
triggered_by=triggered_by,
|
|
result_data=result_data or {},
|
|
started_at=started_at or datetime.now(timezone.utc),
|
|
finished_at=finished_at,
|
|
duration=duration,
|
|
next_run_time=next_run_time,
|
|
)
|
|
db.add(log)
|
|
db.commit()
|
|
db.refresh(log)
|
|
return log.id
|
|
finally:
|
|
db.close()
|
|
except Exception:
|
|
pass
|
|
|
|
def _wrap_task(module_id: str, target_fn, *args, **kwargs):
|
|
started = datetime.now(timezone.utc)
|
|
status = "running"
|
|
error_trace = None
|
|
result_data = None
|
|
try:
|
|
result = target_fn(*args, **kwargs)
|
|
status = "success"
|
|
if isinstance(result, dict):
|
|
result_data = {k: v for k, v in result.items() if isinstance(v, (str, int, float, bool, list, dict)) and k not in ("stdout", "stderr")}
|
|
return result
|
|
except Exception as e:
|
|
status = "failed"
|
|
import traceback
|
|
error_trace = traceback.format_exc()
|
|
raise
|
|
finally:
|
|
_log_task(module_id, status=status, message=None, error_trace=error_trace,
|
|
result_data=result_data, started_at=started,
|
|
finished_at=datetime.now(timezone.utc))
|
|
|
|
class TaskScheduler:
|
|
def __init__(self):
|
|
self.scheduler = BackgroundScheduler()
|
|
self._started = False
|
|
|
|
def start(self):
|
|
if self._started:
|
|
logger.warning("Scheduler already started")
|
|
return
|
|
from ..database import SessionLocal
|
|
from ..models import TaskConfig
|
|
db = SessionLocal()
|
|
try:
|
|
configs = {c.module_id: c for c in db.query(TaskConfig).all()}
|
|
finally:
|
|
db.close()
|
|
|
|
MODULE_JOBS = [
|
|
("scheduled_fetch_trends", self._run_fetch_trends, "热点趋势"),
|
|
("scheduled_collect", self._run_collect, "内容采集"),
|
|
("scheduled_generate", self._run_generate, "内容创作"),
|
|
("scheduled_optimize", self._run_optimize, "合规审查"),
|
|
("scheduled_optimize_sources", self._run_optimize_sources, "信息源优化"),
|
|
("scheduled_metrics_sync", self._run_metrics_sync, "指标同步"),
|
|
("scheduled_reset_search_usage", self._run_reset_search_usage, "搜索用量重置"),
|
|
]
|
|
|
|
for module_id, fn, name in MODULE_JOBS:
|
|
cfg = configs.get(module_id)
|
|
if cfg and not cfg.enabled:
|
|
logger.info(f"跳过禁用任务: {module_id}")
|
|
continue
|
|
schedule = (cfg.schedule if cfg else None) or MODULES.get(module_id, {}).get("cron", "01:00")
|
|
try:
|
|
hour, minute = map(int, schedule.split(":"))
|
|
except (ValueError, AttributeError):
|
|
hour, minute = 1, 0
|
|
self.scheduler.add_job(
|
|
fn,
|
|
CronTrigger(hour=hour, minute=minute),
|
|
id=module_id,
|
|
replace_existing=True,
|
|
max_instances=1,
|
|
coalesce=True
|
|
)
|
|
logger.info(f"调度任务: {module_id} -> {schedule}")
|
|
|
|
# 每小时运行的任务监控：检测卡死/中断任务
|
|
self.scheduler.add_job(
|
|
self._run_task_monitor,
|
|
CronTrigger(hour='*/1'),
|
|
id='scheduled_task_monitor',
|
|
replace_existing=True,
|
|
max_instances=1,
|
|
coalesce=True
|
|
)
|
|
logger.info("调度任务: scheduled_task_monitor -> 每小时")
|
|
|
|
self.scheduler.start()
|
|
self._started = True
|
|
logger.info("Scheduler started with dynamic schedule from TaskConfig")
|
|
def shutdown(self):
|
|
if self.scheduler.running:
|
|
self.scheduler.shutdown()
|
|
logger.info("Scheduler shut down")
|
|
|
|
def _run_fetch_trends(self):
|
|
"""定时刷新热点趋势（百度/微博/知乎实时热搜 + LLM补充）"""
|
|
_set_task_llm_provider("scheduled_fetch_trends")
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _log_task("scheduled_fetch_trends", "running", started_at=started)
|
|
try:
|
|
logger.info("[Scheduled] Fetching hot trends...")
|
|
import subprocess
|
|
result = subprocess.run(
|
|
[sys.executable, str(PROJECT_ROOT / "scripts" / "trends.py")],
|
|
capture_output=True, text=True, timeout=120
|
|
)
|
|
if result.returncode == 0:
|
|
for line in result.stdout.strip().split("\n"):
|
|
if line.strip():
|
|
logger.info("[Trends] %s", line.strip())
|
|
_log_task("scheduled_fetch_trends", "success", log_id=log_id,
|
|
message="趋势刷新成功",
|
|
result_data={"output_lines": len(result.stdout.splitlines())},
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
else:
|
|
_log_task("scheduled_fetch_trends", "failed", log_id=log_id,
|
|
message=f"返回码 {result.returncode}",
|
|
error_trace=result.stderr[-500:],
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
except Exception as e:
|
|
_log_task("scheduled_fetch_trends", "failed", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.exception("[Scheduled] Trends refresh error: %s", e)
|
|
|
|
def _run_generate(self):
|
|
_set_task_llm_provider("scheduled_generate")
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _log_task("scheduled_generate", "running", started_at=started)
|
|
try:
|
|
logger.info("[Scheduled] Starting content generation...")
|
|
result = run_creator_blocking()
|
|
logger.info("[Scheduled] Generation completed: %s", result)
|
|
created_id = result.get("topic_id") if isinstance(result, dict) else None
|
|
review_result = None
|
|
if created_id:
|
|
logger.info("[Scheduled] Running compliance review on %s...", created_id)
|
|
review_result = run_optimizer_blocking([created_id])
|
|
if review_result.get("ok"):
|
|
logger.info("[Scheduled] Review completed for %s", created_id)
|
|
_log_task("scheduled_generate", "success", log_id=log_id,
|
|
message=f"创作完成" + (f", 选题 {created_id}" if created_id else ""),
|
|
result_data={"topic_id": created_id} if created_id else {"note": "无待生成的选题"},
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
except Exception as e:
|
|
_log_task("scheduled_generate", "failed", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.exception("[Scheduled] Generation pipeline failed: %s", e)
|
|
|
|
def _run_optimize(self):
|
|
_set_task_llm_provider("scheduled_optimize")
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _log_task("scheduled_optimize", "running", started_at=started)
|
|
try:
|
|
logger.info("[Scheduled] Starting compliance review...")
|
|
result = run_optimizer_blocking()
|
|
_log_task("scheduled_optimize", "success", log_id=log_id,
|
|
message="合规审查完成",
|
|
result_data={"processed": result.get("processed", 0), "passed": result.get("passed", 0)},
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.info("[Scheduled] Review completed: %s", result)
|
|
except Exception as e:
|
|
_log_task("scheduled_optimize", "failed", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.exception("[Scheduled] Review failed: %s", e)
|
|
|
|
def _run_collect(self):
|
|
_set_task_llm_provider("scheduled_collect")
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _log_task("scheduled_collect", "running", started_at=started)
|
|
try:
|
|
logger.info("[Scheduled] Starting topic collection...")
|
|
result = run_collector_blocking()
|
|
topics_count = result.get("topics_count", 0)
|
|
_log_task("scheduled_collect", "success", log_id=log_id,
|
|
message=f"采集完成，找到 {topics_count} 个选题",
|
|
result_data={"topics_count": topics_count, "output": str(result.get("output", ""))[:200]},
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.info("[Scheduled] Collection completed: %s", result.get("output", "")[-200:])
|
|
except Exception as e:
|
|
_log_task("scheduled_collect", "failed", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.exception("[Scheduled] Collection failed: %s", e)
|
|
|
|
def _run_optimize_sources(self, triggered_by="scheduler"):
|
|
"""AI自动优化采集类别与信息源：对比市场热点和当前配置，给出调整建议"""
|
|
_set_task_llm_provider("scheduled_optimize_sources")
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _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
|
|
from ..database import SessionLocal
|
|
from ..models import CollectorCategory, CollectorSource
|
|
from datetime import date
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
cats = db.query(CollectorCategory).filter(CollectorCategory.is_active == True).all()
|
|
sources = db.query(CollectorSource).filter(CollectorSource.is_active == True).all()
|
|
except Exception:
|
|
logger.warning("[Scheduled] DB not ready for source optimization")
|
|
db.close()
|
|
_log_task("scheduled_optimize_sources", "failed", log_id=log_id,
|
|
message="数据库未就绪", started_at=started, finished_at=datetime.now(timezone.utc))
|
|
return
|
|
|
|
cat_names = [c.name for c in cats]
|
|
src_summary = "\n".join(f"- [{s.source_type}] {s.name}: {s.query or s.url or ''}" for s in sources)
|
|
|
|
prompt = get_prompt("sources_optimization",
|
|
n=len(cat_names),
|
|
cat_names="\n".join(f"- {n}" for n in cat_names),
|
|
n2=len(sources),
|
|
src_summary=src_summary,
|
|
year=datetime.now().year,
|
|
)
|
|
|
|
params = get_prompt_params("sources_optimization")
|
|
resp = call_llm(prompt, temperature=params.get("temperature", 0.5), max_tokens=params.get("max_tokens", 3000))
|
|
if resp.startswith("```"):
|
|
resp = resp.split("\n", 1)[1].rsplit("\n", 1)[0]
|
|
result = json.loads(resp)
|
|
|
|
# 将AI建议写入系统配置（供运营参考，不自动执行）
|
|
from ..models import SystemConfig
|
|
sc = db.query(SystemConfig).filter(SystemConfig.key == "collector_ai_advice").first()
|
|
if sc:
|
|
sc.value = json.dumps(result, ensure_ascii=False)
|
|
else:
|
|
db.add(SystemConfig(key="collector_ai_advice", value=json.dumps(result, ensure_ascii=False), description="AI每日采集优化建议"))
|
|
db.commit()
|
|
logger.info("[Scheduled] Source AI optimization completed: %s", result.get("summary", ""))
|
|
_log_task("scheduled_optimize_sources", "success", log_id=log_id,
|
|
message=result.get("summary", "优化完成"),
|
|
result_data={"categories_assessed": len(result.get("category_assessment", [])),
|
|
"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), triggered_by=triggered_by)
|
|
db.close()
|
|
except Exception as e:
|
|
_log_task("scheduled_optimize_sources", "failed", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
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, triggered_by="scheduler"):
|
|
"""定时从各平台公开API获取发布文章的效果数据（当前仅支持知乎）"""
|
|
_set_task_llm_provider("scheduled_metrics_sync")
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _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
|
|
from ..models import Topic, ContentMetrics
|
|
import re, requests as http_requests
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
topics = db.query(Topic).filter(
|
|
Topic.status.in_(["published", "已发布"])
|
|
).all()
|
|
except Exception:
|
|
logger.warning("[Scheduled] DB not ready for metrics sync")
|
|
db.close()
|
|
_log_task("scheduled_metrics_sync", "failed", log_id=log_id,
|
|
message="数据库未就绪", started_at=started, finished_at=datetime.now(timezone.utc))
|
|
return
|
|
|
|
ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
|
|
count = 0
|
|
for topic in topics:
|
|
platform_urls = topic.platform_urls or {}
|
|
zhihu_url = platform_urls.get("zhihu", "")
|
|
if not zhihu_url:
|
|
continue
|
|
m = re.search(r'zhuanlan\.zhihu\.com/p/(\d+)', zhihu_url)
|
|
if not m:
|
|
continue
|
|
post_id = m.group(1)
|
|
api_url = f"https://zhuanlan.zhihu.com/api/posts/{post_id}"
|
|
try:
|
|
resp = http_requests.get(api_url, headers={"User-Agent": ua}, timeout=10)
|
|
if resp.status_code != 200:
|
|
continue
|
|
raw = resp.json()
|
|
existing = db.query(ContentMetrics).filter(
|
|
ContentMetrics.topic_id == topic.id,
|
|
ContentMetrics.platform == "zhihu"
|
|
).first()
|
|
metric_data = {
|
|
"views": raw.get("voteup_count", raw.get("views_count", 0)),
|
|
"likes": raw.get("voteup_count", 0),
|
|
"favorites": raw.get("favorite_count", 0),
|
|
"comments": raw.get("comment_count", raw.get("comments_count", 0)),
|
|
"shares": raw.get("share_count", 0),
|
|
"last_fetched": datetime.now(),
|
|
"publish_url": zhihu_url,
|
|
"data_snapshot": raw,
|
|
}
|
|
if existing:
|
|
for k, v in metric_data.items():
|
|
setattr(existing, k, v)
|
|
else:
|
|
db.add(ContentMetrics(topic_id=topic.id, platform="zhihu", **metric_data))
|
|
count += 1
|
|
except Exception:
|
|
continue
|
|
if count:
|
|
db.commit()
|
|
logger.info("[Scheduled] Metrics sync completed: synced %d zhihu articles", count)
|
|
_log_task("scheduled_metrics_sync", "success", log_id=log_id,
|
|
message=f"同步完成，{count} 篇知乎文章",
|
|
result_data={"articles_synced": count},
|
|
started_at=started, finished_at=datetime.now(timezone.utc), triggered_by=triggered_by)
|
|
# 生成指标反馈：按 field 聚合表现，写入 metrics_feedback.json 供 collector 读取
|
|
try:
|
|
import json as json_mod
|
|
from sqlalchemy import func as sql_func
|
|
feedback = db.query(
|
|
Topic.field,
|
|
sql_func.avg(ContentMetrics.likes).label("avg_likes"),
|
|
sql_func.avg(ContentMetrics.views).label("avg_views"),
|
|
sql_func.avg(ContentMetrics.comments).label("avg_comments"),
|
|
sql_func.count(ContentMetrics.id).label("article_count"),
|
|
).join(ContentMetrics, ContentMetrics.topic_id == Topic.id
|
|
).filter(Topic.field.isnot(None), Topic.field != ""
|
|
).group_by(Topic.field).all()
|
|
if feedback:
|
|
scored = []
|
|
for row in feedback:
|
|
score = (row.avg_likes or 0) + (row.avg_views or 0) * 0.01 + (row.avg_comments or 0) * 2
|
|
scored.append((row.field, round(score, 1), int(row.article_count)))
|
|
scored.sort(key=lambda x: x[1], reverse=True)
|
|
feedback_data = {
|
|
"updated_at": datetime.now().isoformat(),
|
|
"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 = 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])
|
|
except Exception as e_fb:
|
|
logger.warning("[Scheduled] Metrics feedback generation failed: %s", e_fb)
|
|
else:
|
|
_log_task("scheduled_metrics_sync", "success", log_id=log_id,
|
|
message="无已发布的知乎文章",
|
|
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", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
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):
|
|
"""每日凌晨重置搜索 API 提供商用量计数"""
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _log_task("scheduled_reset_search_usage", "running", started_at=started)
|
|
try:
|
|
from ..database import SessionLocal
|
|
from ..models import SearchProvider
|
|
db = SessionLocal()
|
|
try:
|
|
total = db.query(SearchProvider).update({SearchProvider.usage_today: 0, SearchProvider.last_used_at: None})
|
|
db.commit()
|
|
_log_task("scheduled_reset_search_usage", "success", log_id=log_id,
|
|
message=f"已重置 {total} 个提供商用量",
|
|
result_data={"reset_count": total},
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.info("[Scheduled] Reset %d search providers usage", total)
|
|
finally:
|
|
db.close()
|
|
except Exception as e:
|
|
_log_task("scheduled_reset_search_usage", "failed", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.exception("[Scheduled] Reset search usage failed: %s", e)
|
|
|
|
def _run_task_monitor(self):
|
|
"""每小时检查卡死/中断的任务，标记为失败"""
|
|
started = datetime.now(timezone.utc)
|
|
log_id = _log_task("scheduled_task_monitor", "running", started_at=started)
|
|
stuck_tasklog_timeout = 7200 # 超过2小时视为卡死
|
|
stuck_contenttask_timeout = 10800 # 超过3小时视为卡死
|
|
try:
|
|
from ..database import SessionLocal
|
|
from ..models import TaskLog, ContentTask
|
|
db = SessionLocal()
|
|
try:
|
|
now = datetime.now(timezone.utc)
|
|
cutoff_tasklog = now.timestamp() - stuck_tasklog_timeout
|
|
cutoff_content = now.timestamp() - stuck_contenttask_timeout
|
|
marked = 0
|
|
|
|
# 检查 TaskLog 中卡死的 running 记录
|
|
stuck_logs = db.query(TaskLog).filter(
|
|
TaskLog.status == "running",
|
|
TaskLog.started_at.isnot(None)
|
|
).all()
|
|
for log in stuck_logs:
|
|
if log.started_at.timestamp() < cutoff_tasklog:
|
|
log.status = "failed"
|
|
log.finished_at = now
|
|
log.error_trace = "系统监控：任务运行超时（超过2小时）或进程中断，已自动标记为失败"
|
|
if log.started_at:
|
|
log.duration = int((now - log.started_at).total_seconds())
|
|
marked += 1
|
|
logger.warning("[TaskMonitor] 标记 TaskLog %d (%s) 为失败（超时）", log.id, log.module_id)
|
|
|
|
# 检查 ContentTask 中卡死的 running 记录
|
|
stuck_tasks = db.query(ContentTask).filter(
|
|
ContentTask.status == "running",
|
|
ContentTask.started_at.isnot(None)
|
|
).all()
|
|
for task in stuck_tasks:
|
|
if task.started_at.timestamp() < cutoff_content:
|
|
task.status = "failed"
|
|
task.finished_at = now
|
|
task.error_msg = "系统监控：任务运行超时（超过3小时）或进程中断，已自动标记为失败"
|
|
if task.started_at:
|
|
task.duration = int((now - task.started_at).total_seconds())
|
|
marked += 1
|
|
logger.warning("[TaskMonitor] 标记 ContentTask %s (%s) 为失败（超时）", task.task_id, task.stage)
|
|
|
|
if marked:
|
|
db.commit()
|
|
logger.info("[TaskMonitor] 已标记 %d 个卡死任务为失败", marked)
|
|
|
|
_log_task("scheduled_task_monitor", "success", log_id=log_id,
|
|
message=f"检查完成，标记 {marked} 个卡死任务",
|
|
result_data={"marked_failed": marked},
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
finally:
|
|
db.close()
|
|
except Exception as e:
|
|
import traceback
|
|
_log_task("scheduled_task_monitor", "failed", log_id=log_id,
|
|
message=str(e),
|
|
error_trace=traceback.format_exc(),
|
|
started_at=started, finished_at=datetime.now(timezone.utc))
|
|
logger.exception("[TaskMonitor] 监控检查失败: %s", e)
|
|
|
|
def get_jobs(self):
|
|
"""返回当前所有定时任务的状态"""
|
|
jobs = []
|
|
for job in self.scheduler.get_jobs():
|
|
jobs.append({
|
|
"id": job.id,
|
|
"next_run_time": job.next_run_time.isoformat() if job.next_run_time else None,
|
|
"trigger": str(job.trigger),
|
|
})
|
|
return jobs
|
|
|
|
# 全局单例
|
|
scheduler = TaskScheduler() |