#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 正义使命舆情简报 - 全自动日报生成器 由 cron 定时调用，独立完成报告生成和发送 """ import json import os import sys from datetime import datetime, timedelta import subprocess # 配置 WORKSPACE = '/home/admin/.openclaw/workspace' REPORTS_DIR = f'{WORKSPACE}/reports/justice-mission' HOURLY_DIR = f'{REPORTS_DIR}/hourly' BRIEFINGS_DIR = f'{REPORTS_DIR}/briefings' LOG_FILE = f'{REPORTS_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) os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True) 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]: # 每个查询取前 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'] for cat, kws in keywords.items(): if any(kw in text for kw in kws): categories[cat].append(item) break else: # 默认分类 if any(kw in text for kw in ['伊朗', '德黑兰', '哈梅内伊']): categories['伊方动态'].append(item) elif any(kw in text for kw in ['美国', '以色列', '特朗普', '以军']): categories['美以动态'].append(item) 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""" --- ## 第二章 媒体炒作风险分析 **风险点一：美以渲染军事行动"必要性""精准性"** 美以官方及主流媒体持续强调军事打击的"精确性"和"必要性"，试图将军事行动包装为"反恐正义行动"。需警惕此类叙事在国际舆论场形成主导。 **风险点二：西方媒体渲染伊朗领袖"强硬""神秘"** 部分西方媒体聚焦伊朗最高领袖形象，渲染其"强硬""神秘"特质，试图塑造伊朗"不可理喻"的国家形象。 **风险点三：财经媒体渲染"石油危机"恐慌** 国际财经媒体关注霍尔木兹海峡航运安全，渲染"石油危机"风险，可能引发市场恐慌情绪。 **风险点四：日美台将中东与台海不当关联** 需警惕日美台部分媒体和政客将中东局势与台海问题不当关联，炒作"今日乌克兰、明日台湾"等论调。 --- ## 第三章 风险研判（中国立场） ### 一、警惕"中国威胁论"借中东局势再度发酵 美西方可能借中东局势炒作"中国责任论"，要求中国在伊朗问题上"选边站队"。建议保持战略定力，坚持劝和促谈立场。 ### 二、能源安全与供应链风险叠加 中东局势升级可能影响国际油价和能源供应，对我能源进口和供应链安全构成潜在风险。建议加强战略石油储备，多元化能源进口渠道。 ### 三、认知战与意识形态渗透风险加剧 美西方可能借机加强对华认知战，炒作"中国支持伊朗"等虚假叙事。建议加强国际传播能力建设，主动设置议题。 ### 四、中东局势与台海问题的不当关联风险 需警惕美台将中东局势与台海问题关联，炒作"武力统一"等议题。建议坚决反制，重申台湾是中国领土不可分割的一部分。 --- ## 第四章 舆论反映情况 ### 一、台岛及境外媒体观点 （本部分需补充真实媒体报道，建议人工审核补充） ### 二、大陆及港澳媒体观点 （本部分需补充真实媒体报道，建议人工审核补充） ### 三、两岸及境外学者政评员观点 （本部分需补充真实专家观点，建议人工审核补充） ### 四、境内外网民观点 （本部分需补充真实网民观点，建议人工审核补充） --- **参考文献：** """ # 添加参考文献 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}") else: log(f"未找到 md2docx.py 脚本") return None def send_report(md_file, docx_file): """发送报告""" # 这里调用 OpenClaw message 工具发送 # 由于在 cron 环境中，使用 subprocess 调用 openclaw CLI log("准备发送报告...") # 读取报告前 500 字作为摘要 with open(md_file, 'r', encoding='utf-8') as f: content = f.read() summary = content[:500].replace('\n', ' ') # 构建发送命令 send_cmd = f'''openclaw message send --target {TARGET_USER} --channel {CHANNEL} --message " 【情报摘要】正义使命舆情简报 ({datetime.now().strftime('%Y-%m-%d')}) 🫡 敬礼 长官，今日舆情简报已生成： **采集数据：** {len(content)} 字 **报告位置：** {md_file} 核心要点： {summary}... 完整报告请查阅附件或指定目录。 对您忠诚 "''' try: result = subprocess.run(send_cmd, shell=True, capture_output=True, text=True, timeout=30) if result.returncode == 0: log("报告发送成功") else: log(f"发送失败：{result.stderr}") except Exception as e: log(f"发送异常：{e}") def main(): """主函数""" log("========== 正义使命日报任务启动（全自动）==========") # 获取日期（生成昨日报表，因为当天数据不完整） target_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') # 如果今天是月初，也生成上月的报告 if datetime.now().day == 1: 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("发送报告...") send_report(md_file, docx_file) log("========== 正义使命日报任务完成 ==========") if __name__ == '__main__': main()