feat(kb-bilibili-metadata): Add UP subscription tracking support

- Enhance _extract_up_async() with subscription tracking: - video_count: Number of videos - article_count: Number of articles - subscription_info: Last video/article for update detection - Update CLI to support UP subscription queries - All collection types now support basic_info + subscription status
2026-02-10 14:34:12 +08:00
parent 26accdd726
commit d216ce4ba7
2 changed files with 60 additions and 8 deletions
@@ -378,22 +378,35 @@ class BilibiliExtractor:
        return result

    async def _extract_up_async(self, mid: str) -> Dict[str, Any]:
-        """异步提取UP主信息"""
-        result = {}
+        """
+        异步提取UP主信息（支持订阅跟踪）
+        
+        Args:
+            mid: UP主ID
+            
+        Returns:
+            Dict containing UP主完整信息
+        """
+        result = {
+            'type': 'up',
+            'mid': mid,
+            'subscribed': False,
+            'basic_info': {},
+            'subscription_info': {}
+        }

        if not HTTPX_AVAILABLE:
            return {'error': 'httpx 未安装'}

        try:
            async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client:
-                # 获取UP主信息
+                # 获取UP主基本信息
                response = await client.get('https://api.bilibili.com/x/space/acc/info', params={'mid': mid})
                if response.status_code == 200:
                    data = response.json()
                    if data.get('code') == 0:
                        up = data.get('data', {})
-                        result = {
-                            'mid': mid,
+                        result['basic_info'] = {
                            'name': up.get('name', ''),
                            'sex': up.get('sex', ''),
                            'face': up.get('face', ''),
@@ -411,7 +424,46 @@ class BilibiliExtractor:
                if stats_response.status_code == 200:
                    stats_data = stats_response.json()
                    if stats_data.get('code') == 0:
-                        result['video_count'] = stats_data.get('data', {}).get('video', 0)
+                        result['basic_info']['video_count'] = stats_data.get('data', {}).get('video', 0)
+                
+                # 获取UP主专栏数量
+                article_response = await client.get('https://api.bilibili.com/x/space/article', params={'mid': mid, 'pn': 1, 'ps': 1})
+                if article_response.status_code == 200:
+                    article_data = article_response.json()
+                    if article_data.get('code') == 0:
+                        result['basic_info']['article_count'] = article_data.get('data', {}).get('count', 0)
+                        
+                # 获取UP主最近视频（用于跟踪更新）
+                video_response = await client.get('https://api.bilibili.com/x/space/arc/search', 
+                    params={'mid': mid, 'pn': 1, 'ps': 1, 'order': 'pubtime'})
+                if video_response.status_code == 200:
+                    video_data = video_response.json()
+                    if video_data.get('code') == 0:
+                        videos = video_data.get('data', {}).get('list', {}).get('vlist', [])
+                        if videos:
+                            latest_video = videos[0]
+                            result['subscription_info'] = {
+                                'last_video_id': latest_video.get('bvid', ''),
+                                'last_video_title': latest_video.get('title', ''),
+                                'last_video_time': datetime.fromtimestamp(latest_video.get('created', 0)).strftime('%Y-%m-%d %H:%M:%S') if latest_video.get('created') else '',
+                            }
+                
+                # 获取UP主最近专栏（用于跟踪更新）
+                article_response = await client.get('https://api.bilibili.com/x/space/article', 
+                    params={'mid': mid, 'pn': 1, 'ps': 1, 'sort': 'publish_time'})
+                if article_response.status_code == 200:
+                    article_data = article_response.json()
+                    if article_data.get('code') == 0:
+                        articles = article_data.get('data', {}).get('articles', [])
+                        if articles:
+                            latest_article = articles[0]
+                            if 'last_article' not in result['subscription_info']:
+                                result['subscription_info']['last_article'] = {}
+                            result['subscription_info']['last_article'] = {
+                                'id': latest_article.get('id', 0),
+                                'title': latest_article.get('title', ''),
+                                'time': datetime.fromtimestamp(latest_article.get('publish_time', 0)).strftime('%Y-%m-%d %H:%M:%S') if latest_article.get('publish_time') else '',
+                            }
                        
        except Exception as e:
            logger.error(f"UP主提取错误: {e}")
@@ -49,8 +49,8 @@ def extract_article(extractor: BilibiliExtractor, cvid: str, output_file: str =


 def extract_up(extractor: BilibiliExtractor, mid: str, output_file: str = None) -> Dict[str, Any]:
-    """提取UP主信息"""
-    result = extractor.extract_up_metadata(mid)
+    """提取UP主信息（包含订阅跟踪所需信息）"""
+    result = asyncio.run(extractor._extract_up_async(mid))
    
    if output_file:
        with open(output_file, 'w', encoding='utf-8') as f: