#!/usr/bin/env python3 """ 网络分析模块 功能： 1. 跟踪网络规模变化（粉丝数、关注数变化趋势） 2. 分析关注网络类型 """ import json from datetime import datetime from typing import List, Dict, Any def parse_timestamp(time_str: str) -> datetime: """解析时间戳""" formats = [ "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%d %H:%M:%S", ] for fmt in formats: try: return datetime.strptime(time_str, fmt) except (ValueError, TypeError): continue return None def analyze_network_change(network_history: List[Dict[str, Any]]) -> Dict[str, Any]: """ 分析网络规模变化 Args: network_history: 网络历史数据列表 { "date": "2026-03-01", "followers_count": 100000, "following_count": 1000 } Returns: Dict: 网络变化分析结果 """ if not network_history: return { "initial_followers": 0, "current_followers": 0, "followers_change": 0, "followers_growth_rate": 0, "initial_following": 0, "current_following": 0, "following_change": 0, "trend": "unknown", "daily_changes": [] } # 按日期排序 sorted_history = sorted(network_history, key=lambda x: x.get("date", "")) initial = sorted_history[0] current = sorted_history[-1] # 计算变化 initial_followers = initial.get("followers_count", 0) current_followers = current.get("followers_count", 0) followers_change = current_followers - initial_followers initial_following = initial.get("following_count", 0) current_following = current.get("following_count", 0) following_change = current_following - initial_following # 计算增长率 if initial_followers > 0: followers_growth_rate = (followers_change / initial_followers) * 100 else: followers_growth_rate = 0 # 判断趋势 if abs(followers_change) < initial_followers * 0.05: # 变化小于5% trend = "stable" elif followers_change > 0: trend = "growing" else: trend = "declining" # 生成每日变化 daily_changes = [] for i in range(1, len(sorted_history)): prev = sorted_history[i - 1] curr = sorted_history[i] followers_delta = curr.get("followers_count", 0) - prev.get("followers_count", 0) following_delta = curr.get("following_count", 0) - prev.get("following_count", 0) daily_changes.append({ "date": curr.get("date", ""), "followers_count": curr.get("followers_count", 0), "following_count": curr.get("following_count", 0), "followers_delta": followers_delta, "following_delta": following_delta }) return { "initial_followers": initial_followers, "current_followers": current_followers, "followers_change": followers_change, "followers_growth_rate": round(followers_growth_rate, 2), "initial_following": initial_following, "current_following": current_following, "following_change": following_change, "trend": trend, "daily_changes": daily_changes, "analysis_period_days": len(sorted_history) } def analyze_following_network(following_accounts: List[Dict[str, Any]]) -> Dict[str, Any]: """ 分析关注网络类型 Args: following_accounts: 关注的账号列表 { "id": "account_id", "username": "username", "bio": "bio text", "followers_count": 10000, "verified": true } Returns: Dict: 关注网络分析结果 """ if not following_accounts: return { "total_following": 0, "verified_count": 0, "unverified_count": 0, "influencer_count": 0, "average_followers": 0, "top_influencers": [], "types": {} } total = len(following_accounts) # 认证账号统计 verified_count = sum(1 for acc in following_accounts if acc.get("verified", False)) unverified_count = total - verified_count # 影响力账号（粉丝数 > 100000） influencer_count = sum(1 for acc in following_accounts if acc.get("followers_count", 0) > 100000) # 平均粉丝数 avg_followers = sum(acc.get("followers_count", 0) for acc in following_accounts) / total # Top影响力账号 top_influencers = sorted( following_accounts, key=lambda x: x.get("followers_count", 0), reverse=True )[:10] # 分析账号类型（简化版，基于bio关键词） types = { "media": 0, "government": 0, "ngo": 0, "business": 0, "personal": 0, "other": 0 } type_keywords = { "media": ["news", "media", "journal", "reporter", "press"], "government": ["gov", "official", "ministry", "department", "agency"], "ngo": ["non-profit", "ngo", "organization", "foundation", "charity"], "business": ["company", "ceo", "founder", "startup", "business", "corp"], } for acc in following_accounts: bio = acc.get("bio", "").lower() matched = False for type_name, keywords in type_keywords.items(): if any(keyword in bio for keyword in keywords): types[type_name] += 1 matched = True break if not matched: types["personal"] += 1 return { "total_following": total, "verified_count": verified_count, "unverified_count": unverified_count, "verified_ratio": round(verified_count / total * 100, 2) if total > 0 else 0, "influencer_count": influencer_count, "influencer_ratio": round(influencer_count / total * 100, 2) if total > 0 else 0, "average_followers": round(avg_followers, 2), "top_influencers": [ { "username": acc.get("username"), "followers_count": acc.get("followers_count", 0), "verified": acc.get("verified", False) } for acc in top_influencers ], "types": {k: round(v / total * 100, 2) if total > 0 else 0 for k, v in types.items()} } def main(): """命令行入口，用于测试""" import sys # 示例数据 example_network_history = [ {"date": "2026-03-01", "followers_count": 100000, "following_count": 1000}, {"date": "2026-03-15", "followers_count": 105000, "following_count": 1005}, {"date": "2026-03-30", "followers_count": 115000, "following_count": 1010}, {"date": "2026-04-01", "followers_count": 120000, "following_count": 1015}, ] example_following = [ {"id": "1", "username": "news_org", "bio": "Breaking news from around the world", "followers_count": 500000, "verified": True}, {"id": "2", "username": "gov_official", "bio": "Official government account", "followers_count": 1000000, "verified": True}, {"id": "3", "username": "personal_user", "bio": "Just sharing my thoughts", "followers_count": 1000, "verified": False}, ] if len(sys.argv) > 1: with open(sys.argv[1], 'r', encoding='utf-8') as f: data = json.load(f) example_network_history = data.get("network_history", example_network_history) example_following = data.get("following_accounts", example_following) # 网络变化分析 network_change = analyze_network_change(example_network_history) print("=== 网络变化分析 ===") print(json.dumps(network_change, ensure_ascii=False, indent=2)) print("\n=== 关注网络分析 ===") following_analysis = analyze_following_network(example_following) print(json.dumps(following_analysis, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()