#!/usr/bin/env python3 """ 开源情报目标采集 - 完整工作流 整合采集、解析、清洗全流程 """ import json import argparse import os import sys from pathlib import Path from datetime import datetime from typing import List, Dict, Optional # 添加scripts目录到路径 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from collector import create_collector, CollectionConfig, OSINTCollector from parser import WikipediaParser, ParseConfig, parse_target from cleaner import DataCleaner, CleaningConfig, clean_records from image_collector import ImageCollector, ImageConfig, collect_target_image, collect_batch class OSINTWorkflow: """开源情报采集工作流""" def __init__(self, target_type: str, config: Dict = None): self.target_type = target_type self.config = config or {} self.output_dir = self.config.get('output_dir', './data') Path(self.output_dir).mkdir(parents=True, exist_ok=True) # 初始化各模块 collection_config = CollectionConfig( target_type=target_type, language=self.config.get('language', 'en'), delay_min=self.config.get('delay_min', 8), delay_max=self.config.get('delay_max', 12), save_interval=self.config.get('save_interval', 10), resume=self.config.get('resume', True), output_dir=self.output_dir ) self.collector = OSINTCollector(collection_config) self.parser = WikipediaParser(ParseConfig( language=self.config.get('language', 'en') )) self.cleaner = DataCleaner(CleaningConfig( language=self.config.get('language', 'en') )) # 初始化图片采集器 self.image_collector = ImageCollector(ImageConfig( output_dir=os.path.join(self.output_dir, 'images'), delay_min=self.config.get('delay_min', 3), delay_max=self.config.get('delay_max', 6), save_thumbnails=self.config.get('save_thumbnails', True), )) def load_targets(self, input_file: str) -> List[Dict]: """加载目标列表""" with open(input_file, 'r', encoding='utf-8') as f: data = json.load(f) if isinstance(data, list): return data elif isinstance(data, dict) and 'targets' in data: return data['targets'] else: raise ValueError("输入文件格式不正确") def run(self, input_file: str, output_file: str = None): """运行完整工作流""" print(f"\n{'='*50}") print(f"开源情报目标采集工作流") print(f"目标类型: {self.target_type}") print(f"{'='*50}\n") # 1. 加载目标 print(f"📂 加载目标: {input_file}") targets = self.load_targets(input_file) print(f" 共 {len(targets)} 个目标\n") # 2. 采集数据 print("🔄 阶段1: 数据采集") collected_results = [] for i, target in enumerate(targets): name = target.get('name') or target.get('name_cn') or target.get('name_en') if not name: continue url = target.get('url') or target.get('biography_url') or target.get('wiki链接') lang = target.get('lang', self.config.get('language', 'en')) # 随机延迟 import time, random delay = random.randint(self.config.get('delay_min', 8), self.config.get('delay_max', 12)) time.sleep(delay) # 采集 print(f"[{i+1}/{len(targets)}] 采集: {name}") fetch_result = None if url: try: from crawl4ai import AsyncWebCrawler import asyncio async def fetch_url(target_url): async with AsyncWebCrawler() as crawler: result = await crawler.arun(target_url) return result result = asyncio.run(fetch_url(url)) if result.success: fetch_result = { 'url': url, 'markdown': result.markdown, 'html': result.html, 'collection_time': datetime.now().isoformat() } except Exception as e: print(f" ❌ 采集失败: {e}") # 解析 if fetch_result: parsed = parse_target(name, fetch_result, self.parser.config) target.update(parsed) # 添加数据护照 target['data_source'] = url or '' target['collection_time'] = datetime.now().isoformat() collected_results.append(target) # 自动保存 if (i + 1) % self.config.get('save_interval', 10) == 0: checkpoint_file = f"{self.target_type}_checkpoint.json" with open(os.path.join(self.output_dir, checkpoint_file), 'w', encoding='utf-8') as f: json.dump(collected_results, f, ensure_ascii=False, indent=2) print(f" 💾 检查点已保存") print(f" 采集完成: {len(collected_results)} 条\n") # 3. 采集图片/头像 if self.config.get('collect_images', True): print("🖼️ 阶段2: 图片/头像采集") image_results = [] for i, target in enumerate(collected_results): name = target.get('name') or target.get('name_cn') or target.get('name_en') url = target.get('url') or target.get('data_source') if not name or not url: continue print(f"[{i+1}/{len(collected_results)}] 采集图片: {name}") result = None if 'wikipedia' in url.lower(): result = self.image_collector.collect_from_wikipedia(name, url) else: result = self.image_collector.collect_from_url(name, url) if result: # 添加图片信息到目标记录 target['avatar_filename'] = result.get('filename') target['avatar_path'] = result.get('filepath') target['avatar_thumbnail'] = result.get('thumbnail_path') target['avatar_url'] = result.get('image_url') target['image_collected_time'] = result.get('collection_time') image_results.append(result) print(f" ✅ {result.get('filename')}") else: print(f" ⚪ 未找到图片") print(f" 图片采集完成: {len(image_results)} 张\n") # 4. 清洗数据 print("🧹 阶段3: 数据清洗") cleaned_results = self.cleaner.process_batch(collected_results) stats = self.cleaner.get_stats() print(f" 处理: {stats['records_processed']} 条") print(f" 清洗: {stats['records_cleaned']} 条") print(f" 清理字段: {stats['fields_cleaned']} 个") print(f" 移除无效: {stats['invalid_removed']} 个\n") # 5. 保存结果 if not output_file: output_file = f"{self.target_type}_output.json" output_path = os.path.join(self.output_dir, output_file) with open(output_path, 'w', encoding='utf-8') as f: json.dump(cleaned_results, f, ensure_ascii=False, indent=2) print(f"💾 结果已保存: {output_path}") print(f" 共 {len(cleaned_results)} 条记录\n") # 5. 生成报告 report = self.generate_report(cleaned_results, stats) report_file = f"{self.target_type}_report.md" with open(os.path.join(self.output_dir, report_file), 'w', encoding='utf-8') as f: f.write(report) print(f"📊 报告已生成: {report_file}") return cleaned_results def generate_report(self, results: List[Dict], stats: Dict) -> str: """生成采集报告""" report = f"""# 开源情报目标采集报告 ## 基本信息 - **采集目标**: {self.target_type} - **采集时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - **目标数量**: {len(results)} 条 ## 采集统计 | 指标 | 数值 | |------|------| | 处理记录 | {stats['records_processed']} | | 清洗记录 | {stats['records_cleaned']} | | 清理字段 | {stats['fields_cleaned']} | | 移除无效值 | {stats['invalid_removed']} | ## 字段覆盖率 """ # 统计字段覆盖率 if results: all_fields = set() for r in results: all_fields.update(r.keys()) field_stats = {} for field in all_fields: count = sum(1 for r in results if r.get(field)) field_stats[field] = count for field, count in sorted(field_stats.items(), key=lambda x: -x[1]): pct = count / len(results) * 100 report += f"- **{field}**: {count}/{len(results)} ({pct:.1f}%)\n" return report def main(): parser = argparse.ArgumentParser(description='开源情报目标采集工作流') parser.add_argument('--target-type', '-t', required=True, help='目标类型') parser.add_argument('--input', '-i', required=True, help='输入文件(JSON)') parser.add_argument('--output', '-o', help='输出文件(JSON)') parser.add_argument('--output-dir', '-d', default='./data', help='输出目录') parser.add_argument('--language', '-l', default='en', help='语言: en/zh/ja') parser.add_argument('--delay-min', type=int, default=8, help='最小延迟(秒)') parser.add_argument('--delay-max', type=int, default=12, help='最大延迟(秒)') args = parser.parse_args() workflow = OSINTWorkflow( target_type=args.target_type, config={ 'output_dir': args.output_dir, 'language': args.language, 'delay_min': args.delay_min, 'delay_max': args.delay_max, } ) workflow.run(args.input, args.output) if __name__ == "__main__": main()