Spaces:
Running
Running
File size: 3,324 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | # backend/easter_eggs/engine.py
# PURPOSE: Match user input against the Easter Egg registry and fire the handler.
# This runs BEFORE the normal JARVIS/FRIDAY ReAct pipeline.
# If a match is found, the Easter Egg response is returned directly.
# If no match, fall through to the normal ReAct agent.
from .registry import EASTER_EGG_REGISTRY
import sys as _sys
if getattr(_sys, 'frozen', False):
__import__('easter_eggs.iron_man_eggs') # frozen exe: no 'backend.' prefix
else:
__import__('backend.easter_eggs.iron_man_eggs') # normal Python: full package path
del _sys # clean up — don't pollute module namespace
from difflib import SequenceMatcher
from typing import Optional
import logging
logger = logging.getLogger("easter_eggs")
def fuzzy_match(input_text: str, trigger: str, threshold: float = 0.85) -> bool:
"""Returns True if input matches trigger with >= threshold similarity."""
input_clean = input_text.lower().strip()
trigger_clean = trigger.lower().strip()
if trigger_clean in input_clean:
return True
ratio = SequenceMatcher(None, input_clean, trigger_clean).ratio()
return ratio >= threshold
async def check_easter_eggs(
user_input: str,
ui_trigger: Optional[str],
context: dict
) -> Optional[dict]:
"""
Check user input and UI trigger against all registered Easter Eggs.
Returns the Easter Egg response dict if matched, None otherwise.
"""
for egg in EASTER_EGG_REGISTRY:
# Check phrase triggers
for phrase in egg.trigger_phrases:
if fuzzy_match(user_input, phrase):
logger.info(f"Easter Egg triggered: {egg.id} (phrase: {phrase})")
result = await egg.handler(context)
if result:
result["easter_egg_id"] = egg.id
if egg.discoverable:
await log_egg_triggered(egg.id, context)
return result
# Check UI trigger
if ui_trigger and egg.ui_trigger and ui_trigger == egg.ui_trigger:
logger.info(f"Easter Egg triggered: {egg.id} (ui: {ui_trigger})")
result = await egg.handler(context)
if result:
result["easter_egg_id"] = egg.id
if egg.discoverable:
await log_egg_triggered(egg.id, context)
return result
return None # no match — proceed to normal agent
async def log_egg_triggered(egg_id: str, context: dict):
"""Log triggered Easter Eggs to SQLite so Secret Lab can display them."""
db_path = context.get("db_path")
if not db_path:
return
import sqlite3, time
conn = sqlite3.connect(db_path)
conn.execute('PRAGMA journal_mode=WAL')
conn.execute(
"CREATE TABLE IF NOT EXISTS easter_eggs_triggered (egg_id TEXT PRIMARY KEY, first_triggered_at INTEGER, trigger_count INTEGER DEFAULT 0)"
)
conn.execute(
"INSERT OR IGNORE INTO easter_eggs_triggered (egg_id, first_triggered_at) VALUES (?, ?)",
(egg_id, int(time.time()))
)
conn.execute(
"UPDATE easter_eggs_triggered SET trigger_count = trigger_count + 1 WHERE egg_id = ?",
(egg_id,)
)
conn.commit()
conn.close()
|