import json import os from utils.config_io import if_mission_done, if_mission_fail, load_config_from_json, load_lora_options from utils.logger import logger # 从文件中加载模板 def load_template(template_name, reply_message_id = None): template_path = f"template/{template_name}.json" with open(template_path, 'r', encoding='utf-8') as f: template = json.load(f) if template_name in ['choose_card', 'upload_card', 'start_card']: if template_name == 'choose_card' or template_name == 'upload_card': index = 3 elif template_name == 'start_card': index = 4 lora_options= load_lora_options() if reply_message_id: config_data = load_config_from_json(reply_message_id) template['elements'][index]['actions'][0]['initial_option'] = config_data['action'] template['elements'][index]['actions'][1]['initial_option'] = config_data['lora'] for option, value in lora_options.items(): template['elements'][index]['actions'][1]['options'].append( { "text": { "tag": "lark_md", "content": option }, "value": value } ) return template # 加载用户已选选项 def load_initial_option_from_config(reply_message_id, card_content): config_data = load_config_from_json(reply_message_id) card_content['elements'][4]['actions'][0]['initial_option'] = config_data['action'] card_content['elements'][4]['actions'][1]['initial_option'] = config_data['lora'] return card_content # 保存waiting card def save_process_card(card_content, reply_message_id): output_directory = f"data/input_images/{reply_message_id}" os.makedirs(output_directory, exist_ok=True) json_path = os.path.join(output_directory, "process_card.json") with open(json_path, 'w', encoding='utf-8') as f: json.dump(card_content, f, ensure_ascii=False, indent=4) # 更新waiting card def update_process_card(reply_message_id, img_key, upload_key): output_directory = f"data/input_images/{reply_message_id}" os.makedirs(output_directory, exist_ok=True) json_path = os.path.join(output_directory, "process_card.json") with open(json_path, 'r', encoding='utf-8') as f: card_content = json.load(f) # 更新图片 for column in card_content['elements'][1]['columns']: for i, element in enumerate(column["elements"]): if element.get("img_key") == img_key: column["elements"].append( { "tag": "img", "img_key": upload_key, "alt": { "tag": "plain_text", "content": "" }, "mode": "fit_horizontal", "preview": True } ) break config_data = load_config_from_json(reply_message_id) user_open_id = config_data['user_open_id'] # 更新任务状态 if if_mission_fail(reply_message_id): logger.error(reply_message_id + "任务失败") card_content['header'] = { "title": { "tag": "plain_text", "content": "任务执行出现错误,请检查" }, "template": "red" } card_content['elements'][0] = { "tag": "div", "text": { "content": f"---------------/(ㄒoㄒ)/~~", "tag": "lark_md", } } del card_content['elements'][4] del card_content['elements'][4] else: if if_mission_done(reply_message_id): logger.info(reply_message_id + "任务成功") card_content['header'] = { "title": { "tag": "plain_text", "content": "任务完成" }, "template": "green" } card_content['elements'][0] = { "tag": "div", "text": { "content": f"---------------o(* ̄▽ ̄*)ブ", "tag": "lark_md" } } del card_content['elements'][4] del card_content['elements'][4] with open(json_path, 'w', encoding='utf-8') as f: json.dump(card_content, f, ensure_ascii=False, indent=4) return card_content