zfc/guzhi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(upload): 添加文件上传功能接口和静态文件支持

Browse Source 
refactor(third_party_api): 重构第三方API模块结构和逻辑
feat(third_party_api): 新增OCR图片识别接口
style(third_party_api): 优化API请求参数和响应模型

build: 添加静态文件目录挂载配置
This commit is contained in:
邹方成 2025-10-09 15:19:29 +08:00
parent 258404fa45
commit 11542275f3
11 changed files with 271 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
app/__init__.py
View File

@ -1,6 +1,7 @@
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from tortoise import Tortoise
from app.core.exceptions import SettingNotFound
@ -33,6 +34,8 @@ def create_app() -> FastAPI:
middleware=make_middlewares(),
lifespan=lifespan,
)
# 注册静态文件目录
app.mount("/static", StaticFiles(directory="app/static"), name="static")
register_exceptions(app)
register_routers(app, prefix="/api")
return app

8
app/api/v1/__init__.py
View File

@ -16,6 +16,7 @@ from .menus import menus_router
from .policy.policy import router as policy_router
from .roles import roles_router
from .third_party_api import third_party_api_router
from .upload import router as upload_router
from .users import users_router
from .valuations import router as valuations_router
@ -34,9 +35,10 @@ v1_router.include_router(esg_router, prefix="/esg")
v1_router.include_router(index_router, prefix="/index")
v1_router.include_router(industry_router, prefix="/industry")
v1_router.include_router(policy_router, prefix="/policy")
v1_router.include_router(upload_router, prefix="/upload") # 文件上传路由
v1_router.include_router(
third_party_api_router,
prefix="/third_party_api",
dependencies=[DependAuth, DependPermission],
third_party_api_router,
prefix="/third_party_api",
dependencies=[DependAuth, DependPermission],
)
v1_router.include_router(valuations_router, prefix="/valuations", dependencies=[DependAuth, DependPermission])

47
app/api/v1/third_party_api/third_party_api.py
View File

@ -8,6 +8,7 @@ from app.schemas.base import Success, Fail
from app.schemas.third_party_api import (
BaseAPIRequest,
ChinazAPIRequest,
OCRRequest,
XiaohongshuNoteRequest,
JizhiliaoSearchRequest,
APIResponse,
@ -39,8 +40,7 @@ async def query_copyright_software(request: ChinazAPIRequest):
logger.error(f"查询企业软件著作权失败: {e}")
return Fail(message=f"查询企业软件著作权失败: {str(e)}")
@router.post("/chinaz/patent_info", summary="企业专利信息查询")
@router.post("/chinaz/patent", summary="企业专利信息查询")
async def query_patent_info(request: ChinazAPIRequest):
"""查询企业专利信息"""
try:
@ -57,13 +57,12 @@ async def query_patent_info(request: ChinazAPIRequest):
logger.error(f"查询企业专利信息失败: {e}")
return Fail(message=f"查询企业专利信息失败: {str(e)}")
@router.post("/chinaz/judicial_data", summary="司法综合数据查询")
async def query_judicial_data(request: ChinazAPIRequest):
"""查询司法综合数据"""
@router.post("/chinaz/ocr", summary="OCR图片识别")
async def recognize_ocr(request: OCRRequest):
"""OCR图片识别接口"""
try:
result = await third_party_api_controller.query_judicial_data(
company_name=request.company_name,
result = await third_party_api_controller.recognize_ocr(
image_url=request.url,
chinaz_ver=request.chinaz_ver
)
@ -72,11 +71,11 @@ async def query_judicial_data(request: ChinazAPIRequest):
else:
return Fail(message=result.message)
except Exception as e:
logger.error(f"查询司法综合数据失败: {e}")
return Fail(message=f"查询司法综合数据失败: {str(e)}")
logger.error(f"OCR识别失败: {e}")
return Fail(message=f"OCR识别失败: {str(e)}")
# 小红书API端点
@router.post("/xiaohongshu/note", summary="获取小红书笔记详情")
async def get_xiaohongshu_note(request: XiaohongshuNoteRequest):
"""获取小红书笔记详情"""
@ -90,21 +89,21 @@ async def get_xiaohongshu_note(request: XiaohongshuNoteRequest):
else:
return Fail(message=result.message)
except Exception as e:
logger.error(f"获取小红书笔记失败: {e}")
return Fail(message=f"获取小红书笔记失败: {str(e)}")
logger.error(f"获取小红书笔记详情失败: {e}")
return Fail(message=f"获取小红书笔记详情失败: {str(e)}")
@router.post("/jizhiliao/index_search", summary="极致聊指数搜索")
async def jizhiliao_index_search(request: JizhiliaoSearchRequest):
"""调用极致聊指数搜索接口"""
@router.post("/jizhiliao/search", summary="极致聊指数搜索")
async def search_jizhiliao_index(request: JizhiliaoSearchRequest):
"""执行极致聊指数搜索"""
try:
params = request.model_dump(by_alias=True, exclude_none=True, exclude={"timeout"})
timeout = request.timeout if request.timeout is not None else 30
result = await third_party_api_controller.search_jizhiliao_index(
params=params,
timeout=timeout
)
params = {
"keyword": request.keyword,
"page": request.page,
"size": request.size
}
result = await third_party_api_controller.search_jizhiliao_index(params)
if result.success:
return Success(data=result.data, message=result.message)
else:

5
app/api/v1/upload/__init__.py Normal file
View File

@ -0,0 +1,5 @@
from fastapi import APIRouter
from .upload import router as upload_router
router = APIRouter()
router.include_router(upload_router, prefix="/upload", tags=["文件上传"])

14
app/api/v1/upload/upload.py Normal file
View File

@ -0,0 +1,14 @@
from fastapi import APIRouter, UploadFile, File
from app.controllers.upload import UploadController
from app.schemas.upload import ImageUploadResponse
router = APIRouter()
@router.post("/image", response_model=ImageUploadResponse, summary="上传图片")
async def upload_image(file: UploadFile = File(...)) -> ImageUploadResponse:
"""
上传图片接口
:param file: 图片文件
:return: 图片URL和文件名
"""
return await UploadController.upload_image(file)

99
app/controllers/third_party_api.py
View File

@ -1,53 +1,60 @@
import logging
from typing import Dict, Any, List, Optional
from app.utils.universal_api_manager import universal_api
from app.schemas.third_party_api import (
APIProviderInfo,
APIEndpointInfo,
APIResponse
)
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 = universal_api
"""初始化控制器"""
self.api_manager = UniversalAPIManager()
async def make_api_request(
self,
provider: str,
endpoint: str,
params: Dict[str, Any] = None,
timeout: int = 30
self,
provider: str,
endpoint: str,
params: Dict[str, Any],
timeout: Optional[int] = None
) -> APIResponse:
"""通用API请求方法"""
try:
result = self.api_manager.make_request(
provider=provider,
endpoint=endpoint,
params=params or {},
timeout=timeout
)
"""
发送API请求
Args:
provider: API提供商
endpoint: API端点
params: 请求参数
timeout: 超时时间(秒)
return APIResponse(
success=True,
data=result,
message="请求成功",
provider=provider,
endpoint=endpoint
)
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请求失败: {e}")
logger.error(f"API请求失败: {str(e)}")
return APIResponse(
success=False,
data={},
message=f"请求失败: {str(e)}",
provider=provider,
endpoint=endpoint
message=f"API请求失败: {str(e)}",
data=None
)
# 站长之家API便捷方法
@ -73,19 +80,27 @@ class ThirdPartyAPIController:
"searchKey": company_name,
"pageNo": 1,
"range": 100,
"searchType": "0",
"searchType": 0,
"ChinazVer": chinaz_ver
}
)
async def query_judicial_data(self, company_name: str, chinaz_ver: str = "1") -> APIResponse:
"""查询司法综合数据"""
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="judgement",
endpoint="recognition_ocr",
params={
"q": company_name,
"pageNo": 1,
"url": image_url,
"ChinazVer": chinaz_ver
}
)
@ -100,7 +115,7 @@ class ThirdPartyAPIController:
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(

51
app/controllers/upload.py Normal file
View File

@ -0,0 +1,51 @@
import os
from pathlib import Path
from typing import List
from fastapi import UploadFile
from app.schemas.upload import ImageUploadResponse
class UploadController:
"""文件上传控制器"""
@staticmethod
async def upload_image(file: UploadFile) -> ImageUploadResponse:
"""
上传图片
:param file: 上传的图片文件
:return: 图片URL和文件名
"""
# 检查文件类型
if not file.content_type.startswith('image/'):
raise ValueError("只支持上传图片文件")
# 获取项目根目录
base_dir = Path(__file__).resolve().parent.parent
# 图片保存目录
upload_dir = base_dir / "static" / "images"
# 确保目录存在
if not upload_dir.exists():
upload_dir.mkdir(parents=True, exist_ok=True)
# 生成文件名
filename = file.filename
file_path = upload_dir / filename
# 如果文件已存在，重命名
counter = 1
while file_path.exists():
name, ext = os.path.splitext(filename)
filename = f"{name}_{counter}{ext}"
file_path = upload_dir / filename
counter += 1
# 保存文件
content = await file.read()
with open(file_path, "wb") as f:
f.write(content)
# 返回文件URL
return ImageUploadResponse(
url=f"/static/images/{filename}",
filename=filename
)

87
app/schemas/third_party_api.py
View File

@ -1,72 +1,53 @@
from typing import Dict, Any, Optional, List
from pydantic import BaseModel, Field, ConfigDict
import logging
from typing import List, Optional
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
class BaseAPIRequest(BaseModel):
"""基础API请求模型"""
provider: str = Field(..., description="API提供商", example="chinaz")
endpoint: str = Field(..., description="端点名称", example="icp_info")
params: Dict[str, Any] = Field(default_factory=dict, description="请求参数")
timeout: Optional[int] = Field(30, description="超时时间(秒)")
pass
class ChinazAPIRequest(BaseModel):
class ChinazAPIRequest(BaseAPIRequest):
"""站长之家API请求模型"""
company_name: str = Field(..., description="公司名称", example="百度在线网络技术（北京）有限公司")
chinaz_ver: Optional[str] = Field("1", description="API版本", example="1")
timeout: Optional[int] = Field(30, description="超时时间(秒)")
company_name: Optional[str] = Field(None, description="公司名称")
chinaz_ver: str = Field("1.0", description="API版本号")
class OCRRequest(BaseAPIRequest):
"""OCR识别请求模型"""
url: str = Field(..., description="图片URL地址(支持jpg，png，jpeg，1M以内)")
chinaz_ver: str = Field("1.0", description="API版本号")
class XiaohongshuNoteRequest(BaseModel):
"""小红书笔记详情请求"""
note_id: str = Field(..., description="笔记ID", example="68d2c71d000000000e00e9ea")
class JizhiliaoSearchRequest(BaseModel):
"""极致聊指数搜索请求"""
model_config = ConfigDict(populate_by_name=True)
keyword: str = Field(..., description="搜索关键词", example="人民日报")
mode: int = Field(2, description="搜索模式", example=2)
business_type: int = Field(
8192,
alias="BusinessType",
description="业务类型标识",
example=8192,
)
sub_search_type: int = Field(0, description="子搜索类型", example=0)
verifycode: Optional[str] = Field("", description="验证码", example="")
timeout: Optional[int] = Field(30, description="超时时间(秒)")
class XiaohongshuNoteRequest(BaseAPIRequest):
"""小红书笔记请求模型"""
note_id: str = Field(..., description="笔记ID")
class JizhiliaoSearchRequest(BaseAPIRequest):
"""极致聊搜索请求模型"""
keyword: str = Field(..., description="搜索关键词")
page: int = Field(1, description="页码")
size: int = Field(10, description="每页数量")
class APIResponse(BaseModel):
"""API响应模型"""
success: bool = Field(..., description="请求是否成功")
data: Dict[str, Any] = Field(..., description="响应数据")
message: Optional[str] = Field(None, description="响应消息")
provider: Optional[str] = Field(None, description="API提供商")
endpoint: Optional[str] = Field(None, description="端点名称")
class APIProviderInfo(BaseModel):
"""API提供商信息"""
name: str = Field(..., description="提供商名称")
base_url: str = Field(..., description="基础URL")
description: Optional[str] = Field(None, description="描述")
endpoints: List[str] = Field(default_factory=list, description="可用端点")
success: bool
message: str
data: Optional[dict] = None
class APIEndpointInfo(BaseModel):
"""API端点信息"""
name: str = Field(..., description="端点名称")
path: str = Field(..., description="请求路径")
method: str = Field(..., description="HTTP方法")
description: Optional[str] = Field(None, description="描述")
required_params: List[str] = Field(default_factory=list, description="必需参数")
optional_params: List[str] = Field(default_factory=list, description="可选参数")
path: str
method: str
description: str
required_params: List[str]
optional_params: Optional[List[str]] = None
class APIProviderInfo(BaseModel):
"""API提供商信息"""
name: str
base_url: str
endpoints: dict[str, APIEndpointInfo]
class APIListResponse(BaseModel):
"""API列表响应"""
providers: List[APIProviderInfo] = Field(..., description="API提供商列表")
total_providers: int = Field(..., description="提供商总数")
providers: List[APIProviderInfo]

6
app/schemas/upload.py Normal file
View File

@ -0,0 +1,6 @@
from pydantic import BaseModel
class ImageUploadResponse(BaseModel):
"""图片上传响应模型"""
url: str
filename: str

14
app/utils/api_config.py
View File

@ -29,10 +29,10 @@ class APIConfig:
# 默认配置
default_config = {
"chinaz": {
"api_key": os.getenv("CHINAZ_API_KEY", "YOUR_API_KEY"),
"base_url": "https://openapi.chinaz.net",
"endpoints": {
"copyright_software": {
"api_key": os.getenv("CHINAZ_COPYRIGHT_API_KEY", "YOUR_API_KEY"),
"path": "/v1/1036/copyrightsoftware",
"method": "POST",
"description": "企业软件著作权查询",
@ -40,6 +40,7 @@ class APIConfig:
"optional_params": ["sign"]
},
"patent": {
"api_key": os.getenv("CHINAZ_PATENT_API_KEY", "YOUR_API_KEY"),
"path": "/v1/1036/patent",
"method": "POST",
"description": "企业专利信息查询",
@ -47,12 +48,21 @@ class APIConfig:
"optional_params": ["sign", "searchType"]
},
"judgement": {
"api_key": os.getenv("CHINAZ_JUDGEMENT_API_KEY", "YOUR_API_KEY"),
"path": "/v1/1036/judgementdetailv4",
"method": "POST",
"description": "司法综合数据查询",
"required_params": ["APIKey", "ChinazVer"],
"optional_params": ["sign", "q", "idCardNo", "datatype", "id", "pageNo"]
}
},
"recognition_ocr": {
"api_key": os.getenv("CHINAZ_OCR_API_KEY", "YOUR_API_KEY"),
"path": "/v1/1024/recognition_ocr",
"method": "POST",
"description": "图片OCR识别",
"required_params": ["url", "APIKey", "ChinazVer"],
"optional_params": ["sign"]
},
}
},

65
app/utils/universal_api_manager.py
View File

@ -41,12 +41,69 @@ class UniversalAPIManager:
raise ValueError(f"未找到API提供商配置: {provider}")
return config
def _get_endpoint_config(self, provider: str, endpoint: str) -> Dict[str, Any]:
def _get_endpoint_config(self, provider: str, endpoint: str) -> Optional[Dict[str, Any]]:
"""获取端点配置"""
endpoint_config = self.config.get_endpoint_config(provider, endpoint)
provider_config = self.config.get_api_config(provider)
if not provider_config or 'endpoints' not in provider_config:
return None
return provider_config['endpoints'].get(endpoint)
def _get_api_key(self, provider: str, endpoint: str) -> Optional[str]:
"""获取API密钥"""
endpoint_config = self._get_endpoint_config(provider, endpoint)
if not endpoint_config:
raise ValueError(f"未找到端点配置: {provider}.{endpoint}")
return endpoint_config
return None
return endpoint_config.get('api_key')
def make_request(self, provider: str, endpoint: str, params: Dict[str, Any], timeout: Optional[int] = None) -> Dict[str, Any]:
"""
发送API请求
Args:
provider: API提供商
endpoint: API端点
params: 请求参数
timeout: 超时时间(秒)
Returns:
Dict[str, Any]: API响应
"""
# 获取API配置
endpoint_config = self._get_endpoint_config(provider, endpoint)
if not endpoint_config:
raise ValueError(f"未找到API配置: {provider}/{endpoint}")
# 获取API密钥
api_key = self._get_api_key(provider, endpoint)
if not api_key:
raise ValueError(f"未找到API密钥: {provider}/{endpoint}")
# 获取基础URL
provider_config = self.config.get_api_config(provider)
base_url = provider_config.get('base_url')
if not base_url:
raise ValueError(f"未找到基础URL: {provider}")
# 构建完整URL
url = f"{base_url.rstrip('/')}{endpoint_config['path']}"
# 添加API密钥到参数中
params['APIKey'] = api_key
# 发送请求
try:
response = self.session.request(
method=endpoint_config['method'],
url=url,
json=params if endpoint_config['method'] == 'POST' else None,
params=params if endpoint_config['method'] == 'GET' else None,
timeout=timeout or provider_config.get('timeout', 30)
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"API请求失败: {str(e)}")
raise
def _generate_chinaz_sign(self, date: datetime = None) -> str:
"""