AI_product_Description/test_chat.py

import sys
import time
import os
from PIL import Image
import requests
from  prompt import *
from llm import *
import json
from conf import *
import re
MAX_RETRIES = 5
MAX_HISTORY = 20
MAX_CHAR_LIMIT = 400
MIN_CHAR_LIMIT = 150 # 假设的最小长度限制

history_list=[]

plugins = {
#     "ch_en_selling_points":get_ch_en_selling_points,
#     "en_ch_selling_points":get_en_ch_selling_points,
    "ch_en_selling_title":get_ch_en_selling_title,
#     "en_ch_selling_points_his":get_en_ch_selling_points_his,
#     "TextControl_his":TextControl_his,
#     "TextControl":TextControl
}





def contains_chinese(text):
    pattern = re.compile(r'[\u4e00-\u9fa5]')
    return bool(pattern.search(text))

def format_history(strings, indent="    "):
    result = ""
    for i, string in enumerate(strings, start=1):
        # 拼接序号、缩进和字符串，并添加换行符
        result += f"{indent}{i}. {string}\n"
    return result

def get_history():
    """获取格式化的历史记录（用于原始prompt）"""
    global history_list
    if len(history_list)==0:
        history=''
    else:
        history=format_history(history_list)
    return history


def add_history(input,max_num=20):
    global history_list
    text = re.split(r'[,\.\!\?\;\:]+', input)
    text=text[0].strip()
    history_list.insert(0, text)
    if len(history_list)>max_num:
        history_list=history_list[:max_num]


def generate_text(plm_info,img,graphic_label=None,plat="ali",model_name="mm_qwen"):

    history_string=get_history()
    # print(his)
    if graphic_label:
        tags_sen=",".join(graphic_label)
        plm_info+="\n'      '以下是该衣服的关键点:"+tags_sen
    if plat=="ali":
        key=ali_ky
        model=ali_model[model_name]
    else:
        key=doubao_ky
        model=doubao_model[model_name]

    llm=llm_request(*key,model)
    en,kw='',''
    result_json = None
    for attempt in range(MAX_RETRIES):
        # --- 构造Prompt ---
        if attempt == 0:
            # 第一次尝试：使用您的主Prompt
            usrp = user_prompt.format(basic_info_string=plm_info,history_string=history_string)
        else:
            # 后续尝试：使用“修正Prompt”
            usrp = get_refinement_prompt(plm_info, history_string, result_json)
            
        print(f"--- 尝试 {attempt + 1} ---")
        # print(prompt) # Debug: 打印Prompt

        response_text = llm.llm_mm_request(usrp,img,sys_text=system_prompt)
        try:
     
            is_valid, validation_error, result_json = validate_response(response_text)
            
            if is_valid:
                # 成功！
                print("生成成功！")
                en,kw=result_json['en'],result_json['kw']
                add_history(en)
                break
            
            else:
                # 失败，记录错误，循环将继续
                print(f"尝试 {attempt + 1} 失败: {validation_error}")
                # result_json 已经包含了失败的文本和错误信息，将用于下一次修正
                continue
                
        except Exception as e:
            # API调用本身失败
            print(f"API 调用失败: {e}")
            result_json = {"error": "API_FAILURE", "raw_response": str(e)}
            continue   
 
    if result_json and result_json.get("error") == "EN_TOO_LONG":
        # 如果是因为超长而失败，且 raw_response 有效
        try:
            failed_data = json.loads(result_json.get("raw_response", "{}"))
            long_en_text = failed_data.get("en")
            
            if long_en_text and len(long_en_text) > MAX_CHAR_LIMIT:
                en = smart_truncate_by_sentence(long_en_text, max_chars=MAX_CHAR_LIMIT)
                kw = failed_data.get("kw", '')
                
                print("执行智能截断成功，返回修正后的文案。")
                
                add_history(en)
        
        except (json.JSONDecodeError, KeyError, TypeError):
            # 无法截断，进入最终错误
            pass

    if isinstance(kw,str):
        kw = [item.strip() for item in kw.split('.') if item.strip()]
    return en,kw


def validate_response(response_text):
    """验证模型的输出是否符合所有规则"""
    try:
        # 规则1: 是否是有效JSON?
        data = json.loads(response_text.strip())
    except json.JSONDecodeError:
        return False, "INVALID_JSON", {"error": "INVALID_JSON", "raw_response": response_text}

    # 规则2: 键是否齐全?
    if not all(k in data for k in ["en", "ch", "kw"]):
        return False, "MISSING_KEYS", {"error": "MISSING_KEYS", "raw_response": json.dumps(data)}
        
    en_text = data.get("en", "")
    
    # 规则3: 长度是否超标? 
    if len(en_text) > MAX_CHAR_LIMIT:
        return False, "EN_TOO_LONG", {"error": "EN_TOO_LONG", "raw_response": json.dumps(data)}
        
    # 规则4: 长度是否太短?
    if len(en_text) < MIN_CHAR_LIMIT:
        return False, "EN_TOO_SHORT", {"error": "EN_TOO_SHORT", "raw_response": json.dumps(data)}
    if contains_chinese(en_text):
        return False, "EN_CONTAINS_CHINESE", {"error": "EN_CONTAINS_CHINESE", "raw_response": json.dumps(data)}
    return True, "SUCCESS", data

def smart_truncate_by_sentence(text, max_chars=MAX_CHAR_LIMIT):
    if len(text) <= max_chars:
        return text
    sentence_endings = re.compile(r'[.!?](?:\s+|$)') 
    best_truncate_point = -1 

    for match in sentence_endings.finditer(text):
        end_position = match.end()
        
        if end_position <= max_chars:
            best_truncate_point = end_position
        else:
            break

    if best_truncate_point > 0:
        truncated_text = text[:best_truncate_point].strip()
        if not truncated_text.endswith(('.', '!', '?')):
            truncated_text += '.'
        return truncated_text.strip()
    else:
        return text[:max_chars-3].strip() + '...'


def get_refinement_prompt(basic_info_string, history_string, failed_result):
    """
    根据上一次的失败原因，生成一个“引导式修正”的Prompt
    """
    failure_reason = failed_result.get("error", "UNKNOWN")
    raw_response = failed_result.get("raw_response", "")

    feedback = ""
    # 尝试提取上次失败的文案
    last_text_en = ""
    try:
        if raw_response:
            last_text_en = json.loads(raw_response).get("en", "")
    except json.JSONDecodeError:
        pass # 无法解析，last_text_en 保持空

    if failure_reason == "INVALID_JSON":
        feedback = f"你上次的输出不是一个有效的JSON。请【严格】按照JSON格式输出。你上次的错误输出是：\n{raw_response}"
    elif failure_reason == "EN_TOO_LONG":
        feedback = f"""
        你上次生成的 "en" 描述【超过了{MAX_CHAR_LIMIT}个字符】！
        【你生成的超长原文】:\n{last_text_en}
        【修正任务】: 请【大幅精简】上述原文，保留核心卖点，使其长度【绝对】在{MIN_CHAR_LIMIT}-{MAX_CHAR_LIMIT}字符以内。
        """
    elif failure_reason == "EN_TOO_SHORT":
        feedback = f"""
        你上次生成的 "en" 描述太短了（小于{MIN_CHAR_LIMIT}字符）。
        【你生成的原文】:\n{last_text_en}
        【修正任务】: 请在原文案基础上，围绕核心卖点再丰富一些细节，使其达到{MIN_CHAR_LIMIT}-{MAX_CHAR_LIMIT}字符。
        """
    elif failure_reason == "MISSING_KEYS":
        feedback = f"你上次输出的JSON缺少 'en', 'ch' 或 'kw' 键。请确保三者齐全。"
    elif failure_reason == "TOO_SIMILAR":
        feedback = "你上次生成的文案与历史记录太相似了。请换一个角度（比如从'材质'或'穿搭场景'）重新构思，字数保持在要求的范围内。"
    elif failure_reason == "EN_CONTAINS_CHINESE":
        feedback = f"""
        你上次生成的 "en" 描述中包含了中文汉字（例如：{last_text_en}）。
        【修正任务】: "en" 字段【必须是纯英文】，【绝对禁止】出现任何中文字符。请严格修正并重新输出。
        """
    else:
        feedback = "你上次的生成失败了。请重新严格按照所有规则生成一次。"

    # 修正Prompt模板
    refinement_prompt = f"""## 角色
你是一个文案修正专家。

## 原始任务
根据以下信息和随消息传入的图片生成文案：{basic_info_string}

## 上次失败的反馈 (你必须修正！)
{feedback}

## 核心规则 (必须再次遵守)
1.  【必须】输出严格的JSON格式。
2.  "en" 描述【必须严格在{MAX_CHAR_LIMIT}字符以内】。
3.  【不要】使用历史开篇：\n{history_string}

## 最终输出
请直接输出修正后的、严格符合要求的JSON字典。
"""
    return refinement_prompt




def gen_title(info,tags=None,referencr_title=None,method="ch_en_selling_title",plat="ali",model_name="text_dsv3"):
    
    if tags:
        tags_sen=",".join(tags)
        info="\n'      '以下是该衣服的关键点:"+tags_sen
    if referencr_title:
        info="\n'      '请以这条标题样例的结构作为借鉴来写这条标题："+referencr_title
    sysp,usrp = plugins[method](info)
    if plat=="ali":
        key=ali_ky
        model=ali_model[model_name]
    else:
        key=doubao_ky
        model=doubao_model[model_name]

    llm=llm_request(*key,model)
    res=llm.llm_text_request(usrp,sysp)
    res_dict = json.loads(res)
    return {"title":res_dict["en_tile"]}

if __name__ == "__main__":
    
    # inf="'Meet your new best friend in fashion—this unisex sweater that whispers comfort and style. Crafted from premium cotton, it feels like a gentle hug on your skin. The heart embroidery adds a touch of whimsy, making you the star of any casual outing. Perfect for layering or wearing solo, this soft companion keeps you cozy all season long."
    # print(gen_title(inf))
    # id_image,id_price, id_color, id_ingredient, id_selling_point, id_details=search_json_files("1A6H4K7V0")
    # id_image=id_image[2:]
    # id_image=os.path.join("/data/data/luosy/project/sku_search",id_image)
    id_image="https://img2.goelia.com.au/prod/product/1ENC6E220/material/main/Shopify/-1/72736752b0ad405382d5ed277dabc660.jpg"
    graphic_label=['-100% Merino wool', '-With pockets', '-H-line fit']
    plm_info='1、手工流苏边设计 \xa0 2、贴袋设计 \xa0 3、金属纽扣'
    # print(id_details,id_image)
    for _ in range(3):
        result=generate_text(plm_info,id_image,graphic_label)
        # result=gen_title("This maxi dress features unparalleled comfort and a unique texture with its <b>tencel blend fabric</b>. The square neckline and smocked bodice create a flattering silhouette, while the layered skirt adds romantic flair. <b>Side pockets and an included scarf scrunchie</b> enhance both style and functionality, elevating its versatility for everyday wear and beyond.")
        print(result)
    # from tqdm import tqdm
    # def image_to_base64(image):
    #     # 将Image对象转换为BytesIO对象
    #     image_io = io.BytesIO()
    #     image.save(image_io, format='PNG')
    #     image_io.seek(0)

    #     # 使用base64编码
    #     image_base64 = base64.b64encode(image_io.read()).decode('utf-8')

    #     return image_base64
    
    # def create_html_with_base64_images(root, output_html):
    #     with open(output_html, 'w', encoding='utf-8') as html_file:
    #         html_file.write('<!DOCTYPE html>\n<html>\n<head>\n<title>Images in Table</title>\n')
    #         html_file.write('<meta charset="UTF-8">\n')  # 添加字符编码声明
    #         html_file.write('<style>\n')
    #         html_file.write('table {\nborder-collapse: collapse;\nwidth: 100%;\n}\n')
    #         html_file.write('table, th, td {\nborder: 1px solid black;\n}\n')
    #         html_file.write('img {\nmax-width: 100%;\nheight: auto;\ndisplay: block;\nmargin-left: auto;\nmargin-right: auto;\n}\n')
    #         html_file.write('</style>\n')
    #         html_file.write('</head>\n<body>\n')
    #         html_file.write('<table>\n')
    #         html_file.write('<tr>\n')
    #         html_file.write('<th>输入的图片</th>\n')  # 第一列：索引
    #         html_file.write('<th>输入的描述</th>\n')  # 第二列：标题
    #         html_file.write('<th>输出的商品详情</th>\n')  # 第二列：标题
    #         html_file.write('<th>输出的商品详情（翻译）</th>\n')  # 第三列：图表
    #         html_file.write('<th>输出的卖点</th>\n')  # 第三列：图表
    #         # for i in range(1, 100):  # 添加序号列1到13
    #         #     html_file.write(f'<th>{i}</th>\n')
    #         html_file.write('</tr>\n')

    #         for file in tqdm(os.listdir(root)[:100], desc="Processing", unit="iter"):
    #             if '.ipynb_checkpoints' in file:
    #                 continue
    #             file_path = os.path.join(root, file)
    #             with open(file_path, 'r') as f:
    #                 data = json.load(f)
    #             if data and "商品图像" in data.keys():
    #                 id_image,id_details=data["商品图像"][2:],  data["商品细节"]
    #             else:
    #                 continue
    #             id_image=os.path.join("/data/data/luosy/project/sku_search",id_image)
                
    #             img_base64 = image_to_base64(Image.open(id_image))

    #             ch,en,kw=generate_text_new(id_details,id_image)
    #             html_file.write('<tr>\n')
    #             # html_file.write(f'<td>{index+1}</td>\n')  # 添加序号
                
    # #             html_file.write('<td>\n')
    # #             html_file.write(f'<img src="data:image/png;base64,{frame_title_img}" alt="Image">\n')
    # #             html_file.write('</td>\n')
    #             html_file.write('<td>\n')
    #             html_file.write(f'<img src="data:image/png;base64,{img_base64}" alt="Image">\n')
    #             html_file.write('</td>\n')
    #             html_file.write(f'<td>{id_details}</td>\n')  # 添加序号
    #             html_file.write(f'<td>{en}</td>\n')  # 添加序号
    #             html_file.write(f'<td>{ch}</td>\n')  # 添加序号
    #             html_file.write(f'<td>{kw}</td>\n')  # 添加序号
    #             # html_file.write('</td>\n')
    #             # for img in image_data:
    #             #     html_file.write('<td>\n')
    #             #     html_file.write(f'<img src="data:image/jpeg;base64,{img}" alt="Image" style="max-width: 100px; max-height: 100px; margin: 5px;">\n')
    #             #     html_file.write('</td>\n')
    #             # html_file.write('</td>\n')
    #             html_file.write('</tr>\n')

    #         html_file.write('</table>\n')
    #         html_file.write('</body>\n</html>')
    # root='/data/data/luosy/project/sku_search/database/meta'
    # create_html_with_base64_images(root, "out——qw_v6.html")



    # app.run(host="0.0.0.0",port=2222,debug=True)

    # print(gen_title(info= "This sweatshirt is a wardrobe essential with its simple yet stylish design and 3D heart pattern that adds a fun visual pop. <b>The unisex design is perfect for couples</b>, and it pairs effortlessly with jeans, cargo pants, or a pleated skirt. Ideal for school, work, or casual outings, it's comfortable and trendy all day long!"))
#     from PIL import Image
#     img1=Image.open("/data/data/luosy/project/sku_search/temp_img/企业微信截图_17372766091671.png")

#     ch_sen,en_sen,key_point,id_image,id_price, id_color, id_ingredient, id_selling_point, id_details=generate_text("",img1,"""-With elastic waistband
# -With hairband
# -X-line fit
# 1.腰部橡筋 2.袖子橡
# 筋 3.前中绳子可调
# 节大小""")
#     print(len(en_sen),end=" ")
#     print(ch_sen,en_sen,key_point)
#     ###############################
#     img2=Image.open("/data/data/luosy/project/sku_search/temp_img/企业微信截图_17389065463149[1](1).png")

#     ch_sen,en_sen,key_point,id_image,id_price, id_color, id_ingredient, id_selling_point, id_details=generate_text("",img2,"""-Washable wool
# -Unisex
# -With silver threads
# 1.后中开衩；2.双扣可调节袖袢；3.暗门筒设计，天然果实扣；4.可水洗羊毛含银葱人字纹面料；5.里面左右两侧均有内袋，左侧最外层内袋是手机袋，防丢失""")
#     print(len(en_sen),end=" ")
#     print(ch_sen,en_sen,key_point)
#     ###############################
#     img3=Image.open("/data/data/luosy/project/sku_search/temp_img/企业微信截图_17392379937637.png")

#     ch_sen,en_sen,key_point,id_image,id_price, id_color, id_ingredient, id_selling_point, id_details=generate_text("",img3,"""-Acetate
# -With pockets
# -Workwear
# 1.描述二醋酸面料:2.扣子为镶钻布包扣;3.半裙后腰包橡筋;4.半裙有
# 侧插袋;5.半裙有侧开隐形拉链，这是两件套套装""")
#     print(len(en_sen),end=" ")
#     print(ch_sen,en_sen,key_point)