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 tencel blend fabric. The square neckline and smocked bodice create a flattering silhouette, while the layered skirt adds romantic flair. Side pockets and an included scarf scrunchie 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('\n\n\nImages in Table\n') # html_file.write('\n') # 添加字符编码声明 # html_file.write('\n') # html_file.write('\n\n') # html_file.write('\n') # html_file.write('\n') # html_file.write('\n') # 第一列：索引 # html_file.write('\n') # 第二列：标题 # html_file.write('\n') # 第二列：标题 # html_file.write('\n') # 第三列：图表 # html_file.write('\n') # 第三列：图表 # # for i in range(1, 100): # 添加序号列1到13 # # html_file.write(f'\n') # html_file.write('\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('\n') # # html_file.write(f'\n') # 添加序号 # # html_file.write('\n') # html_file.write('\n') # html_file.write(f'\n') # 添加序号 # html_file.write(f'\n') # 添加序号 # html_file.write(f'\n') # 添加序号 # html_file.write(f'\n') # 添加序号 # # html_file.write('\n') # # for img in image_data: # # html_file.write('\n') # # html_file.write('\n') # html_file.write('\n') # html_file.write('
输入的图片输入的描述输出的商品详情输出的商品详情（翻译）输出的卖点{i}
{index+1}\n') # # html_file.write(f'Image\n') # # html_file.write('\n') # html_file.write(f'Image\n') # html_file.write('{id_details}{en}{ch}{kw}\n') # # html_file.write(f'Image\n') # # html_file.write('
\n') # html_file.write('\n') # 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. The unisex design is perfect for couples, 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)