""" 定时任务调度器 基于 APScheduler，支持在 FastAPI 生命周期内运行定时任务 """ import os import logging from datetime import datetime 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 logger = logging.getLogger(__name__) class TaskScheduler: def __init__(self): self.scheduler = BackgroundScheduler() self._started = False def start(self): if self._started: logger.warning("Scheduler already started") return # 使用 CronTrigger 设置每日固定时间点 self.scheduler.add_job( self._run_collect, CronTrigger(hour=1, minute=30), id='scheduled_collect', ) self.scheduler.add_job( self._run_generate, CronTrigger(hour=3, minute=30), id='scheduled_generate', replace_existing=True, max_instances=1, coalesce=True ) self.scheduler.add_job( self._run_optimize, CronTrigger(hour=4, minute=30), id='scheduled_optimize', replace_existing=True, max_instances=1, coalesce=True ) self.scheduler.add_job( self._run_optimize_sources, CronTrigger(hour=5, minute=0), id='scheduled_optimize_sources', replace_existing=True, max_instances=1, coalesce=True ) self.scheduler.add_job( self._run_metrics_sync, CronTrigger(hour=6, minute=0), id='scheduled_metrics_sync', replace_existing=True, max_instances=1, coalesce=True ) self.scheduler.start() self._started = True logger.info("Scheduler started: 01:30 collect, 03:30 generate, 04:30 review, 05:00 optimize_sources, 06:00 metrics_sync") def shutdown(self): if self.scheduler.running: self.scheduler.shutdown() logger.info("Scheduler shut down") def _run_generate(self): try: logger.info("[Scheduled] Starting content generation...") result = run_creator() 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]) if review_result.get("ok"): logger.info("[Scheduled] Review completed for %s", created_id) else: logger.warning("[Scheduled] Review failed: %s", review_result.get("error")) except Exception as e: logger.exception("[Scheduled] Generation pipeline failed: %s", e) def _run_optimize(self): try: logger.info("[Scheduled] Starting compliance review...") result = run_optimizer() logger.info("[Scheduled] Review completed: %s", result) except Exception as e: logger.exception("[Scheduled] Review failed: %s", e) def _run_collect(self): try: logger.info("[Scheduled] Starting topic collection...") result = run_collector() logger.info("[Scheduled] Collection completed: %s", result.get("output", "")[-200:]) except Exception as e: logger.exception("[Scheduled] Collection failed: %s", e) def _run_optimize_sources(self): """AI自动优化采集类别与信息源：对比市场热点和当前配置，给出调整建议""" 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() 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 = f"""你是一个内容策略分析师。分析当前中文互联网可持续生活领域的真实热点，与以下配置进行对比。 当前配置的类别（{len(cat_names)}个）： {chr(10).join(f'- {n}' for n in cat_names)} 当前配置的信息源（{len(sources)}个）： {src_summary} 请完成以下任务： 1. 评估每个类别是否仍符合2026年中国市场真实热点（基于你的知识） 2. 评估每个信息源是否可能在中国正常访问 3. 建议新增或删除的类别（最多2条） 4. 建议新增的信息源搜索词（最多3条，包含具体搜索词） 输出 JSON 格式： {{ "category_assessment": [{{"name": "类别名", "status": "保留/淘汰/合并", "reason": "原因"}}], "source_assessment": [{{"name": "源名", "status": "保留/淘汰/替换", "reason": "原因"}}], "suggested_new_categories": [{{"name": "类别名", "search_query": "搜索词", "reason": "推荐原因"}}], "suggested_new_sources": [{{"name": "源名", "type": "web_search", "query": "搜索词", "focus": "聚焦领域"}}], "summary": "一句话总结本次优化建议" }} 只输出JSON，不要其他文字。""" resp = call_llm(prompt, temperature=0.5, max_tokens=2000) 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", "")) db.close() except Exception as e: logger.exception("[Scheduled] Source AI optimization failed: %s", e) def _run_metrics_sync(self): """定时从各平台公开API获取发布文章的效果数据（当前仅支持知乎）""" 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() 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) else: logger.info("[Scheduled] Metrics sync: no zhihu articles to sync") db.close() except Exception as e: logger.exception("[Scheduled] Metrics sync failed: %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()