""" 热点宝工具 提供抖音视频画像数据分析功能。 """ import json from typing import Optional, Dict, Any import httpx from agent.tools import tool, ToolResult # API 基础配置 HOTSPOT_API = "http://your-hotspot-api.com" # 替换为实际API地址 DEFAULT_TIMEOUT = 60.0 @tool( display={ "zh": { "name": "获取视频画像数据", "params": { "video_url": "抖音视频链接", "platform_video_id": "平台视频ID" } } } ) async def hotspot_get_video_profile( video_url: Optional[str] = None, platform_video_id: Optional[str] = None, ) -> ToolResult: """ 获取抖音视频的受众画像数据 Args: video_url: 抖音视频链接（二选一） platform_video_id: 平台视频ID（二选一） Returns: ToolResult 包含画像数据： { "code": 0, "message": "success", "data": { "video_id": "7123456789", "basic_stats": { "like_count": 12700, "comment_count": 856, "share_count": 432, "play_count": 125000 }, "audience_profile": { "age_distribution": { "18-24": 0.15, "25-34": 0.25, "35-44": 0.20, "45-54": 0.25, "55+": 0.15 }, "gender_distribution": { "male": 0.45, "female": 0.55 }, "region_distribution": { "一线城市": 0.30, "二线城市": 0.35, "三线及以下": 0.35 }, "device_distribution": { "ios": 0.40, "android": 0.60 } }, "engagement_metrics": { "avg_watch_time": 145, "completion_rate": 0.78, "interaction_rate": 0.12 }, "content_tags": ["历史", "教育", "感动", "正能量"], "emotional_analysis": { "primary_emotion": "感动", "emotion_scores": { "感动": 0.85, "敬佩": 0.70, "悲伤": 0.45 } } } } """ try: if not video_url and not platform_video_id: return ToolResult( title="参数错误", output="", error="必须提供 video_url 或 platform_video_id 之一" ) url = f"{HOTSPOT_API}/api/video/profile" payload = {} if video_url: payload["video_url"] = video_url if platform_video_id: payload["platform_video_id"] = platform_video_id async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client: response = await client.post( url, json=payload, headers={"Content-Type": "application/json"}, ) response.raise_for_status() data = response.json() video_id = data.get("data", {}).get("video_id", "Unknown") return ToolResult( title=f"视频画像: {video_id}", output=json.dumps(data, ensure_ascii=False, indent=2), long_term_memory=f"Retrieved video profile for {video_id}" ) except httpx.HTTPStatusError as e: return ToolResult( title="获取画像失败", output="", error=f"HTTP error {e.response.status_code}: {e.response.text}" ) except Exception as e: return ToolResult( title="获取画像失败", output="", error=str(e) ) @tool( display={ "zh": { "name": "分析受众匹配度", "params": { "video_profile": "视频画像数据", "target_audience": "目标受众定义" } } } ) async def hotspot_analyze_audience( video_profile: Dict[str, Any], target_audience: Dict[str, Any], ) -> ToolResult: """ 分析视频受众与目标受众的匹配度 Args: video_profile: 视频画像数据（从 hotspot_get_video_profile 获取） target_audience: 目标受众定义，例如： { "age_range": ["45-54", "55+"], "min_age_percentage": 0.30, "primary_emotion": "感动", "min_engagement_rate": 0.10 } Returns: ToolResult 包含匹配度分析： { "code": 0, "message": "success", "data": { "match_score": 0.85, "match_level": "高度匹配", "details": { "age_match": { "score": 0.90, "target_percentage": 0.40, "actual_percentage": 0.40, "status": "符合" }, "emotion_match": { "score": 0.85, "target_emotion": "感动", "actual_emotion": "感动", "emotion_score": 0.85, "status": "符合" }, "engagement_match": { "score": 0.80, "target_rate": 0.10, "actual_rate": 0.12, "status": "符合" } }, "recommendation": "该视频高度符合目标受众特征，建议引入" } } """ try: # 这里实现匹配度计算逻辑 # 实际项目中可以调用API或本地计算 audience_data = video_profile.get("audience_profile", {}) age_dist = audience_data.get("age_distribution", {}) emotional_data = video_profile.get("emotional_analysis", {}) engagement_data = video_profile.get("engagement_metrics", {}) # 计算年龄匹配度 target_age_ranges = target_audience.get("age_range", []) target_age_percentage = sum(age_dist.get(age, 0) for age in target_age_ranges) min_age_percentage = target_audience.get("min_age_percentage", 0.30) age_match = target_age_percentage >= min_age_percentage # 计算情感匹配度 target_emotion = target_audience.get("primary_emotion", "") actual_emotion = emotional_data.get("primary_emotion", "") emotion_score = emotional_data.get("emotion_scores", {}).get(target_emotion, 0) emotion_match = actual_emotion == target_emotion or emotion_score >= 0.70 # 计算互动匹配度 actual_engagement = engagement_data.get("interaction_rate", 0) min_engagement = target_audience.get("min_engagement_rate", 0.10) engagement_match = actual_engagement >= min_engagement # 综合评分 match_score = ( (0.90 if age_match else 0.50) * 0.4 + (emotion_score if emotion_match else 0.50) * 0.4 + (0.80 if engagement_match else 0.50) * 0.2 ) match_level = "高度匹配" if match_score >= 0.80 else "中度匹配" if match_score >= 0.60 else "低度匹配" result = { "code": 0, "message": "success", "data": { "match_score": round(match_score, 2), "match_level": match_level, "details": { "age_match": { "score": 0.90 if age_match else 0.50, "target_percentage": min_age_percentage, "actual_percentage": round(target_age_percentage, 2), "status": "符合" if age_match else "不符合" }, "emotion_match": { "score": round(emotion_score, 2), "target_emotion": target_emotion, "actual_emotion": actual_emotion, "emotion_score": round(emotion_score, 2), "status": "符合" if emotion_match else "不符合" }, "engagement_match": { "score": 0.80 if engagement_match else 0.50, "target_rate": min_engagement, "actual_rate": round(actual_engagement, 2), "status": "符合" if engagement_match else "不符合" } }, "recommendation": f"该视频{match_level}，{'建议引入' if match_score >= 0.70 else '建议进一步评估'}" } } return ToolResult( title=f"受众匹配分析: {match_level}", output=json.dumps(result, ensure_ascii=False, indent=2), long_term_memory=f"Analyzed audience match, score: {match_score:.2f}" ) except Exception as e: return ToolResult( title="分析失败", output="", error=str(e) )