720402ffe7
🎉 主要更新:
后端:
- 全新华为应用市场爬虫系统
- 三表分离数据库设计 (app_info, app_metrics, app_rating)
- 完整的API接口 (搜索、分类、热门、上新等)
- 元服务自动识别和分类
- 智能Token管理和数据处理
- 修复热门应用重复显示问题
前端:
- 全新首页设计 (今日上架、热门应用)
- 应用页面 (彩色分类磁贴、智能图标匹配)
- 今日上新页面 (日期切换)
- 热门应用页面 (卡片布局)
- 应用详情页面 (完整信息展示)
- Apple风格搜索栏
- Footer组件
- 底部导航栏优化 (4个导航项)
- 骨架屏加载效果
- FontAwesome图标集成
UI/UX:
- 统一浅色背景 (#F5F5F7)
- 流畅的过渡动画
- 响应式设计
- 毛玻璃效果
文档:
- CHANGELOG.md - 完整更新日志
- QUICKSTART.md - 快速开始
- 多个技术文档和使用指南
版本: v2.0.0
107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
import httpx
|
||
import json
|
||
from typing import Optional, Dict, Any
|
||
from app.config import settings
|
||
from app.crawler.token_manager import TokenManager
|
||
|
||
class HuaweiAPI:
|
||
def __init__(self):
|
||
self.base_url = "https://web-drcn.hispace.dbankcloud.com/edge"
|
||
self.locale = "zh_CN"
|
||
self.token_manager = TokenManager()
|
||
self.client = httpx.AsyncClient(timeout=30.0)
|
||
|
||
async def get_app_info(self, pkg_name: Optional[str] = None, app_id: Optional[str] = None) -> Dict[str, Any]:
|
||
"""获取应用基本信息"""
|
||
if not pkg_name and not app_id:
|
||
raise ValueError("必须提供 pkg_name 或 app_id")
|
||
|
||
# 获取token
|
||
tokens = await self.token_manager.get_token()
|
||
|
||
# 构建请求
|
||
url = f"{self.base_url}/webedge/appinfo"
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "HuaweiMarketCrawler/1.0",
|
||
"interface-code": tokens["interface_code"],
|
||
"identity-id": tokens["identity_id"]
|
||
}
|
||
|
||
body = {"locale": self.locale}
|
||
if pkg_name:
|
||
body["pkgName"] = pkg_name
|
||
else:
|
||
body["appId"] = app_id
|
||
|
||
# 发送请求
|
||
response = await self.client.post(url, headers=headers, json=body)
|
||
response.raise_for_status()
|
||
|
||
data = response.json()
|
||
|
||
# 数据清洗
|
||
return self._clean_data(data)
|
||
|
||
async def get_app_rating(self, app_id: str) -> Optional[Dict[str, Any]]:
|
||
"""获取应用评分详情"""
|
||
# 跳过元服务
|
||
if app_id.startswith("com.atomicservice"):
|
||
return None
|
||
|
||
tokens = await self.token_manager.get_token()
|
||
|
||
url = f"{self.base_url}/harmony/page-detail"
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "HuaweiMarketCrawler/1.0",
|
||
"interface-code": tokens["interface_code"],
|
||
"identity-id": tokens["identity_id"]
|
||
}
|
||
|
||
body = {
|
||
"pageId": f"webAgAppDetail|{app_id}",
|
||
"pageNum": 1,
|
||
"pageSize": 100,
|
||
"zone": ""
|
||
}
|
||
|
||
try:
|
||
response = await self.client.post(url, headers=headers, json=body)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
# 解析评分数据
|
||
layouts = data["pages"][0]["data"]["cardlist"]["layoutData"]
|
||
comment_cards = [l for l in layouts if l.get("type") == "fl.card.comment"]
|
||
|
||
if not comment_cards:
|
||
return None
|
||
|
||
star_info_str = comment_cards[0]["data"][0]["starInfo"]
|
||
return json.loads(star_info_str)
|
||
|
||
except Exception as e:
|
||
print(f"获取评分失败: {e}")
|
||
return None
|
||
|
||
def _clean_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""清洗数据"""
|
||
# 移除 \0 字符
|
||
for key, value in data.items():
|
||
if isinstance(value, str):
|
||
data[key] = value.replace('\x00', '')
|
||
|
||
# 移除 AG-TraceId
|
||
data.pop('AG-TraceId', None)
|
||
|
||
# 验证 appId 长度
|
||
if len(data.get('appId', '')) < 15:
|
||
raise ValueError("appId长度小于15，可能是安卓应用")
|
||
|
||
return data
|
||
|
||
async def close(self):
|
||
"""关闭客户端"""
|
||
await self.client.aclose()
|