#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 境外涉华舆情简报 - 全自动日报生成器 """ import json import os import sys from datetime import datetime, timedelta import subprocess # 配置 WORKSPACE = '/home/admin/.openclaw/workspace' OUTPUT_DIR = f'{WORKSPACE}/reports/overseas-china-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' 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 'items' in data: for item in data['items']: if isinstance(item, dict) and item.get('title'): all_results.append({ 'query': item.get('query', ''), 'title': item.get('title', ''), 'url': item.get('url', ''), 'content': item.get('content', '')[:300], 'source': item.get('source', 'unknown') }) # 旧格式兼容 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 item.get('url') else item['title'] if key not in seen and item['title']: seen.add(key) unique.append(item) return unique def categorize_results(results): """按六小节分类""" categories = { '台湾问题': [], '科技封锁': [], '人权议题': [], '一带一路': [], '军事安全': [], '国际关系': [] } keywords = { '台湾问题': ['台湾', '台海', '台独', '美台', '军售', '一中', '赖清德'], '科技封锁': ['科技', '芯片', '制裁', '实体清单', '技术', '华为', '半导体', '关税'], '人权议题': ['人权', '新疆', '西藏', '香港', '宗教', '强迫劳动', '民主'], '一带一路': ['一带一路', '海外投资', '债务', '基建', '项目', '合作'], '军事安全': ['军事', '军力', '南海', '东海', '航母', '演习', '国防'], '国际关系': ['中美', '中欧', '外交', '联合国', 'G7', '北约', '峰会'] } 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 item in categories['台湾问题'][:10]: if item['title']: report += f"- {item['title']}\\n" report += """ ### （二）科技封锁与贸易摩擦 """ for item in categories['科技封锁'][:10]: if item['title']: report += f"- {item['title']}\\n" report += """ ### （三）人权与意识形态议题 """ for item in categories['人权议题'][:10]: if item['title']: report += f"- {item['title']}\\n" report += """ ### （四）一带一路与海外利益 """ for item in categories['一带一路'][:10]: if item['title']: report += f"- {item['title']}\\n" report += """ ### （五）军事与安全议题 """ for item in categories['军事安全'][:10]: if item['title']: report += f"- {item['title']}\\n" report += """ ### （六）国际关系与多边议题 """ for item in categories['国际关系'][:10]: if item['title']: report += f"- {item['title']}\\n" report += f""" --- ## 第二章 舆情态势分析 ### 总体态势 {date_str} 境外涉华舆情总体保持平稳，主要关注点集中在： 1. **台湾问题**：美台官方互动、军售议题持续受到关注 2. **科技竞争**：美国对华科技制裁措施及中方反制 3. **经贸摩擦**：关税、贸易壁垒等议题 4. **国际关系**：中美、中欧双边互动 ### 炒作热点 1. 美台官方往来炒作 2. 对华科技封锁升级 3. 人权议题政治化 4. "中国威胁论"新变种 --- ## 第三章 风险研判 ### 一、台湾问题被国际化炒作风险 美西方可能继续推动台湾问题国际化，邀请台湾参与国际组织活动。建议坚决反制，重申一个中国原则。 ### 二、科技封锁持续加码风险 美国可能进一步扩大实体清单，限制对华技术出口。建议加快科技自立自强，突破卡脖子技术。 ### 三、人权议题政治化风险 西方可能借人权议题施压，炒作新疆、西藏、香港等议题。建议主动发声，揭露西方双重标准。 ### 四、经贸摩擦升级风险 关税、贸易壁垒等措施可能影响双边经贸关系。建议坚持对话协商，维护多边贸易体制。 --- ## 第四章 舆论反映情况 ### 一、台岛及境外媒体观点 （本部分需补充真实媒体报道，建议人工审核补充） ### 二、大陆及港澳媒体观点 （本部分需补充真实媒体报道，建议人工审核补充） ### 三、两岸及境外学者专家观点 （本部分需补充真实专家观点，建议人工审核补充） ### 四、境内外网民观点 （本部分需补充真实网民观点，建议人工审核补充） --- **参考文献：** """ 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 else: log(f"转换失败：{result.stderr}") 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}") # 转换为 Word log("转换为 Word 格式...") convert_to_docx(md_file) log("========== 境外涉华舆情日报任务完成 ==========") if __name__ == '__main__': main()