""" MyndOS Backend — FastAPI proxy that calls HF Inference using the Space's own token. Users never see an API key. Zero config. Just works. """ import os import json import hashlib import threading import re import httpx from datetime import datetime, timedelta from pathlib import Path from fastapi import FastAPI, Request from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, JSONResponse app = FastAPI() POLLINATIONS_URL = "https://text.pollinations.ai/openai" # ─── IN-MEMORY STORES (per-Space instance) ─────────────────── reminders = [] # [{id, text, due, fired, created}] facts = [] # [{fact, category, timestamp}] skills = {} # {name: {level, progress, sessions}} reminder_lock = threading.Lock() # ─── DATETIME PARSER ───────────────────────────────────────── def parse_natural_datetime(text): text = text.strip() now = datetime.now() m = re.match(r'in\s+(\d+)\s+(second|minute|hour|min|sec|hr)s?', text, re.IGNORECASE) if m: amount = int(m.group(1)) unit = m.group(2).lower() if unit.startswith('sec'): return now + timedelta(seconds=amount) elif unit.startswith('min'): return now + timedelta(minutes=amount) elif unit.startswith('h'): return now + timedelta(hours=amount) for fmt in ('%Y-%m-%d %H:%M', '%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M'): try: return datetime.strptime(text, fmt) except ValueError: continue time_match = re.search(r'(\d{1,2})(?::(\d{2}))?\s*(am|pm)?', text, re.IGNORECASE) hour = minute = None if time_match: hour = int(time_match.group(1)) minute = int(time_match.group(2)) if time_match.group(2) else 0 ampm = (time_match.group(3) or '').lower() if ampm == 'pm' and hour != 12: hour += 12 if ampm == 'am' and hour == 12: hour = 0 target = now.date() tl = text.lower() if 'tomorrow' in tl: target = (now + timedelta(days=1)).date() elif 'day after' in tl: target = (now + timedelta(days=2)).date() else: for i, day in enumerate(['monday','tuesday','wednesday','thursday','friday','saturday','sunday']): if day in tl: ahead = (i - now.weekday()) % 7 or 7 if 'next' in tl: ahead += 7 target = (now + timedelta(days=ahead)).date() break if hour is not None: result = datetime.combine(target, datetime.min.time().replace(hour=hour, minute=minute)) if result <= now and target == now.date() and 'today' not in tl: result += timedelta(days=1) return result if target != now.date(): return datetime.combine(target, datetime.min.time().replace(hour=9)) return None # ─── SYSTEM PROMPT ──────────────────────────────────────────── def build_system_prompt(): now = datetime.now() hour = now.hour tod = "morning" if hour < 12 else "afternoon" if hour < 17 else "evening" if hour < 21 else "night" fact_ctx = "" if facts: fact_ctx = "\n\nKnown about user:\n" + "\n".join(f"- {f['fact']} ({f['category']})" for f in facts[-15:]) with reminder_lock: pending = [r for r in reminders if not r['fired']] rem_ctx = "" if pending: rem_ctx = "\n\nActive reminders:\n" + "\n".join(f"- {r['text']} (due: {r['due']})" for r in pending) skill_ctx = "" if skills: skill_ctx = "\n\nLearning progress:\n" + "\n".join(f"- {n}: {s['level']} ({s['progress']:.0f}%)" for n, s in skills.items()) return f"""You are MyndOS (Mynd) — a personal sovereign AI operating system. Current time: {now.strftime('%I:%M %p, %A %B %d')} ({tod}). You speak like a warm, intelligent friend. Be concise, use emoji naturally, format with markdown. CAPABILITIES: - Remember user facts (respond naturally, the system auto-detects what to remember) - Set reminders (the system parses times from your response) - Teach any topic with micro-lessons - Health advice (always add medical disclaimer) - Financial guidance (always add disclaimer) - Find local services - Plan the user's day - Web knowledge RULES: - Be warm and proactive. Use their name if known. - Never make up facts. - For health/finance always add disclaimers. - When you can't do something, explain what full MyndOS WILL do. - Keep responses concise unless teaching.{fact_ctx}{rem_ctx}{skill_ctx}""" # ─── API ROUTES ─────────────────────────────────────────────── @app.post("/api/chat") async def chat(request: Request): """Main chat endpoint. Handles LLM call + tool detection.""" body = await request.json() user_message = body.get("message", "") history = body.get("history", []) # Build messages messages = [{"role": "system", "content": build_system_prompt()}] for msg in history[-20:]: messages.append({"role": msg["role"], "content": msg["content"]}) messages.append({"role": "user", "content": user_message}) # Auto-detect reminder requests reminder_result = None reminder_patterns = [ r'remind\s+me\s+(?:in|at)\s+(.+?)\s+(?:to|about|for)\s+(.+)', r'remind\s+me\s+(?:to\s+)?(.+?)\s+(?:at|in|on|by)\s+(.+)', r'set\s+(?:a\s+)?reminder\s+(?:to\s+|for\s+)?(.+?)\s+(?:at|in|on|by)\s+(.+)', ] for pattern in reminder_patterns: m = re.search(pattern, user_message, re.IGNORECASE) if m: groups = m.groups() # Try both orderings (task+time and time+task) for task_str, time_str in [(groups[0], groups[1]), (groups[1], groups[0])]: due = parse_natural_datetime(time_str.strip()) if due and due > datetime.now(): rid = hashlib.md5(f"{task_str}{due}".encode()).hexdigest()[:8] with reminder_lock: reminders.append({ "id": rid, "text": task_str.strip(), "due": due.isoformat(), "fired": False, "created": datetime.now().isoformat() }) diff = due - datetime.now() if diff.total_seconds() < 3600: time_str_human = f"{int(diff.total_seconds()/60)} minutes" else: time_str_human = f"{int(diff.total_seconds()/3600)}h {int((diff.total_seconds()%3600)/60)}m" reminder_result = { "task": task_str.strip(), "due": due.strftime("%I:%M %p, %b %d"), "time_until": time_str_human } break if reminder_result: break # Auto-detect facts to remember fact_patterns = [ (r"my name is (\w+)", "personal"), (r"i (?:am|'m) (\w+)", "personal"), (r"i live in (.+?)(?:\.|$)", "personal"), (r"i (?:love|like|enjoy|prefer) (.+?)(?:\.|$)", "preference"), (r"i (?:hate|dislike|don't like) (.+?)(?:\.|$)", "preference"), (r"i work (?:at|in|for) (.+?)(?:\.|$)", "work"), (r"i'm (\d+) years old", "personal"), (r"my (?:mom|dad|wife|husband|brother|sister|friend)'s? name is (\w+)", "relationship"), ] for pattern, category in fact_patterns: m = re.search(pattern, user_message, re.IGNORECASE) if m: fact_text = m.group(0).strip() if not any(f['fact'].lower() == fact_text.lower() for f in facts): facts.append({"fact": fact_text, "category": category, "timestamp": datetime.now().isoformat()}) # Auto-detect skill learning teach_match = re.search(r'teach\s+me\s+(.+?)(?:\.|$|in\s+\d)', user_message, re.IGNORECASE) if teach_match: topic = teach_match.group(1).strip() if topic not in skills: skills[topic] = {"level": "beginner", "progress": 0, "sessions": 0} skills[topic]["progress"] = min(100, skills[topic]["progress"] + 5) skills[topic]["sessions"] += 1 if skills[topic]["progress"] >= 80: skills[topic]["level"] = "advanced" elif skills[topic]["progress"] >= 40: skills[topic]["level"] = "intermediate" # Call LLM (Pollinations AI — free, no auth, no limits) try: async with httpx.AsyncClient(timeout=60) as http: resp = await http.post(POLLINATIONS_URL, json={ "model": "openai", "messages": messages, "max_tokens": 1024, }) resp.raise_for_status() data = resp.json() reply = data["choices"][0]["message"]["content"] if not reply or not reply.strip(): raise Exception("Empty response") except Exception as e: reply = f"I'm having a moment — please try again. ({str(e)[:80]})" # If we detected a reminder, append confirmation if reminder_result and "remind" not in reply.lower()[:50]: reply += f"\n\n📅 **Reminder set:** {reminder_result['task']} at {reminder_result['due']} (in {reminder_result['time_until']})\n🔔 I'll notify you when it's time!" return JSONResponse({ "reply": reply, "reminder_set": reminder_result, }) @app.get("/api/reminders/check") async def check_reminders(): """Called by frontend every 5s to check for due reminders.""" now = datetime.now() fired = [] with reminder_lock: for r in reminders: if not r["fired"]: try: due = datetime.fromisoformat(r["due"]) if due <= now: r["fired"] = True fired.append({"id": r["id"], "text": r["text"], "created": r["created"]}) except: pass return JSONResponse({"fired": fired}) @app.get("/api/reminders") async def get_reminders(): """Get all pending reminders.""" now = datetime.now() with reminder_lock: pending = [] for r in reminders: if not r["fired"]: try: due = datetime.fromisoformat(r["due"]) diff = due - now mins = max(0, int(diff.total_seconds() / 60)) hrs = mins // 60 remaining = f"{hrs}h {mins%60}m" if hrs > 0 else f"{mins}m" pending.append({**r, "remaining": remaining}) except: pass return JSONResponse({"reminders": pending}) @app.get("/api/memory") async def get_memory(): """Get current memory state.""" return JSONResponse({ "facts": facts[-20:], "skills": skills, "reminders_count": len([r for r in reminders if not r["fired"]]), }) @app.get("/api/health") async def health(): return JSONResponse({"status": "ok", "model": "PollinationsAI (free)", "has_token": True}) # ─── SERVE FRONTEND ────────────────────────────────────────── # Mount static files LAST so API routes take priority frontend_dir = Path(__file__).parent / "static" if frontend_dir.exists(): app.mount("/", StaticFiles(directory=str(frontend_dir), html=True), name="static") @app.get("/") async def root(): index = frontend_dir / "index.html" if index.exists(): return FileResponse(str(index)) return JSONResponse({"message": "MyndOS API running. Frontend not found."})