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: dict
    basic_data: dict
    judgement_type: list
    
judge_type_map = {
    '1': 'is_keyword',                      # 关键词引流
    '2': 'is_similarity_logo',              # 近似商标
    '3': 'is_brand_1to1',                   # 商标一比一侵权
    '4': 'is_low_price',                    # 低价
    '5': 'is_in_authorize_sale_list'        # 未生成产品
}
    
@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):
    """侵权判定全流程接口"""
    judgement_result = {}
    for type in request.judgement_type:
        if type == 1:
            # 关键词引流判断
            is_keyword = ApiService.keyword_judgement(request.url_data, request.basic_data)
            judgement_result[judge_type_map[str(type)]] = is_keyword
        elif type == 2:
            # 近似商标侵权判断
            is_similarity_logo = ApiService.similarity_logo_judgement(request.url_data, request.basic_data)
            judgement_result[judge_type_map[str(type)]] = is_similarity_logo
        elif type == 3:
            # 商标1比1侵权判断
            is_brand_1to1 = ApiService.logo_1to1_judgement(request.url_data, request.basic_data)
            judgement_result[judge_type_map[str(type)]] = is_brand_1to1
        elif type == 4:
            # 低价判断
            is_low_price = ApiService.low_price_judgement(request.url_data, request.basic_data)
            print(is_low_price)
            judgement_result[judge_type_map[str(type)]] = is_low_price
        elif type == 5:
            # 授权商品判断
            is_in_authorize_sale_list = ApiService.license_judgement(request.url_data, request.basic_data)
            judgement_result[judge_type_map[str(type)]] = is_in_authorize_sale_list
        else:
            judgement_result[f'类型{type}'] = '未制定侵权逻辑，无法判定'
    
    return {
        'code': 200,
        'message': 'sucess',
        'data': {
            'juedgement_result': judgement_result
        }
    }

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