import os import gspread from pathlib import Path from pydantic import BaseModel from google.oauth2.service_account import Credentials from config import prompt_settings from app.utils.logger import get_logger from app.utils.prompts.schemas import * from functools import lru_cache logger = get_logger("prompt") _SCOPES = [ "https://www.googleapis.com/auth/spreadsheets.readonly" ] _sheet_cache: dict[str, tuple[str, str]] = {} def _read_sheet_data(sheet_name: str) -> tuple[str, str]: if sheet_name not in _sheet_cache: creds = Credentials.from_service_account_file( prompt_settings.GOOGLE_SERVICE_ACCOUNT_JSON, scopes=_SCOPES ) gc = gspread.authorize(creds) ws = gc.open_by_key(prompt_settings.PROMPT_SPREADSHEET).worksheet(sheet_name) _sheet_cache[sheet_name] = (ws.cell(3, 2).value, ws.cell(2, 2).value) return _sheet_cache[sheet_name] class Prompt(): sheet_name: str prompt_template: str prompt_model: str prompt_input_class = BaseModel prompt_output_class = BaseModel def __init__(self, sheet_name, prompt_input_class, prompt_output_class, template_file: str = None, prompt_model: str = None): self.sheet_name = sheet_name self.prompt_input_class = prompt_input_class self.prompt_output_class = prompt_output_class if template_file: self.prompt_template = Path(template_file).read_text(encoding="utf-8") self.prompt_model = prompt_model self._template_file = template_file else: self.prompt_template, self.prompt_model = _read_sheet_data(sheet_name) self._template_file = None def _reload_prompt(self): if self._template_file: self.prompt_template = Path(self._template_file).read_text(encoding="utf-8") else: _sheet_cache.pop(self.sheet_name, None) self.prompt_template, self.prompt_model = _read_sheet_data(self.sheet_name) def build_prompt(self, input_data:dict, silent:bool = False) -> str: verified_input = self.prompt_input_class(**input_data) build_template = self.prompt_template build_template = build_template.format(**verified_input.model_dump()) if not silent: logger.debug(f"build_template: {build_template}") logger.debug(f"input_data: {input_data}") return build_template marketing_prompt = Prompt( sheet_name="marketing", prompt_input_class=MarketingPromptInput, prompt_output_class=MarketingPromptOutput, ) lyric_prompt = Prompt( sheet_name="lyric", prompt_input_class=LyricPromptInput, prompt_output_class=LyricPromptOutput, ) yt_upload_prompt = Prompt( sheet_name="yt_upload", prompt_input_class=YTUploadPromptInput, prompt_output_class=YTUploadPromptOutput, ) image_autotag_prompt = Prompt( sheet_name="image_tag", prompt_input_class=ImageTagPromptInput, prompt_output_class=ImageTagPromptOutput, ) _SUBTITLE_TEMPLATE_PATH = str(Path(__file__).parent / "templates" / "subtitle_prompt.txt") @lru_cache() def create_dynamic_subtitle_prompt(length: int) -> Prompt: return Prompt( sheet_name="subtitle", prompt_input_class=SubtitlePromptInput, prompt_output_class=SubtitlePromptOutput[length], template_file=_SUBTITLE_TEMPLATE_PATH, prompt_model=os.getenv("SUBTITLE_PROMPT_MODEL", "gpt-4o-mini"), ) def reload_all_prompt(): marketing_prompt._reload_prompt() lyric_prompt._reload_prompt() yt_upload_prompt._reload_prompt() image_autotag_prompt._reload_prompt()