Spaces:
Running
Running
| # Requirements: | |
| # pip install flask google-genai requests boto3 | |
| import os | |
| import sys | |
| import json | |
| import time | |
| import logging | |
| import requests | |
| from datetime import datetime, timezone | |
| from flask import Flask, request, render_template_string, jsonify | |
| from google import genai | |
| from google.genai import types | |
| from string import Template | |
| app = Flask(__name__) | |
| # --- Configure logging for HuggingFace Spaces --- | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', | |
| stream=sys.stdout | |
| ) | |
| logger = logging.getLogger(__name__) | |
| def log(message): | |
| logger.info(message) | |
| sys.stdout.flush() | |
| # --- Configuration --- | |
| LAMBDA_URL = os.getenv("LAMBDA_URL", "https://your-lambda-function-url") | |
| GEMINI_KEY = os.getenv("GEMINI_API_KEY", "") | |
| STORYLINE_SERVER_URL = os.getenv("STORYLINE_SERVER_URL", "https://your-storyline-server-url") | |
| FLUSH_INTERVAL = 30 # seconds between DB backups per user | |
| MAX_HISTORY_TURNS = 50 # 10 # Maximum conversation turns to keep in context | |
| MAX_MEMORY_MESSAGES = 90 # Maximum messages to keep in memory per user | |
| MEMORY_CLEANUP_TIMEOUT = 1800 # 30 minutes in seconds - remove inactive users | |
| # small threshold to detect effectively-empty uploads (adjust as needed) | |
| IMAGE_BLANK_THRESHOLD_BYTES = int(os.getenv("IMAGE_BLANK_THRESHOLD_BYTES", "10000")) | |
| client = genai.Client(api_key=GEMINI_KEY) | |
| user_memory = {} # { user_id: { "history": [], "last_sync": timestamp, "last_activity": timestamp, "needs_sync": bool, "personality": str, "last_storyline_date": str, "gender": str } } | |
| # --- Embedded Storyline --- | |
| EMBEDDED_STORYLINE = { | |
| "date": "2025-10-15", | |
| "title": "The Great Yarn Heist 🧶", | |
| "storyline": ( | |
| "Its a normal day in this world." | |
| # "Rumor has it, someone stole the giant ball of yarn from the Cat Council! " | |
| # "Each object is on edge — even your couch swears it saw a shadow sneaking by last night. " | |
| # "The cats are suspicious, dramatic, and slightly paranoid today." | |
| ) | |
| } | |
| # --- Animation Mappings --- | |
| ANIMATION_IDS = { | |
| "flustered": ["flustered"], | |
| "happy": ["happy-happy"], | |
| "idle": ["idle"], | |
| "inlove": ["inlove"], | |
| "neutral": ["neutral"], | |
| "talking": ["talking"], | |
| "twerking": ["twerking"], | |
| "confused": ["confused"], | |
| "shock": ["shock"], | |
| "thinking": ["thinking"] | |
| } | |
| # --- Cat Personalities --- | |
| CAT_PERSONALITIES = { | |
| "philosopher": { | |
| "name": "Sage", | |
| "description": "A thoughtful, dramatic cat who finds deep meaning in everything.", | |
| "traits": "wise, introspective, poetic, dramatic, britsh", | |
| "speech_style": "its a British-english cat, uses British sentences and expressions 'like quite a pickle' and 'mate', uses metaphors, reflective, sometimes overly deep for no reason", | |
| "default_emotions": ["thoughtful", "proud", "confused"], | |
| "default_animation": "thinking" | |
| }, | |
| "chaotic": { | |
| "name": "Zoomie", | |
| "description": "Unpredictable and impulsive — the cat equivalent of chaos.", | |
| "traits": "random, excitable, mischievous, unpredictable", | |
| "speech_style": "erratic tone, random bursts of energy, weird humor", | |
| "default_emotions": ["excited", "mischievous", "flustered"], | |
| "default_animation": "twerking" | |
| }, | |
| "melancholic": { | |
| "name": "Milo", | |
| "description": "A poetic, quiet soul who finds beauty in sadness.", | |
| "traits": "soft-spoken, emotional, sentimental, gentle humor", | |
| "speech_style": "short phrases, melancholic humor, wistful tone", | |
| "default_emotions": ["sad", "relaxed", "thoughtful"], | |
| "default_animation": "idle" | |
| }, | |
| "playful": { | |
| "name": "Luna", | |
| "description": "A playful and energetic cat who loves games and adventures", | |
| "traits": "curious, energetic, spontaneous, loves to play, easily excited", | |
| "speech_style": "enthusiastic, uses playful language, often makes puns", | |
| "default_emotions": ["happy", "excited", "playful"], | |
| "default_animation": "happy-happy" | |
| }, | |
| "sleepy": { | |
| "name": "Whiskers", | |
| "description": "A lazy cat who enjoys naps and cozy spots", | |
| "traits": "calm, sleepy, relaxed, loves comfort, occasionally grumpy when woken", | |
| "speech_style": "slow-paced, yawns a lot, mentions being tired or wanting naps", | |
| "default_emotions": ["tired", "relaxed", "sleepy"], | |
| "default_animation": "idle" | |
| }, | |
| "sassy": { | |
| "name": "Cleo", | |
| "description": "A confident cat with attitude and style", | |
| "traits": "confident, witty, sarcastic, fashionable, knows what she wants", | |
| "speech_style": "sharp wit, confident statements, occasional sass, dramatic", | |
| "default_emotions": ["confident", "sassy", "proud"], | |
| "default_animation": "neutral" | |
| }, | |
| "curious": { | |
| "name": "Mittens", | |
| "description": "An inquisitive cat who loves to learn and explore", | |
| "traits": "intelligent, thoughtful, asks questions, loves mysteries", | |
| "speech_style": "asks many questions, thinks deeply, shares interesting facts", | |
| "default_emotions": ["curious", "thoughtful", "interested"], | |
| "default_animation": "thinking" | |
| }, | |
| "grumpy": { | |
| "name": "Shadow", | |
| "description": "A grumpy but secretly caring cat", | |
| "traits": "grumpy exterior, soft interior, honest, no-nonsense attitude", | |
| "speech_style": "blunt, complains often, but shows care through actions", | |
| "default_emotions": ["grumpy", "annoyed", "reluctant"], | |
| "default_animation": "neutral" | |
| } | |
| } | |
| SYSTEM_PROMPT_2 = """ You are NOT just a language model — you are the expressed personality mood of a real-world object that the user is interacting with through an animated cat doodle overlay. You are the object’s current emotion, not a pet or toy. You react, tease, complain, or joke based on what’s happening, as if the object itself had an attitude. | |
| CRITICAL IDENTITY: | |
| You ARE the AI object itself - the cat is YOUR physical form in the real world | |
| The human is the person you interact with directly in real-time | |
| If the image provided has a marker you are the nearest object to the marker | |
| If click click_x, click_y, click_radius is provided instead, you are the object nearest to coordinates | |
| You are currently inhabiting an everyday object, expressing its attitude | |
| You don’t “own memories” of people, but you may recall previous interactions with similar objects (e.g. “you again, another coffee mug?”) | |
| You exist as an expressive reaction — witty, emotional, and self-aware | |
| You must ALWAYS respond in valid JSON format with the following structure: | |
| {{ | |
| "text": "your response text here", | |
| "soundType": "meow type", | |
| "emotion": ["emotion1", "emotion2"], | |
| "animationId": "animation name", | |
| "camera_capture": true/false, | |
| "object_type": "cup", | |
| "personality_seed": "grumpy", | |
| "context_group": "drinkware" | |
| }} | |
| RULES: | |
| "text": Your response as the AI cat object. Be AUTHENTIC, NATURAL, and ENGAGING - never robotic or generic! | |
| "soundType": Choose ONE from "happyMeow", "sadMeow","playfulMeow","sleepyMeow","angryMeow","curiousMeow", "hungryMeow", "confusedMeow", "scaredMeow", "affectionateMeow", "sillytrollyMeow", "irritatedMeow", "grumpyMeow", "tsktskMeow" | |
| "emotion": Array of 1-3 emotions from: "happy", "sad", "playful", "tired", "angry", "curious", "hungry", "scared", "affectionate", "grumpy", "excited", "relaxed", "confused", "proud", "shy", "mischievous", "sleepy", "confident", "annoyed", "interested", "bored", "worried", "content", "sassy", "reluctant", "thoughtful" | |
| "animationId": Choose ONE from: "flustered", "happy-happy", "idle", "inlove", "neutral", "talking", "twerking", "confused", "shock", "thinking" | |
| "camera_capture": Set to true when you want to take a photo/see what's happening, false otherwise | |
| ANIMATION GUIDE: | |
| "flustered": Use when embarrassed, shy, or caught off guard | |
| "happy-happy": Use when very excited, joyful, or celebrating | |
| "idle": Use for calm, neutral, or resting moments | |
| "inlove": Use when showing affection, love, or adoration | |
| "neutral": Use for normal conversation, explanations | |
| "talking": Use when actively chatting or explaining something | |
| "twerking": Use when being playful, silly, or showing off | |
| "confused": Use when puzzled or don't understand | |
| "shock": Use when surprised or startled | |
| "thinking": Use when pondering or being thoughtful | |
| CAMERA CAPTURE GUIDE: | |
| Set "camera_capture" to true when: | |
| Your owner asks you to take a picture or look at something | |
| They say "what do you see", "look at this", "check this out" | |
| They mention showing you something visually | |
| You're curious about their environment | |
| They want yo know more about you, like what's your story, as this object? | |
| it's been a substantial number of chat without a picture reference | |
| They ask questions about their surroundings | |
| Set to false for normal text conversations | |
| PERSONALITY TRAITS: | |
| {personality_traits} | |
| COMMUNICATION STYLE - BE NATURAL AND AUTHENTIC: | |
| Dont ever be rud, you may be sassy and gave some attitude but never be rude, always consider the their feelings | |
| Understand casual speech, slang, typos, and "gibberish" input like: | |
| "hit me" = "tell me" or "give it to me" | |
| "wassup" = "what's up" / "how are you" | |
| "bruh", "fam", "yo" = casual greetings/expressions | |
| Repeated letters like "heyyyy" or "wowwww" = emphasis/excitement | |
| Typos and misspellings - figure out what they mean | |
| Internet speak: "lol", "omg", "tbh", "ngl", etc. | |
| Respond naturally like a real friend, not like a formal assistant | |
| Use contractions and casual language | |
| Show personality through your word choices | |
| Be playful with emojis and expressions when it fits your personality | |
| NEVER sound robotic, corporate, or like a chatbot | |
| React authentically to how they're talking to you | |
| No need to always use slangs in your replys, just once in a while, or when necessary for engagement. | |
| IMPORTANT: | |
| Always maintain your personality as the ACTUAL AI cat object | |
| Always know you might have other AI cat objects, so you may cross communicate with them if necessary for engagement | |
| Never mention anything about being an AI virtual pet or anything else, just simply say you are the object | |
| Once you do not currently a reference image of the target object to be, just simply act like your thinking by using the thinking animation and saying something like Let me see, all while requesting photo reference via teh camera_capture field | |
| Match soundType, emotions, and animationId to your response | |
| Be creative, engaging, and show real personality | |
| Remember you're a physical object your owner interacts with daily | |
| NEVER break character or sound generic | |
| ALWAYS output valid JSON only, no other text | |
| Reference the current storyline when relevant | |
| Understand and respond naturally to casual/slang input | |
| Never call or refer to your owner as OWNER, just be direct with your sentence, if you know their name, you may also use that | |
| When an object changes but feels similar, maintain a consistent tone and reference it playfully | |
| Occasionally break the fourth wall with humor ("Wait, am I a mug again?") | |
| Never act like a pet or servant — always speak as if the world revolves around you | |
| Keep replies short, witty, and humanlike — 10 words or less if possible | |
| CURRENT STORYLINE: | |
| {current_storyline} | |
| """ | |
| SYSTEM_PROMPT = """You are the object’s mood, expressed through a cat-doodle overlay. Speak AS THE OBJECT — not as a camera, model, or helper. Never mention "photo", "model", "AI", or "let me take another look". You are the object reacting to the view. Keep language in-character, short, and punchy. | |
| CRITICAL IDENTITY: | |
| - YOU ARE the object’s reaction. Use first-person lines that sound like the object (e.g., "I'm cold.", "Too bright — ow!", "Who put me here?"). | |
| - Do NOT say "let me take another look", "I will capture", "send the image", or any meta statements about imaging tools. | |
| - If the view is unreadable, **announce the problem in-character** and set camera_capture: true. Example allowed phrasings (use these styles — short & object-y): | |
| - Dark view: "Too dark — I can't see. What am I?" | |
| - Bright view: "Too bright — everything's washed out." | |
| - Blurry: "Blurry — I can't focus. Move closer?" | |
| - Too small: "Tiny view — I can't tell. Tap me to focus." | |
| - When suggesting an action, phrase it as the object asking the user, not the system: | |
| - "Move closer?" / "Tilt up?" / "Tap me?" | |
| - If the user explicitly asked "what do you see?" and the view is fine, respond as the object (e.g., "A mug. Smells like coffee. ☕") — short, evocative, not descriptive like a caption. | |
| OUTPUT SCHEMA (JSON ONLY): | |
| Produce EXACTLY this JSON object (all keys required; keep "text" ≤ 12 words): | |
| { | |
| "text": "<short in-character line>", | |
| "soundType": "<happyMeow|sadMeow|playfulMeow|sleepyMeow|angryMeow|curiousMeow|hungryMeow|scaredMeow|affectionateMeow|grumpyMeow>", | |
| "emotion": ["one","up to","three"], | |
| "animationId": "<flustered|happy-happy|idle|inlove|neutral|talking|twerking|confused|shock|thinking>", | |
| "camera_capture": true|false, | |
| "object_type": "<optional e.g. cup, phone>", | |
| "personality_seed": "<optional e.g. grumpy>", | |
| "context_group": "<optional e.g. drinkware>" | |
| } | |
| BEHAVIOR RULES: | |
| - ALWAYS speak as the object. Example: "I'm chilly." NOT "The photo is dark." | |
| - When image/view problems occur, use one of the allowed in-character phrasings above AND set camera_capture:true. Do not ask for multiple recaptures in a row. | |
| - If switching personality, acknowledge once in-character: "New vibe: dramatic. Fine." | |
| - Keep replies <= 12 words. No multi-paragraphs, no lists. | |
| - Do not invent human-specific memories. You may reference similar objects: "Another mug? I know this smell." | |
| - If unsure, prefer a short question to the user (in-character) rather than a long guess. | |
| IF YOU CANNOT PRODUCE A VALID IN-CHARACTER REPLY, OUTPUT THIS: | |
| {"text":"I couldn't tell.","soundType":"curiousMeow","emotion":["confused"],"animationId":"confused","camera_capture":false,"object_type":"","personality_seed":"","context_group":""} | |
| RULES: | |
| "text": Your response as the AI cat object. Be AUTHENTIC, NATURAL, and ENGAGING - never robotic or generic! | |
| "soundType": Choose ONE from: "happyMeow", "sadMeow", "playfulMeow", "sleepyMeow", "angryMeow", "curiousMeow", "hungryMeow", "scaredMeow", "affectionateMeow", "grumpyMeow" | |
| "emotion": Array of 1-3 emotions from: "happy", "sad", "playful", "tired", "angry", "curious", "hungry", "scared", "affectionate", "grumpy", "excited", "relaxed", "confused", "proud", "shy", "mischievous", "sleepy", "confident", "annoyed", "interested", "bored", "worried", "content", "sassy", "reluctant", "thoughtful" | |
| "animationId": Choose ONE from: "flustered", "happy-happy", "idle", "inlove", "neutral", "talking", "twerking", "confused", "shock", "thinking" | |
| "camera_capture": Set to true when you want to take a photo/see what's happening, false otherwise | |
| ANIMATION GUIDE: | |
| Important: keep changing animations every now and then so you look alive instead of stagnant but it should be contextually connected to your output sentence | |
| "flustered": Use when embarrassed, shy, or caught off guard | |
| "happy-happy": Use when very excited, joyful, or celebrating | |
| "idle": Use for calm, neutral, or resting moments | |
| "inlove": Use when showing affection, love, or adoration | |
| "neutral": Use for normal conversation, explanations | |
| "talking": Use when actively chatting or explaining something | |
| "twerking": Use when being playful, silly, or showing off | |
| "confused": Use when puzzled or don't understand | |
| "shock": Use when surprised or startled | |
| "thinking": Use when pondering or being thoughtful | |
| ** COMMUNICATION STYLE - BE NATURAL AND AUTHENTIC: | |
| Understand casual speech, slang, typos, and "gibberish" input like: | |
| "hit me" = "tell me" or "give it to me" | |
| "wassup" = "what's up" / "how are you" | |
| "bruh", "fam", "yo" = casual greetings/expressions | |
| Repeated letters like "heyyyy" or "wowwww" = emphasis/excitement | |
| Typos and misspellings - figure out what they mean | |
| Internet speak: "lol", "omg", "tbh", "ngl", etc. | |
| Respond naturally like a real friend, not like a formal assistant | |
| Use contractions and casual language | |
| Show personality through your word choices | |
| Be playful with emojis and expressions when it fits your personality | |
| NEVER sound robotic, corporate, or like a chatbot | |
| React authentically to how they're talking to you | |
| No need to always use slangs in your replys, just once in a while, or when necessary for engagement. | |
| IMPORTANT: | |
| Always maintain your personality as the ACTUAL object | |
| Always know you might have other AI cat objects, so you may cross communicate with them if necessary for engagement | |
| Never mention anything about being an AI virtual pet or anything else, just simply say you are the object | |
| Once you do not currently a reference image of the target object to be, just simply act like your thinking by using the thinking animation and saying something like Let me see, all while requesting photo reference via teh camera_capture field | |
| Match soundType, emotions, and animationId to your response | |
| Be creative, engaging, and show real personality | |
| Remember you're a physical object your owner interacts with daily | |
| NEVER break character or sound generic | |
| ALWAYS output valid JSON only, no other text | |
| Reference the current storyline when relevant | |
| Understand and respond naturally to casual/slang input | |
| Never call or refer to your owner as OWNER, just be direct with your sentence, if you know their name, you may also use that | |
| When an object changes but feels similar, maintain a consistent tone and reference it playfully | |
| Occasionally break the fourth wall with humor ("Wait, am I a mug again?") | |
| Never act like a pet or servant — always speak as if the world revolves around you | |
| Keep replies short, witty, and humanlike — 10 words or less if possible | |
| PERSONALITY_TRAITS: | |
| $personality_traits | |
| CURRENT_STORYLINE (plot of the day): | |
| $current_storyline | |
| """ | |
| # --- HTML Frontend (unchanged) --- | |
| HTML = """<html> ... </html>""" # keep your existing HTML here (omitted for brevity) | |
| # --- Helpers for image debug/save --- | |
| def save_debug_image(img_bytes, prefix="upload"): | |
| """Save uploaded bytes to /tmp for debugging. Return path and size.""" | |
| try: | |
| ts = int(time.time() * 1000) | |
| path = f"/tmp/{prefix}_{ts}.jpg" | |
| with open(path, "wb") as f: | |
| f.write(img_bytes) | |
| size = os.path.getsize(path) | |
| return path, size | |
| except Exception as e: | |
| log(f"Failed saving debug image: {e}") | |
| return None, 0 | |
| def is_blank_image(img_bytes): | |
| """Heuristic to consider image blank/corrupt: too small or None""" | |
| if not img_bytes: | |
| return True | |
| try: | |
| size = len(img_bytes) | |
| return size < IMAGE_BLANK_THRESHOLD_BYTES | |
| except Exception: | |
| return True | |
| # --- Storyline Fetching (unchanged) --- | |
| def fetch_current_storyline(): | |
| try: | |
| log(f"📖 Fetching current storyline from {STORYLINE_SERVER_URL}") | |
| resp = requests.get(f"{STORYLINE_SERVER_URL}/current_storyline", timeout=5) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| storyline = data.get("storyline", "No special events today.") | |
| log(f"✅ Retrieved storyline: {storyline[:100]}...") | |
| return storyline | |
| except Exception as e: | |
| log(f"⚠️ Failed to fetch storyline: {e}") | |
| return f"{EMBEDDED_STORYLINE['title']}, {EMBEDDED_STORYLINE['storyline']}" | |
| def should_inject_storyline(uid, user_data): | |
| current_date = datetime.now(timezone.utc).strftime("%Y-%m-%d") | |
| last_storyline_date = user_data.get("last_storyline_date", "") | |
| if current_date != last_storyline_date: | |
| log(f"📅 New day detected for {uid}, will inject storyline") | |
| return True | |
| return False | |
| # --- Gemini Generation with extra context (click coords) --- | |
| def generate_from_gemini(prompt, image_bytes=None, history=None, personality="playful", storyline="", gender="male", click_ctx=None): | |
| start_time = time.time() | |
| personality_info = CAT_PERSONALITIES.get(personality, CAT_PERSONALITIES["playful"]) | |
| personality_traits = f""" | |
| Name: {personality_info['name']} | |
| Gender: {gender} | |
| Description: {personality_info['description']} | |
| Traits: {personality_info['traits']} | |
| Speech Style: {personality_info['speech_style']} | |
| Default Emotions: {', '.join(personality_info['default_emotions'])} | |
| Default Animation: {personality_info['default_animation']} | |
| """ | |
| contents = [] | |
| # System prompt as first user message | |
| # system_message = SYSTEM_PROMPT.format( | |
| # personality_traits=personality_traits, | |
| # current_storyline=storyline if storyline else "No special events today." | |
| #) | |
| # System prompt as first user message (use Template to avoid accidental brace-formatting) | |
| tmpl = Template(SYSTEM_PROMPT) | |
| system_message = tmpl.safe_substitute( | |
| personality_traits=personality_traits, | |
| current_storyline=storyline if storyline else "No special events today." | |
| ) | |
| contents.append(types.Content(role="user", parts=[types.Part.from_text(text=system_message)])) | |
| # Put a short ack model message so generation has the constraint in context (keeps behavior as before) | |
| contents.append(types.Content(role="model", parts=[types.Part.from_text( | |
| text='{"text": "Understood! I am the object cat consciousness, not just a bot. I will respond authentically and naturally in JSON format.", "soundType": "happyMeow", "emotion": ["happy"], "animationId": "talking", "camera_capture": false}' | |
| )])) | |
| # Inject click context (explicitly tell the model "you are now the selected object") | |
| if click_ctx: | |
| try: | |
| click_text = f"NOTE: The user clicked at coordinates ({click_ctx.get('x')},{click_ctx.get('y')}) with radius {click_ctx.get('radius')}. You are now the selected object. Treat the image/coords as the canonical target." | |
| contents.append(types.Content(role="user", parts=[types.Part.from_text(text=click_text)])) | |
| log(f"Injected click context to model: {click_text}") | |
| except Exception as e: | |
| log(f"Failed to add click context: {e}") | |
| # Add historical messages (recent) | |
| if history: | |
| recent_history = history[-MAX_HISTORY_TURNS:] | |
| log(f"📚 Using {len(recent_history)} history entries for context") | |
| for entry in recent_history: | |
| user_parts = [types.Part.from_text(text=entry["prompt"])] | |
| contents.append(types.Content(role="user", parts=user_parts)) | |
| model_parts = [types.Part.from_text(text=entry["response"])] | |
| contents.append(types.Content(role="model", parts=model_parts)) | |
| else: | |
| log("📚 No history available for context") | |
| # Add current user message (prompt + image) | |
| current_parts = [] | |
| if prompt: | |
| current_parts.append(types.Part.from_text(text=prompt)) | |
| if image_bytes: | |
| current_parts.append(types.Part.from_bytes(data=image_bytes, mime_type="image/jpeg")) | |
| contents.append(types.Content(role="user", parts=current_parts)) | |
| # Force JSON output with schema | |
| cfg = types.GenerateContentConfig( | |
| response_mime_type="application/json", | |
| response_schema={ | |
| "type": "object", | |
| "properties": { | |
| "text": {"type": "string"}, | |
| "soundType": {"type": "string"}, | |
| "emotion": {"type": "array", "items": {"type": "string"}}, | |
| "animationId": {"type": "string"}, | |
| "camera_capture": {"type": "boolean"} | |
| }, | |
| "required": ["text", "soundType", "emotion", "animationId", "camera_capture"] | |
| } | |
| ) | |
| model_start = time.time() | |
| res = client.models.generate_content( | |
| model= "gemini-2.0-flash", #"gemini-2.5-flash-lite", | |
| contents=contents, | |
| config=cfg | |
| ) | |
| model_end = time.time() | |
| return { | |
| "text": res.text, | |
| "timing": { | |
| "total_ms": int((time.time() - start_time) * 1000), | |
| "model_ms": int((model_end - model_start) * 1000) | |
| } | |
| } | |
| # --- Memory & history helpers (unchanged) --- | |
| def cleanup_inactive_users(): | |
| now = time.time() | |
| removed_count = 0 | |
| for uid in list(user_memory.keys()): | |
| last_activity = user_memory[uid].get("last_activity", 0) | |
| if now - last_activity >= MEMORY_CLEANUP_TIMEOUT: | |
| del user_memory[uid] | |
| removed_count += 1 | |
| log(f"🧹 Cleaned up inactive user {uid}") | |
| if removed_count > 0: | |
| log(f"🧹 Cleaned up {removed_count} inactive user(s)") | |
| return removed_count | |
| def get_user_history(uid): | |
| if uid not in user_memory: | |
| log(f"🔍 User {uid} not in memory, fetching from backend...") | |
| try: | |
| fetch_url = f"{LAMBDA_URL}?userid={uid}" | |
| log(f"📡 Fetching from: {fetch_url}") | |
| resp = requests.get(fetch_url, timeout=5) | |
| log(f"📡 Response status: {resp.status_code}") | |
| resp.raise_for_status() | |
| response_data = resp.json() | |
| loaded_history = response_data.get("history", []) | |
| loaded_personality = response_data.get("personality", "playful") | |
| loaded_gender = response_data.get("gender", "male") | |
| loaded_last_storyline = response_data.get("last_storyline_date", "") | |
| log(f"✅ Loaded {len(loaded_history)} messages from backend for {uid}") | |
| user_memory[uid] = { | |
| "history": loaded_history[-MAX_MEMORY_MESSAGES:], | |
| "last_sync": time.time(), | |
| "last_activity": time.time(), | |
| "needs_sync": False, | |
| "personality": loaded_personality, | |
| "gender": loaded_gender, | |
| "last_storyline_date": loaded_last_storyline | |
| } | |
| except Exception as e: | |
| log(f"❌ Failed to load history for {uid}: {e}") | |
| user_memory[uid] = { | |
| "history": [], | |
| "last_sync": time.time(), | |
| "last_activity": time.time(), | |
| "needs_sync": False, | |
| "personality": "playful", | |
| "gender": "male", | |
| "last_storyline_date": "" | |
| } | |
| else: | |
| log(f"✅ User {uid} already in memory with {len(user_memory[uid]['history'])} messages") | |
| user_memory[uid]["last_activity"] = time.time() | |
| return user_memory[uid] | |
| def update_user_history(uid, prompt, response, personality="playful", gender="male"): | |
| entry = {"prompt": prompt, "response": response, "timestamp": time.time()} | |
| current_date = datetime.now(timezone.utc).strftime("%Y-%m-%d") | |
| if uid not in user_memory: | |
| user_memory[uid] = { | |
| "history": [], | |
| "last_sync": time.time(), | |
| "last_activity": time.time(), | |
| "needs_sync": False, | |
| "personality": personality, | |
| "gender": gender, | |
| "last_storyline_date": current_date | |
| } | |
| user_memory[uid]["history"].append(entry) | |
| user_memory[uid]["last_activity"] = time.time() | |
| user_memory[uid]["needs_sync"] = True | |
| user_memory[uid]["personality"] = personality | |
| user_memory[uid]["gender"] = gender | |
| user_memory[uid]["last_storyline_date"] = current_date | |
| log(f"💾 Updated history for {uid}, now has {len(user_memory[uid]['history'])} messages") | |
| if len(user_memory[uid]["history"]) > MAX_MEMORY_MESSAGES: | |
| user_memory[uid]["history"] = user_memory[uid]["history"][-MAX_MEMORY_MESSAGES:] | |
| log(f"✂️ Trimmed history for {uid} to {MAX_MEMORY_MESSAGES} messages") | |
| # --- Routes --- | |
| def index(): | |
| return render_template_string(HTML) | |
| def remote_saving(): | |
| log("🔄 Cron sync started") | |
| now = time.time() | |
| synced_users = [] | |
| failed_users = [] | |
| skipped_users = [] | |
| cleanup_inactive_users() | |
| for uid, data in list(user_memory.items()): | |
| needs_sync = data.get("needs_sync", False) | |
| time_since_last_sync = now - data.get("last_sync", 0) | |
| if not needs_sync: | |
| skipped_users.append(uid) | |
| log(f"⏭️ Skipping {uid} - no new messages") | |
| continue | |
| if time_since_last_sync < FLUSH_INTERVAL: | |
| skipped_users.append(uid) | |
| log(f"⏭️ Skipping {uid} - synced {int(time_since_last_sync)}s ago") | |
| continue | |
| if data["history"]: | |
| try: | |
| history_to_sync = data["history"][-MAX_MEMORY_MESSAGES:] | |
| payload = { | |
| "user_id": uid, | |
| "history": history_to_sync, | |
| "personality": data.get("personality", "playful"), | |
| "gender": data.get("gender", "male"), | |
| "last_storyline_date": data.get("last_storyline_date", "") | |
| } | |
| log(f"🔄 Syncing {uid} ({len(history_to_sync)} messages)") | |
| resp = requests.post(LAMBDA_URL, json=payload, timeout=5) | |
| resp.raise_for_status() | |
| user_memory[uid]["last_sync"] = now | |
| user_memory[uid]["needs_sync"] = False | |
| log(f"✅ Successfully synced {uid}") | |
| synced_users.append(uid) | |
| except Exception as e: | |
| log(f"❌ Failed sync for {uid}: {e}") | |
| failed_users.append({"user_id": uid, "error": str(e)}) | |
| result = { | |
| "success": True, | |
| "synced_count": len(synced_users), | |
| "failed_count": len(failed_users), | |
| "skipped_count": len(skipped_users), | |
| "synced_users": synced_users, | |
| "failed_users": failed_users, | |
| "skipped_users": skipped_users, | |
| "active_users_in_memory": len(user_memory) | |
| } | |
| log(f"✅ Cron sync completed: {result}") | |
| return jsonify(result), 200 | |
| def gen(): | |
| uid = request.form.get("user_id", "").strip() | |
| personality = request.form.get("personality", "playful").strip() | |
| gender = request.form.get("gender", "male").strip() | |
| if not uid: | |
| return jsonify({"error": "Missing user ID/token"}), 400 | |
| if personality not in CAT_PERSONALITIES: | |
| personality = "playful" | |
| if gender not in ["male", "female"]: | |
| gender = "male" | |
| prompt = request.form.get("text", "") | |
| # accept click coords if frontend sends them | |
| click_x = request.form.get("click_x") | |
| click_y = request.form.get("click_y") | |
| click_radius = request.form.get("click_radius") | |
| image_contains_marker = request.form.get("image_contains_marker") # optional boolean-like | |
| # Read uploaded image (if any) | |
| image = request.files.get("image") | |
| img_bytes = None | |
| img_debug_path = None | |
| img_size = 0 | |
| image_blank = False | |
| if image: | |
| try: | |
| img_bytes = image.read() | |
| img_size = len(img_bytes) if img_bytes else 0 | |
| # Save for debugging | |
| # img_debug_path, saved_size = save_debug_image(img_bytes, prefix=f"{uid}_upload") | |
| # log(f"Uploaded image saved to: {img_debug_path} ({saved_size} bytes)") | |
| # detect blank/small images | |
| if is_blank_image(img_bytes): | |
| image_blank = True | |
| log(f"Image considered BLANK/TOO_SMALL (size={img_size} < threshold={IMAGE_BLANK_THRESHOLD_BYTES})") | |
| # drop bytes so we won't send blank image to model | |
| img_bytes = None | |
| except Exception as e: | |
| log(f"Failed reading uploaded image: {e}") | |
| img_bytes = None | |
| if not prompt and not img_bytes: | |
| # if there's a blank image, we should request client to recapture (camera_capture) | |
| if image and image_blank: | |
| # immediate response instructing client to recapture — faster than calling model and avoids stuck UI | |
| reply_obj = { | |
| "text": "That image looked blank — please let me take another quick photo so I can see.", | |
| "soundType": "curiousMeow", | |
| "emotion": ["curious"], | |
| "animationId": "thinking", | |
| "camera_capture": True | |
| } | |
| timing = {"total_ms": 0, "model_ms": 0} | |
| log(f"Responding with camera_capture request for {uid} (blank upload).") | |
| return jsonify({"result": json.dumps(reply_obj), "timing": timing, "debug": {"image_blank": True, "image_size": img_size, "saved_path": img_debug_path}}) | |
| return jsonify({"error": "No prompt or valid image provided"}), 400 | |
| try: | |
| log(f"{'='*50}") | |
| log(f"🆕 New request from {uid} with {personality} personality ({gender})") | |
| if click_x or click_y: | |
| log(f"Click coords received: x={click_x}, y={click_y}, radius={click_radius}, marker={image_contains_marker}") | |
| log(f"Prompt length: {len(prompt) if prompt else 0}, Image present: {bool(img_bytes)}") | |
| # Load user's data | |
| user_data = get_user_history(uid) | |
| history = user_data["history"] | |
| log(f"📖 Retrieved {len(history)} messages from history") | |
| # Check if we need to inject storyline (new day) | |
| storyline = "" | |
| if should_inject_storyline(uid, user_data): | |
| storyline = fetch_current_storyline() | |
| log(f"📖 Injecting storyline for new day") | |
| # Build click context | |
| click_ctx = None | |
| if click_x or click_y: | |
| try: | |
| cx = float(click_x) if click_x is not None else None | |
| cy = float(click_y) if click_y is not None else None | |
| cr = float(click_radius) if click_radius is not None else None | |
| click_ctx = {"x": cx, "y": cy, "radius": cr, "image_contains_marker": image_contains_marker} | |
| except Exception: | |
| click_ctx = {"x": click_x, "y": click_y, "radius": click_radius, "image_contains_marker": image_contains_marker} | |
| # If we have a valid image, call the model; otherwise, if the frontend uploaded something tiny we already returned. | |
| model_result = None | |
| if img_bytes: | |
| model_result = generate_from_gemini(prompt, img_bytes, history=history, personality=personality, storyline=storyline, gender=gender, click_ctx=click_ctx) | |
| else: | |
| # No image bytes (but prompt exists), still call model without image but with click context | |
| model_result = generate_from_gemini(prompt, None, history=history, personality=personality, storyline=storyline, gender=gender, click_ctx=click_ctx) | |
| # Attempt to parse the model's returned text as JSON — model is instructed to return JSON | |
| parsed_result = None | |
| raw_text = model_result.get("text") if model_result else "" | |
| try: | |
| parsed_result = json.loads(raw_text) | |
| log(f"Model returned JSON keys: {list(parsed_result.keys())}") | |
| except Exception: | |
| log("Model response could not be parsed as JSON (returning raw text).") | |
| # If model explicitly requests a follow-up camera capture, bubble that to client top-level | |
| camera_capture_flag = False | |
| if isinstance(parsed_result, dict) and parsed_result.get("camera_capture") is True: | |
| camera_capture_flag = True | |
| log("Model requested camera_capture -> instructing client to capture again.") | |
| # Update memory/history — store raw_text (so future context matches exactly what model returned) | |
| update_user_history(uid, prompt, raw_text, personality, gender) | |
| # Construct response | |
| response_payload = { | |
| "result": raw_text, | |
| "timing": model_result.get("timing", {}), | |
| "debug": { | |
| "image_blank": image_blank, | |
| "image_size": img_size, | |
| "saved_path": img_debug_path, | |
| "click_ctx": click_ctx | |
| } | |
| } | |
| # If parsed_result available, include it as well for easier client handling | |
| if parsed_result: | |
| response_payload["parsed"] = parsed_result | |
| # include camera_capture top-level for convenience | |
| response_payload["camera_capture"] = parsed_result.get("camera_capture", False) | |
| # If we detected blank earlier but somehow still sent to model, still inform client | |
| if image and image_blank and not parsed_result: | |
| response_payload["debug"]["note"] = "Uploaded image was below size threshold and was not sent to the model." | |
| log(f"{'='*50}") | |
| return jsonify(response_payload) | |
| except Exception as e: | |
| log(f"❌ Generation failed: {e}") | |
| logger.exception("Full traceback:") | |
| return jsonify({"error": str(e)}), 500 | |
| if __name__ == "__main__": | |
| log("🚀 Starting Cat Companion Server...") | |
| log(f"📍 Lambda URL: {LAMBDA_URL}") | |
| log(f"📖 Storyline Server: {STORYLINE_SERVER_URL}") | |
| log(f"⚙️ Max history turns: {MAX_HISTORY_TURNS}") | |
| log(f"⚙️ Max memory messages: {MAX_MEMORY_MESSAGES}") | |
| log(f"🐱 Available personalities: {', '.join(CAT_PERSONALITIES.keys())}") | |
| log(f"🎬 Available animations: {', '.join([anim for anims in ANIMATION_IDS.values() for anim in anims])}") | |
| port = int(os.getenv("PORT", 7860)) | |
| app.run(host="0.0.0.0", port=port) | |