#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 美国对华制裁监测预警技能主程序 """ import asyncio import json import sys from pathlib import Path from datetime import datetime # 添加当前目录到路径 sys.path.insert(0, str(Path(__file__).parent)) from collector import SanctionsCollector from analyzer import SanctionsAnalyzer from early_warning import EarlyWarningSystem class SanctionsMonitoringSkill: """美国对华制裁监测预警技能主类""" def __init__(self, use_async: bool = True): self.use_async = use_async self.collector = SanctionsCollector(use_async=use_async) self.analyzer = SanctionsAnalyzer() self.warning_system = EarlyWarningSystem() self.analysis_results = {} async def monitor(self, **kwargs) -> dict: """执行制裁监测和预警""" print("=== 开始美国对华制裁监测预警 ===\n") # 第一步：数据采集 print("步骤 1/4：数据采集...") await self._collect_data(**kwargs) print("✅ 数据采集完成\n") # 第二步：制裁分析 print("步骤 2/4：制裁分析...") self._analyze_sanctions() print("✅ 制裁分析完成\n") # 第三步：风险预警 print("步骤 3/4：风险预警...") self._generate_warnings(**kwargs) print("✅ 风险预警完成\n") # 第四步：生成报告 print("步骤 4/4：生成监测报告...") report = self._generate_report() print("✅ 监测报告生成完成\n") return { "analysis_results": self.analysis_results, "report": report } async def _collect_data(self, **kwargs): """数据采集""" if "input_data" in kwargs and kwargs["input_data"]: input_data = kwargs["input_data"] if "entities" in input_data: self.collector.collected_data["entities"] = input_data["entities"] if "monitored_entities" in input_data: self.collector.collected_data["monitored_entities"] = input_data["monitored_entities"] print(" 使用用户提供的数据") print(f" - 已制裁实体数量：{len(input_data.get('entities', []))}") print(f" - 监测实体数量：{len(input_data.get('monitored_entities', []))}") else: print(" ⚠️ 未提供输入数据，跳过数据采集") def _analyze_sanctions(self): """制裁分析""" entities = self.collector.collected_data.get("entities", []) if entities: # 转换为字典格式 entities_dicts = [ { "name": e.name if hasattr(e, 'name') else e.get("name", ""), "entity_id": e.entity_id if hasattr(e, 'entity_id') else e.get("entity_id", ""), "country": e.country if hasattr(e, 'country') else e.get("country", ""), "list_type": e.list_type if hasattr(e, 'list_type') else e.get("list_type", ""), "added_date": e.added_date if hasattr(e, 'added_date') else e.get("added_date", ""), "reason": e.reason if hasattr(e, 'reason') else e.get("reason", ""), "source": e.source if hasattr(e, 'source') else e.get("source", "") } for e in entities ] self.analyzer.load_entities(entities_dicts) analysis_result = self.analyzer.generate_analysis_report() self.analysis_results["sanctions_analyzer"] = analysis_result print(f" - 分析了 {len.()entities} 个制裁实体") else: print(" ⚠️ 无制裁实体数据，跳过制裁分析") def _generate_warnings(self, **kwargs): """风险预警""" sanctioned_entities = self.collector.collected_data.get("entities", []) monitored_entities = kwargs.get("monitored_entities", []) if "input_data" in kwargs and kwargs["input_data"]: monitored_entities = kwargs["input_data"].get("monitored_entities", monitored_entities) if sanctioned_entities: sanctioned_dicts = [ { "name": e.name if hasattr(e, 'name') else e.get("name", ""), "reason": e.reason if hasattr(e, 'reason') else e.get("reason", "") } for e in sanctioned_entities ] self.warning_system.load_sanctioned_entities(sanctioned_dicts) self.warning_system.load_monitored_entities(monitored_entities) # 生成预警 alerts = self.warning_system.generate_alerts() warning_result = self.warning_system.generate_warning_report() self.analysis_results["early_warning"] = warning_result print(f" - 生成了 {len(alerts)} 条风险预警") else: print(" ⚠️ 无已制裁实体数据，跳过风险预警") def _generate_report(self) -> dict: """生成监测报告""" # 生成Markdown报告 md_report = self._generate_markdown_report() # 生成JSON报告 json_report = self.analysis_results return { "markdown": md_report, "json": json_report } def _generate_markdown_report(self) -> str: """生成Markdown格式报告""" md = [] # 报告标题 md.append("# 美国对华制裁监测预警报告\n") md.append(f"> **生成时间：** {datetime.now().strftime('%Y-%m-%d %H:%M:%S GMT')}") md.append(f"> **监测工具：** 开源情报-对华制裁监测技能 v1.0.0") md.append(f"> **开发者：** 瞰宇 (Kàn Yǔ)") md.append("") # 制裁分析部分 md.append("---\n") md.append("# 一、制裁动态分析\n") if "sanctions_analyzer" in self.analysis_results: analyzer_result = self.analysis_results["sanctions_analyzer"] md.append("## 1.1 制裁模式识别\n") patterns = analyzer_result.get("制裁模式识别", []) if patterns: md.append("| 模式名称 | 描述 | 目标产业 | 触发事件 | 出现频率 |") md.append("|---------|------|----------|----------|----------|") for pattern in patterns[:10]: md.append(f"| {pattern['模式名称']} | {pattern['描述']} | {pattern['目标产业']} | {pattern['触发事件']} | {pattern['出现频率']} |") md.append("") else: md.append("> 暂无制裁模式数据\n") md.append("") md.append("## 1.2 产业影响分析\n") industry_impact = analyzer_result.get("产业影响分析", {}) if industry_impact: md.append("| 产业 | 实体数量 | 影响比例 |") md.append("|------|----------|----------|") for industry, data in industry_impact.items(): md.append(f"| {industry} | {data['实体数量']} | {data['影响比例']} |") md.append("") md.append("## 1.3 统计摘要\n") stats = analyzer_result.get("统计信息", {}) for key, value in stats.items(): md.append(f"- **{key}：** {value}") md.append("") # 风险预警部分 md.append("---\n") md.append("# 二、风险预警\n") if "early_warning" in self.analysis_results: warning_result = self.analysis_results["early_warning"] md.append("## 2.1 预警摘要\n") summary = warning_result.get("预警摘要", {}) for key, value in summary.items(): md.append(f"- **{key}：** {value}") md.append("") md.append("## 2.2 详细预警\n") alerts = warning_result.get("详细预警", []) if alerts: for i, alert in enumerate(alerts, 1): md.append(f"### 预警 #{i}: {alert['实体名称']}") md.append(f"- **实体名称：** {alert['实体名称']}") md.append(f"- **风险等级：** {alert['风险等级']}") md.append(f"- **风险类型：** {alert['风险类型']}") md.append(f"- **触发因素：** {alert['触发因素']}") md.append(f"- **潜在制裁措施：** {alert['潜在制裁措施']}") md.append(f"- **应对建议：** {alert['应对建议']}") md.append("") else: md.append("> 暂无预警数据\n") md.append("") # 报告尾部 md.append("---\n") md.append(f"**报告生成时间：** {datetime.now().strftime('%Y-%m-%d %H:%M:%S GMT+8')}") md.append(f"**监测工具：** 开源情报-对华制裁监测技能 v1.0.0") md.append("") md.append("---\n") md.append("**瞰宇 | 穿越数字国界，瞭望全球局势 · 解构认知域，守护认知安全**") return "\n".join(md) def save_report(self, report: dict, output_dir: str, format: str = "both"): """保存报告""" import os if not os.path.exists(output_dir): os.makedirs(output_dir) if format in ["markdown", "both"]: md_file = os.path.join(output_dir, "sanctions_monitoring_report.md") with open(md_file, 'w', encoding='utf-8') as f: f.write(report["markdown"]) print(f"✅ Markdown报告已保存到：{md_file}") if format in ["json", "both"]: json_file = os.path.join(output_dir, "sanctions_monitoring_report.json") with open(json_file, 'w', encoding='utf-8') as f: json.dump(report["json"], f, ensure_ascii=False, indent=2) print(f"✅ JSON报告已保存到：{json_file}") async def main(): """主函数""" import argparse parser = argparse.ArgumentParser(description="美国对华制裁监测预警技能") parser.add_argument("--input-data", help="输入数据JSON文件路径") parser.add_argument("--output-dir", default="./reports", help="输出目录") parser.add_argument("--format", default="both", choices=["markdown", "json", "both"], help="输出格式") args = parser.parse_args() # 初始化技能 skill = SanctionsMonitoringSkill(use_async=True) # 加载输入数据 input_data = None if args.input_data: with open(args.input_data, 'r', encoding='utf-8') as f: input_data = json.load(f) # 执行监测 result = await skill.monitor(input_data=input_data) # 保存报告 skill.save_report(result["report"], args.output_dir, args.format) print(f"\n=== 监测完成 ===") print(f"输出目录：{args.output_dir}") if __name__ == "__main__": asyncio.run(main())