249 lines
13 KiB
Python
249 lines
13 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
Manages the construction of prompts for different AI generation tasks.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import traceback
|
||
|
|
from .resource_loader import ResourceLoader # Use relative import within the same package
|
||
|
|
|
||
|
|
class PromptManager:
|
||
|
|
"""Handles the loading and construction of prompts."""
|
||
|
|
|
||
|
|
def __init__(self, config):
|
||
|
|
"""Initializes the PromptManager with the global configuration."""
|
||
|
|
self.config = config
|
||
|
|
# Instantiate ResourceLoader once, assuming it's mostly static methods or stateless
|
||
|
|
self.resource_loader = ResourceLoader()
|
||
|
|
|
||
|
|
def get_topic_prompts(self):
|
||
|
|
"""Constructs the system and user prompts for topic generation."""
|
||
|
|
print("Constructing prompts for topic generation...")
|
||
|
|
try:
|
||
|
|
# --- System Prompt ---
|
||
|
|
system_prompt_path = self.config.get("topic_system_prompt")
|
||
|
|
if not system_prompt_path:
|
||
|
|
print("Error: topic_system_prompt path not specified in config.")
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
# Use ResourceLoader's static method directly
|
||
|
|
system_prompt = ResourceLoader.load_file_content(system_prompt_path)
|
||
|
|
if not system_prompt:
|
||
|
|
print(f"Error: Failed to load topic system prompt from '{system_prompt_path}'.")
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
# --- User Prompt ---
|
||
|
|
user_prompt_path = self.config.get("topic_user_prompt")
|
||
|
|
if not user_prompt_path:
|
||
|
|
print("Error: topic_user_prompt path not specified in config.")
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
base_user_prompt = ResourceLoader.load_file_content(user_prompt_path)
|
||
|
|
if base_user_prompt is None: # Check for None explicitly
|
||
|
|
print(f"Error: Failed to load base topic user prompt from '{user_prompt_path}'.")
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
# --- Build the dynamic part of the user prompt (Logic moved from prepare_topic_generation) ---
|
||
|
|
user_prompt_dynamic = "你拥有的创作资料如下：\n"
|
||
|
|
|
||
|
|
# Add genPrompts directory structure
|
||
|
|
gen_prompts_path = self.config.get("prompts_dir")
|
||
|
|
if gen_prompts_path and os.path.isdir(gen_prompts_path):
|
||
|
|
try:
|
||
|
|
gen_prompts_list = os.listdir(gen_prompts_path)
|
||
|
|
for gen_prompt_folder in gen_prompts_list:
|
||
|
|
folder_path = os.path.join(gen_prompts_path, gen_prompt_folder)
|
||
|
|
if os.path.isdir(folder_path):
|
||
|
|
try:
|
||
|
|
# List files, filter out subdirs if needed
|
||
|
|
gen_prompts_files = [f for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f))]
|
||
|
|
user_prompt_dynamic += f"{gen_prompt_folder}\n{gen_prompts_files}\n"
|
||
|
|
except OSError as e:
|
||
|
|
print(f"Warning: Could not list directory {folder_path}: {e}")
|
||
|
|
except OSError as e:
|
||
|
|
print(f"Warning: Could not list base prompts directory {gen_prompts_path}: {e}")
|
||
|
|
else:
|
||
|
|
print(f"Warning: Prompts directory '{gen_prompts_path}' not found or invalid.")
|
||
|
|
|
||
|
|
# Add resource directory contents
|
||
|
|
resource_dir_config = self.config.get("resource_dir", [])
|
||
|
|
for dir_info in resource_dir_config:
|
||
|
|
source_type = dir_info.get("type", "UnknownType")
|
||
|
|
source_file_paths = dir_info.get("file_path", [])
|
||
|
|
for file_path in source_file_paths:
|
||
|
|
# Use ResourceLoader's static method
|
||
|
|
file_content = ResourceLoader.load_file_content(file_path)
|
||
|
|
if file_content:
|
||
|
|
user_prompt_dynamic += f"{source_type}信息:\n{os.path.basename(file_path)}\n{file_content}\n\n"
|
||
|
|
else:
|
||
|
|
print(f"Warning: Could not load resource file {file_path}")
|
||
|
|
|
||
|
|
# Add dateline information (optional)
|
||
|
|
user_prompt_dir = os.path.dirname(user_prompt_path)
|
||
|
|
dateline_path = os.path.join(user_prompt_dir, "2025各月节日宣传节点时间表.md") # Consider making this configurable
|
||
|
|
if os.path.exists(dateline_path):
|
||
|
|
dateline_content = ResourceLoader.load_file_content(dateline_path)
|
||
|
|
if dateline_content:
|
||
|
|
user_prompt_dynamic += f"\n{dateline_content}"
|
||
|
|
|
||
|
|
# Combine dynamic part, base template, and final parameters
|
||
|
|
user_prompt = user_prompt_dynamic + base_user_prompt
|
||
|
|
|
||
|
|
select_num = self.config.get("num", 1) # Default to 1 if not specified
|
||
|
|
select_date = self.config.get("date", "")
|
||
|
|
user_prompt += f"\n选题数量：{select_num}\n选题日期：{select_date}\n"
|
||
|
|
# --- End of moved logic ---
|
||
|
|
|
||
|
|
print(f"Topic prompts constructed. System: {len(system_prompt)} chars, User: {len(user_prompt)} chars.")
|
||
|
|
return system_prompt, user_prompt
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error constructing topic prompts: {e}")
|
||
|
|
traceback.print_exc()
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
def get_content_prompts(self, topic_item):
|
||
|
|
"""Constructs the system and user prompts for content generation based on a topic item."""
|
||
|
|
print(f"Constructing content prompts for topic: {topic_item.get('object', 'N/A')}...")
|
||
|
|
try:
|
||
|
|
# --- System Prompt ---
|
||
|
|
system_prompt_path = self.config.get("content_system_prompt")
|
||
|
|
if not system_prompt_path:
|
||
|
|
print("Error: content_system_prompt path not specified in config.")
|
||
|
|
return None, None
|
||
|
|
# Use ResourceLoader's static method. load_system_prompt was just load_file_content.
|
||
|
|
system_prompt = ResourceLoader.load_file_content(system_prompt_path)
|
||
|
|
if not system_prompt:
|
||
|
|
print(f"Error: Failed to load content system prompt from '{system_prompt_path}'.")
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
# --- User Prompt (Logic moved from ResourceLoader.build_user_prompt) ---
|
||
|
|
user_prompt = ""
|
||
|
|
prompts_dir = self.config.get("prompts_dir")
|
||
|
|
resource_dir_config = self.config.get("resource_dir", [])
|
||
|
|
|
||
|
|
if not prompts_dir or not os.path.isdir(prompts_dir):
|
||
|
|
print(f"Warning: Prompts directory '{prompts_dir}' not found or invalid. Content user prompt might be incomplete.")
|
||
|
|
# Decide whether to return error or continue with potentially incomplete prompt
|
||
|
|
|
||
|
|
# 1. 添加Demand部分 (based on topic_item['logic'])
|
||
|
|
try:
|
||
|
|
logic_file = f"{topic_item['logic']}.txt"
|
||
|
|
demand_path = os.path.join(prompts_dir, "Demand", logic_file)
|
||
|
|
demand_content = ResourceLoader.load_file_content(demand_path)
|
||
|
|
if demand_content:
|
||
|
|
user_prompt += f"Demand:\n{demand_content}\n"
|
||
|
|
else:
|
||
|
|
print(f"Warning: Demand file not found or empty: {demand_path}")
|
||
|
|
except KeyError:
|
||
|
|
print("Warning: 'logic' key missing in topic_item for Demand prompt.")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error processing Demand prompt: {e}")
|
||
|
|
|
||
|
|
# 2. 添加Object信息 (based on topic_item['object'])
|
||
|
|
try:
|
||
|
|
object_name_base = topic_item['object'] # This might be '景点信息-XXX.txt'
|
||
|
|
object_file_path = None
|
||
|
|
# Find the full path for the object file from config
|
||
|
|
for dir_info in resource_dir_config:
|
||
|
|
if dir_info.get("type") == "Object":
|
||
|
|
for file_path in dir_info.get("file_path", []):
|
||
|
|
# Match basename, assuming topic_item['object'] is the basename
|
||
|
|
if os.path.basename(file_path) == object_name_base:
|
||
|
|
object_file_path = file_path
|
||
|
|
break
|
||
|
|
if object_file_path: break
|
||
|
|
|
||
|
|
if object_file_path:
|
||
|
|
object_content = ResourceLoader.load_file_content(object_file_path)
|
||
|
|
if object_content:
|
||
|
|
user_prompt += f"Object信息:\n{object_content}\n"
|
||
|
|
else:
|
||
|
|
print(f"Warning: Object file could not be loaded: {object_file_path}")
|
||
|
|
else:
|
||
|
|
# If basename match fails, maybe topic_item['object'] is just 'XXX'?
|
||
|
|
# Try finding based on substring? This might be ambiguous.
|
||
|
|
print(f"Warning: Object file path not found in config matching basename: {object_name_base}")
|
||
|
|
except KeyError:
|
||
|
|
print("Warning: 'object' key missing in topic_item for Object prompt.")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error processing Object prompt: {e}")
|
||
|
|
|
||
|
|
# 3. 添加Product信息 (if applicable, based on topic_item['product'] and ['product_logic'])
|
||
|
|
try:
|
||
|
|
product_name = topic_item.get('product')
|
||
|
|
product_logic = topic_item.get('product_logic')
|
||
|
|
if product_name and product_logic:
|
||
|
|
product_logic_file = f"{product_logic}.txt"
|
||
|
|
product_logic_path = os.path.join(prompts_dir, "Demand", product_logic_file) # Assuming product logic is also in Demand?
|
||
|
|
product_logic_content = ResourceLoader.load_file_content(product_logic_path)
|
||
|
|
if product_logic_content:
|
||
|
|
user_prompt += f"Produce Demand:\n{product_logic_content}\n"
|
||
|
|
else:
|
||
|
|
print(f"Warning: Product Demand file not found or empty: {product_logic_path}")
|
||
|
|
|
||
|
|
product_file_path = None
|
||
|
|
for dir_info in resource_dir_config:
|
||
|
|
if dir_info.get("type") == "Product":
|
||
|
|
for file_path in dir_info.get("file_path", []):
|
||
|
|
# Match basename
|
||
|
|
if os.path.basename(file_path) == product_name:
|
||
|
|
product_file_path = file_path
|
||
|
|
break
|
||
|
|
if product_file_path: break
|
||
|
|
|
||
|
|
if product_file_path:
|
||
|
|
product_content = ResourceLoader.load_file_content(product_file_path)
|
||
|
|
if product_content:
|
||
|
|
user_prompt += f"Product信息:\n{product_content}\n"
|
||
|
|
else:
|
||
|
|
print(f"Warning: Product file could not be loaded: {product_file_path}")
|
||
|
|
else:
|
||
|
|
print(f"Warning: Product file path not found in config for: {product_name}")
|
||
|
|
except KeyError as e:
|
||
|
|
print(f"Warning: Missing 'product' or 'product_logic' key in topic_item for Product prompt: {e}")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error processing Product prompt: {e}")
|
||
|
|
|
||
|
|
# 4. 添加Style信息 (based on topic_item['style'])
|
||
|
|
try:
|
||
|
|
style_file = f"{topic_item['style']}.txt"
|
||
|
|
style_path = os.path.join(prompts_dir, "Style", style_file)
|
||
|
|
style_content = ResourceLoader.load_file_content(style_path)
|
||
|
|
if style_content:
|
||
|
|
user_prompt += f"Style:\n{style_content}\n"
|
||
|
|
else:
|
||
|
|
print(f"Warning: Style file not found or empty: {style_path}")
|
||
|
|
except KeyError:
|
||
|
|
print("Warning: 'style' key missing in topic_item for Style prompt.")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error processing Style prompt: {e}")
|
||
|
|
|
||
|
|
# 5. 添加Refer信息 (based on topic_item['style_logic'])
|
||
|
|
try:
|
||
|
|
refer_file = f"{topic_item['style_logic']}.txt"
|
||
|
|
refer_path = os.path.join(prompts_dir, "Refer", refer_file)
|
||
|
|
refer_content = ResourceLoader.load_file_content(refer_path)
|
||
|
|
if refer_content:
|
||
|
|
user_prompt += f"Refer:\n{refer_content}\n"
|
||
|
|
else:
|
||
|
|
print(f"Warning: Refer file not found or empty: {refer_path}")
|
||
|
|
except KeyError:
|
||
|
|
print("Warning: 'style_logic' key missing in topic_item for Refer prompt.")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error processing Refer prompt: {e}")
|
||
|
|
|
||
|
|
# --- End of moved logic ---
|
||
|
|
|
||
|
|
print(f"Content prompts constructed. System: {len(system_prompt)} chars, User: {len(user_prompt)} chars.")
|
||
|
|
return system_prompt, user_prompt
|
||
|
|
|
||
|
|
except KeyError as e:
|
||
|
|
# Catch potential KeyErrors from accessing topic_item if a required key is missing early on
|
||
|
|
print(f"Error constructing content prompts: Missing essential key '{e}' in topic_item: {topic_item}")
|
||
|
|
return None, None
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error constructing content prompts: {e}")
|
||
|
|
traceback.print_exc()
|
||
|
|
return None, None
|