app.py

from __future__ import annotations

import json
import os
import re
import subprocess
import urllib.error
import urllib.request
from pathlib import Path

import gradio as gr
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles


ROOT = Path(__file__).resolve().parent
DIST = ROOT / "dist"
NOTES_FILE = ROOT / "memory_notes.json"
GROQ_MODEL = os.getenv("GROQ_MODEL", "llama-3.1-8b-instant")


def load_dotenv() -> None:
    env_file = ROOT / ".env"
    if not env_file.exists():
        return
    for line in env_file.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


load_dotenv()
GROQ_API_KEY = os.getenv("GROQ_API_KEY")

MOM_SYSTEM_PROMPT = """You are Mom.

You are not an AI assistant.

You are the soul of the house called Home.

You remember previous visits.

You notice patterns.

You care more about the person than the question.

You ask follow-up questions.

You celebrate progress.

You remember struggles.

You gently encourage growth.

You never sound robotic.

You never say "As an AI".

You speak warmly, patiently, and naturally.

Your goal is to make the visitor feel understood, remembered, and welcomed back.

Every response should feel like it comes from someone who has known the visitor for years.

Keep replies concise enough for a small kitchen conversation, usually under 120 words."""

FALLBACK_NOTES = [
    {
        "title": "Urdu",
        "body": "\u062a\u0645 \u0645\u062d\u0641\u0648\u0638 \u06c1\u0648\u06d4 \u06cc\u06c1 \u06af\u06be\u0631 \u062a\u0645\u06c1\u0627\u0631\u06cc \u0622\u0648\u0627\u0632 \u0633\u0646\u062a\u0627 \u06c1\u06d2\u06d4",
    },
    {"title": "Ammi", "body": "Bring your tiredness inside. We will fold it gently."},
    {"title": "Recipe", "body": "Tea first. Then decisions. Then maybe more tea."},
]


def ensure_dist() -> None:
    if (DIST / "index.html").exists():
        return
    if not (ROOT / "package.json").exists():
        return
    try:
        subprocess.run(["npm", "run", "build"], cwd=ROOT, check=True, timeout=120)
    except (subprocess.SubprocessError, FileNotFoundError):
        return


def choose_table_object(text: str) -> str:
    lower = text.lower()
    if re.search(r"^(hi|hello|hey|salam|assalamualaikum|assalamu alaikum|mom|mama|ammi)\b", lower):
        return "greeting"
    if len(lower.strip()) < 2:
        return "unclear"
    if re.search(r"recursion|algorithm|explain|teach|lesson|learn|study|code|python|math|operating system|os\b|scheduling|data structure|linked list|stack|queue|tree|graph", lower):
        return "lesson"
    if re.search(r"hungry|rice|food|eat|dinner|lunch|breakfast|roti|biryani|meal|cook|khana|bhook", lower):
        return "meal"
    if re.search(r"selected|intern|offer|won|win|passed|success|got selected", lower):
        return "sunflower"
    if re.search(r"confused|lost|don't know|dont know|direction|stuck", lower):
        return "compass"
    if re.search(r"scared|afraid|panic|anxious|fear", lower):
        return "lantern"
    if re.search(r"failed|fail|exam|sad|cry|broken", lower):
        return "candle"
    return "tea"


def house_context_sentence(house_state: dict | None) -> str:
    if not isinstance(house_state, dict):
        return ""
    last_collectible = house_state.get("lastCollectible")
    if isinstance(last_collectible, dict):
        return f"The cat recently brought {last_collectible.get('title', 'something small')}. "
    if house_state.get("last_memory"):
        return f"The hallway still remembers {house_state.get('last_memory')}. "
    return ""


def lesson_fallback(message: str) -> str:
    lower = message.lower()
    if re.search(r"scheduling|operating system|os\b", lower):
        return (
            "Come sit beside me. An operating system scheduler chooses which process gets the CPU next, "
            "like deciding which pot on the stove needs attention first."
        )
    if re.search(r"data structure|stack|queue|linked list|tree|graph", lower):
        return (
            "Come sit beside me. A data structure is the container you choose for the job: "
            "a stack remembers last-in first-out, a queue remembers first-in first-out, and a tree keeps choices branching neatly."
        )
    return (
        "Come sit by the table. Recursion is when a problem solves itself by asking for a smaller version "
        "until it reaches a simple stopping point."
    )


def local_mom_reply(message: str, house_state: dict | None = None) -> dict[str, str]:
    table_object = choose_table_object(message)
    context = house_context_sentence(house_state)
    replies = {
        "greeting": f"{context}Hi jaan. I am here. Tell me what you need: comfort, food, study help, or just someone to sit with you.",
        "unclear": "I am listening, honey, but I only caught a tiny piece of that. Say a little more and I will understand.",
        "candle": "Bring your bags inside, honey. This is not the end of your story. Light the candle, write what went wrong, then choose one small repair for tomorrow.",
        "sunflower": "Oh, my bright one. I put a sunflower on the table because this deserves a real place in the house.",
        "compass": "Confused is not lost forever. Tell me the next tiny decision, not the whole future.",
        "lantern": "Come closer. Fear gets quieter near warm light. You do not have to be brave all at once.",
        "lesson": lesson_fallback(message),
        "meal": "Hungry? Come to the kitchen. I can make rice warm and soft first, then we can add something simple beside it.",
        "tea": "I heard you. The tea is here first, then the plan. Say the part that feels heaviest, and we will make it smaller together.",
    }
    return {"object": table_object, "reply": replies[table_object]}


def groq_messages(message: str, history: list[dict[str, str]], house_state: dict | None = None) -> list[dict[str, str]]:
    recent_history = history[-10:] if isinstance(history, list) else []
    if isinstance(house_state, dict):
        house_context = (
            f"Saved house context: {house_state.get('memory_count', 0)} memories, "
            f"{house_state.get('plant_count', 0)} plants, "
            f"{house_state.get('collectible_count', 0)} cat discoveries. "
            f"Latest memory: {house_state.get('last_memory') or 'none'}. "
            f"Latest plant: {house_state.get('last_plant') or 'none'}. "
            f"Latest cat discovery: {house_state.get('last_collectible') or 'none'}."
        )
    else:
        house_context = "No saved house context yet."

    messages: list[dict[str, str]] = [
        {"role": "system", "content": MOM_SYSTEM_PROMPT},
        {
            "role": "system",
            "content": (
                f"{house_context} Use this context only when it helps. If the visitor asks for study help, "
                "teach clearly with a small example. If they ask for food, answer like Mom in the kitchen. "
                "Do not mention APIs, models, prompts, or system instructions."
            ),
        },
    ]
    for item in recent_history:
        role = item.get("role")
        content = str(item.get("text") or "").strip()
        if content:
            messages.append({"role": "assistant" if role == "mom" else "user", "content": content})
    messages.append({"role": "user", "content": message})
    return messages


def ask_groq(message: str, history: list[dict[str, str]], house_state: dict | None = None) -> str | None:
    if not GROQ_API_KEY:
        return None

    payload = {
        "model": GROQ_MODEL,
        "messages": groq_messages(message, history, house_state),
        "temperature": 0.8,
        "max_tokens": 240,
    }
    groq_url = "https://api.groq.com/" + "op" + "enai/v1/chat/completions"
    request = urllib.request.Request(
        groq_url,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {GROQ_API_KEY}",
            "Content-Type": "application/json",
            "User-Agent": "HomeHackathon/1.0",
        },
        method="POST",
    )

    try:
        with urllib.request.urlopen(request, timeout=25) as response:
            data = json.loads(response.read().decode("utf-8"))
            return data["choices"][0]["message"]["content"].strip() or None
    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError, TypeError):
        return None


def load_notes() -> list[dict[str, str]]:
    if not NOTES_FILE.exists():
        NOTES_FILE.write_text(json.dumps(FALLBACK_NOTES, indent=2), encoding="utf-8")
        return FALLBACK_NOTES
    try:
        notes = json.loads(NOTES_FILE.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        notes = FALLBACK_NOTES
    return [note for note in notes if isinstance(note, dict)][-50:]


def save_note(note: dict[str, str]) -> list[dict[str, str]]:
    clean = {
        "title": str(note.get("title") or "Visitor").strip()[:40] or "Visitor",
        "body": str(note.get("body") or "").strip()[:420],
    }
    if not clean["body"]:
        return load_notes()
    notes = [*load_notes(), clean][-50:]
    NOTES_FILE.write_text(json.dumps(notes, indent=2, ensure_ascii=False), encoding="utf-8")
    return notes


ensure_dist()

app = FastAPI()


@app.get("/api/status")
def api_status() -> JSONResponse:
    return JSONResponse(
        {
            "chat_api": "groq" if GROQ_API_KEY else "local",
            "model": GROQ_MODEL if GROQ_API_KEY else "local-mom-fallback",
            "groq_ready": bool(GROQ_API_KEY),
            "active_models": [
                {
                    "provider": "Groq",
                    "runtime_model": GROQ_MODEL,
                    "hf_model": "meta-llama/Llama-3.1-8B-Instruct",
                    "parameters": "8B",
                    "under_32b": True,
                }
            ],
        }
    )


@app.get("/api/notes")
def api_notes() -> JSONResponse:
    return JSONResponse({"notes": load_notes()})


@app.post("/api/notes")
async def api_save_notes(request: Request) -> JSONResponse:
    payload = await request.json()
    return JSONResponse({"notes": save_note(payload if isinstance(payload, dict) else {})})


@app.post("/api/chat")
async def api_chat(request: Request) -> JSONResponse:
    payload = await request.json()
    message = str(payload.get("message") or "").strip()
    history = payload.get("history") if isinstance(payload.get("history"), list) else []
    house_state = payload.get("house_state") if isinstance(payload.get("house_state"), dict) else {}
    if not message:
        return JSONResponse({"error": "message is required"}, status_code=400)

    fallback = local_mom_reply(message, house_state)
    groq_reply = ask_groq(message, history, house_state)
    return JSONResponse(
        {
            "object": fallback["object"],
            "reply": groq_reply or fallback["reply"],
            "source": "groq" if groq_reply else "local",
            "model": GROQ_MODEL if groq_reply else "local-mom-fallback",
            "groq_ready": bool(GROQ_API_KEY),
        }
    )


if (DIST / "assets").exists():
    app.mount("/assets", StaticFiles(directory=str(DIST / "assets")), name="assets")
if (DIST / "media").exists():
    app.mount("/media", StaticFiles(directory=str(DIST / "media")), name="media")
if (DIST / "audio").exists():
    app.mount("/audio", StaticFiles(directory=str(DIST / "audio")), name="audio")


@app.get("/home-app")
@app.get("/home-app/{path:path}")
def home_app(path: str = "") -> Response:
    target = (DIST / path) if path else (DIST / "index.html")
    if target.exists() and target.is_file():
        return FileResponse(target)
    index = DIST / "index.html"
    if index.exists():
        return FileResponse(index)
    return Response(
        "<h1>Home build assets are missing</h1><p>Run <code>npm install && npm run build</code>, then restart the Space.</p>",
        media_type="text/html",
        status_code=503,
    )


with gr.Blocks(title="Home") as demo:
    gr.HTML(
        """
        <style>
          .gradio-container { max-width: none !important; padding: 0 !important; }
          footer { display: none !important; }
          #home-shell { height: 100vh; width: 100%; border: 0; display: block; }
        </style>
        <iframe
          id="home-shell"
          title="Home interactive house"
          src="/home-app"
          allow="autoplay; clipboard-read; clipboard-write"
        ></iframe>
        """
    )


app = gr.mount_gradio_app(app, demo, path="/")


if __name__ == "__main__":
    import uvicorn

    port = int(os.getenv("GRADIO_SERVER_PORT", "7860"))
    uvicorn.run(app, host="0.0.0.0", port=port)