BrandAnalysis/api.py

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
import json
from pydantic import BaseModel
from utils import Service
import uvicorn

app = FastAPI()

# 添加全局异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={
            "code": 400,
            "msg": "请求参数错误",
            "data": {
                "detail": exc.errors(),
                "body": exc.body
            }
        },
    )

# 定义请求体
class KeyWordRequest(BaseModel):
    brand_name: str                             # 判定品牌名称
    shop_detail_info_brand_name: str            # 商品详情页的品牌名称
    title: str                                  # 商品名称
    
class LicenseRequest(BaseModel):
    brand_name: str                             # 判定品牌名称
    title: str                                  # 商品名称
    
class LogoRequest(BaseModel):
    brand_name: str                             # 判定品牌名称
    image_url: str                              # 图片链接 
    
@app.post("/brandanalysis/api/v1/keyword")
async def brand_key_word_judgement(request: KeyWordRequest):
    """关键词引流判定api"""
    if request.shop_detail_info_brand_name not in request.title:
        title = request.shop_detail_info_brand_name + request.title
    else:
        title = request.title
        
    judgement_res = Service.agent.brand_key_word_judgement(request.brand_name, title)
    
    result = json.loads(judgement_res)
    return {"code": 200, "message": "success", "data": {"judgement_info": result}}

@app.post("/brandanalysis/api/v1/license")
async def license_judgement(request: LicenseRequest):
    license_list = Service.get_license_list(request.brand_name)
    license_judgement = json.loads(Service.agent.license_product_judgement(request.title, license_list))
    return {"code": 200, "message": "success", "data": {"judgement_info": license_judgement}}

@app.post("/brandanalysis/api/v1/logo")
async def logo_judgement(request: LogoRequest):
    logo_path_map = {
        "李宁": "./logo/lining.jpg"
    }
    logo_path = logo_path_map[request.brand_name]
    logo_judgement_res = json.loads(Service.agent.image_logo_judgement(logo_path, request.image_url))
    
    return {"code": 200, "message": "success", "data": {"judgement_info": logo_judgement_res}}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)