邹方成
9d950ba368
- 将估值计算改为后台任务执行，提高响应速度 - 添加估值评估记录的软删除功能 - 更新评估状态字段值从approved/rejected改为success/fail - 修复注册接口的HTTP状态码问题 - 更新API版本号和服务器配置 - 禁用FastAPI尾部斜杠重定向
144 lines
4.4 KiB
Python
144 lines
4.4 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.0") -> 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 = "2.0") -> APIResponse:
|
||
"""查询企业专利信息"""
|
||
return await self.make_api_request(
|
||
provider="chinaz",
|
||
endpoint="patent",
|
||
params={
|
||
"searchKey": company_name,
|
||
"pageNo": "1",
|
||
"range": "100",
|
||
"searchType": '1',
|
||
"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
|
||
)
|
||
|
||
async def run_dify_workflow(
|
||
self,
|
||
zl_img: str,
|
||
response_mode: str = "blocking",
|
||
user: str = "default_user"
|
||
) -> APIResponse:
|
||
"""运行Dify工作流"""
|
||
params = {
|
||
"inputs": {"zl_img": zl_img},
|
||
"response_mode": response_mode,
|
||
"user": user
|
||
}
|
||
return await self.make_api_request("dify", "workflows_run", params)
|
||
|
||
|
||
# 创建全局控制器实例
|
||
third_party_api_controller = ThirdPartyAPIController() |