AIServ/services/ai_service.py

#1.引入numpy
import numpy as np
#2.引入模型
from models.model_loader import model_loader
#3.引入scikit-learn的余弦相似度函数
from sklearn.metrics.pairwise import cosine_similarity
#4.引入json
import json
#5.引入缓存机制
from cachetools import TTLCache
#6.引入哈希值
import hashlib
#7.引入标题提取
import utils.matching as TitleMatcher
#8.引入模板数据
from services.templateFun import templateData, SectorScheduler
# 引入样式数据
from services.style.style1to4data import sectorStyle1to4Data
from services.style.style5data import sectorStyle5Data
#9.引入随机数
import random
#10.引入请求
from utils.request import post_request,generate_error
#11.引入时间
import time
#12.引入输入类型
import utils.inputType as inputType

# 创建AI处理类
class AIService:
    """第一步：初始化"""
    def __init__(self):
        # 1.1加载文本处理模型
        self.text_model = model_loader.load_text_model()
        # 1.2使用TTLCache，最大缓存50个模板，1小时过期，缓存模板嵌入向量
        self.template_embeddings_cache = TTLCache(maxsize=50, ttl=3600)
        # 1.3用户输入文本
        self.user_input = ""
        # 1.4组件样式数据
        self.component_style_data = []
        # 1.5推理过程
        self.reasoning = []
        # 1.6 当前样式数据对象
        self.current_style_data = None

    """第二步：调用模型能力"""
    #2.1 文字转换成向量
    def text_to_embedding(self, text):
        # 2.1.1如果text是字符串，则转换成列表
        if isinstance(text, str):
            text = [text]
        
        embeddings = self.text_model.encode(text)
        # 2.2.2返回numpy数组，不要转成list
        return embeddings

    #2.2生成文字摘要
    def userMessage_to_title(self, userMessage):
        return TitleMatcher.title_matcher.generate_title(userMessage)

    #2.3筛选通栏
    def filter_sectors(self, userMessage,matched_template_id):
        return inputType.inputType_matcher.generate_inputType(userMessage,matched_template_id)
 
    """第三步：查找最匹配的皮肤模板"""
    def find_best_matching_template(self, user_text, skin_data):
        # 3.1如果皮肤数据为空，则返回None
        if not skin_data:
            return None
        
        # 3.2提取用户输入中的颜色关键词
        color_keywords = self.extract_color_keywords(user_text)
        print(f"提取到用户颜色关键词: {color_keywords}")
        # 更新推理过程
        self.reasoning.append(f"我提取到用户需求中的颜色关键词是: {color_keywords}")
        
        # 3.3根据颜色关键词过滤模板
        filtered_skin_data = skin_data
        if color_keywords:
            filtered_skin_data = self.filter_templates_by_color(skin_data, color_keywords)
            print(f"根据颜色过滤后模板数量: {len(filtered_skin_data)}")
            # 更新推理过程
            self.reasoning.append(f"根据颜色过滤数据以后有: {len(filtered_skin_data)}个符合该需求的模板")
        
        # 3.4获取模板信息（使用缓存）
        template_infos, template_embeddings = self.get_template_embeddings(filtered_skin_data)
        
        # 3.5如果模板信息为空，则返回None
        if not template_infos:
            return None
        
        # 3.6用户输入的文本转化为向量
        user_embedding = self.text_to_embedding(user_text)
        
        # 3.7如果用户输入的文本的维度是1，则转换成2维
        if len(user_embedding.shape) == 1:
            user_embedding = user_embedding.reshape(1, -1)
        if len(template_embeddings.shape) == 1:
            template_embeddings = template_embeddings.reshape(1, -1)
        
        # 3.8通过余弦函数批量计算相似度
        similarities = cosine_similarity(user_embedding, template_embeddings)[0]
        best_index = np.argmax(similarities)
        best_similarity = similarities[best_index]
        
        return {
            'template_info': template_infos[best_index],
            'similarity_score': best_similarity
        }

    # 3.9提取用户输入中的颜色关键词
    def extract_color_keywords(self, text):
        color_map = {
            "红": "红色",
            "绿": "绿色",
            "蓝": "蓝色",
            "黄": "黄色",
            "橙": "橙色",
            "紫": "紫色",
            "黑": "黑色",
            "白": "白色",
            "灰": "灰色"
        }
        
        found_colors = []
        for char, color in color_map.items():
            if char in text:
                found_colors.append(color)
        
        return found_colors
    
    # 3.10根据颜色关键词过滤模板
    def filter_templates_by_color(self, skin_data, color_keywords):
        filtered = []
        for template in skin_data:
            try:
                keywords = json.loads(template.get('template_keyword', '[]'))
                keyword_text = ' '.join(keywords)
                
                # 检查模板关键词是否包含任何颜色关键词
                for color in color_keywords:
                    if color in keyword_text:
                        filtered.append(template)
                        break
            except:
                continue
        
        return filtered

    """第四步：缓存"""
    # 4.1读取或者新建缓存
    def get_template_embeddings(self, skin_data):
        
        if not skin_data:
            return None, []
        
        # 4.1.2准备模板关键词文本
        template_texts = []
        template_infos = []
        
        for template in skin_data:
            try:
                keywords = json.loads(template.get('template_keyword', '[]'))
                keyword_text = ' '.join(keywords)
                template_texts.append(keyword_text)
                template_infos.append(template)
            except json.JSONDecodeError:
                continue
        
        if not template_texts:
            return None, []
        
        # 4.1.3检查缓存
        cache_key = self._get_template_cache_key(skin_data)
        if cache_key in self.template_embeddings_cache:
            print("📦 从缓存加载模板嵌入向量")
            template_embeddings = self.template_embeddings_cache[cache_key]
            return template_infos, template_embeddings
        
        print("🔄 计算新的模板嵌入向量")
        template_embeddings = self.text_model.encode(template_texts)
        
        # 4.1.4存储到缓存（cachetools自动处理过期）
        self.template_embeddings_cache[cache_key] = template_embeddings
        return template_infos, template_embeddings

    # 4.2 生成模板数据的缓存键
    def _get_template_cache_key(self, skin_data):
        template_keys = []
        for template in skin_data:
            template_keys.append(f"{template.get('id')}_{template.get('template_keyword')}")
        combined_key = '|||'.join(sorted(template_keys))
        return hashlib.md5(combined_key.encode('utf-8')).hexdigest()

    # 4.3 清空模板缓存
    def clear_cache(self):
        self.template_embeddings_cache.clear()
        print("🗑️ 模板嵌入向量缓存已清空")
    
     # 4.4 获取模板缓存
    def get_cache_info(self):
        return {
            'template_embeddings_cache_size': len(self.template_embeddings_cache),
            'cached_templates_count': len(self.template_embeddings_cache),
            'cache_maxsize': self.template_embeddings_cache.maxsize,
            'cache_ttl': self.template_embeddings_cache.ttl
        }


    """第五步：生成通栏数据"""
    # 5.1 随机生成通栏数据
    def get_sectors(self,website_id,matched_template_id,token):
        print("开始生成通栏数据！")
        print(matched_template_id)
        # 根据matched_template_id选择样式数据
        if int(matched_template_id) == 5:
            print("正在使用第5套皮肤的数据！")
            self.current_style_data = sectorStyle5Data
        else:
            print("正在使用前4套皮肤的数据！")
            self.current_style_data = sectorStyle1to4Data

        sectors_config = self.current_style_data.sectors_config
        scheduler = SectorScheduler(sectors_config)
        result = scheduler.generate_sequence(10)
        
        # 处理强制位置（倒数位置）
        forced_map = {}
        for sector in sectors_config:
            if 'force_pos' in sector:
                # force_pos: 1(倒数第一), 2(倒数第二), 3(倒数第三)
                forced_map[sector['force_pos']] = sector['name']
        
        # 替换结果中的对应位置
        if forced_map and result:
            for pos, name in forced_map.items():
                idx = -1 * pos
                if abs(idx) <= len(result):
                    print(f"强制调整位置: 将 {name} 放入索引 {idx}")
                    result[idx] = name

        # print("结果:", result)
        sectorNames = []

        # 第一步：获取到筛选结果
        for sector_name in result:
            for sector in sectors_config:
                if sector['name'] == sector_name:
                    sectorNames.append(sector['name'])
                    break

        # 第二步：使用筛选结果把sectors_webSiteData重组为模板数据
        for name in sectorNames:
            # 从字典中直接获取对应的sector数据
            if name in self.current_style_data.sectors_data:
                matched_sector = self.current_style_data.sectors_data[name]
                # 添加到templateData.sectors_webSiteData.index
                # templateData.sectors_webSiteData["template"]["index"].append(matched_sector)
                # print(matched_sector)
                # print(templateData.sectors_webSiteData["template"]["index"])
                templateData.sectors_webSiteData["template"]["index"].append(matched_sector)

        #print("模板数据：" + json.dumps(templateData.sectors_webSiteData["template"]["index"], ensure_ascii=False))
        #获得通栏名称
        ai_service.test_normal_case_cnames()

        # 第三步：选择组件样式
        ai_service.updata_component_style(website_id,matched_template_id,token)

        # 第四步：依照用户需求修改样式
        ai_service.updata_component_style_to_user(website_id,matched_template_id,token)

        # 第五步：生成画布数据
        # 循环templateData.sectors_webSiteData["template"]["index"]，生成画布数据
        # templateData.canvas_data["template"]["index"].append()
        all_y = 0  # 初始化累计高度
        for index,sectorData in enumerate(templateData.sectors_webSiteData["template"]["index"]):
            # 获取当前通栏的高度
            sector_height = sectorData["sectorCanvasHeight"]
            # 生成画布数据并添加到canvas_data
            canvas_item = ai_service.get_canvans_data(sector_height, all_y, sectorData,index)
            templateData.canvas_data["template"]["index"].append(canvas_item)
            # 更新累计高度
            all_y += sector_height

    # 5.2 根据通栏数据生成画布数据
    def get_canvans_data(self,sector_height,all_y,sectorData,sort):
        data = {
            "i": int(time.time() * 1000) * 1000 + random.randint(0, 9999),  # 毫秒级时间戳后加随机4位数
            "x": 0,#通栏的x坐标恒定为0
            "y": all_y,#通栏的y坐标为当前所有通栏的高度之和
            "w": 12,#通栏的宽度恒定为12
            "h": sector_height,#通栏的高度为sectorData["sectorCanvasHeight"]
            "type": sectorData["sectorName"],
            "content":sectorData,
            "dataSort": sort,
            "moved": False
        }

        return data

    # 5.3 生成随机的样式数据
    def updata_component_style(self,website_id,matched_template_id,token):
        getComponentStyle = post_request("public/getAllSectorComponentStyle", {
            "website_id":website_id,
            "template_id": matched_template_id,
        },token)
        # 请补全此方法
        # 检查响应是否成功
        if getComponentStyle.get('code') != 200:
            print(f"获取组件样式失败: {getComponentStyle.get('message')}")
            return
        
        # 获取组件样式数据
        style_data = getComponentStyle.get('data', [])
        self.component_style_data = style_data
        
        # 需要跳过的通栏类型
        skip_sectors = ["adSector", "linkSector", "linkCxfwSector"]

        # 更新推理过程
        self.reasoning.append(f"通栏创建完毕")
        
        # 遍历模板中的每个通栏
        for sector in templateData.sectors_webSiteData["template"]["index"]:
            sector_name = sector["sectorName"]
            
            # 跳过不需要修改样式的通栏
            if sector_name in skip_sectors:
                print(f"跳过 {sector_name} 的组件样式修改")
                continue
            
            # 在样式数据中查找匹配的通栏
            matched_style = next((s for s in style_data if s["sector_id"] == sector_name), None)
            if not matched_style:
                print(f"警告: 未找到 {sector_name} 的组件样式配置")
                continue
            
            # 获取该通栏的组件列表配置
            component_list_config = matched_style.get("component_list", [])
            
            # 遍历通栏中的每个组件
            for idx, component in enumerate(sector["componentList"]):
                if idx < len(component_list_config):
                    comp_config = component_list_config[idx]
                    if comp_config and isinstance(comp_config, list) and len(comp_config) > 0:
                        comp_style_data = comp_config[0].get("component_style_data", [])
                        
                        # 如果有可用的样式数据，随机选择一个样式
                        if comp_style_data:
                            style_count = len(comp_style_data)
                            random_style = random.randint(1, style_count)
                            component["component_style"] = random_style
                            print(f"初始化 {sector_name} 的第{idx+1}个组件样式为: {random_style}")
                            # 更新推理过程
                            #self.reasoning.append(f"初始化 {sector_name} 的第{idx+1}个组件样式为: {random_style}号样式")
                        else:
                            print(f"警告: {sector_name} 的第{idx+1}个组件没有可用的样式数据")
                    else:
                        print(f"警告: {sector_name} 的第{idx+1}个组件配置无效")
                else:
                    print(f"警告: {sector_name} 的组件数量({len(sector['componentList'])})超过配置数量({len(component_list_config)})")
    
    # 5.4 根据用户需求修改样式数据
    def updata_component_style_to_user(self, website_id, matched_template_id, token):
        # 需要跳过的通栏类型
        skip_sectors = ["adSector", "linkSector", "linkCxfwSector"]
        
        # 1. 将用户输入转换为向量
        user_embedding = self.text_to_embedding(self.user_input)
        if len(user_embedding.shape) == 1:
            user_embedding = user_embedding.reshape(1, -1)

        # 更新推理过程
        self.reasoning.append(f"最后让我挑选一下符合用户需求的组件样式")
        
        # 2. 遍历每个通栏
        for sector in templateData.sectors_webSiteData["template"]["index"]:
            sector_name = sector["sectorName"]
            
            # 跳过不需要修改样式的通栏
            if sector_name in skip_sectors:
                print(f"跳过 {sector_name} 的组件样式优化")
                # 更新推理过程
                #self.reasoning.append(f"跳过 {sector_name} 的组件样式优化")
                continue
            
            # 3. 在样式数据中查找匹配的通栏
            matched_style = next((s for s in self.component_style_data if s["sector_id"] == sector_name), None)
            if not matched_style:
                print(f"警告: 未找到 {sector_name} 的组件样式配置")
                continue
            
            # 4. 遍历通栏中的每个组件
            for idx, component in enumerate(sector["componentList"]):
                comp_config_list = matched_style.get("component_list", [])
                if idx >= len(comp_config_list) or not comp_config_list[idx]:
                    continue
                
                comp_config = comp_config_list[idx][0]  # 取第一个配置
                comp_style_data = comp_config.get("component_style_data", [])
                
                if not comp_style_data:
                    continue
                
                # 5. 提取所有样式的描述文本
                style_descriptions = [style["img_name"] for style in comp_style_data]
                
                # 6. 将样式描述转换为向量
                style_embeddings = self.text_to_embedding(style_descriptions)
                
                # 7. 计算用户输入与每个样式的相似度
                similarities = cosine_similarity(user_embedding, style_embeddings)[0]
                best_index = np.argmax(similarities)
                best_style = comp_style_data[best_index]
                
                # 8. 更新组件样式
                component["component_style"] = best_style["img_id"]
                print(f"优化 {sector_name} 的第{idx+1}个组件样式为: {best_style['img_name']} (ID: {best_style['img_id']})")
                # 更新推理过程
                self.reasoning.append(f"选择 {sector_name} 的第{idx+1}个组件样式为: {best_style['img_name']} (ID: {best_style['img_id']})")
    
    
    """其他功能 模拟推理过程"""
    # 获得选择通栏的中文名称
    def test_normal_case_cnames(self):
        #获得生成的模板数据
        result = templateData.sectors_webSiteData["template"]["index"]
        #更新推理过程
        self.reasoning.append(f"接着我要创建一套符合用户需求的通栏组合")
        #遍历result获得所有通栏的Cname
        cnnames = []
        
        # 遍历每个通栏，使用enumerate获取索引
        for i, sector in enumerate(result):
            sector_name = sector["sectorName"]
            # 在sectors_config中查找对应的中文名称
            if self.current_style_data:
                for config in self.current_style_data.sectors_config:
                    if config["name"] == sector_name:
                        cnnames.append(config["CNname"])
                        # 使用通栏在结果列表中的索引+1作为序号
                        self.reasoning.append(f"第{i+1}个通栏选择: {config['CNname']}")
                        break  # 找到后跳出内层循环
         
    # 清理数据
    def clear_data(self):
        #清理模板数据
        templateData.canvas_data["template"]["index"] = []
        templateData.sectors_webSiteData["template"]["index"] = []
        #清理推理数据
        self.reasoning = []
        self.current_style_data = None
        #初始化权重
        sectorStyle1to4Data.reset_sectors_config()
        sectorStyle5Data.reset_sectors_config()

    #添加推理过程
    def add_reasoning(self, reasoning):
        self.reasoning.append(reasoning)

ai_service = AIService()