BrandAnalysis/utils/service.py

from agent.agent import Agent
from db import MongoDao
from data import BrandInfo
import gradio as gr
import json
import pandas as pd
from utils import load_image_from_url, load_image_from_cos

product_dao = MongoDao("vbrand-ec")
license_dao = MongoDao("ProductStandard")

class Service:
    product_map = {}
    agent = Agent()
    @staticmethod
    def get_brand_list():
        """获取数据库中存储的品牌列表"""
        brand_list = product_dao.get_one_field_data("brandName")
        brand_list = list(dict.fromkeys(brand_list))
        
        return brand_list
    
    @staticmethod
    def get_init_state_product_list(brand_name):
        """获取初始状态下的产品列表，并返回列表中第一个产品的信息"""
        brand_info = BrandInfo(brand_name)
        prodct_titles = [p.title for p in brand_info.product_list]
        Service.product_map[brand_name] = brand_info.product_list
        
        # 初始商品列表的第一个商品
        init_product = brand_info.product_list[0]
        # init_product.images = [load_image_from_url(image_url) for image_url in init_product.images]
        return prodct_titles, init_product
    
    @staticmethod
    def on_brand_change(brand_name):
        """加载选定品牌的所有商品"""
        brand_info = BrandInfo(brand_name)
        produt_titles = [p.title for p in brand_info.product_list]
        Service.product_map[brand_name] = brand_info.product_list
        return gr.update(choices=produt_titles, value=produt_titles[0])
    
    @staticmethod
    def on_product_change(title, brand_name):
        product_list = Service.product_map.get(brand_name, [])
        product = next((p for p in product_list if p.title == title), None)
        if not product:
            return [], [["商品未找到", ""]]
        
        # 加载图片
        images = [load_image_from_cos(image_url) for image_url in product.images]
        
        # 产品信息
        info = [
            ["商品链接", product.url],
            ["平台", product.platFormName],
            ["商户名称", product.shopTitle],
            ["颜色", product.colors],
            ["尺寸", product.sizes],
            ["价格", product.price],
        ]
        
        product_df = pd.DataFrame(info, columns=["", ""])
        
        return images, product_df, product
    
    @staticmethod
    def get_license_list(brand_name):
        """获取品牌方授权生成的商品列表"""
        license_list = []
        records = license_dao.get_records_by_query({"BrandName":brand_name})
        for record in records:
            if "ProductSeries" not in record.keys():
                record["ProductSeries"] = ""
            product = f"{record['ProductSeries']} {record['Category']}"
            if product not in license_list:
                license_list.append(product)
        return license_list
    
    @staticmethod
    def infringement_judgement(brand_name, product):
        """侵权判定逻辑"""
        # 关键词引流判定
        # if brand_name not in product.title:
        #     # 如果标题中不包含品牌名称，拿到该商品的品牌名称
        key_word_judgement = json.loads(Service.agent.brand_key_word_judgement(brand_name, product.title))
        
        # 未授权商品判定
        license_list = Service.get_license_list("李宁")
        license_judgement = json.loads(Service.agent.license_product_judgement(product.title, license_list))
        
        # 商标侵权判定
        logo_judgement = json.loads(Service.agent.image_logo_judgement("./logo/lining.jpg", product.images[0]))
        
        key_word_falg = key_word_judgement["key_word_flag"]
        license_judgement_flag = license_judgement["in_list"]
        
        result = f"""
            关键词引流: {key_word_falg}
            是否为授权生产产品: {license_judgement_flag}
            
            产品LOGO图像判定:
            图像中是否包含logo: {logo_judgement["is_contain_logo"]}
        """
        
        if logo_judgement["is_contain_logo"]:
            result += f"""是否是指定品牌LOGO: {logo_judgement["is_jugement_logo"]}
            LOGO名称: {logo_judgement["brand_name"]}
            """
        return result

if __name__ == "__main__":
    brand_list = Service.get_brand_list()
    print(brand_list)