Yuzhiran Dev
468c9066cb
_log_task 在任务开始（running）和结束（success/failed）时分别 INSERT，导致 running 行永不更新，被 task monitor 标记为失败。 修复：新增 log_id 参数，传入时 UPDATE 已有行。所有 _run_* 方法 save running log_id 并在结束时传入。早期 return 处也标记失败。
610 lines
30 KiB
Python
610 lines
30 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__)
|
||
|
||
MODULES = {
|
||
"scheduled_refresh_search_cache": {"name": "🔍 搜索缓存", "cron": "01:00"},
|
||
"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_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_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_refresh_search_cache", self._run_refresh_search_cache, "搜索缓存"),
|
||
("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补充）"""
|
||
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_refresh_search_cache(self):
|
||
"""定时刷新搜索缓存（通过 opencode webfetch）"""
|
||
started = datetime.now(timezone.utc)
|
||
log_id = _log_task("scheduled_refresh_search_cache", "running", started_at=started)
|
||
try:
|
||
logger.info("[Scheduled] Refreshing search cache via opencode...")
|
||
import subprocess
|
||
result = subprocess.run(
|
||
[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"):
|
||
if line.strip():
|
||
logger.info("[SearchCache] %s", line.strip())
|
||
for line in result.stderr.strip().split("\n"):
|
||
if line.strip():
|
||
logger.warning("[SearchCache] %s", line.strip())
|
||
if result.returncode == 0:
|
||
_log_task("scheduled_refresh_search_cache", "success", log_id=log_id,
|
||
message="搜索缓存刷新成功",
|
||
result_data={"output_lines": len(result.stdout.splitlines())},
|
||
started_at=started, finished_at=datetime.now(timezone.utc))
|
||
logger.info("[Scheduled] Search cache refreshed")
|
||
else:
|
||
_log_task("scheduled_refresh_search_cache", "failed", log_id=log_id,
|
||
message="部分失败",
|
||
error_trace=result.stderr[-500:],
|
||
started_at=started, finished_at=datetime.now(timezone.utc))
|
||
logger.warning("[Scheduled] Search cache refresh may have partial failures")
|
||
except subprocess.TimeoutExpired:
|
||
_log_task("scheduled_refresh_search_cache", "failed", log_id=log_id,
|
||
message="超时",
|
||
started_at=started, finished_at=datetime.now(timezone.utc))
|
||
logger.warning("[Scheduled] Search cache refresh timed out")
|
||
except Exception as e:
|
||
_log_task("scheduled_refresh_search_cache", "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] Search cache refresh error: %s", e)
|
||
|
||
def _run_generate(self):
|
||
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, "review_ok": review_result.get("ok") if review_result else None},
|
||
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):
|
||
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):
|
||
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自动优化采集类别与信息源：对比市场热点和当前配置，给出调整建议"""
|
||
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获取发布文章的效果数据（当前仅支持知乎）"""
|
||
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() |