from parsers.base import BaseParser from typing import Dict from urllib.parse import urlencode class KuaishouBugPKParser(BaseParser): """快手解析器 - BugPK API""" def parse(self, video_url: str) -> Dict: """解析快手视频""" try: url = f"{self.api_url}/api/kuaishou?{urlencode({'url': video_url})}" response = self._make_request(url) data = response.json() return self._extract_data(data) except Exception as e: raise Exception(f"快手解析失败(BugPK API): {str(e)}") def _extract_data(self, data: Dict) -> Dict: """提取并标准化数据 实际返回格式: { "code": 200, "msg": "解析成功", "data": { "title": "视频标题", "cover": "封面URL", "url": "视频URL" } } """ try: if data.get("code") == 200: video_data = data.get("data", {}) cover = video_data.get("cover", "") video_url = video_data.get("url", "") title = video_data.get("title", "") description = "" # BugPK API 不返回简介 author = "" # BugPK API 不返回作者 return self._normalize_response(cover, video_url, title, description, author) else: raise Exception(f"解析失败: {data.get('msg', '未知错误')}") except Exception as e: raise Exception(f"数据提取失败: {str(e)}") class KuaishouUctbParser(BaseParser): """快手解析器 - 优创 API""" def parse(self, video_url: str) -> Dict: """解析快手视频""" try: url = f"{self.api_url}/api/videojx?{urlencode({'url': video_url})}" response = self._make_request(url) data = response.json() return self._extract_data(data) except Exception as e: raise Exception(f"快手解析失败(优创 API): {str(e)}") def _extract_data(self, data: Dict) -> Dict: """提取并标准化数据 实际返回格式: { "code": 200, "data": { "desc": "视频描述", "cover": "封面URL", "playurl": "视频URL" }, "msg": "请求成功" } """ try: if data.get("code") == 200: video_data = data.get("data", {}) cover = video_data.get("cover", "") video_url = video_data.get("playurl", "") title = video_data.get("desc", "") description = "" author = "" # 优创API不返回作者 return self._normalize_response(cover, video_url, title, description, author) else: raise Exception(f"解析失败: {data.get('msg', '未知错误')}") except Exception as e: raise Exception(f"数据提取失败: {str(e)}")