#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 高校舆情分析 - 报告生成脚本 读取采集数据，按事件逐一罗列生成 Markdown 报告 """ import json import os import re import sys from datetime import datetime, timedelta from collections import defaultdict # 配置 OUTPUT_DIR = os.environ.get('UNIVERSITY_SENTIMENT_OUTPUT_DIR', os.path.expanduser('~/.openclaw/workspace/reports/university-sentiment')) HOURLY_DIR = os.path.join(OUTPUT_DIR, 'hourly') BRIEFINGS_DIR = os.path.join(OUTPUT_DIR, 'briefings') LOG_FILE = os.path.join(OUTPUT_DIR, 'generate.log') # 事件分类关键词 EVENT_KEYWORDS = { '高校管理': ['外卖', '食堂', '宿舍', '校规', '管理', 'canteen', 'dormitory', 'campus management', 'food', 'meal'], '意识形态': ['节日', '圣诞节', '万圣节', '情人节', '母亲节', '洋节', '意识形态', '宣传', 'holiday', 'Christmas', 'ideology', 'propaganda'], '学生权益': ['学费', '就业', '实习', '贷款', 'tuition', 'employment', 'internship', 'student rights'], '校园安全': ['暴力', '骚扰', '安全', '事故', 'violence', 'harassment', 'safety', 'incident'], '教师学术': ['教授', '学术', '不端', '师德', 'professor', 'academic', 'misconduct', 'teacher'], '招生教育': ['招生', '高考', '留学生', '加分', 'admission', 'enrollment', 'quota'], '敏感话题': ['女权', 'LGBT', '宗教', '民族', 'feminism', 'LGBTQ', 'religion', 'ethnic'], '科研科技': ['科研', '经费', '造假', 'research', 'funding', 'fraud'], } def log(msg): ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S') line = f'[{ts}] {msg}' print(line, flush=True) with open(LOG_FILE, 'a', encoding='utf-8') as f: f.write(line + '\n') def load_recent_data(days=1): """加载最近几天的采集数据""" all_items = [] now = datetime.now() for day_offset in range(days): target = now - timedelta(days=day_offset) date_prefix = target.strftime('%Y%m%d') if not os.path.exists(HOURLY_DIR): continue for fname in os.listdir(HOURLY_DIR): if fname.startswith(date_prefix) and fname.endswith('.json'): fpath = os.path.join(HOURLY_DIR, fname) try: with open(fpath, 'r', encoding='utf-8') as f: data = json.load(f) # 提取唯一项 for item in data.get('unique_items', []): if item.get('title') and item.get('url'): all_items.append(item) except Exception as e: log(f'读取 {fname} 失败：{e}') # 去重 seen = set() unique = [] for item in all_items: url = item.get('url', '') if url not in seen: seen.add(url) unique.append(item) return unique def classify_event(item): """根据关键词分类事件""" text = (item.get('title', '') + ' ' + item.get('content', '')).lower() for category, keywords in EVENT_KEYWORDS.items(): if any(kw.lower() in text for kw in keywords): return category return '其他' def group_into_events(items): """将采集条目聚类为独立事件""" events = defaultdict(list) for item in items: category = classify_event(item) events[category].append(item) # 每个分类内按标题相似度进一步细分 final_events = [] for category, cat_items in events.items(): # 简单策略：同一分类下的条目归为同一事件 # 如有多个明显不同主题，可进一步细分 if cat_items: final_events.append({ 'category': category, 'items': cat_items, }) return final_events def generate_event_section(event, index): """生成单个事件的 Markdown 内容""" items = event['items'] category = event['category'] if not items: return '' # 构建事件标题 first_item = items[0] title = first_item.get('title', '未知事件') # 截取标题前 30 字作为事件编号标题 event_label = f'（{chr(0x4e00 + index - 1)}）' # （一）（二）（三）... section = f'### {event_label}{title}\n\n' # 罗列境外媒体报道和社交媒体信息 for i, item in enumerate(items[:5]): # 每事件最多 5 条 source = item.get('engine', '未知来源') date = item.get('publishedDate', '') content = item.get('content', '') url = item.get('url', '') if date: date_str = date[:10] if len(date) >= 10 else date else: date_str = '近日' section += f'{date_str}，{source}报道，' if content: section += content[:200] section += f'①\n\n' # 风险研判 risk_assessments = { '高校管理': '该事件涉及高校后勤管理与学生日常生活矛盾，已在社交媒体引发讨论。需关注舆情进一步发酵，以及校方后续改进措施。同时注意防范境外势力借机炒作校园管理负面情绪。', '意识形态': '该事件涉及高校意识形态管理与文化认同等敏感话题，境外势力可能借此渲染"思想管控"叙事。需注意个别势力恶意煽宣，鼓噪高校学生炒作抵抗等动向。', '学生权益': '该事件涉及学生切身利益，可能引发集体行动。需关注学生诉求是否得到合理回应，防止境外势力介入煽动。', '校园安全': '该事件涉及校园安全问题，可能引发家长和社会广泛关注。需关注校方处置是否及时得当，防止舆情升级。', '教师学术': '该事件涉及教师师德和学术规范，可能影响高校声誉。需关注事件调查进展及舆论走向。', '招生教育': '该事件涉及教育公平议题，社会关注度高。需关注政策调整及各方反应，防止被境外势力利用。', '敏感话题': '该事件涉及女权、宗教等敏感话题，境外势力可能借此推进"价值观渗透"叙事。需高度警惕恶意煽宣和认知操纵。', '科研科技': '该事件涉及科研诚信，可能被境外媒体放大炒作。需关注调查进展，维护学术声誉。', '其他': '该事件已在社交媒体引发关注，需持续跟踪舆情发展，防止境外势力借题发挥。', } risk = risk_assessments.get(category, risk_assessments['其他']) section += f'{risk}\n\n' return section def generate_report(date_str, events): """生成完整 Markdown 报告""" now = datetime.now().strftime('%Y-%m-%d %H:%M') report = f"""# 高校舆情分析简报 **报告日期：** {date_str} **编制时间：** {now} **密级：** 内部 **事件数量：** {len(events)} 件 --- 长官： 现将近期境外关于中国高校舆情分析简报呈上，请审阅。 --- """ # 按事件逐一罗列 for i, event in enumerate(events, 1): report += generate_event_section(event, i) # 参考文献 report += '---\n\n## 参考文献\n\n' ref_num = 1 seen_urls = set() for event in events: for item in event['items'][:3]: url = item.get('url', '') if url and url not in seen_urls and url.startswith('http'): seen_urls.add(url) title = item.get('title', '未知标题') source = item.get('engine', '未知来源') report += f'① {title}，{url}（{source}，[引用日期 {date_str}]）\n\n' ref_num += 1 if ref_num > 20: break if ref_num > 20: break report += f"""--- 对您忠诚 瞰宇 {now} """ return report def main(): log('========== 高校舆情报告生成启动 ==========') date_str = datetime.now().strftime('%Y-%m-%d') os.makedirs(BRIEFINGS_DIR, exist_ok=True) # 加载数据 items = load_recent_data(days=1) log(f'加载 {len(items)} 条采集数据') if not items: log('警告：无有效采集数据，生成空报告') items = [] # 聚类事件 events = group_into_events(items) log(f'识别 {len(events)} 个独立事件') # 生成报告 report = generate_report(date_str, events) # 保存 Markdown md_file = os.path.join(BRIEFINGS_DIR, f'{date_str}-高校舆情分析简报.md') with open(md_file, 'w', encoding='utf-8') as f: f.write(report) log(f'Markdown 报告已保存：{md_file}') # 转换 Word docx_file = os.path.join(BRIEFINGS_DIR, f'{date_str}-高校舆情分析简报.docx') md2docx_script = os.path.join(os.path.dirname(__file__), 'md2docx.py') if os.path.exists(md2docx_script): try: import subprocess result = subprocess.run( ['python3', md2docx_script, md_file, docx_file], capture_output=True, text=True, timeout=60 ) if result.returncode == 0: log(f'Word 文档已生成：{docx_file}') else: log(f'Word 转换失败：{result.stderr}') except Exception as e: log(f'Word 转换异常：{e}') log('========== 报告生成完成 ==========') return md_file if __name__ == '__main__': main()