#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Manages the construction of prompts for different AI generation tasks. """ import os import traceback import logging # Add logging from .resource_loader import ResourceLoader # Use relative import within the same package class PromptManager: """Handles the loading and construction of prompts.""" def __init__(self, topic_system_prompt_path: str, topic_user_prompt_path: str, content_system_prompt_path: str, prompts_dir: str, resource_dir_config: list, topic_gen_num: int = 1, # Default values if needed topic_gen_date: str = "" ): self.topic_system_prompt_path = topic_system_prompt_path self.topic_user_prompt_path = topic_user_prompt_path self.content_system_prompt_path = content_system_prompt_path self.prompts_dir = prompts_dir self.resource_dir_config = resource_dir_config self.topic_gen_num = topic_gen_num self.topic_gen_date = topic_gen_date def get_topic_prompts(self): """Constructs the system and user prompts for topic generation.""" logging.info("Constructing prompts for topic generation...") try: # --- System Prompt --- if not self.topic_system_prompt_path: logging.error("Topic system prompt path not provided during PromptManager initialization.") return None, None system_prompt = ResourceLoader.load_file_content(self.topic_system_prompt_path) if not system_prompt: logging.error(f"Failed to load topic system prompt from '{self.topic_system_prompt_path}'.") return None, None # --- User Prompt --- if not self.topic_user_prompt_path: logging.error("Topic user prompt path not provided during PromptManager initialization.") return None, None base_user_prompt = ResourceLoader.load_file_content(self.topic_user_prompt_path) if base_user_prompt is None: logging.error(f"Failed to load base topic user prompt from '{self.topic_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 if self.prompts_dir and os.path.isdir(self.prompts_dir): try: gen_prompts_list = os.listdir(self.prompts_dir) for gen_prompt_folder in gen_prompts_list: folder_path = os.path.join(self.prompts_dir, 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: logging.warning(f"Could not list directory {folder_path}: {e}") except OSError as e: logging.warning(f"Could not list base prompts directory {self.prompts_dir}: {e}") else: logging.warning(f"Prompts directory '{self.prompts_dir}' not found or invalid.") # Add resource directory contents for dir_info in self.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: logging.warning(f"Could not load resource file {file_path}") # Add dateline information (optional) user_prompt_dir = os.path.dirname(self.topic_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 user_prompt += f"\n选题数量：{self.topic_gen_num}\n选题日期：{self.topic_gen_date}\n" # --- End of moved logic --- logging.info(f"Topic prompts constructed. System: {len(system_prompt)} chars, User: {len(user_prompt)} chars.") return system_prompt, user_prompt except Exception as e: logging.exception("Error constructing topic prompts:") return None, None def get_content_prompts(self, topic_item): """Constructs the system and user prompts for content generation based on a topic item.""" logging.info(f"Constructing content prompts for topic: {topic_item.get('object', 'N/A')}...") try: # --- System Prompt --- if not self.content_system_prompt_path: logging.error("Content system prompt path not provided during PromptManager initialization.") return None, None system_prompt = ResourceLoader.load_file_content(self.content_system_prompt_path) if not system_prompt: logging.error(f"Failed to load content system prompt from '{self.content_system_prompt_path}'.") return None, None # --- User Prompt (Logic moved from ResourceLoader.build_user_prompt) --- user_prompt = "" prompts_dir = self.prompts_dir resource_dir_config = self.resource_dir_config if not prompts_dir or not os.path.isdir(prompts_dir): logging.warning(f"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部分 (直接使用 topic_item['logic'] 的描述性文本) try: demand_description = topic_item.get('logic') if demand_description: user_prompt += f"Demand Logic:\n{demand_description}\n" else: logging.warning("Warning: 'logic' key missing or empty in topic_item for Demand prompt.") except Exception as e: logging.exception("Error processing Demand description:") # 2. Object Info - 先列出所有可用文件，再注入匹配文件的内容 try: object_name_from_topic = topic_item.get('object') # e.g., "尚书第建筑群" object_file_basenames = [] matched_object_file_path = None matched_object_basename = None # 遍历查找 Object 文件 for dir_info in resource_dir_config: if dir_info.get("type") == "Object": for file_path in dir_info.get("file_path", []): basename = os.path.basename(file_path) object_file_basenames.append(basename) # 尝试匹配当前 topic 的 object (仅当尚未找到匹配时) if object_name_from_topic and not matched_object_file_path: cleaned_resource_name = basename if cleaned_resource_name.startswith("景点信息-"): cleaned_resource_name = cleaned_resource_name[len("景点信息-"):] if cleaned_resource_name.endswith(".txt"): cleaned_resource_name = cleaned_resource_name[:-len(".txt")] if cleaned_resource_name and cleaned_resource_name in object_name_from_topic: matched_object_file_path = file_path matched_object_basename = basename # 注意：这里不 break，继续收集所有文件名 # 构建提示词 - Part 1: 文件列表 if object_file_basenames: user_prompt += "Object信息:\n" # user_prompt += f"{object_file_basenames}\n\n" # 直接打印列表可能不够清晰 for fname in object_file_basenames: user_prompt += f"- {fname}\n" user_prompt += "\n" # 加一个空行 logging.info(f"Listed {len(object_file_basenames)} available object files.") else: logging.warning("No resource directory entry found with type 'Object', or it has no file paths.") # 构建提示词 - Part 2: 注入匹配文件内容 if matched_object_file_path: logging.info(f"Attempting to load content for matched object file: {matched_object_basename}") matched_object_content = ResourceLoader.load_file_content(matched_object_file_path) if matched_object_content: user_prompt += f"{matched_object_basename}：\n{matched_object_content}\n\n" logging.info(f"Successfully loaded and injected content for: {matched_object_basename}") else: logging.warning(f"Object file matched ({matched_object_basename}) but could not be loaded or is empty.") elif object_name_from_topic: # 只有当 topic 中指定了 object 但没找到匹配文件时才警告 logging.warning(f"Could not find a matching Object resource file to inject content for '{object_name_from_topic}'. Only the list of files was provided.") except KeyError: logging.warning("Warning: 'object' key potentially missing in topic_item.") except Exception as e: logging.exception("Error processing Object prompt section:") # 3. 添加Product信息 (if applicable) try: product_name = topic_item.get('product') product_logic_description = topic_item.get('product_logic') # Directly use this description if product_name: # Add Product Logic description first (if available) if product_logic_description: user_prompt += f"Product Logic:\n{product_logic_description}\n" else: logging.warning(f"Warning: 'product_logic' key missing or empty for product '{product_name}'.") # Then, load Product Info file 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", []): if product_name in os.path.basename(file_path): 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 Info:\n{product_content}\n" else: logging.warning(f"Product file could not be loaded: {product_file_path}") else: logging.warning(f"Product file path not found in config for: {product_name}") # Removed KeyError check for product_logic as it's now optional text except KeyError: logging.warning("Warning: Missing 'product' key in topic_item for Product prompt.") except Exception as e: logging.exception("Error processing Product prompt:") # 4. 添加Style信息 (加载文件 based on topic_item['style'] from Style/ folder) try: style_filename = topic_item.get('style') if style_filename: # Ensure .txt extension if not style_filename.lower().endswith('.txt'): style_file = f"{style_filename}.txt" else: style_file = style_filename style_path = os.path.join(prompts_dir, "Style", style_file) # Look in Style/ subfolder style_content = ResourceLoader.load_file_content(style_path) if style_content: user_prompt += f"Style Info:\n{style_content}\n" # Changed label for clarity else: logging.warning(f"Style file not found or empty: {style_path}") else: logging.warning("Warning: 'style' key missing or empty in topic_item.") except Exception as e: logging.exception("Error processing Style prompt:") # 5. 添加Target Audience信息 (加载文件 based on topic_item['target_audience'] from Demand/ folder) try: target_audience_filename = topic_item.get('target_audience') if target_audience_filename: # Ensure .txt extension (or use the check as done for style) if not target_audience_filename.lower().endswith('.txt'): target_audience_file = f"{target_audience_filename}.txt" else: target_audience_file = target_audience_filename # Assuming target audience files are in the 'Demand' subdirectory based on old prompt # Load from Demand/ subfolder as confirmed target_audience_path = os.path.join(prompts_dir, "Demand", target_audience_file) target_audience_content = ResourceLoader.load_file_content(target_audience_path) if target_audience_content: user_prompt += f"Target Audience Info:\n{target_audience_content}\n" else: logging.warning(f"Target Audience file not found or empty: {target_audience_path}") else: logging.warning("Warning: 'target_audience' key missing or empty in topic_item.") except Exception as e: logging.exception("Error processing Target Audience prompt:") # 6. 添加Refer信息 (加载 Refer/ 目录下所有文件的内容) try: refer_dir = os.path.join(prompts_dir, "Refer") if os.path.isdir(refer_dir): refer_content_all = "" refer_files = [f for f in os.listdir(refer_dir) if os.path.isfile(os.path.join(refer_dir, f))] logging.info(f"Found {len(refer_files)} files in Refer directory: {refer_files}") for refer_file in refer_files: refer_path = os.path.join(refer_dir, refer_file) content = ResourceLoader.load_file_content(refer_path) if content: refer_content_all += f"--- Refer File: {refer_file} ---\n{content}\n\n" else: logging.warning(f"Could not load Refer file: {refer_path}") if refer_content_all: user_prompt += f"Refer Info:\n{refer_content_all}" else: logging.warning("No content loaded from Refer directory.") else: logging.warning(f"Refer directory not found: {refer_dir}") except Exception as e: logging.exception("Error processing Refer directory:") # --- End of prompt construction logic --- logging.info(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 logging.error(f"Error constructing content prompts: Missing essential key '{e}' in topic_item: {topic_item}") return None, None except Exception as e: logging.exception("Error constructing content prompts:") return None, None