diff --git a/app/__init__.py b/app/__init__.py index f3712f6..cbfdb40 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -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 diff --git a/app/api/v1/__init__.py b/app/api/v1/__init__.py index 5443bb0..b64975f 100644 --- a/app/api/v1/__init__.py +++ b/app/api/v1/__init__.py @@ -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]) diff --git a/app/api/v1/third_party_api/third_party_api.py b/app/api/v1/third_party_api/third_party_api.py index 16ebe6c..43b7ea5 100644 --- a/app/api/v1/third_party_api/third_party_api.py +++ b/app/api/v1/third_party_api/third_party_api.py @@ -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: diff --git a/app/api/v1/upload/__init__.py b/app/api/v1/upload/__init__.py new file mode 100644 index 0000000..708e3f9 --- /dev/null +++ b/app/api/v1/upload/__init__.py @@ -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=["文件上传"]) \ No newline at end of file diff --git a/app/api/v1/upload/upload.py b/app/api/v1/upload/upload.py new file mode 100644 index 0000000..9abd6f2 --- /dev/null +++ b/app/api/v1/upload/upload.py @@ -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) \ No newline at end of file diff --git a/app/controllers/third_party_api.py b/app/controllers/third_party_api.py index f377da7..9e9a70e 100644 --- a/app/controllers/third_party_api.py +++ b/app/controllers/third_party_api.py @@ -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( diff --git a/app/controllers/upload.py b/app/controllers/upload.py new file mode 100644 index 0000000..a1af0d1 --- /dev/null +++ b/app/controllers/upload.py @@ -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 + ) \ No newline at end of file diff --git a/app/schemas/third_party_api.py b/app/schemas/third_party_api.py index eb6dad8..6130de1 100644 --- a/app/schemas/third_party_api.py +++ b/app/schemas/third_party_api.py @@ -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="提供商总数") \ No newline at end of file + providers: List[APIProviderInfo] \ No newline at end of file diff --git a/app/schemas/upload.py b/app/schemas/upload.py new file mode 100644 index 0000000..3db93b3 --- /dev/null +++ b/app/schemas/upload.py @@ -0,0 +1,6 @@ +from pydantic import BaseModel + +class ImageUploadResponse(BaseModel): + """图片上传响应模型""" + url: str + filename: str \ No newline at end of file diff --git a/app/utils/api_config.py b/app/utils/api_config.py index 275c234..b162f63 100644 --- a/app/utils/api_config.py +++ b/app/utils/api_config.py @@ -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"] + }, } }, diff --git a/app/utils/universal_api_manager.py b/app/utils/universal_api_manager.py index b7d3d82..441253b 100644 --- a/app/utils/universal_api_manager.py +++ b/app/utils/universal_api_manager.py @@ -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: """