content-finder-agent/examples/content_finder/tools/hotspot.py

"""
热点宝工具

提供抖音视频画像数据分析功能。
"""

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)
        )