#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 人物画像技能主程序 整合所有分析模块，生成完整的人物画像报告 创建时间：2026-04-03 开发者：瞰宇 (Kàn Yǔ) """ import asyncio import json import sys from pathlib import Path # 添加当前目录到路径 sys.path.insert(0, str(Path(__file__).parent)) from data_collector import DataCollector from text_analyzer import TextAnalyzer from media_analyzer import MediaAnalyzer from public_opinion_analyzer import PublicOpinionAnalyzer from psychology_analyzer import PsychologyAnalyzer from behavior_analyzer import BehaviorAnalyzer from report_generator import ReportGenerator class PersonaProfileSkill: """ 人物画像技能主类 整合所有分析模块，生成完整的人物画像报告 """ def __init__(self, use_osint_tools: bool = False): """ 初始化人物画像技能 Args: use_osint_tools: 是否使用OSINT工具进行数据采集 """ self.use_osint_tools = use_osint_tools # 初始化各模块 self.collector = DataCollector(use_osint_tools=use_osint_tools) self.text_analyzer = TextAnalyzer() self.media_analyzer = MediaAnalyzer() self.public_opinion_analyzer = PublicOpinionAnalyzer() self.psychology_analyzer = PsychologyAnalyzer() self.behavior_analyzer = BehaviorAnalyzer() self.report_generator = ReportGenerator() self.analysis_results = {} async def analyze(self, person_name: str, **kwargs) -> Dict: """ 执行完整的人物画像分析 Args: person_name: 人物姓名 **kwargs: 其他参数 - time_range: 时间范围（默认"10y"） - include_social: 是否包含社交媒体数据（默认True） - include_media: 是否包含媒体报道（默认True） - input_data: 用户提供的输入数据（可选） Returns: 分析结果字典 """ print(f"=== 开始分析人物：{person_name} ===\n") # 第一步：数据采集 print("步骤 1/7：数据采集...") await self._collect_data(person_name, **kwargs) print("✅ 数据采集完成\n") # 第二步：文本立场分析 print("步骤 2/7：文本立场分析...") self._analyze_text_stance() print("✅ 文本立场分析完成\n") # 第三步：媒体叙事分析 print("步骤 3/7：媒体叙事分析...") self._analyze_media() print("✅ 媒体叙事分析完成\n") # 第四步：公众情绪分析 print("步骤 4/7：公众情绪分析...") self._analyze_public_opinion() print("✅ 公众情绪分析完成\n") # 第五步：心理特征推断 print("步骤 5/7：心理特征推断...") self._analyze_psychology() print("✅ 心理特征推断完成\n") # 第六步：行为模式识别 print("步骤 6/7：行为模式识别...") self._analyze_behavior() print("✅ 行为模式识别完成\n") # 第七步：生成报告 print("步骤 7/7：生成分析报告...") report = self._generate_report(person_name) print("✅ 分析报告生成完成\n") return { "person_name": person_name, "analysis_results": self.analysis_results, "report": report } async def _collect_data(self, person_name: str, **kwargs): """数据采集""" # 如果用户提供了输入数据，直接使用 if "input_data" in kwargs and kwargs["input_data"]: input_data = kwargs["input_data"] if "person_profile" in input_data: self.collector.collected_data["person_profile"] = input_data["person_profile"] if "statements" in input_data: self.collector.collected_data["statements"] = input_data["statements"] if "media_reports" in input_data: self.collector.collected_data["media_reports"] = input_data["media_reports"] if "social_posts" in input_data: self.collector.collected_data["social_posts"] = input_data["social_posts"] print(f" 使用用户提供的数据") print(f" - 言论数量：{len(input_data.get('statements', []))}") print(f" - 媒体报道数量：{len(input_data.get('media_reports', []))}") print(f" - 社交帖子数量：{len(input_data.get('social_posts', []))}") else: # 使用OSINT工具采集数据 time_range = kwargs.get("time_range", "10y") include_social = kwargs.get("include_social", True) include_media = kwargs.get("include_media", True) if self.use_osint_tools: await self.collector.collect_all( person_name=person_name, time_range=time_range, include_social=include_social, include_media=include_media ) print(f" 使用OSINT工具采集数据") print(f" - 时间范围：{time_range}") print(f" - 包含社交媒体：{'是' if include_social else '否'}") print(f" - 包含媒体报道：{'是' if include_media else '否'}") else: print(" ⚠️ 未启用OSINT工具，跳过自动数据采集") print(" 💡 提示：可以通过input_data参数提供数据") def _analyze_text_stance(self): """文本立场分析""" statements = self.collector.collected_data.get("statements", []) if statements: # 转换为字典格式 statements_dicts = [ { "content": s.content if hasattr(s, 'content') else s.get("content", ""), "date": s.date if hasattr(s, 'date') else s.get("date", "") } for s in statements ] self.text_analyzer.load_statements(statements_dicts) text_result = self.text_analyzer.generate_spectrum_report() self.analysis_results["text_analyzer"] = text_result print(f" - 政治光谱：经济 {text_result['政治光谱坐标']['经济立场']:.2f}, 社会 {text_result['政治光谱坐标']['社会立场']:.2f}") else: print(" ⚠️ 无言论数据，跳过文本立场分析") def _analyze_media(self): """媒体叙事分析""" reports = self.collector.collected_data.get("media_reports", []) if reports: # 转换为字典格式 reports_dicts = [ { "media_name": r.media_name if hasattr(r, 'media_name') else r.get("media_name", ""), "title": r.title if hasattr(r, 'title') else r.get("title", ""), "content": r.content if hasattr(r, 'content') else r.get("content", ""), "date": r.date if hasattr(r, 'date') else r.get("date", ""), "sentiment": r.sentiment if hasattr(r, 'sentiment') else r.get("sentiment", "neutral") } for r in reports ] self.media_analyzer.load_media_reports(reports_dicts) media_result = self.media_analyzer.generate_media_report() self.analysis_results["media_analyzer"] = media_result print(f" - 分析了 {len(reports)} 篇媒体报道") else: print(" ⚠️ 无媒体报道数据，跳过媒体叙事分析") def _analyze_public_opinion(self): """公众情绪分析""" posts = self.collector.collected_data.get("social_posts", []) if posts: # 转换为字典格式 posts_dicts = [ { "content": p.get("content", ""), "date": p.get("date", ""), "author": p.get("author", ""), "likes": p.get("likes", 0) } for p in posts ] self.public_opinion_analyzer.load_social_data(posts_dicts) opinion_result = self.public_opinion_analyzer.generate_public_opinion_report() self.analysis_results["public_opinion_analyzer"] = opinion_result print(f" - 分析了 {len(posts)} 条社交媒体帖子") else: print(" ⚠️ 无社交媒体数据，跳过公众情绪分析") def _analyze_psychology(self): """心理特征推断""" statements = self.collector.collected_data.get("statements", []) if statements: statements_dicts = [ { "content": s.content if hasattr(s, 'content') else s.get("content", ""), "date": s.date if hasattr(s, 'date') else s.get("date", "") } for s in statements ] self.psychology_analyzer.load_statements(statements_dicts) psych_result = self.psychology_analyzer.generate_psychology_report() self.analysis_results["psychology_analyzer"] = psych_result print(f" - 大五人格：外向性 {psych_result['大五人格特质']['外向性']}, 宜人性 {psych_result['大五人格特质']['宜人性']}") else: " ⚠️ 无言论数据，跳过心理特征推断" def _analyze_behavior(self): """行为模式识别""" behaviors = self.collector.collected_data.get("behaviors", []) if behaviors: self.behavior_analyzer.load_behaviors(behaviors) behavior_result = self.behavior_analyzer.generate_behavior_report() self.analysis_results["behavior_analyzer"] = behavior_result print(f" - 分析了 {len(behaviors)} 条行为记录") else: print(" ⚠️ 无行为记录数据，跳过行为模式识别") def _generate_report(self, person_name: str) -> Dict: """生成分析报告""" self.report_generator.load_analysis_results(self.analysis_results) # 生成Markdown报告 md_report = self.report_generator.generate_markdown_report(person_name) # 生成JSON报告 json_report = self.report_generator.generate_json_report(person_name) return { "markdown": md_report, "json": json_report } def save_report(self, report: Dict, output_dir: str, format: str = "both"): """ 保存报告 Args: report: 报告字典 output_dir: 输出目录 format: 输出格式（"markdown", "json", "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, "persona_profile_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, "persona_profile_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("person_name", help="人物姓名") 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="输出格式") parser.add_argument("--use-osint", action="store_true", help="使用OSINT工具采集数据") args = parser.parse_args() # 初始化技能 skill = PersonaProfileSkill(use_osint_tools=args.use_osint) # 加载输入数据 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.analyze( person_name=args.person_name, input_data=input_data ) # 保存报告 skill.save_report(result["report"], args.output_dir, args.format) print(f"\n=== 分析完成 ===") print(f"人物：{args.person_name}") print(f"输出目录：{args.output_dir}") if __name__ == "__main__": asyncio.run(main())