#!/home/ubuntu/project/we-mp-rss-main/venv/bin/python3 """选题日报 — 纯脚本生成，无 LLM 依赖 (no_agent=true) 用法: python3 daily_topic_report.py [hours=36] 输出: stdout → Hermes cron deliver 投递 用于 Hermes cron job 218c1e953f63 (选题日报)""" import sqlite3, json, sys, re from datetime import datetime, timezone, timedelta from collections import Counter, defaultdict HOURS = int(sys.argv[1]) if len(sys.argv) > 1 else 36 DB_PATH = "/home/ubuntu/project/we-mp-rss-data/db.db" CALENDAR_PATH = "/home/ubuntu/project/config/exam_calendar.json" beijing = timezone(timedelta(hours=8)) now = datetime.now(beijing) cutoff_ts = int((now - timedelta(hours=HOURS)).timestamp()) # ═══════════════════════════════════════ # 1. 查询文章 # ═══════════════════════════════════════ try: db = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) db.row_factory = sqlite3.Row rows = db.execute(""" SELECT a.title, a.url, a.publish_time, a.description, f.mp_name FROM articles a JOIN feeds f ON a.mp_id = f.id WHERE a.publish_time > ? ORDER BY a.publish_time DESC """, (cutoff_ts,)).fetchall() db.close() except Exception as e: print(f"DB_ERROR: {e}") sys.exit(1) if not rows: db2 = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) latest_ts = db2.execute("SELECT MAX(publish_time) FROM articles").fetchone()[0] db2.close() if latest_ts: latest_dt = datetime.fromtimestamp(latest_ts, beijing) gap_hours = (now - latest_dt).total_seconds() / 3600 if gap_hours > 48: print(f"""## 📭 近 {HOURS} 小时无新文章 > 数据库最新: {latest_dt.strftime('%m-%d %H:%M')}（{gap_hours:.0f}h 前） > 可能原因: 公众号停更 / we-mp-rss 登录过期 👉 [检查登录状态](https://rss.gdcjgk.net) （gdcjgk / admin888）""") sys.exit(1) print(f"📭 近 {HOURS} 小时无新文章") sys.exit(0) articles = [dict(r) for r in rows] # ═══════════════════════════════════════ # 2. 考试日历 # ═══════════════════════════════════════ try: with open(CALENDAR_PATH) as f: calendar = json.load(f) except Exception: calendar = {"events": []} future_events = [] for ev in calendar.get("events", []): if not isinstance(ev, dict) or "date" not in ev: continue parts = ev["date"].split("-") if len(parts) != 2: continue ev_date = datetime(now.year, int(parts[0]), int(parts[1]), tzinfo=beijing) if ev_date < now: ev_date = datetime(now.year + 1, int(parts[0]), int(parts[1]), tzinfo=beijing) days_left = (ev_date - now).days if 0 <= days_left <= 14: future_events.append((days_left, ev)) future_events.sort() # ═══════════════════════════════════════ # 3. 关键词提取 & 聚类 # ═══════════════════════════════════════ TOPIC_PATTERNS = { "分数线|投档|录取分|省控线": ("分数线/投档解读", ["各校分数线对比", "分数涨跌分析", "不同分数段择校建议"]), "排名|排行|登顶|第.*名": ("院校排名/实力榜单", ["排名变化解读", "各校王牌专业", "择校参考"]), "职业本科|本科.*扩容|升格": ("职业本科政策利好", ["政策逐条解读", "候选校实力盘点", "中职→本科路径"]), "真题|试卷|备考|复习|教材|刷题": ("27届备考干货", ["复习顺序规划", "真题使用技巧", "各科提分策略"]), "报名|报考|自行.*报|应届生.*材料": ("报名指南/流程", ["报名材料清单", "自行报名流程", "常见踩坑"]), "计算机.*等级|技能.*课程|证书.*报名|证书.*要求|护士.*证": ("证书考试相关", ["证书要求解读", "报名时间提醒", "备考资源"]), "录取.*通知书|寄出": ("录取通知书动态", ["各校寄发时间", "信息核对提醒", "签收注意事项"]), "中考|分数.*公布|成绩.*公布": ("中考放榜/中职择校", ["分数线汇总", "中职学校推荐", "选校避坑"]), "英语|语文|数学.*复习|名词|句式|语法": ("语数英专项技巧", ["题型拆解", "高频考点", "满分技巧"]), "停招|招生.*名单|不在名单": ("中职招生变化", ["停招学校名单", "择校避坑指南", "资质验证方法"]), "暑假|暑期": ("暑期学习规划", ["每日学习计划", "自律方法", "开学前准备"]), "上岸|逆袭|考上|升本": ("升学逆袭故事", ["备考经验", "心态调整", "时间管理"]), "宿舍|校区|投入.*使用|新.*校": ("高校硬件升级", ["新校区盘点", "宿舍条件对比", "校园环境"]), "新生.*开学|报到": ("新生入学指南", ["报到时间汇总", "必备物品清单", "军训准备"]), "学费|公办.*民办|费用": ("学费/择校经济账", ["公办学费对比", "民办性价比", "奖助学金"]), } STOP_WORDS = {"最新消息","点击查看","转发周知","一起来看看","广东","关注","查看","视频","来了", "月","日","年","3+证书","中职生","中职菌","高职","高考","2026","2027", "公布","终于","定了","官宣","注意","重要","最新","消息","刚刚","速看","重磅"} def extract_keywords(title): clean = title for sw in STOP_WORDS: clean = clean.replace(sw, "") words = re.findall(r'[\u4e00-\u9fff]{2,5}', clean) words.sort(key=len, reverse=True) return words[:5] def match_topic(title): for pattern, (topic_name, angles) in TOPIC_PATTERNS.items(): if re.search(pattern, title): return topic_name, angles return None, None topic_groups = defaultdict(list) unmatched = [] for a in articles: topic_name, angles = match_topic(a["title"]) if topic_name: topic_groups[topic_name].append((a, angles)) else: unmatched.append(a) # ═══════════════════════════════════════ # 4. 统计概况 # ═══════════════════════════════════════ source_counts = Counter(a["mp_name"] for a in articles) source_str = "、".join(f"「{k}」{v}篇" for k, v in source_counts.items()) first_time = datetime.fromtimestamp(articles[-1]["publish_time"], beijing) last_time = datetime.fromtimestamp(articles[0]["publish_time"], beijing) # ═══════════════════════════════════════ # 5. 生成报告 # ═══════════════════════════════════════ lines = [] lines.append(f"## 📊 选题日报 ({now.strftime('%m-%d %H:%M')})") lines.append("") lines.append(f"> 近{HOURS}h收录 **{len(articles)}篇** 文章（{source_str}）") lines.append(f"> 时段: {first_time.strftime('%m-%d %H:%M')} ~ {last_time.strftime('%m-%d %H:%M')}") lines.append("") all_keywords = [] for a in articles: all_keywords.extend(extract_keywords(a["title"])) kw_counter = Counter(all_keywords) top_kw = [kw for kw, cnt in kw_counter.most_common(8) if cnt >= 2] if top_kw: lines.append(f"> 🔥 热词: {' · '.join(top_kw)}") lines.append("") lines.append("---") lines.append("") if topic_groups: lines.append("### 🔥 值得追的选题") lines.append("") sorted_topics = sorted(topic_groups.items(), key=lambda x: len(x[1]), reverse=True) for i, (topic_name, group) in enumerate(sorted_topics[:5], 1): arts = group sources = Counter(a[0]["mp_name"] for a in arts) cross_source = len(sources) >= 2 lines.append(f"**选题{i}：{topic_name}**") lines.append(f"> 相关: {len(arts)}篇" + (" · 多号同追 🔥" if cross_source else "")) angles = arts[0][1] if angles: lines.append(f"> 切入: {' / '.join(angles[:3])}") for a, _ in arts[:2]: title_short = a["title"][:40] lines.append(f"> 参考: {a['mp_name']}《{title_short}》") lines.append("") if unmatched: lines.append("### 💡 可储备选题") lines.append("") for a in unmatched[:4]: title_short = a["title"][:45] lines.append(f"> {title_short} — {a['mp_name']}") lines.append("") lines.append("---") lines.append("") if future_events: lines.append("### ⚠️ 临近节点") for days_left, ev in future_events: topics = ev.get("topics", []) lines.append(f"> ⏰ **{days_left}天后**: {ev['title']}") for t in topics[:2]: lines.append(f"> - {t}") else: lines.append("### ⚠️ 考试日历") lines.append("> 未来14天无临近节点") lines.append("") lines.append(f"> 🤖 脚本自动生成 · {now.strftime('%m-%d %H:%M')}") report = "\n".join(lines) report_bytes = report.encode("utf-8") if len(report_bytes) > 3800: trunc_marker = "\n> ...(更多选题请查看数据源)" while len(report.encode("utf-8")) > 3800: lines = report.split("\n") if len(lines) > 10: lines = lines[:-3] report = "\n".join(lines) + trunc_marker else: break print(report)