#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 提示词构建服务 负责根据选题信息构建完整的提示词 """ import logging from typing import Dict, Any, Optional, Tuple, List from pathlib import Path from core.config import ConfigManager, GenerateContentConfig, GenerateTopicConfig, PosterConfig from utils.prompts import PromptTemplate from api.services.prompt_service import PromptService logger = logging.getLogger(__name__) class PromptBuilderService: """提示词构建服务类""" def __init__(self, config_manager: ConfigManager, prompt_service: PromptService): """ 初始化提示词构建服务 Args: config_manager: 配置管理器 prompt_service: 提示词服务 """ self.config_manager = config_manager self.prompt_service = prompt_service def _ensure_content_config(self) -> GenerateContentConfig: """确保内容生成配置已加载""" # 按需加载内容生成配置 if not self.config_manager.load_task_config('content_gen'): logger.warning("未找到内容生成配置，将使用默认配置") return self.config_manager.get_config('content_gen', GenerateContentConfig) def _ensure_topic_config(self) -> GenerateTopicConfig: """确保选题生成配置已加载""" # 按需加载选题生成配置 if not self.config_manager.load_task_config('topic_gen'): logger.warning("未找到选题生成配置，将使用默认配置") return self.config_manager.get_config('topic_gen', GenerateTopicConfig) def _ensure_poster_config(self) -> PosterConfig: """确保海报生成配置已加载""" # 按需加载海报生成配置 if not self.config_manager.load_task_config('poster_gen'): logger.warning("未找到海报生成配置，将使用默认配置") return self.config_manager.get_config('poster_gen', PosterConfig) def build_content_prompt(self, topic: Dict[str, Any], step: str = "content") -> Tuple[str, str]: """ 构建内容生成提示词 (已重构) 此方法现在依赖于一个预先填充好完整信息的topic对象。 """ content_config = self._ensure_content_config() template = PromptTemplate(content_config.content_system_prompt, content_config.content_user_prompt) # 从预填充的topic对象中直接获取信息，不再调用prompt_service style_obj = topic.get('style_object', {}) style_content = f"{style_obj.get('styleName', '')}\n{style_obj.get('description', '')}" audience_obj = topic.get('audience_object', {}) demand_content = f"{audience_obj.get('audienceName', '')}\n{audience_obj.get('description', '')}" spot_obj = topic.get('scenic_spot_object', {}) object_content = f"{spot_obj.get('name', '')}\n{spot_obj.get('description', '')}" product_obj = topic.get('product_object', {}) product_content = f"{product_obj.get('productName', '')}\n{product_obj.get('detailedDescription', '')}" # 获取通用的参考内容 refer_content = self.prompt_service.get_refer_content(step) system_prompt = template.get_system_prompt() user_prompt = template.build_user_prompt( style_content=style_content, demand_content=demand_content, object_content=object_content, product_content=product_content, refer_content=refer_content ) return system_prompt, user_prompt def build_content_prompt_with_params(self, topic: Dict[str, Any], styles: Optional[List[str]] = None, audiences: Optional[List[str]] = None, scenic_spots: Optional[List[str]] = None, products: Optional[List[str]] = None, step: str = "content") -> Tuple[str, str]: """ 使用额外参数构建内容生成提示词 Args: topic: 选题信息 styles: 风格列表 audiences: 受众列表 scenic_spots: 景区列表 products: 产品列表 step: 当前步骤，用于过滤参考内容 Returns: 系统提示词和用户提示词的元组 """ # 获取内容生成配置 content_config = self._ensure_content_config() # 加载系统提示词和用户提示词模板 system_prompt_path = content_config.content_system_prompt user_prompt_path = content_config.content_user_prompt # 创建提示词模板 template = PromptTemplate(system_prompt_path, user_prompt_path) # 获取风格内容 style_content = '' if styles: style_content = '\n'.join([f"{style}: {self.prompt_service.get_style_content(style)}" for style in styles]) else: style_filename = topic.get("style", "") style_content = f"{style_filename}\n{self.prompt_service.get_style_content(style_filename)}" # 获取目标受众内容 demand_content = '' if audiences: demand_content = '\n'.join([f"{audience}: {self.prompt_service.get_audience_content(audience)}" for audience in audiences]) else: demand_filename = topic.get("targetAudience", "") demand_content = f"{demand_filename}\n{self.prompt_service.get_audience_content(demand_filename)}" # 获取景区信息 object_content = '' if scenic_spots: object_content = '\n'.join([f"{spot}: {self.prompt_service.get_scenic_spot_info(spot)}" for spot in scenic_spots]) else: object_name = topic.get("object", "") object_content = f"{object_name}\n{self.prompt_service.get_scenic_spot_info(object_name)}" # 获取产品信息 product_content = '' if products: product_content = '\n'.join([f"{product}: {self.prompt_service.get_product_info(product)}" for product in products]) else: product_name = topic.get("product", "") product_content = f"{product_name}\n{self.prompt_service.get_product_info(product_name)}" # 获取参考内容 refer_content = self.prompt_service.get_refer_content(step) # 构建系统提示词 system_prompt = template.get_system_prompt() # 构建用户提示词 user_prompt = template.build_user_prompt( style_content=style_content, demand_content=demand_content, object_content=object_content, product_content=product_content, refer_content=refer_content ) return system_prompt, user_prompt def build_poster_prompt(self, topic: Dict[str, Any], content: Dict[str, Any]) -> Tuple[str, str]: """ 构建海报生成提示词 Args: topic: 选题信息 content: 生成的内容 Returns: 系统提示词和用户提示词的元组 """ # 获取海报生成配置 poster_config = self._ensure_poster_config() # 从配置中获取海报提示词模板路径 system_prompt_path = poster_config.poster_system_prompt user_prompt_path = poster_config.poster_user_prompt if not system_prompt_path or not user_prompt_path: raise ValueError("海报提示词模板路径不完整") # 创建提示词模板 template = PromptTemplate(system_prompt_path, user_prompt_path) # 获取景区信息 object_name = topic.get("object", "") object_content = self.prompt_service.get_scenic_spot_info(object_name) # 构建系统提示词 system_prompt = template.get_system_prompt() # 构建用户提示词 user_prompt = template.build_user_prompt( content=content.get("content", ""), title=content.get("title", ""), object_name=object_name, object_content=object_content ) return system_prompt, user_prompt def build_topic_prompt(self, products: Optional[List[str]] = None, scenic_spots: Optional[List[str]] = None, styles: Optional[List[str]] = None, audiences: Optional[List[str]] = None, dates: Optional[str] = None, numTopics: int = 5, style_objects: Optional[List[Dict[str, Any]]] = None, audience_objects: Optional[List[Dict[str, Any]]] = None, scenic_spot_objects: Optional[List[Dict[str, Any]]] = None, product_objects: Optional[List[Dict[str, Any]]] = None) -> Tuple[str, str]: """ 构建选题生成提示词 Args: products: 产品列表 scenic_spots: 景区列表 styles: 风格列表 audiences: 受众列表 dates: 日期字符串，可能为单个日期、多个日期用逗号分隔或范围如'2023-01-01 to 2023-01-31' numTopics: 要生成的选题数量 Returns: 系统提示词和用户提示词的元组 """ # 获取选题生成配置 topic_config = self._ensure_topic_config() system_prompt_path = topic_config.topic_system_prompt user_prompt_path = topic_config.topic_user_prompt if not system_prompt_path or not user_prompt_path: raise ValueError("选题提示词模板路径不完整") template = PromptTemplate(system_prompt_path, user_prompt_path) month = dates or '' if dates and ' to ' in dates: start_date, end_date = dates.split(' to ') month = f"从 {start_date} 到 {end_date}" elif dates and ',' in dates: month = ', '.join(dates.split(',')) # 使用传入的完整对象构建内容，避免重复查询 if style_objects: style_content = '\n'.join([f"{obj['styleName']}: {obj.get('description', '')}" for obj in style_objects]) elif styles: style_content = '\n'.join([f"{style}: {self.prompt_service.get_style_content(style)}" for style in styles]) else: all_styles = self.prompt_service.get_all_styles() style_content = "Style文件列表:\n" + "\n".join([f"- {style['name']}" for style in all_styles]) if audience_objects: demand_content = '\n'.join([f"{obj['audienceName']}: {obj.get('description', '')}" for obj in audience_objects]) elif audiences: demand_content = '\n'.join([f"{audience}: {self.prompt_service.get_audience_content(audience)}" for audience in audiences]) else: all_audiences = self.prompt_service.get_all_audiences() demand_content = "Demand文件列表:\n" + "\n".join([f"- {audience['name']}" for audience in all_audiences]) if scenic_spot_objects: object_content = '\n'.join([f"{obj['name']}: {obj.get('description', '')}" for obj in scenic_spot_objects]) elif scenic_spots: object_content = '\n'.join([f"{spot}: {self.prompt_service.get_scenic_spot_info(spot)}" for spot in scenic_spots]) else: all_spots = self.prompt_service.get_all_scenic_spots() object_content = "Object信息:\n" + "\n".join([f"- {spot['name']}" for spot in all_spots]) if product_objects: product_content = '\n'.join([f"{obj['productName']}: {obj.get('detailedDescription', '')}" for obj in product_objects]) elif products: product_content = '\n'.join([f"{product}: {self.prompt_service.get_product_info(product)}" for product in products]) else: product_content = '' refer_content = self.prompt_service.get_refer_content("topic") system_prompt = template.get_system_prompt() creative_materials = ( f"你拥有的创作资料如下：\n" f"风格信息:\n{style_content}\n\n" f"受众信息:\n{demand_content}\n\n" f"参考内容:\n{refer_content}\n\n" f"景区信息:\n{object_content}\n\n" f"产品信息:\n{product_content}" ) user_prompt = template.build_user_prompt( creative_materials=creative_materials, numTopics=numTopics, month=month ) return system_prompt, user_prompt def build_judge_prompt(self, topic: Dict[str, Any], content: Dict[str, Any]) -> Tuple[str, str]: """ 构建内容审核提示词 (已重构) 此方法现在依赖于一个预先填充好完整信息的topic对象。 """ content_config = self._ensure_content_config() template = PromptTemplate(content_config.judger_system_prompt, content_config.judger_user_prompt) # 从预填充的topic对象中直接获取信息 spot_obj = topic.get('scenic_spot_object', {}) object_content = f"{spot_obj.get('name', '')}\n{spot_obj.get('description', '')}" product_obj = topic.get('product_object', {}) product_content = f"{product_obj.get('productName', '')}\n{product_obj.get('detailedDescription', '')}" refer_content = self.prompt_service.get_refer_content("judge") system_prompt = template.get_system_prompt() import json tweet_content = json.dumps(content, ensure_ascii=False, indent=4) user_prompt = template.build_user_prompt( tweet_content=tweet_content, object_content=object_content, product_content=product_content, refer_content=refer_content ) return system_prompt, user_prompt def build_judge_prompt_with_params(self, topic: Dict[str, Any], content: Dict[str, Any], styles: Optional[List[str]] = None, audiences: Optional[List[str]] = None, scenic_spots: Optional[List[str]] = None, products: Optional[List[str]] = None) -> Tuple[str, str]: """ 使用额外参数构建内容审核提示词 Args: topic: 选题信息 content: 生成的内容 styles: 风格列表 audiences: 受众列表 scenic_spots: 景区列表 products: 产品列表 Returns: 系统提示词和用户提示词的元组 """ # 获取内容生成配置 content_config = self._ensure_content_config() # 从配置中获取审核提示词模板路径 system_prompt_path = content_config.judger_system_prompt user_prompt_path = content_config.judger_user_prompt # 创建提示词模板 template = PromptTemplate(system_prompt_path, user_prompt_path) # 获取景区信息 object_content = '' if scenic_spots: object_content = '\n'.join([f"{spot}: {self.prompt_service.get_scenic_spot_info(spot)}" for spot in scenic_spots]) else: object_name = topic.get("object", "") object_content = f"{object_name}\n{self.prompt_service.get_scenic_spot_info(object_name)}" # 获取产品信息 product_content = '' if products: product_content = '\n'.join([f"{product}: {self.prompt_service.get_product_info(product)}" for product in products]) else: product_name = topic.get("product", "") product_content = f"{product_name}\n{self.prompt_service.get_product_info(product_name)}" # 获取参考内容 refer_content = self.prompt_service.get_refer_content("judge") # 构建系统提示词 system_prompt = template.get_system_prompt() # 格式化内容 import json tweet_content = json.dumps(content, ensure_ascii=False, indent=4) # 构建用户提示词 user_prompt = template.build_user_prompt( tweet_content=tweet_content, object_content=object_content, product_content=product_content, refer_content=refer_content ) return system_prompt, user_prompt def build_judge_prompt_simple(self, topic: Dict[str, Any], content: Dict[str, Any]) -> Tuple[str, str]: """ 构建简化的内容审核提示词（只需要产品信息、景区信息和文章） Args: topic: 选题信息 content: 生成的内容 Returns: 系统提示词和用户提示词的元组 """ # 获取内容生成配置 content_config = self._ensure_content_config() # 从配置中获取审核提示词模板路径 system_prompt_path = content_config.judger_system_prompt user_prompt_path = content_config.judger_user_prompt # 创建提示词模板 template = PromptTemplate(system_prompt_path, user_prompt_path) # 获取景区信息 object_name = topic.get("object", "") object_content = self.prompt_service.get_scenic_spot_info(object_name) # 获取产品信息 product_name = topic.get("product", "") product_content = self.prompt_service.get_product_info(product_name) # 构建系统提示词 system_prompt = template.get_system_prompt() # 格式化内容 import json tweet_content = json.dumps(content, ensure_ascii=False, indent=4) # 构建用户提示词（简化版，不包含参考内容） user_prompt = template.build_user_prompt( tweet_content=tweet_content, object_content=object_content, product_content=product_content, refer_content="" # 简化版不使用参考内容 ) return system_prompt, user_prompt