#!/usr/bin/env python3 """ 社交媒体账号综合评估主脚本 整合：活跃度 + 内容 + 网络 + 主题叙事 + 粉丝风险 + 报告生成 """ import json import sys from pathlib import Path from datetime import datetime from typing import List, Dict, Any # 添加脚本目录到路径 sys.path.insert(0, str(Path(__file__).parent)) from analyze_activity import calculate_activity_metrics from analyze_content import analyze_content_distribution, analyze_engagement_heat from analyze_network import analyze_network_change, analyze_following_network from analyze_topic_narrative import analyze_topics_and_sentiment from analyze_follower_risk import analyze_follower_and_risk def determine_account_role(content_dist: Dict, topic_analysis: Dict, network_analysis: Dict) -> str: """ 判定账号角色 Returns: str: 角色类型 """ original_ratio = content_dist.get("distribution", {}).get("original", 0) retweet_ratio = content_dist.get("distribution", {}).get("retweet", 0) primary_topic = topic_analysis.get("topic_classification", {}).get("primary_topic", "") # 判定逻辑 if original_ratio >= 70: return "信息源" elif retweet_ratio >= 60: return "放大器" elif primary_topic in ["政治", "争议", "社会"]: return "意见领袖" else: return "混合型" def assess_influence_quality( engagement: Dict, network_change: Dict, follower_quality: Dict ) -> Dict[str, Any]: """ 评估影响力质量 Returns: Dict: 影响力质量评估 """ avg_engagement = engagement.get("average_total_engressment", 0) followers_growth = network_change.get("followers_growth_rate", 0) high_quality_ratio = follower_quality.get("high_quality_ratio", 0) bot_ratio = follower_quality.get("bot_suspect_ratio", 0) # 影响力真实性评分 reality_score = 100 # 机器人粉丝多 = 影响力可能虚假 if bot_ratio >= 50: reality_score -= 50 elif bot_ratio >= 30: reality_score -= 30 elif bot_ratio >= 10: reality_score -= 10 # 互动量与粉丝数不成比例（假设平均互动量 < 粉丝数的0.1%） # 简化判断：如果平均互动量太低，可能存在买粉 if avg_engagement < 100 and followers_growth > 100: reality_score -= 20 # 判定等级 if reality_score >= 80: quality_level = "真实" elif reality_score >= 50: quality_level = "基本真实" elif reality_score >= 30: quality_level = "部分虚假" else: quality_level = "高度可疑" return { "reality_score": max(0, reality_score), "quality_level": quality_level, "factors": { "average_engagement": avg_engagement, "followers_growth_rate": followers_growth, "high_quality_follower_ratio": high_quality_ratio, "bot_suspect_ratio": bot_ratio } } def generate_recommendations( risk_signals: Dict, influence_quality: Dict, account_role: str ) -> Dict[str, Any]: """ 生成后续行动建议 Args: risk: 风险信号 influence_quality: 影响力质量 account_role: 账号角色 Returns: Dict: 建议 """ recommendations = [] overall_risk = risk_signals.get("overall_risk_level", "未知") influence_level = influence_quality.get("quality_level", "未知") # 1. 基于风险等级的建议 if overall_risk in ["极高", "高"]: recommendations.append({ "type": "监控", "priority": "紧急", "action": "纳入高危监控列表", "description": "账号存在高风险行为，需要24小时实时监控" }) # 2. 基于影响力质量度的建议 if influence_level in ["部分虚假", "高度可疑"]: recommendations.append({ "type": "调查", "priority": "高", "action": "反向调查", "description": "影响力真实性存疑，建议反向调查背后实体和资金来源" }) # 3. 基于账号角色的建议 if account_role == "信息源": recommendations.append({ "type": "内容分析", "priority": "中", "action": "持续内容监测", "description": "作为信息源，需要持续监测其内容质量和可信度" }) # 4. 基于具体风险信号的建议 for signal in risk_signals.get("risk_signals", []): if signal.get("risk_level") == "高": recommendations.append({ "type": "风险应对", "priority": "高", "action": f"应对{signal.get('type')}风险", "description": f"检测到{signal.get('type')}风险信号，需要制定针对性应对策略" }) # 去重（基于action字段） seen_actions = set() unique_recommendations = [] for rec in recommendations: if rec["action"] not in seen_actions: unique_recommendations.append(rec) seen_actions.add(rec["action"]) return unique_recommendations def evaluate_social_account_comprehensive( account_info: Dict[str, Any], tweets: List[Dict[str, Any]], network_history: List[Dict[str, Any]] = None, following_accounts: List[Dict[str, Any]] = None, followers_sample: List[Dict[str, Any]] = None, interactions: List[Dict[str, Any]] = None, days_back: int = 90 ) -> Dict[str, Any]: """ 社交媒体账号综合评估主入口 Args: account_info: 账号信息 tweets: 推文列表 network_history: 网络历史数据 following_accounts: 关注的账号列表 followers_sample: 粉丝样本 interactions: 交互记录 days_back: 分析天数 Returns: Dict: 综合评估结果 """ # Step 1: 活跃度分析 activity_metrics = calculate_activity_metrics(tweets, days=days_back) # Step 2: 内容分析 content_dist = analyze_content_distribution(tweets) engagement_heat = analyze_engagement_heat(tweets, top_n=10) # Step 3: 网络分析 network_change = analyze_network_change(network_history or []) following_analysis = analyze_following_network(following_accounts or []) # Step 4: 主题与叙事分析 topic_analysis = analyze_topics_and_sentiment(tweets) # Step 5: 粉丝与风险分析 follower_risk_analysis = analyze_follower_and_risk( tweets, followers_sample or [], interactions or [] ) # Step 6: 综合评估 # 6.1 账号角色判定 account_role = determine_account_role(content_dist, topic_analysis, following_analysis) # 6.2 影响力质量评估 influence_quality = assess_influence_quality( engagement_heat, network_change, follower_risk_analysis.get("follower_quality", {}) ) # 6.3 后续行动建议 recommendations = generate_recommendations( follower_risk_analysis.get("risk_signals", {}), influence_quality, account_role ) # Step 7: 整合结果 result = { "account_info": account_info, "analysis_summary": { "account_role": account_role, "influence_quality": influence_quality, "overall_risk_level": follower_risk_analysis.get("risk_signals", {}).get("overall_risk_level", "未知"), }, "detailed_analysis": { "activity_metrics": activity_metrics, "content_distribution": content_dist, "engagement_heat": engagement_heat, "network_change": network_change, "following_analysis": following_analysis, "topic_analysis": topic_analysis, "follower_risk_analysis": follower_risk_analysis, }, "recommendations": recommendations, "generated_at": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"), "parameters": { "days_back": days_back } } return result def main(): """命令行入口""" if len(sys.argv) > 1: # 从文件读取 with open(sys.argv[1], 'r', encoding='utf-8') as f: input_data = json.load(f) else: # 使用示例数据 input_data = { "account_info": { "id": "example_account", "username": "Example User", "bio": "Bio text here", "location": "Location", "created_at": "2020-01-01", "followers_count": 100000, "following_count": 1000, "verified": True, }, "tweets": [ { "id": "1", "text": "Original tweet about topic", "created_at": "2026-04-01T10:00:00Z", "like_count": 1000, "retweet_count": 500, "reply_count": 100, } ], "network_history": [], "following_accounts": [], "followers_sample": [], "interactions": [], } result = evaluate_social_account_comprehensive( account_info=input_data.get("account_info", {}), tweets=input_data.get("tweets", []), network_history=input_data.get("network_history"), following_accounts=input_data.get("following_accounts"), followers_sample=input_data.get("followers_sample"), interactions=input_data.get("interactions"), days_back=input_data.get("days_back", 90) ) print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) if __name__ == "__main__": main()