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