""" 定时任务调度器 基于 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 .sync import sync_all_topics 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', replace_existing=True, max_instances=1, coalesce=True ) self.scheduler.add_job( self._run_sync, CronTrigger(hour=2, minute=30), id='scheduled_sync', replace_existing=True, max_instances=1, coalesce=True ) 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 with daily cron triggers (01:30 collect, 02:30 sync, 03:30 generate, 04:30 optimize, 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 optimization...") result = run_optimizer() logger.info("[Scheduled] Optimization completed: %s", result) except Exception as e: logger.exception("[Scheduled] Optimization 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_sync(self): try: logger.info("[Scheduled] Starting data sync...") sync_all_topics() logger.info("[Scheduled] Sync completed") except Exception as e: logger.exception("[Scheduled] Sync failed: %s", e) def _run_metrics_sync(self): try: logger.info("[Scheduled] Starting metrics sync...") from ..database import SessionLocal from ..models import Topic, ContentMetrics, PublishRecord import random, math from datetime import date 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 random.seed(42) multipliers = { "zhihu": {"v": 1.0, "l": 1.2, "f": 0.6, "c": 1.5, "s": 0.3}, "wechat": {"v": 1.8, "l": 0.6, "f": 0.4, "c": 0.3, "s": 2.0}, "xiaohongshu": {"v": 2.5, "l": 1.5, "f": 1.8, "c": 1.0, "s": 1.5}, } count = 0 for topic in topics: platforms = set() records = db.query(PublishRecord).filter( PublishRecord.topic_id == topic.id, PublishRecord.action == "publish", PublishRecord.status == "success" ).all() for rec in records: platforms.add(rec.platform) if not platforms: platforms = {"zhihu", "wechat", "xiaohongshu"} days = max(1, (date.today() - (topic.published_at or date.today())).days) quality = (topic.compliance_score or 70) / 100.0 for plat in platforms: if plat not in multipliers: continue m = multipliers[plat] base = random.randint(30, 200) growth = 1 + math.log(days + 1, 2) * 0.5 views = int(base * m["v"] * growth) likes = int(views * quality * 0.08 * m["l"]) favs = int(likes * 0.5 * m["f"]) comm = int(views * quality * 0.02 * m["c"]) shar = int(views * quality * 0.03 * m["s"]) existing = db.query(ContentMetrics).filter( ContentMetrics.topic_id == topic.id, ContentMetrics.platform == plat ).first() if existing: existing.views = views existing.likes = likes existing.favorites = favs existing.comments = comm existing.shares = shar existing.last_fetched = datetime.now() else: db.add(ContentMetrics( topic_id=topic.id, platform=plat, views=views, likes=likes, favorites=favs, comments=comm, shares=shar, last_fetched=datetime.now() )) count += 1 db.commit() db.close() logger.info("[Scheduled] Metrics sync completed: %d entries for %d topics", count, len(topics)) 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()