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, ApiService
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                              # 图片链接 
    
class JudgementRequest(BaseModel):
    url_data: str
    basic_data: 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}}

@app.post("/brandanalysis/api/v1/infringe_judgement")
async def infringe_judgement(request: JudgementRequest):
    """侵权判定全流程接口"""
    # 先解析链接和基础数据
    url_data = json.loads(request.url_data)
    basic_data = json.loads(request.basic_data)
    
    # 关键词引流判断
    is_keyword = ApiService.keyword_judgement(url_data, basic_data)
    # 近似商标侵权判断
    is_similarity_logo = ApiService.similarity_logo_judgement(url_data, basic_data)
    # 商标1比1侵权判断
    is_brand_1to1 = ApiService.logo_1to1_judgement(url_data, basic_data)
    # 低价判断
    is_low_price = ApiService.low_price_judgement(url_data, basic_data)
    # 授权商品判断
    is_in_authorize_sale_list = ApiService.license_judgement(url_data, basic_data)
    
    return {
        'code': 200,
        'message': 'sucess',
        'data': {
            'juedgement_result': {
                'is_keyword': is_keyword,
                'is_similarity_logo': is_similarity_logo,
                'is_brand_1to1': is_brand_1to1,
                'is_low_price': is_low_price,
                'is_in_authorize_sale_list': is_in_authorize_sale_list,
            }
        }
    }

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