import sys # print(sys.path) # sys.path.append("/data/data/luosy/project/sku_search") import time import os from PIL import Image import uuid from prompt import * from llm import * import json from conf import * import re 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 } def search_json_files(filename): directory = "/data/data/luosy/project/sku_search/database/meta" data = {} # im=Image.new('RGB', (800, 800), 'white') for file in os.listdir(directory): if file.endswith('.json') and filename in file: with open(os.path.join(directory, file), 'r') as f: data = json.load(f) if data =={}: return None,"暂无数据","暂无数据","暂无数据","暂无数据","暂无数据" return data["商品图像"], data["价格"], data["色号"], data["成分"], data["关键词"], data["商品细节"] def is_valid_response(response): """ 检查响应是否是有效的JSON格式且包含所需字段 """ try: response_dict = json.loads(response) required_fields = ["服装描述(中文)", "服装描述(英文)", "3个关键点(英文)"] return all(field in response_dict for field in required_fields) except Exception as e: return False def extract_response_values(response): """ 从响应中提取各个字段的值 Args: response: JSON格式的响应字符串 Returns: tuple: (中文描述, 英文描述, 关键点) """ try: response_dict = json.loads(response) cn_desc = response_dict.get("服装描述(中文)", "") en_desc = response_dict.get("服装描述(英文)", "") key_points = response_dict.get("3个关键点(英文)", "") return cn_desc, en_desc, key_points except Exception as e: return "", "", "" # def doubao_request(pre_signed_url_output, info,method, max_retries=1): # """ # 带重试机制的豆包请求 # Args: # pre_signed_url_output: 预签名URL # info: 商品信息 # max_retries: 最大重试次数 # Returns: # tuple: (中文描述, 英文描述, 关键点) # """ # info = plugins[method](info) # for attempt in range(max_retries): # try: # result = doubao_MMM_request( pre_signed_url_output, info) # if is_valid_response(result): # ch,en,key=extract_response_values(result) # if len(en)>350 and method=="ch_en_selling_points": # sysp,usrp=TextControl(en) # en=get_lm_text(sysp,usrp) # return ch,en,key # except Exception as e: # print(f"Attempt {attempt + 1} failed with error: {str(e)}") # if attempt == max_retries - 1: # raise e # raise Exception("Failed to get valid response after maximum retries") # def doubao_generate(image,info,method): # if not os.path.exists(MMMConfig.root_path): # os.makedirs(MMMConfig.root_path) # file_name=f"{uuid.uuid4()}.jpg" # original_file=os.path.join(MMMConfig.root_path,file_name) # # 添加这一行,将RGBA转换为RGB # if image.mode == 'RGBA': # image = image.convert('RGB') # image.save(original_file) # compressed_file = original_file # object_key = file_name # quality = compress_image(original_file, compressed_file) # pre_signed_url_output = upload_tos(compressed_file,object_key) # if pre_signed_url_output and os.path.exists(compressed_file): # os.remove(compressed_file) # return doubao_request(pre_signed_url_output,info,method) # def generate_text(id,image=None,info="",method="ch_en_selling_points"): # if len(id) == 9: # id_image,id_price, id_color, id_ingredient, id_selling_point, id_details=search_json_files(id) # else: # id_image,id_price, id_color, id_ingredient, id_selling_point, id_details=None,None,None,None,None,None # info=id_details+info if id_details else info # if image == None and id_image != None: # image=Image.open(id_image) # ch_sen,en_sen,key_point=doubao_generate(image,info,method) # return ch_sen,en_sen,key_point,id_image,id_price, id_color, id_ingredient, id_selling_point, id_details # def gen_title(id,info=None,method="ch_en_selling_title"): # info_new=None # if len(id) == 9: # info_new=generate_text(id)[1] # if info_new ==None and info !=None: # info_new=info # sysp,usrp = plugins[method](info_new) # res=get_lm_text(sysp,usrp) # res_dict = json.loads(res) # # print(res) # return res_dict["en_tile"],res_dict["ch_title"] 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(): 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,method="en_ch_selling_points_his",plat="ali",model_name="mm_tyqw"): # his="""1.Embrace the spirit of freedom with this Hanma denim\n # 2.Meet Aria, your new wardrobe staple that exudes elegance and comfort\n # 3.Embrace the elegance of this 16 momme mulberry silk wide-leg pant\n # 4.Meet your new best friend, the ultra-lightweight and versatile 'Transformer' down jacket\n # 5.Embrace the allure of this enchanting blouse, where every detail whispers sophistication\n # 6.Meet Gloria\n # 7.Discover Gloria # 8.Step into the spotlight with Gloria""" his=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] usrp = plugins[method](plm_info,his) # print(usrp) llm=llm_request(*key,model) for _ in range(3): try: result = llm.llm_mm_request(usrp,img) print(result) if is_valid_response(result): ch,en,kw=extract_response_values(result) add_history(en) break except Exception as e: print(f"Attempt failed with error: {str(e)}") # result={"descr":en,"keywords":kw} return en,kw def generate_text_new(plm_info,img,graphic_label=None,method="TextControl_his",plat="ali",model_name1="mm_tyqw",model_name2="text_qwen"): his=get_history() print(his) if graphic_label: plm_info+="\n以下是该衣服的关键点:"+graphic_label if plat=="ali": key=ali_ky model1=ali_model[model_name1] model2=ali_model[model_name2] else: key=doubao_ky model=doubao_model[model_name1] sim_prompt="如果图片里有文字的话,请结合图片里的衣服和文本信息进行描述一下衣服,还要具体到衣服的风格,细节,版型,设计等" # print(usrp) llm_mm=llm_request(*key,model1) llm_lm=llm_request(*key,model2) res1 = llm_mm.llm_mm_request(sim_prompt,img) sys_prompt,usr_prompt = plugins[method](res1,his) print(usr_prompt) for _ in range(3): try: # print(sys_prompt,usr_prompt) result=llm_lm.llm_text_request(usr_prompt,sys_prompt) print(result) if is_valid_response(result): ch,en,kw=extract_response_values(result) add_history(en) break except Exception as e: print(f"Attempt failed with error: {str(e)}") # result={"descr":en,"keywords":kw} return ch,en,kw def gen_title(info,tags=None,referencr_title=None,method="ch_en_selling_title",plat="doubao",model_name="text_ds"): 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"]} # @app.route('/title', methods=['POST']) # def request_title(): # try: # spu = request.form.get("spu", None) # referencr_title = request.form.get("referencr_title", None) # desc = request.form.get("desc") # if not desc: # return jsonify({"error": "desc is required"}), 400 # tags=request.form.get("tags", None) # result=gen_title(desc,tags,referencr_title) # response = { # "spu": spu, # "result": result # } # return jsonify(response), 200 # except Exception as e: # return jsonify({"error": str(e)}), 500 # @app.route('/description', methods=['POST']) # def request_description(): # try: # spu = request.form.get("spu", None) # reference_url = request.form.get("reference_url", None) # graphic_label = request.form.get("graphic_label", None) # img = request.form.get("img") # if not img: # return jsonify({"error": "img is required"}), 400 # plm_info = request.form.get("plm_info") # if not plm_info: # return jsonify({"error": "plm_info is required"}), 400 # result=generate_text(plm_info,img,graphic_label) # response = { # "spu": spu, # "result": result # } # return jsonify(response), 200 # except Exception as e: # return jsonify({"error": str(e)}), 500 if __name__ == "__main__": 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) print(id_details,id_image) for _ in range(3): result=generate_text_new('1、16mm平纹香云纱;2、香蕉裤型;3、裤身立体收省;4、后腰包橡筋;5、两侧插袋;两边插袋里面有网布贴袋,一大一小,大的可套手机,小的可套耳机或卡片;6、真门牌,大溪地贝壳扣',"/data/data/Mia/product_env_project/gen_sellpoint/企业微信截图_17439887354277.png") 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)