# -*- coding: utf-8 -*- """ 🎃 Pumpkin AI Console — Full Version (Text + Image + Hint System) Author: Cherry Leung, Jay Luk Date: 2025-10-13 """ import sys import os import re import base64 import random import unicodedata as _ud # for robust Han-only normalization from pathlib import Path import openai import logging from datetime import datetime # Try to load .env file if python-dotenv is available try: from dotenv import load_dotenv load_dotenv() except ImportError: pass # .env file won't be loaded automatically, use system environment variables # ======================== # LOGGING SETUP # ======================== # Configure logging to show DEBUG level messages with timestamp logging.basicConfig( level=logging.DEBUG, # Changed to DEBUG for detailed logging format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) # ======================== # CONFIGURATION # ======================== POE_API_KEY = os.getenv("POE_API_KEY") if not POE_API_KEY: logger.error("POE_API_KEY environment variable not set!") raise ValueError("POE_API_KEY environment variable is required. Please set it in HuggingFace Space Secrets.") BASE_URL = "https://api.poe.com/v1" TEXT_MODEL = "gpt-4o-mini" IMAGE_MODEL = "Nano-Banana" # Switched to Nano-Banana for better pumpkin integration TIMEOUT = 60 logger.info(f"Initializing OpenAI client with base URL: {BASE_URL}") client = openai.OpenAI(api_key=POE_API_KEY, base_url=BASE_URL) logger.info("OpenAI client initialized successfully") # ======================== # REGEX # ======================== STATUS_NO_RX = re.compile(r"\{status(?:\s*code)?:\s*no\}", re.IGNORECASE) STATUS_YES_RX = re.compile(r"\{status(?:\s*code)?:\s*yes\}", re.IGNORECASE) DATA_URL_RX = re.compile(r"data:image/(png|jpeg|jpg|webp);base64,([A-Za-z0-9+/=]+)", re.IGNORECASE) # ======================== # PROMPTS / HINTS / RESCUE LINES # ======================== def load_text(path: Path) -> str: """Read text file safely (UTF-8-SIG to remove BOM).""" logger.info(f"Loading text file: {path}") return path.read_text(encoding="utf-8-sig", errors="ignore") def load_prompts(prompt_dir: Path): classify = load_text(prompt_dir / "text_or_image_classification.txt") textrep = load_text(prompt_dir / "text_replies.txt") imagerep = load_text(prompt_dir / "image_replies.txt") hintsraw = load_text(prompt_dir / "hints.txt") return classify, textrep, imagerep, hintsraw def load_hints_simple(file_path: str, limit: int = 10) -> list[str]: """ Read hints.txt formatted as one hint per line. Lines starting with # or empty lines are ignored. Returns up to `limit` hints. """ path = Path(file_path) if not path.exists(): logger.warning(f"File not found: {file_path}") print(f"⚠️ File not found: {file_path}") return [] logger.info(f"Loading hints from: {file_path}") text = path.read_text(encoding="utf-8-sig", errors="ignore") lines = [ ln.strip() for ln in text.splitlines() if ln.strip() and not ln.strip().startswith("#") ] logger.info(f"Loaded {len(lines)} hints (returning up to {limit})") return lines[:limit] def load_rescue_lines(file_path: str) -> list[str]: """ Load preset rescue lines from a text file (rescue_line.txt). Ignores empty lines and lines starting with '#'. """ path = Path(file_path) if not path.exists(): logger.warning(f"Rescue lines file not found: {file_path}") print(f"⚠️ File not found: {file_path}") return [] logger.info(f"Loading rescue lines from: {file_path}") lines = [ ln.strip() for ln in path.read_text(encoding="utf-8-sig", errors="ignore").splitlines() if ln.strip() and not ln.strip().startswith("#") ] logger.info(f"Loaded {len(lines)} rescue lines") return lines # ======================== # MODEL CALL HELPERS # ======================== def chat_completion(model: str, system_prompt: str, user_content: str, temperature: float = 0.0) -> str: logger.info(f"Making chat completion request to model: {model} (temp={temperature})") logger.debug(f"User content preview: {user_content[:100]}...") chat = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}, ], temperature=temperature, timeout=TIMEOUT, ) response = chat.choices[0].message.content.strip() logger.info(f"Received response from {model}, length: {len(response)} chars") logger.debug(f"Response preview: {response[:200]}...") return response def classify_input(user_input: str, system_prompt: str) -> str: logger.info("Classifying user input (text vs image request)") out = chat_completion( model=TEXT_MODEL, system_prompt=system_prompt, user_content=user_input, temperature=0.0, ).lower() result = "image" if "image" in out else "text" logger.info(f"Classification result: {result}") return result def generate_text_reply(user_input: str, text_reply_prompt: str, threaded_context: str = None) -> str: """ Generate text reply with optional threaded conversation context. If threaded_context is provided, it will be prepended to the user input. """ logger.info("Generating text reply") # If we have threaded context, include it if threaded_context: logger.info("Using threaded context for reply generation") user_content = threaded_context else: user_content = user_input return chat_completion( model=TEXT_MODEL, system_prompt=text_reply_prompt, user_content=user_content, temperature=0.9, ) def generate_image_data_url(user_input: str, image_reply_prompt: str) -> str: """ Ask the image model to generate image and return as data URL. Handles both data URLs and HTTP URLs from the model. """ logger.info(f"Generating image with model: {IMAGE_MODEL}") prompt = ( image_reply_prompt + "\n\n【使用者提示】" + user_input + "\n\n請只輸出一條 data URL（data:image/png;base64,XXXXX）。不要文字、不要Markdown、不要說明。" ) logger.debug(f"Image generation prompt length: {len(prompt)} chars") msg = client.chat.completions.create( model=IMAGE_MODEL, messages=[ {"role": "system", "content": "你是嚴格的影像生成器。只產出base64圖像的data URL，不要任何額外文字。"}, {"role": "user", "content": prompt}, ], temperature=0.8, timeout=TIMEOUT, ) response = msg.choices[0].message.content.strip() logger.info(f"Image generation response received, length: {len(response)} chars") logger.debug(f"Response preview: {response[:200]}...") # Check if response already has data URL if "data:image/" in response: logger.info("Response contains data URL") return response # Extract HTTP URL from response (markdown or plain) image_url = extract_image_url_from_response(response) if image_url: logger.info(f"Found HTTP URL, attempting to download: {image_url}") data_url = download_image_as_data_url(image_url) if data_url: logger.info("Successfully converted HTTP URL to data URL") return data_url else: logger.error("Failed to download image from URL") return response # Return original response for debugging logger.warning("No image URL or data URL found in response") return response # Return as-is for debugging def extract_image_url_from_response(response: str) -> str: """ Extract image URL from markdown or plain text response. Handles responses like: ![...](https://pfst.cf2.poecdn.net/...) """ logger.info("Attempting to extract image URL from response") logger.debug(f"Full response to parse: {response}") # First, normalize the response by removing newlines to handle broken markdown normalized_response = response.replace('\n', ' ').replace('\r', '') # Try markdown format first: ![text](url) # Use non-greedy match and capture everything until closing paren markdown_match = re.search(r'!\[.*?\]\((https://[^\)]+)\)', normalized_response, re.DOTALL) if markdown_match: url = markdown_match.group(1).strip() logger.info(f"Extracted URL from markdown: {url}") return url # Try plain URL format with query parameters (with or without newlines) # Match the entire URL including query params (?w=..&h=..) url_match = re.search(r'(https://pfst\.cf2\.poecdn\.net/[^\s\)\]]+)', normalized_response) if url_match: url = url_match.group(1).strip() logger.info(f"Extracted plain URL: {url}") return url # Try to find ANY https URL any_url_match = re.search(r'(https://[^\s\)\]<>"\']+)', normalized_response) if any_url_match: url = any_url_match.group(1).strip() # Remove trailing punctuation url = url.rstrip('.,;!?') logger.info(f"Extracted generic URL: {url}") return url logger.warning("No image URL found in response") return None def download_image_as_data_url(url: str) -> str: """ Download image from URL and convert to base64 data URL. Includes proper headers to bypass CDN restrictions. """ logger.info(f"Downloading image from: {url}") try: import urllib.request # Create request with headers to bypass 403 Forbidden req = urllib.request.Request( url, headers={ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'image/avif,image/webp,image/apng,image/*,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9', 'Referer': 'https://poe.com/', 'Origin': 'https://poe.com' } ) response = urllib.request.urlopen(req, timeout=30) image_data = response.read() logger.info(f"Downloaded {len(image_data)} bytes") # Convert to base64 b64 = base64.b64encode(image_data).decode('utf-8') data_url = f"data:image/png;base64,{b64}" logger.info(f"Converted to data URL, length: {len(data_url)}") return data_url except Exception as e: logger.error(f"Failed to download image: {e}") return None def analyze_image_detailed(image_path: str) -> str: """ Analyze an image with comprehensive detail extraction. Returns structured description of all elements in the image. """ logger.info(f"Analyzing image with detailed prompt: {image_path}") # Read image and convert to base64 image_data = Path(image_path).read_bytes() b64_image = base64.b64encode(image_data).decode('utf-8') # Detect image format ext = Path(image_path).suffix.lower() mime_type = "image/jpeg" if ext in [".jpg", ".jpeg"] else f"image/{ext[1:]}" detailed_prompt = """請詳細分析這張圖片，並提供全面的描述。請按以下結構組織你的描述： 【人物分析】 - 人數統計：共有幾位人物 - 對於每位人物，請描述： * 年齡範圍（例如：嬰兒、兒童、青少年、青年、中年、老年） * 性別 * 族裔特徵 * 面部特徵（例如：髮型、髮色、眼睛顏色、表情） * 服裝描述（顏色、風格、配飾） * 姿勢與動作 * 整體氛圍/感覺 * 如果可辨識為知名人物，請提及姓名 【動物分析】（如果有） - 動物種類與品種 - 年齡階段（幼年/成年/老年） - 顏色與花紋 - 姿勢與動作 - 與環境或人物的互動 【環境與場景】 - 地點類型（室內/室外、具體場所） - 時間（白天/夜晚、季節） - 光線與氛圍 - 背景元素描述 【物品與道具】 - 主要物品列表及其特徵 - 物品的顏色、材質、狀態 - 物品在畫面中的位置與作用 【整體構圖】 - 拍攝角度與視角 - 畫面重點與焦點 - 色調與風格 - 情緒與氛圍 請盡可能詳細且有條理地描述所有可見元素。""" try: response = client.chat.completions.create( model=IMAGE_MODEL, messages=[ { "role": "user", "content": [ {"type": "text", "text": detailed_prompt}, { "type": "image_url", "image_url": { "url": f"data:{mime_type};base64,{b64_image}" } } ] } ], temperature=0.3, timeout=TIMEOUT, ) analysis = response.choices[0].message.content.strip() logger.info(f"Image analysis completed, length: {len(analysis)} chars") return analysis except Exception as e: logger.error(f"Failed to analyze image: {e}") return f"⚠️ 圖片分析失敗：{str(e)}" def analyze_image_from_data_url(data_url: str) -> str: """ Analyze image from data URL format. Saves temporarily, analyzes, then cleans up. """ logger.info("Analyzing image from data URL") temp_path = "temp_analysis.png" try: # Save temporarily ok = save_data_url_to_file(data_url, temp_path) if not ok: logger.error("Failed to save data URL to temp file") return "⚠️ 無法處理上傳的圖片" # Analyze analysis = analyze_image_detailed(temp_path) # Cleanup try: Path(temp_path).unlink() logger.info("Cleaned up temp analysis file") except Exception as e: logger.warning(f"Could not delete temp file: {e}") return analysis except Exception as e: logger.error(f"Failed to analyze image from data URL: {e}") return f"⚠️ 圖片分析失敗：{str(e)}" def generate_with_nano_banana(prompt: str) -> str: """ Generate image using Nano-Banana model. Returns data URL or HTTP URL (which will be converted). """ logger.info("Generating image with Nano-Banana model") logger.debug(f"Prompt length: {len(prompt)} chars") try: msg = client.chat.completions.create( model="Nano-Banana", # Use Nano-Banana instead of Gemini messages=[ {"role": "system", "content": "你是圖像生成助手。根據描述生成圖片。"}, {"role": "user", "content": prompt}, ], temperature=0.8, timeout=TIMEOUT, ) response = msg.choices[0].message.content.strip() logger.info(f"Nano-Banana response received, length: {len(response)} chars") # Check if response already has data URL if "data:image/" in response: logger.info("Response contains data URL") return response # Extract and download HTTP URL image_url = extract_image_url_from_response(response) if image_url: logger.info(f"Found HTTP URL, downloading: {image_url}") data_url = download_image_as_data_url(image_url) if data_url: logger.info("Successfully converted HTTP URL to data URL") return data_url logger.warning("No valid image data in Nano-Banana response") return response # Return as-is for debugging except Exception as e: logger.error(f"Nano-Banana generation failed: {e}") return f"⚠️ 圖片生成失敗：{str(e)}" def generate_image_with_analysis(user_input: str, image_reply_prompt: str, uploaded_image_data: str = None, conversation_context: str = None) -> str: """ Generate image from text description with optional image analysis and conversation threading. If uploaded_image_data provided: Analyze the image first, then use that analysis to generate a new image If conversation_context provided: Include previous conversation history to generate contextually relevant images If no uploaded_image_data: Use text description directly """ logger.info("Starting image generation flow") if conversation_context: logger.info("Using conversation threading for image generation") logger.debug(f"Conversation context length: {len(conversation_context)} chars") if uploaded_image_data: # User uploaded image - try to analyze it first logger.info("User uploaded image - attempting to analyze image content...") # Try to analyze the uploaded image analysis_result = analyze_image_from_data_url(uploaded_image_data) logger.info("=" * 80) logger.info("IMAGE ANALYSIS RESULT:") logger.info("=" * 80) logger.info(analysis_result) logger.info("=" * 80) # Check if analysis failed if "無法分析" in analysis_result or "無法識別" in analysis_result or "抱歉" in analysis_result: logger.warning("Image analysis failed - POE vision API not working properly") logger.info("Will generate pumpkin image based on text description only") # Fallback: Use text description only, mention that image was uploaded # Include conversation context if available context_section = f"\n\n{conversation_context}" if conversation_context else "" enhanced_prompt = f"""{image_reply_prompt} 【使用者提示】 {user_input} （用戶上傳了一張圖片，但圖片分析功能暫時不可用。請根據文字描述生成一張充滿南瓜元素的創意圖片。） {context_section} 請生成一張包含南瓜元素的新圖片。 請只輸出一張圖片URL或data URL。""" else: # Analysis succeeded - use it logger.info("Image analysis successful - using results for generation") # Include conversation context if available context_section = f"\n\n{conversation_context}" if conversation_context else "" enhanced_prompt = f"""{image_reply_prompt} 【使用者提示】 {user_input} 【圖片分析結果】 以下是用戶上傳圖片的詳細分析： {analysis_result} {context_section} 請根據以上圖片分析結果，生成一張包含南瓜元素的新圖片。 保留原圖的主要構圖和元素，但加入南瓜相關的創意元素。 {f'請考慮對話歷史中的上下文，讓圖片與之前的對話內容相關聯。' if conversation_context else ''} 請只輸出一張圖片URL或data URL。""" logger.info("=" * 80) logger.info("FULL PROMPT SENT TO NANO-BANANA:") logger.info("=" * 80) logger.info(enhanced_prompt) logger.info("=" * 80) else: # No image uploaded, use text description logger.info("No uploaded image - generating from text description") # Include conversation context if available context_section = f"\n\n{conversation_context}" if conversation_context else "" enhanced_prompt = f"""{image_reply_prompt} 【使用者提示】 {user_input} {context_section} 請生成包含南瓜元素的圖片。 {f'請考慮對話歷史中的上下文，讓圖片與之前的對話內容相關聯。' if conversation_context else ''} 請只輸出一張圖片URL或data URL。""" logger.info("=" * 80) logger.info("PROMPT SENT TO NANO-BANANA:") logger.info("=" * 80) logger.info(enhanced_prompt) logger.info("=" * 80) # Use Nano-Banana model for generation logger.info("Calling Nano-Banana for image generation...") return generate_with_nano_banana(enhanced_prompt) def save_data_url_to_file(data_url: str, output_path: str) -> bool: logger.info(f"Attempting to save data URL to file: {output_path}") m = DATA_URL_RX.search(data_url) if not m: logger.error("Failed to extract base64 data from data URL") return False b64 = m.group(2) Path(output_path).write_bytes(base64.b64decode(b64)) logger.info(f"Successfully saved image to: {output_path}") return True # ======================== # CONVERSATION THREADING # ======================== class ConversationHistory: """ Manages conversation history and threading. Maintains recent messages and determines when to thread related conversations. """ def __init__(self, max_history: int = 5): self.max_history = max_history self.history = [] # List of (user_msg, ai_reply) tuples def add_turn(self, user_msg: str, ai_reply: str): """Add a conversation turn to history.""" self.history.append({"user": user_msg, "ai": ai_reply}) if len(self.history) > self.max_history: self.history.pop(0) logger.debug(f"Added turn to history. Total turns: {len(self.history)}") def get_history(self): """Get all conversation history.""" return self.history def check_threading(self, new_message: str) -> tuple[bool, list]: """ Check if new message relates to recent conversation. Returns (should_thread, related_messages). """ if len(self.history) == 0: logger.info("No conversation history - processing message in isolation") return False, [] logger.info(f"Checking threading relationships for new message (history size: {len(self.history)})") # Build a prompt to check relationships history_text = "" for i, turn in enumerate(self.history[-3:], 1): # Check last 3 messages history_text += f"user_msg{i}: \"{turn['user']}\"\n" history_text += f"reply_by_llm{i}: \"{turn['ai']}\"\n\n" relationship_prompt = f"""分析以下對話歷史和新訊息之間的關係。 對話歷史： {history_text} 新訊息："{new_message}" 請判斷新訊息是否與對話歷史中的任何訊息有關聯。關聯包括： - 繼續討論相同話題 - 追問或補充說明 - 引用或回應之前的內容 - 相關的上下文 如果有關聯，回覆 "RELATED" 並說明與哪些訊息相關。 如果無關聯，回覆 "ISOLATED"。 格式：[RELATED/ISOLATED]: 簡短說明""" try: response = chat_completion( model=TEXT_MODEL, system_prompt="你是對話關係分析專家。分析訊息之間的關聯性。", user_content=relationship_prompt, temperature=0.3, ) logger.info(f"Threading analysis result: {response[:100]}") if "RELATED" in response.upper(): # Get related turns (last 3 for context) related_turns = self.history[-3:] logger.info(f"Threading detected - including {len(related_turns)} previous turns") return True, related_turns else: logger.info("No threading relationship detected") return False, [] except Exception as e: logger.error(f"Threading check failed: {e}") return False, [] def format_threaded_context(self, related_turns: list, new_message: str) -> str: """ Format conversation history for threaded context. """ context = "【對話歷史】\n" for i, turn in enumerate(related_turns, 1): context += f"user_msg{i}: \"{turn['user']}\"\n" context += f"reply_by_llm{i}: \"{turn['ai']}\"\n\n" context += f"【最新訊息】\nnew_msg: \"{new_message}\"\n\n" context += "請主要回應最新訊息，同時考慮對話歷史提供的上下文。" return context # ======================== # GAME LOGIC # ======================== # keep only CJK Han (Chinese characters); drop punctuation/emoji/spaces/latin/etc. def _normalize_han_only(s: str) -> str: return "".join(ch for ch in s if "CJK UNIFIED IDEOGRAPH" in _ud.name(ch, "")) def is_target_lyric(text: str) -> bool: """ True if, after stripping all non-Chinese chars, the sequence matches target: 做過幾分鐘公主搭著南瓜車亦有過愛人來接浪漫度午夜 (Punctuation/whitespace/emoji inside the user's input are ignored.) """ target = "做過幾分鐘公主搭著南瓜車亦有過愛人來接浪漫度午夜" normalized_input = _normalize_han_only(text) normalized_target = _normalize_han_only(target) is_match = normalized_input == normalized_target logger.info(f"Checking if input is target lyric: {is_match}") if is_match: logger.info("🎉 TARGET LYRIC DETECTED! User wins!") return is_match def add_rescue_and_hint(base_message: str, hints_pool: list[str], rescue_lines: list[str]) -> str: """ Shared function to append rescue line + hint to any message. Used for both text replies (when status: no) and image generation replies. """ rescue_line = random.choice(rescue_lines) if rescue_lines else "快啲救我啦～我要變南瓜湯喇！" hint = random.choice(hints_pool) if hints_pool else "（未載入提示）" separator = "\n===============================================\n" logger.debug(f"Selected rescue line: {rescue_line}") logger.debug(f"Selected hint: {hint}") return f"{base_message}{separator}{rescue_line} 咒語提示：{hint}" def handle_not_guessing(reply: str, hints_pool: list[str], rescue_lines: list[str]) -> str: """ When {status: no}: - strip the tag - append one random rescue line + one random hint (same line) """ logger.info("User is not guessing - adding rescue line and hint") clean_reply = STATUS_NO_RX.sub("", reply).strip() return add_rescue_and_hint(clean_reply, hints_pool, rescue_lines) def handle_text_turn(user_input: str, text_reply_prompt: str, hints_pool: list[str], rescue_lines: list[str], threaded_context: str = None) -> str: logger.info("Processing text reply") reply = generate_text_reply(user_input, text_reply_prompt, threaded_context) if STATUS_NO_RX.search(reply): logger.info("Detected {status: no} in reply") return handle_not_guessing(reply, hints_pool, rescue_lines) if STATUS_YES_RX.search(reply): logger.info("Detected {status: yes} in reply - user is guessing!") reply = STATUS_YES_RX.sub("", reply).strip() return reply def handle_image_turn(user_input: str, image_reply_prompt: str, output_path: str, hints_pool: list[str], rescue_lines: list[str], uploaded_image_data: str = None, conversation_history: 'ConversationHistory' = None) -> tuple[str, str]: """ Handle image generation with optional image analysis and conversation threading. If uploaded_image_data is provided, analyze it first before generating. If conversation_history is provided, check for threading relationships. Returns tuple of (text_response, data_url_or_none) """ logger.info("Processing image generation request") # Check for conversation threading conversation_context = None if conversation_history: should_thread, related_turns = conversation_history.check_threading(user_input) if should_thread: conversation_context = conversation_history.format_threaded_context(related_turns, user_input) logger.info(f"[IMAGE THREADING] Using threaded context with {len(related_turns)} previous turns") else: logger.info("[IMAGE THREADING] No threading relationship detected") # Use new analysis-aware generation with optional conversation context raw = generate_image_with_analysis(user_input, image_reply_prompt, uploaded_image_data, conversation_context) # Try to save to file for backward compatibility (console mode) ok = save_data_url_to_file(raw, output_path) logger.info(f"Image save result: ok={ok}") # Extract data URL from raw response for frontend data_url_match = DATA_URL_RX.search(raw) data_url = data_url_match.group(0) if data_url_match else None if data_url: logger.info(f"Found data URL in response (length: {len(data_url)} chars)") else: logger.warning("No data URL found in response") # ALWAYS add rescue line + hint, regardless of success/failure if ok or data_url: base_message = f"🎃 圖像已生成" logger.info(f"Image successfully generated") else: base_message = f"⚠️ 圖像生成遇到問題，但我會繼續嘗試幫你！" logger.warning(f"Image generation failed") logger.info("Adding rescue line and hint to image response") # ALWAYS use shared function to add rescue line + hint final_response = add_rescue_and_hint(base_message, hints_pool, rescue_lines) # If failed, append debug info if not (ok or data_url): final_response += f"\n\n[Debug] Raw response: {raw[:300]}..." logger.info(f"Final response length: {len(final_response)} chars") logger.debug(f"Final response preview: {final_response[:200]}...") return final_response, data_url # ======================== # UI TEXT # ======================== def greeting() -> str: return """🎃 你好，我係一個患咗南瓜強迫症嘅AI。你可以： - 當我正常AI咁叫我做嘢 (但我可能會病發亂答一通，唔建議) - 當我正常AI咁整圖 (但我會忍唔住用南瓜污染幅圖，唔建議) - 畀幅圖我，我會自動用南瓜污染你幅圖 (唔建議，無謂啦) - 嘗試用一句24個中文字組成嘅「驅瓜咒語」醫好我 (強烈建議) ────────────────────────────────────────────── """ def winning_text() -> str: return """公主！你終於出現喇！ 我有南瓜強迫症，係因為畀一隻南瓜精靈上咗身！ 南瓜精靈終日幻想自己係灰姑娘故事入面嘅一架南瓜車， 只有等到變成公主後嘅灰姑娘出現，佢先肯離開！ 你唱出咗南瓜界經典金曲《南瓜車》嘅歌詞，超渡咗南瓜精靈，解救咗我！ 而家就去(url)接收我畀你嘅小禮物啦！Happy Halloween！ """ # ======================== # CONSOLE LOOP # ======================== def process_user_input( user_input: str, classify_prompt: str, text_reply_prompt: str, image_reply_prompt: str, hints_pool: list[str], rescue_lines: list[str], conversation_history: ConversationHistory, image_output_path: str = "pumpkin_output.png", ) -> tuple[str, str]: """ Process user input with conversation threading support. Returns tuple of (response, raw_ai_reply_for_history) """ logger.info(f"Processing user input: {user_input[:50]}..." if len(user_input) > 50 else f"Processing user input: {user_input}") if user_input.lower() in ["exit", "quit"]: logger.info("User requested exit") return "EXIT", "" if is_target_lyric(user_input): return "WIN", "" # Check for conversation threading should_thread, related_turns = conversation_history.check_threading(user_input) threaded_context = None if should_thread: threaded_context = conversation_history.format_threaded_context(related_turns, user_input) logger.info("Using threaded conversation context") result = classify_input(user_input, classify_prompt) if result == "image": body = handle_image_turn(user_input, image_reply_prompt, image_output_path, hints_pool, rescue_lines) # For image responses, store simplified reply in history return body, "🎃 [生成了南瓜主題圖片]" body = handle_text_turn(user_input, text_reply_prompt, hints_pool, rescue_lines, threaded_context) full_response = f"🎃 南瓜AI：{body}" return full_response, body def run_console(): logger.info("=" * 60) logger.info("🎃 Pumpkin AI Console Starting...") logger.info("=" * 60) PROMPT_DIR = Path(r"C:\Users\cherry.leung\OneDrive - CMRS Digital Solutions Limited\Pumpkin\New\Prompts") logger.info(f"Loading prompts from: {PROMPT_DIR}") classify_prompt, text_reply_prompt, image_reply_prompt, hints_raw = load_prompts(PROMPT_DIR) # file paths hints_path = PROMPT_DIR / "hints.txt" rescue_path = PROMPT_DIR / "rescue_line.txt" # << your file name # load pools hints_pool = load_hints_simple(hints_path, limit=10) rescue_lines = load_rescue_lines(rescue_path) # Initialize conversation history manager conversation_history = ConversationHistory(max_history=5) logger.info("Conversation history manager initialized") logger.info(f"Initialization complete: {len(hints_pool)} hints, {len(rescue_lines)} rescue lines") logger.info("=" * 60) print(f"(Loaded {len(hints_pool)} hints, {len(rescue_lines)} rescue lines)\n") print(greeting()) while True: user_input = input("👤 你: ").strip() if not user_input: logger.debug("Empty input received, skipping") continue logger.info(f"New user input received (length: {len(user_input)} chars)") outcome, ai_reply = process_user_input( user_input=user_input, classify_prompt=classify_prompt, text_reply_prompt=text_reply_prompt, image_reply_prompt=image_reply_prompt, hints_pool=hints_pool, rescue_lines=rescue_lines, conversation_history=conversation_history, image_output_path="pumpkin_output.png", ) if outcome == "EXIT": logger.info("Session ended by user") print("🎃 南瓜AI：再見呀公主，下次見～") sys.exit(0) if outcome == "WIN": logger.info("🎉 USER WON THE GAME!") print(winning_text()) sys.exit(0) print(outcome + "\n") logger.info("Response delivered to user") # Add this turn to conversation history if ai_reply: conversation_history.add_turn(user_input, ai_reply) logger.debug(f"Turn added to conversation history") # ======================== # ENTRYPOINT # ======================== if __name__ == "__main__": run_console()