123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115 |
- 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"<at id=\"{user_open_id}\"></at>---------------/(ㄒ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"<at id=\"{user_open_id}\"></at>---------------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
|