import os import time import uuid import re from typing import Literal from config.prompt import INTENT_PROMPT, ANSWER_PROMPT from utils.tools import read_json_file, save_dict_to_json, add_suffix, extract_fields, string_to_json from utils.post_data import upload_file_to_tos from utils.logger_config import setup_logger from utils.qa_robot import text_qa, intent_reg, large_order_qa from modules.filter_goods import filter_goods from modules.get_image import get_sku_image from modules.export_excel import export_to_excel, batch_export_to_excel from modules.query_db import get_sku_info, get_sku_id, get_core_sku_freq, get_fusion_order, get_sales logger = setup_logger(__name__) class LargeOrderService: """ 大单匹配业务逻辑服务 """ def __init__(self, output_path="./output/"): self.output_path = output_path self.intent_prompt = INTENT_PROMPT self.answer_prompt = ANSWER_PROMPT def check_file_exists(self, directory, filename): """ 检查特定目录下是否存在指定文件。 :param directory: 目录路径 :param filename: 文件名 :return: 如果文件存在返回 True，否则返回 False """ file_path = os.path.join(directory, filename) return os.path.isfile(file_path) def has_alphanumeric_mix(self, text): """判断字符串中是否包含字母和数字混合的子字符串""" pattern = r'(?=.*[a-zA-Z])(?=.*\d)' return bool(re.search(pattern, text)) def has_number_isdigit(self, text): """使用isdigit()方法检查字符串中是否包含数字""" return any(char.isdigit() for char in text) # 根据SKC查询商品信息 def get_skc_info(self, skc, json_file, get_fields): """ 根据skc值从JSON文件中获取指定多个字段的信息 参数: skc: 要匹配的skc值 json_file: JSON文件路径 get_fields (list): 需要获取的字段名列表（如 ["price", "name"]） 返回: dict: 包含所有指定字段的字典（字段不存在时值为None）；若未找到skc则返回None """ # 读取JSON文件信息 file_info = read_json_file(json_file) # 定义所有需要遍历的goods_info路径（扁平化处理） goods_info_sources = [ [file_info["primary_goods_info"]["goods_info"]], # 单个字典的goods_info [item["goods_info"] for item in file_info["combine_two_info"]], # 二级组合 *[item["goods_info"] for item in file_info["combine_three_info"]] # 三级组合（展开嵌套列表） ] # 遍历所有数据源，匹配skc后提取指定字段 for info_list in goods_info_sources: for item in info_list: if item.get("skc") == skc: # 提取get_fields列表中的所有字段，不存在则为None return {field: item.get(field) for field in get_fields} # 未找到匹配的skc时返回None return None # 获取主推款商品信息 def get_primary_goods_info(self, primary_sku: str, start_date:str = None, end_date: str = None): # 1. 获取商品ID、颜色ID goods_code = primary_sku[:-3] color_code = primary_sku[-3:] combine_id = get_sku_id(goods_code, color_code) goods_id = combine_id[0] color_id = combine_id[1] # 2. 获取主推款频次 result_freq_two = get_core_sku_freq( goods_code = goods_code, color_code = color_code, goods_id = goods_id, color_id = color_id, query_type = "22", start_date = start_date, end_date = end_date ) result_freq_three = get_core_sku_freq( goods_code = goods_code, color_code = color_code, goods_id = goods_id, color_id = color_id, query_type = "33", start_date = start_date, end_date = end_date ) # 3. 获取主推款商品信息 result_info = get_sku_info(goods_id, color_id) result_image = get_sku_image(goods_code+color_code) logger.info(f"\nfreq22:{result_freq_two}\nfreq33:{result_freq_three}\ninfo:{result_info}") goods_info = { "sku": result_info[0], "skc": str(result_info[0]) + str(result_info[2]), "name": result_info[1], "color_code": result_info[2], "color": result_info[3], "season": result_info[4], "category": result_info[5], "price": float(result_info[6]), "image_url": result_image["image_url"], "image_path": result_image["image_path"] } return { "goods_info": goods_info, "goods_code": goods_code, "color_code": color_code, "goods_id": goods_id, "color_id": color_id, "freq_two": int(result_freq_two), "freq_three": int(result_freq_three) } # 获取连带款商品信息 def get_sub_goods_info(self, fusion_order: list, combine_type: Literal["22", "33"]): if combine_type == "22": goods_color = fusion_order[0] goods_count = fusion_order[1] goods_id = goods_color.split('-')[0] color_id = goods_color.split('-')[1] result_info = get_sku_info(goods_id, color_id) goods_info = { "sku": result_info[0], "name": result_info[1], "color_code": result_info[2], "color": result_info[3], "season": result_info[4], "category": result_info[5], "price": float(result_info[6]), "goods_id": goods_id, "color_id": color_id, "skc": str(result_info[0]) + str(result_info[2]) } return { "goods_info": goods_info, "freq": int(goods_count), "count": int(goods_count)*2 } elif combine_type == "33": goods_color = fusion_order[:-1] goods_count = fusion_order[-1] goods_info = [] for combine_id in goods_color: goods_id = combine_id.split('-')[0] color_id = combine_id.split('-')[1] result_info = get_sku_info(goods_id, color_id) result = { "sku": result_info[0], "name": result_info[1], "color_code": result_info[2], "color": result_info[3], "season": result_info[4], "category": result_info[5], "price": float(result_info[6]), "goods_id": goods_id, "color_id": color_id, "skc": str(result_info[0]) + str(result_info[2]) } goods_info.append(result) return { "goods_info": goods_info, "freq": int(goods_count), "count": int(goods_count)*3 } # 获取主推款&连带款 def get_all_goods_info(self, primary_sku: str, start_date: str = None, end_date: str = None): logger.info(f"-----------------执行连带计算：{primary_sku}-----------------") # 1. 获取主商品信息 primary_goods_info = self.get_primary_goods_info(primary_sku, start_date, end_date) # 2. 获取连带商品 combine_results = { type_: get_fusion_order( goods_code = primary_goods_info["goods_code"], color_code = primary_goods_info["color_code"], goods_id = primary_goods_info["goods_id"], color_id = primary_goods_info["color_id"], query_type = type_, start_date = start_date, end_date = end_date ) for type_ in ["22", "33"] } combine_two = combine_results["22"] combine_three = combine_results["33"] # 3. 获取连带商品信息 combine_two_info = [] for item in combine_two: item_info = self.get_sub_goods_info(item, '22') # 筛选22组合 # can_combine = filter_goods.can_combine([primary_goods_info["goods_info"], item_info["goods_info"]]) # if can_combine: # 筛选22组合-开始 sales = get_sales( goods_code = primary_goods_info["goods_code"], color_code = primary_goods_info["color_code"], goods_id = item_info["goods_info"]["goods_id"], color_id = item_info["goods_info"]["color_id"], query_type = "22", sub_goods_id = None, sub_color_id = None, start_date = start_date, end_date = end_date ) result_image = get_sku_image(item_info["goods_info"]["sku"]+item_info["goods_info"]["color_code"]) item_info["goods_info"]["image_url"] = result_image["image_url"] item_info["goods_info"]["image_path"] = result_image["image_path"] item_info["main_freq"] = primary_goods_info["freq_two"] item_info["sales"] = sum([float(sale) for sale in sales]) item_info["combine_rate"] = item_info["freq"] / primary_goods_info["freq_two"] combine_two_info.append(item_info) # else: # logger.info(f"不符合22组合逻辑！") combine_three_info = [] for item in combine_three: item_info = self.get_sub_goods_info(item, '33') # 筛选33组合 # can_combine = filter_goods.can_combine([primary_goods_info["goods_info"], item_info["goods_info"][0], item_info["goods_info"][1]]) # if can_combine: # 筛选33组合 sales = get_sales( goods_code = primary_goods_info["goods_code"], color_code = primary_goods_info["color_code"], goods_id = item_info["goods_info"][0]["goods_id"], color_id = item_info["goods_info"][0]["color_id"], query_type = "33", sub_goods_id = item_info["goods_info"][1]["goods_id"], sub_color_id = item_info["goods_info"][1]["color_id"], start_date = start_date, end_date = end_date ) for item in item_info["goods_info"]: result_image = get_sku_image(item["sku"]+item["color_code"]) item["image_url"] = result_image["image_url"] item["image_path"] = result_image["image_path"] item_info["main_freq"] = primary_goods_info["freq_three"] item_info["sales"] = sum([float(sale) for sale in sales]) if primary_goods_info["freq_three"] == 0: logger.info(f"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx：除数不能为0") item_info["combine_rate"] = 0.0001 else: item_info["combine_rate"] = item_info["freq"] / primary_goods_info["freq_three"] combine_three_info.append(item_info) # else: # logger.info(f"不符合33组合逻辑！") result = { "primary_goods_info": primary_goods_info, "combine_two_info": combine_two_info, "combine_three_info": combine_three_info } file_name = primary_goods_info["goods_code"] + primary_goods_info["color_code"] + '.json' file_path = os.path.join('./output', file_name) save_dict_to_json(result, file_path) logger.info(f"保存连带计算结果：{file_path}") # xlsx_file = export_to_excel(result) # xlsx_url = upload_file_to_tos(xlsx_file) return { "primary_sku": primary_sku, "json_file": file_path # "xlsx_file": xlsx_file, # "xlsx_url": xlsx_url } # 大单搭配 def generate_outfit_orders(self, script_path): """ 生成包含商品详情的搭配订单列表 参数: script_path (str): 存储搭配规则数据的JSON文件路径 product_info_file (str): 存储商品详细信息的JSON文件路径 返回: dict: 搭配订单结果，格式为 {"outfit_orders": [搭配1, 搭配2, ...]} 其中每个搭配配是包含商品详情的字典列表 """ # 1. 读取搭配数据并提取核心商品信息 try: outfit_data = read_json_file(script_path) except Exception as e: raise ValueError(f"读取搭配数据文件失败: {str(e)}") from e # 提取主推款商品基础信息 primary_outfit = outfit_data.get("primary_goods_info", []) # 提取两套搭配中的商品基础信息 two_outfits = outfit_data.get("combine_two_info", []) # 提取三套搭配中的商品基础信息 three_outfits = outfit_data.get("combine_three_info", []) flatten = lambda lst: [item for sublist in lst for item in sublist] outfits_list = [ [item.get("goods_info", {}) for item in [primary_outfit]], [item.get("goods_info", {}) for item in two_outfits], flatten([*[item.get("goods_info", {}) for item in three_outfits ]]) ] # 筛选需要的核心字段（用于后续QA模型） target_core_fields = ["skc", "name", "color", "season", "category"] outfits_info_list = [extract_fields(item, target_core_fields) for item in outfits_list] outfits_info = { "主推款": outfits_info_list[0], "连带款": outfits_info_list[1] + outfits_info_list[2] } # 2. 通过QA模型生成搭配组合和理由 try: qa_response = text_qa(str(outfits_info)) parsed_qa_result = string_to_json(qa_response) logger.info(f"AI搭配结果：{parsed_qa_result}") except Exception as e: raise RuntimeError(f"生成或解析搭配结果失败: {str(e)}") from e # 安全获取搭配组合和对应理由（默认空列表避免报错） outfit_combinations = parsed_qa_result.get("outfit_combine", []) combination_reasons = parsed_qa_result.get("combine_reason", []) logger.info(f"outfit_combinations: \n{outfit_combinations}") logger.info(f"combination_reasons: \n{combination_reasons}") # 3. 为每个搭配组合补充商品详细信息（价格、图片路径等） outfit_orders = [] for skc_combination, reason in zip(outfit_combinations, combination_reasons): # 构建单个搭配组合的完整信息 detailed_combination = [] for product_skc in skc_combination: # 获取商品详细字段（确保返回字典，避免None导致KeyError） product_details = self.get_skc_info(product_skc, script_path, ["price", "image_path"]) or {} # 补充SKC作为唯一标识 product_details["skc"] = product_skc product_details["reason"] = reason logger.info(f"product_details：\n{product_details}") detailed_combination.append(product_details) outfit_orders.append(detailed_combination) outfit_data["outfit_orders"] = outfit_orders save_dict_to_json(outfit_data, script_path) return outfit_data # 主流程 def pipeline(self, primary_sku: str, start_date: str = None, end_date: str = None): qa_sku = primary_sku.upper() process_success = False # 1. 计算TOP20连带款 mid_result = self.get_all_goods_info(qa_sku, start_date, end_date) # 2. 执行大单搭配 mid_result_json = mid_result.get("json_file") final_result = self.generate_outfit_orders(mid_result_json) # 3. 搭配结果写入表格 try: xlsx_file = export_to_excel(final_result) # 4. 搭配结果上传云端 xlsx_url = upload_file_to_tos(xlsx_file) process_success = True logger.info(f"成功完成主推款搭配：{qa_sku}") except Exception as e: process_success = False logger.info(f"主推款搭配组合失败：{qa_sku}, 报错：{e}") # 5. 构建QA内容返回 result = { "primary_sku": mid_result.get("primary_sku"), "json_file": mid_result_json, "xlsx_file": xlsx_file, "xlsx_url": xlsx_url, "success": process_success } # 6. 构建QA知识库 qa_file = add_suffix(mid_result_json, "_qa") qa_content = { "主推款": mid_result.get("primary_sku"), "结果链接": xlsx_url, "是否成功": process_success } save_dict_to_json(qa_content, qa_file) return result def large_order_robot(self, user_query): # 意图识别 intent_result = intent_reg(user_query) intent_json = string_to_json(intent_result) # 提取意图识别结果 query_intent = intent_json.get("意图类型") query_sku = intent_json.get("咨询款号") query_time = intent_json.get("咨询时间") # 非咨询服装搭配 if query_sku == None or not self.has_alphanumeric_mix(str(query_sku)): logger.info(f"一般性问题，快速生成答复中！") response = large_order_qa(user_query, query_intent) # 咨询服装搭配 else: excel_file = f"{uuid.uuid4()}.xlsx" # 指定查询时间：计算所有查询 if query_time and self.has_number_isdigit(str(query_time)): start_date = query_time.split('_')[0] end_date = query_time.split('_')[-1] logger.info(f"基于指定查询时间：{query_time}，执行对：{str(query_sku)} 的统计计算！") for primary_sku in query_sku: self.pipeline(primary_sku, start_date, end_date) # 未指定查询时间：计算未统计的 else: exist_dict = {sku: self.check_file_exists(directory="./output", filename=f"{sku}.xlsx") for sku in query_sku} failed_sku = [sku for sku, exists in exist_dict.items() if not exists] if len(failed_sku) == 0: logger.info(f"查询款号统计结果已预生成！") for primary_sku in failed_sku: logger.info(f"基于当前一个月时间，执行对：{str(failed_sku)} 的统计计算！") self.pipeline(primary_sku) logger.info(f"数据上传中，请稍等！") excel_path = batch_export_to_excel(query_sku, excel_file) excel_url = upload_file_to_tos(excel_path) response = large_order_qa(user_query, query_intent, ANSWER_PROMPT.format(query=str(query_sku), answer=excel_url)) return response large_order_service = LargeOrderService() if __name__ == "__main__": large_order_service = LargeOrderService() # query = "请给出1B9L6E04086Y、1ACLAB10A86Y在七月十六到八月十六期间的大单组合结果" # # query = "你可以为我做些什么呢？" # response = large_order_service.large_order_robot(query) # print(response) priamry_sku = ["1ECRAC07007X", "1CNC6N23008H", "1ECR6J0Z011R", "10NL5J89032H", "1A9L1A20086Y", "1E9CCD23005W"] # "1ECR6J0Z011R", "1ENR8B33005W" # 1ECRAC07007X 1CNC6N23008H 1ECR6J0Z011R 10CL6E57038H 1ENC8B17005W 10NL5J89032H 1A9L1A20086Y 1E9CCD23005W for sku in priamry_sku: try: result = large_order_service.pipeline(sku, "2025-11-03", "2025-11-09") print(result) except Exception as e: logger.error(f"Error processing {sku}: {e}") # logger.info(result["primary_goods_info"]) # two = result["combine_two_info"] # for item in two: # logger.info(item) # three = result["combine_three_info"] # for item in three: # logger.info(item)