邹方成
11542275f3
refactor(third_party_api): 重构第三方API模块结构和逻辑 feat(third_party_api): 新增OCR图片识别接口 style(third_party_api): 优化API请求参数和响应模型 build: 添加静态文件目录挂载配置
130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
import logging
|
||
from typing import Dict, Any, Optional
|
||
from app.utils.universal_api_manager import UniversalAPIManager
|
||
from app.schemas.third_party_api import APIResponse
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class ThirdPartyAPIController:
|
||
"""第三方API控制器"""
|
||
|
||
def __init__(self):
|
||
"""初始化控制器"""
|
||
self.api_manager = UniversalAPIManager()
|
||
|
||
async def make_api_request(
|
||
self,
|
||
provider: str,
|
||
endpoint: str,
|
||
params: Dict[str, Any],
|
||
timeout: Optional[int] = None
|
||
) -> APIResponse:
|
||
"""
|
||
发送API请求
|
||
|
||
Args:
|
||
provider: API提供商
|
||
endpoint: API端点
|
||
params: 请求参数
|
||
timeout: 超时时间(秒)
|
||
|
||
Returns:
|
||
APIResponse: API响应
|
||
"""
|
||
try:
|
||
# 发送请求
|
||
result = self.api_manager.make_request(provider, endpoint, params)
|
||
|
||
# 检查响应
|
||
if isinstance(result, dict) and result.get('code') == '200':
|
||
return APIResponse(
|
||
success=True,
|
||
message="请求成功",
|
||
data=result
|
||
)
|
||
else:
|
||
return APIResponse(
|
||
success=False,
|
||
message=f"请求失败: {result.get('msg', '未知错误')}",
|
||
data=result
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"API请求失败: {str(e)}")
|
||
return APIResponse(
|
||
success=False,
|
||
message=f"API请求失败: {str(e)}",
|
||
data=None
|
||
)
|
||
|
||
# 站长之家API便捷方法
|
||
async def query_copyright_software(self, company_name: str, chinaz_ver: str = "1") -> APIResponse:
|
||
"""查询企业软件著作权"""
|
||
return await self.make_api_request(
|
||
provider="chinaz",
|
||
endpoint="copyright_software",
|
||
params={
|
||
"entName": company_name,
|
||
"pageNo": 1,
|
||
"range": 100,
|
||
"ChinazVer": chinaz_ver
|
||
}
|
||
)
|
||
|
||
async def query_patent_info(self, company_name: str, chinaz_ver: str = "1") -> APIResponse:
|
||
"""查询企业专利信息"""
|
||
return await self.make_api_request(
|
||
provider="chinaz",
|
||
endpoint="patent",
|
||
params={
|
||
"searchKey": company_name,
|
||
"pageNo": 1,
|
||
"range": 100,
|
||
"searchType": 0,
|
||
"ChinazVer": chinaz_ver
|
||
}
|
||
)
|
||
|
||
async def recognize_ocr(self, image_url: str, chinaz_ver: str = "1.0") -> APIResponse:
|
||
"""
|
||
OCR图片识别
|
||
|
||
Args:
|
||
image_url: 图片URL地址(支持jpg，png，jpeg，1M以内)
|
||
chinaz_ver: API版本号
|
||
|
||
Returns:
|
||
APIResponse: OCR识别结果
|
||
"""
|
||
return await self.make_api_request(
|
||
provider="chinaz",
|
||
endpoint="recognition_ocr",
|
||
params={
|
||
"url": image_url,
|
||
"ChinazVer": chinaz_ver
|
||
}
|
||
)
|
||
|
||
# 小红书API便捷方法
|
||
async def get_xiaohongshu_note(self, note_id: str) -> APIResponse:
|
||
"""获取小红书笔记详情"""
|
||
params = {"noteId": note_id}
|
||
|
||
return await self.make_api_request(
|
||
provider="xiaohongshu",
|
||
endpoint="xiaohongshu_note_detail",
|
||
params=params
|
||
)
|
||
|
||
async def search_jizhiliao_index(self, params: Dict[str, Any], timeout: int = 30) -> APIResponse:
|
||
"""执行极致聊指数搜索"""
|
||
return await self.make_api_request(
|
||
provider="jizhiliao",
|
||
endpoint="index_search",
|
||
params=params,
|
||
timeout=timeout
|
||
)
|
||
|
||
|
||
# 创建全局控制器实例
|
||
third_party_api_controller = ThirdPartyAPIController() |