Viral Shorts

Instructions

Implementation Reference

• src/server.py:649-796 (handler)

  Main handler implementation of discover_niche_trends tool. Analyzes video titles to extract keyword combinations (bigrams/trigrams), calculates average VPH (Views Per Hour) for each niche, filters by minimum video count, ranks by average VPH, and returns a detailed Markdown report of top performing sub-niches within a main topic.
  async def discover_niche_trends( main_topic: str, hours_ago: int = 24, min_videos: int = 3, top_niches: int = 10 ) -> list[TextContent]: """ 工具 5 的实现:发现细分爆款领域 算法流程: 1. 搜索主题关键词,获取大量视频 2. 提取标题中的关键词组合(bigram/trigram) 3. 统计每个细分领域的视频数量和平均 VPH 4. 返回 VPH 最高的细分领域 """ try: from collections import defaultdict from datetime import datetime, timezone client = YouTubeClient() analyzer = ViralAnalyzer() # 获取更多视频用于分析(最多 50 个) videos = client.get_trending_shorts( keyword=main_topic, hours_ago=hours_ago, max_results=50 ) if not videos: return [TextContent( type="text", text=f"未找到 '{main_topic}' 相关的视频" )] # 分析视频(计算 VPH) videos = analyzer.analyze_videos(videos) # 提取关键词组合 niche_data = defaultdict(lambda: { 'videos': [], 'total_vph': 0, 'total_views': 0, 'count': 0 }) for video in videos: # 提取标题中的关键词组合 title_lower = video.title.lower() words = re.findall(r'\b\w+\b', title_lower) # 生成 bigram (2个词的组合) for i in range(len(words) - 1): bigram = f"{words[i]} {words[i+1]}" # 必须包含主题词 if main_topic.lower() in bigram: niche_data[bigram]['videos'].append(video) niche_data[bigram]['total_vph'] += video.vph niche_data[bigram]['total_views'] += video.views niche_data[bigram]['count'] += 1 # 生成 trigram (3个词的组合) for i in range(len(words) - 2): trigram = f"{words[i]} {words[i+1]} {words[i+2]}" if main_topic.lower() in trigram: niche_data[trigram]['videos'].append(video) niche_data[trigram]['total_vph'] += video.vph niche_data[trigram]['total_views'] += video.views niche_data[trigram]['count'] += 1 # 过滤:至少有 min_videos 个视频 filtered_niches = { niche: data for niche, data in niche_data.items() if data['count'] >= min_videos } if not filtered_niches: return [TextContent( type="text", text=f"未找到符合条件的细分领域(至少 {min_videos} 个视频)\n\n建议:\n- 降低 min_videos 参数\n- 扩大时间范围\n- 尝试其他主题关键词" )] # 计算平均 VPH 并排序 niche_rankings = [] for niche, data in filtered_niches.items(): avg_vph = data['total_vph'] / data['count'] avg_views = data['total_views'] / data['count'] niche_rankings.append({ 'niche': niche, 'avg_vph': avg_vph, 'avg_views': avg_views, 'video_count': data['count'], 'top_video': max(data['videos'], key=lambda v: v.vph) }) # 按平均 VPH 排序 niche_rankings.sort(key=lambda x: x['avg_vph'], reverse=True) top_niches_list = niche_rankings[:top_niches] # 生成报告 report = f"# 🔍 '{main_topic}' 主题细分爆款领域分析\n\n" report += f"**分析参数**\n" report += f"- 主题: {main_topic}\n" report += f"- 时间范围: 最近 {hours_ago} 小时\n" report += f"- 分析视频数: {len(videos)}\n" report += f"- 发现细分领域: {len(filtered_niches)} 个\n" report += f"- 最少视频数阈值: {min_videos}\n\n" report += "---\n\n" report += "## 📊 高 VPH 细分领域排行\n\n" for idx, niche_info in enumerate(top_niches_list, 1): report += f"### {idx}. **{niche_info['niche'].title()}**\n\n" report += f"- **平均 VPH**: {niche_info['avg_vph']:,.0f} 次/小时\n" report += f"- **平均播放量**: {niche_info['avg_views']:,.0f}\n" report += f"- **视频数量**: {niche_info['video_count']}\n" top_vid = niche_info['top_video'] report += f"- **代表视频**: [{top_vid.title[:50]}...]({top_vid.url})\n" report += f" - 播放量: {top_vid.views:,}\n" report += f" - VPH: {top_vid.vph:,.0f}\n" report += f" - 互动率: {top_vid.engagement_rate:.2f}%\n\n" # 添加洞察 report += "---\n\n## 💡 洞察建议\n\n" if top_niches_list: top_niche = top_niches_list[0] report += f"1. **最强细分领域**: '{top_niche['niche'].title()}' (平均 VPH {top_niche['avg_vph']:,.0f})\n" report += f"2. **内容策略**: 可以围绕这些细分领域创作类似内容\n" report += f"3. **竞争程度**: 视频数量 {top_niche['video_count']} 个,属于" if top_niche['video_count'] < 5: report += "**蓝海市场**(竞争少)\n" elif top_niche['video_count'] < 10: report += "**成长市场**(适度竞争)\n" else: report += "**红海市场**(竞争激烈)\n" return [TextContent(type="text", text=report)] except Exception as e: return [TextContent( type="text", text=f"❌ 发现细分领域失败: {str(e)}" )]
• src/server.py:256-294 (registration)

  Tool registration in list_tools() function. Defines the discover_niche_trends tool with its name, description, inputSchema including main_topic (required), hours_ago, min_videos, and top_niches parameters with their types, defaults, and validation constraints.
  Tool( name="discover_niche_trends", description=( "智能发现某个主题下的细分爆款领域。" "通过分析大量视频标题,自动识别高 VPH 的关键词组合," "例如在 'AI' 主题下发现 'AI Kpop'、'AI 印度故事' 等细分趋势。" ), inputSchema={ "type": "object", "properties": { "main_topic": { "type": "string", "description": "主题关键词(如 'AI', 'tutorial', 'funny')" }, "hours_ago": { "type": "integer", "description": "时间范围(小时),默认 24 小时", "default": 24, "minimum": 1, "maximum": 720 }, "min_videos": { "type": "integer", "description": "细分领域最少视频数量,默认 3", "default": 3, "minimum": 2, "maximum": 20 }, "top_niches": { "type": "integer", "description": "返回前 N 个细分领域,默认 10", "default": 10, "minimum": 3, "maximum": 30 } }, "required": ["main_topic"] } )
• src/server.py:349-355 (registration)

  Call routing in call_tool() function. Routes tool invocation with name 'discover_niche_trends' to the handler function, extracting arguments from the request and passing them to discover_niche_trends().
  elif name == "discover_niche_trends": return await discover_niche_trends( main_topic=arguments["main_topic"], hours_ago=arguments.get("hours_ago", 24), min_videos=arguments.get("min_videos", 3), top_niches=arguments.get("top_niches", 10) )
• src/models/video.py:7-54 (schema)

  VideoData schema using Pydantic BaseModel. Defines the data structure for YouTube videos with fields including video_id, title, channel_name, views, likes, comments, published_at, engagement_rate, and includes helper methods to_markdown_row() and markdown_header() used for report generation.
  class VideoData(BaseModel): """YouTube Shorts 视频数据模型""" video_id: str = Field(..., description="视频 ID") title: str = Field(..., description="视频标题") channel_name: str = Field(..., description="频道名称") channel_id: str = Field(..., description="频道 ID") channel_subscribers: Optional[int] = Field(None, description="频道订阅数") views: int = Field(..., description="播放量") likes: int = Field(default=0, description="点赞数") comments: int = Field(default=0, description="评论数") published_at: datetime = Field(..., description="发布时间") duration: str = Field(..., description="视频时长") url: str = Field(..., description="视频链接") thumbnail_url: Optional[str] = Field(None, description="缩略图链接") description: Optional[str] = Field(None, description="视频描述") engagement_rate: float = Field(default=0.0, description="互动率 (%)") class Config: """Pydantic 配置""" json_encoders = { datetime: lambda v: v.isoformat() } def to_markdown_row(self) -> str: """转换为 Markdown 表格行""" published_time = self.published_at.strftime('%Y-%m-%d %H:%M') return ( f"| [{self.title[:40]}...]({self.url}) " f"| {self.channel_name[:20]} " f"| {self.views:,} " f"| {self.likes:,} " f"| {self.comments:,} " f"| {self.engagement_rate:.2f}% " f"| {published_time} |" ) @staticmethod def markdown_header() -> str: """Markdown 表格头""" return ( "| 标题 | 频道 | 播放量 | 点赞数 | 评论数 | 互动率 | 发布时间 |\n" "|------|------|--------|--------|--------|--------|----------|" )
• src/youtube/analyzer.py:58-69 (helper)

  ViralAnalyzer.analyze_videos() helper method used by discover_niche_trends to compute engagement rates for all videos. This is called to prepare video metrics before niche analysis.
  @classmethod def analyze_videos(cls, videos: List[VideoData]) -> List[VideoData]: """ 批量分析视频 Args: videos: 视频列表 Returns: 分析后的视频列表 """ return [cls.analyze_video(video) for video in videos]