#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 台湾每日舆情简报 - 全自动日报生成器 """ import json import os from datetime import datetime, timedelta import subprocess # 配置 WORKSPACE = '/home/admin/.openclaw/workspace' OUTPUT_DIR = f'{WORKSPACE}/reports/taiwan-daily-briefing' HOURLY_DIR = f'{OUTPUT_DIR}/hourly' BRIEFINGS_DIR = f'{OUTPUT_DIR}/briefings' LOG_FILE = f'{OUTPUT_DIR}/cron.log' TARGET_USER = '020251091339931228' CHANNEL = 'dingtalk' # 扩大采集源 - 国际主流媒体 + 国家主流媒体 MEDIA_SOURCES = [ # 中国大陆主流媒体 'site:xinhuanet.com 台湾 OR 台海 OR 台独', 'site:people.com.cn 台湾 OR 统一', 'site:chinanews.com 台湾 OR 台海', 'site:globaltimes.cn Taiwan OR 台独', 'site:observer.cn 台湾 OR 台独', 'site:thepaper.cn 台湾', 'site:chinamil.com.cn 台湾 OR 台海', # 国际主流媒体 'site:reuters.com Taiwan China OR cross-strait', 'site:apnews.com Taiwan China', 'site:bbc.com Taiwan China', 'site:scmp.com Taiwan China', 'site:nytimes.com Taiwan China', 'site:ft.com Taiwan China', 'site:wsj.com Taiwan China', # 港澳台 'site:cna.com.tw 台湾 OR 两岸', 'site:udn.com 台湾 OR 两岸', 'site:hk01.com 台湾 OR 两岸', 'site:takungpao.com 台湾', ] def log(msg): timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') line = f"[{timestamp}] {msg}" print(line) with open(LOG_FILE, 'a', encoding='utf-8') as f: f.write(line + '\n') def load_hourly_data(date_str): """读取指定日期的所有小时素材""" all_results = [] date_prefix = date_str.replace('-', '') for i in range(24): filename = f"{date_prefix}_{i:02d}00.json" filepath = os.path.join(HOURLY_DIR, filename) if os.path.exists(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) if 'queries' in data: for q in data['queries']: if 'data' in q and 'results' in q['data']: for r in q['data']['results'][:5]: if r.get('title') and r.get('url'): all_results.append({ 'query': q.get('query', ''), 'title': r.get('title', ''), 'url': r.get('url', ''), 'content': r.get('content', '')[:300], 'source': r.get('engine', 'unknown') }) except Exception as e: log(f"读取 {filename} 失败：{e}") # 去重 seen = set() unique = [] for item in all_results: key = item['url'] if key not in seen and item['title']: seen.add(key) unique.append(item) return unique def categorize_results(results): """按议题分类""" categories = { '两岸关系': [], '台独动向': [], '美台往来': [], '军事动态': [], '岛内政治': [], '国际舆论': [] } keywords = { '两岸关系': ['两岸', '统一', '九二共识', '交流', '经贸', '三通'], '台独动向': ['台独', '赖清德', '民进党', '去中国化', '正名', '制宪'], '美台往来': ['美台', '美国', '军售', '访台', '过境', '国会'], '军事动态': ['军演', '台海', '解放军', '巡航', '国防部', '战机'], '岛内政治': ['选举', '国民党', '民众党', '立法院', '民调', '内阁'], '国际舆论': ['国际', '媒体', '智库', '学者', '外交', '邦交'] } for item in results: text = item['title'] + item['content'] + item.get('query', '') for cat, kws in keywords.items(): if any(kw in text for kw in kws): categories[cat].append(item) break else: categories['两岸关系'].append(item) return categories def generate_report(date_str, results): """生成 Markdown 报告""" categories = categorize_results(results) total = sum(len(v) for v in categories.values()) report = f"""# 台湾每日舆情简报 **报告日期：** {date_str} **编制时间：** {datetime.now().strftime('%Y-%m-%d %H:%M')} **密级：** 内部 **采集数据量：** {total} 条 --- 长官： 现将 {date_str} 台湾舆情简报呈上，请审阅。 ## 第一章 舆情概述 """ for cat_name, items in categories.items(): if items: report += f"### {cat_name}\n\n" for item in items[:10]: if item['title']: report += f"- {item['title']}\n" report += "\n" report += f""" --- ## 第二章 重点议题分析 ### 一、台独分裂动向 （需补充详细分析） ### 二、美台官方往来 （需补充详细分析） ### 三、岛内政局变化 （需补充详细分析） --- ## 第三章 风险研判 ### 一、台独冒险挑衅风险 （需补充详细分析） ### 二、外部势力干涉风险 （需补充详细分析） ### 三、台海军事冲突风险 （需补充详细分析） --- ## 第四章 舆论反映情况 ### 一、大陆及港澳媒体观点 （需补充真实来源） ### 二、台湾媒体观点 （需补充真实来源） ### 三、国际媒体观点 （需补充真实来源） ### 四、专家学者观点 （需补充真实来源） --- **参考文献：** """ ref_num = 1 seen_urls = set() for cat_items in categories.values(): for item in cat_items[:5]: if item['url'] and item['url'] not in seen_urls and item['url'].startswith('http'): seen_urls.add(item['url']) report += f"{ref_num}. {item['title']} - {item['url']} [{item['source']}] [引用日期 {date_str}]\n" ref_num += 1 if ref_num > 30: break if ref_num > 30: break report += f""" --- 对您忠诚 **明见** 战略分析师 {datetime.now().strftime('%Y-%m-%d %H:%M')} """ return report def convert_to_docx(md_file): """转换为 Word 格式""" docx_file = md_file.replace('.md', '.docx') md2docx_script = f'{BRIEFINGS_DIR}/md2docx.py' if os.path.exists(md2docx_script): try: result = subprocess.run( ['python3', md2docx_script, os.path.basename(md_file)], cwd=BRIEFINGS_DIR, capture_output=True, text=True, timeout=60 ) if result.returncode == 0: log(f"Word 文档已生成：{docx_file}") return docx_file except Exception as e: log(f"转换异常：{e}") return None def main(): """主函数""" log("========== 台湾每日舆情日报任务启动（全自动）==========") target_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') log(f"目标日期：{target_date}") md_file = f"{BRIEFINGS_DIR}/{target_date}-台湾舆情简报.md" docx_file = f"{BRIEFINGS_DIR}/{target_date}-台湾舆情简报.docx" if os.path.exists(docx_file): log(f"今日报告已存在：{docx_file}，跳过生成") return log("加载小时采集数据...") results = load_hourly_data(target_date) log(f"采集到 {len(results)} 条唯一结果") if len(results) == 0: log("警告：未采集到有效数据") results = [{'title': '数据采集中，请稍后查看完整报告', 'url': '', 'content': '', 'source': 'system'}] log("生成 Markdown 报告...") report = generate_report(target_date, results) os.makedirs(BRIEFINGS_DIR, exist_ok=True) with open(md_file, 'w', encoding='utf-8') as f: f.write(report) log(f"Markdown 报告已保存：{md_file}") log("转换为 Word 格式...") convert_to_docx(md_file) log("========== 台湾每日舆情日报任务完成 ==========") if __name__ == '__main__': main()