memory.py

"""Short-term structured memory for LifeOS.

One shared JSON store at data/memory.json:
  meals:    [{date, dish, ingredients[]}]
  workouts: [{date, type, duration_min}]
  finances: {income_monthly, subscriptions: [{name, cost, last_used}]}
  user_profile: {name, dietary_prefs, fitness_goal, budget_weekly}

A real install starts blank (empty_data). Set LIFEOS_DEMO=1 to instead seed a
sample persona whose dates are generated relative to "today" so the demo never
goes stale. memory.json is gitignored and created on first import / app start.
"""

import json
import os
import threading
import uuid
from datetime import date, timedelta

import config

DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
MEMORY_PATH = os.path.join(DATA_DIR, "memory.json")

_lock = threading.Lock()


def _d(days_ago: int) -> str:
    return (date.today() - timedelta(days=days_ago)).isoformat()


def _new_id() -> str:
    return uuid.uuid4().hex[:8]


def _week_day(weekday: int) -> str:
    """Date of the given weekday (0=Mon) in the current week."""
    today = date.today()
    return (today + timedelta(days=weekday - today.weekday())).isoformat()


def _months_out(months: int) -> str:
    """YYYY-MM roughly `months` months from today."""
    y, m = date.today().year, date.today().month + months
    y += (m - 1) // 12
    m = (m - 1) % 12 + 1
    return f"{y:04d}-{m:02d}"


def empty_data() -> dict:
    """A blank store for a brand-new user. Same schema as seed_data() but with
    no personal history — the user fills it in via the Profile/Memory UI."""
    return {
        "meals": [],
        "workouts": [],
        "calendar": [],
        "workout_schedule": {"days": [], "time": "07:00"},
        "goals": [],
        "finances": {
            "income_monthly": 0,
            "monthly_payments": [],
            "subscriptions": [],
        },
        "user_profile": {
            "first_name": "",
            "last_name": "",
            "name": "",
            "dietary_prefs": [],
            "fitness_goal": "",
            "budget_weekly": 0,
            "address": "",
            "city": "",
            "postal_code": "",
            "country": "",
        },
    }


def first_run_data() -> dict:
    """What a fresh install starts with: demo persona if LIFEOS_DEMO, else blank."""
    return seed_data() if config.DEMO else empty_data()


def seed_data() -> dict:
    """Demo persona: Awais, a busy student on a tight budget."""
    return {
        "meals": [
            {"date": _d(7), "dish": "Chicken karahi", "ingredients": ["chicken thighs", "tomatoes", "ginger", "green chili", "yogurt"]},
            {"date": _d(6), "dish": "Spaghetti aglio e olio", "ingredients": ["spaghetti", "garlic", "olive oil", "chili flakes"]},
            {"date": _d(5), "dish": "Egg fried rice", "ingredients": ["rice", "eggs", "frozen peas", "soy sauce", "green onion"]},
            {"date": _d(4), "dish": "Spaghetti bolognese", "ingredients": ["spaghetti", "ground beef", "tomato sauce", "onion"]},
            {"date": _d(3), "dish": "Chickpea curry", "ingredients": ["chickpeas", "coconut milk", "tomatoes", "curry powder", "rice"]},
            {"date": _d(2), "dish": "Grilled cheese and tomato soup", "ingredients": ["bread", "cheddar", "canned tomato soup"]},
            {"date": _d(1), "dish": "Chicken caesar wrap", "ingredients": ["chicken breast", "tortilla", "romaine", "caesar dressing", "parmesan"]},
        ],
        "workouts": [
            {"date": _d(6), "type": "run", "duration_min": 30},
            {"date": _d(5), "type": "push (chest/triceps)", "duration_min": 50},
            {"date": _d(3), "type": "pull (back/biceps)", "duration_min": 45},
            {"date": _d(2), "type": "run", "duration_min": 25},
            {"date": _d(1), "type": "legs", "duration_min": 55},
        ],
        "calendar": [
            {"id": _new_id(), "title": "CS lecture: Distributed Systems", "date": _week_day(0), "start": "10:00", "end": "12:00", "kind": "class"},
            {"id": _new_id(), "title": "Cafe shift", "date": _week_day(1), "start": "16:00", "end": "21:00", "kind": "work"},
            {"id": _new_id(), "title": "Stats tutorial", "date": _week_day(2), "start": "09:00", "end": "10:30", "kind": "class"},
            {"id": _new_id(), "title": "Cafe shift", "date": _week_day(5), "start": "08:00", "end": "14:00", "kind": "work"},
        ],
        "workout_schedule": {"days": ["mon", "wed", "sat"], "time": "07:00"},
        "goals": [
            {
                "id": _new_id(),
                "title": "Used car",
                "target_amount": 8000,
                "saved": 1200,
                "deadline": _months_out(14),
                "notes": "",
            }
        ],
        "finances": {
            "income_monthly": 1400,
            "monthly_payments": [
                {"name": "Rent (shared)", "amount": 650, "due_day": 1},
                {"name": "Phone plan", "amount": 45, "due_day": 15},
                {"name": "Transit pass", "amount": 128, "due_day": 1},
            ],
            "subscriptions": [
                {"name": "Netflix", "cost": 16.99, "last_used": _d(2)},
                {"name": "Spotify Premium", "cost": 11.99, "last_used": _d(0)},
                {"name": "Crunchyroll", "cost": 11.99, "last_used": _d(54)},
                {"name": "FitnessPal Pro", "cost": 19.99, "last_used": _d(88)},
                {"name": "Disney+", "cost": 13.99, "last_used": _d(41)},
                {"name": "iCloud 200GB", "cost": 3.99, "last_used": _d(0)},
                {"name": "Adobe CC Student", "cost": 22.99, "last_used": _d(9)},
            ],
        },
        "user_profile": {
            "first_name": "Awais",
            "last_name": "Aziz",
            "name": "Awais Aziz",
            "dietary_prefs": ["halal", "high-protein", "no pork"],
            "fitness_goal": "build muscle, run a 10K in September",
            "budget_weekly": 80,
            "address": "",
            "city": "Toronto",
            "postal_code": "",
            "country": "Canada",
        },
    }


def load() -> dict:
    with _lock:
        first_run = not os.path.exists(MEMORY_PATH)
        if first_run:
            mem = first_run_data()
            os.makedirs(DATA_DIR, exist_ok=True)
            with open(MEMORY_PATH, "w", encoding="utf-8") as f:
                json.dump(mem, f, indent=2, ensure_ascii=False)
            return mem
        with open(MEMORY_PATH, "r", encoding="utf-8") as f:
            mem = json.load(f)
    # Backfill keys added by schema upgrades so old memory.json files keep
    # working. Defaults come from empty_data() — never inject demo persona data
    # into a real user's store.
    blank = empty_data()
    changed = False
    for key in ("calendar", "workout_schedule", "goals"):
        if key not in mem:
            mem[key] = blank[key]
            changed = True
    if "monthly_payments" not in mem.get("finances", {}):
        mem.setdefault("finances", {})["monthly_payments"] = blank["finances"]["monthly_payments"]
        changed = True
    profile = mem.setdefault("user_profile", {})
    if "first_name" not in profile:
        # Upgrade legacy single-`name` profiles: first word -> first_name.
        parts = (profile.get("name") or "").split(" ", 1)
        profile["first_name"] = parts[0]
        profile.setdefault("last_name", parts[1] if len(parts) > 1 else "")
        changed = True
    profile.setdefault("last_name", "")
    for key in ("address", "city", "postal_code", "country"):
        if key not in profile:
            profile[key] = blank["user_profile"][key]
            changed = True
    full_name = f"{profile['first_name']} {profile['last_name']}".strip()
    if profile.get("name") != full_name:
        profile["name"] = full_name
        changed = True
    for g in mem.get("goals", []):
        if not g.get("id"):
            g["id"] = _new_id()
            changed = True
    if changed:
        save(mem)
    return mem


def save(mem: dict) -> None:
    with _lock:
        os.makedirs(DATA_DIR, exist_ok=True)
        with open(MEMORY_PATH, "w", encoding="utf-8") as f:
            json.dump(mem, f, indent=2, ensure_ascii=False)


def _write(mem: dict) -> dict:
    """Persist a full store to disk and return it (used by the reset helpers).
    Caller must not hold _lock — save() takes it."""
    save(mem)
    return mem


def reset_to_empty() -> dict:
    """Wipe the store back to a blank slate (real-user reset)."""
    return _write(empty_data())


def reset_to_seed() -> dict:
    """Reset the store to the demo persona (used by demo mode / reset button)."""
    return _write(seed_data())


def log_meal(dish: str, ingredients: list[str], when: str | None = None) -> dict:
    mem = load()
    mem["meals"].append({"date": when or date.today().isoformat(), "dish": dish, "ingredients": ingredients})
    save(mem)
    return mem


def log_workout(workout_type: str, duration_min: int, when: str | None = None) -> dict:
    mem = load()
    mem["workouts"].append({"date": when or date.today().isoformat(), "type": workout_type, "duration_min": int(duration_min)})
    save(mem)
    return mem


def set_subscriptions(subs: list[dict]) -> dict:
    """Replace detected subscriptions (from the Money feature)."""
    mem = load()
    mem["finances"]["subscriptions"] = subs
    save(mem)
    return mem


def add_event(ev: dict) -> dict:
    """Add a calendar event. Fills in an id if missing. Returns full mem."""
    mem = load()
    event = {
        "id": ev.get("id") or _new_id(),
        "title": ev.get("title", ""),
        "date": ev.get("date", date.today().isoformat()),
        "start": ev.get("start", "09:00"),
        "end": ev.get("end", "10:00"),
        "kind": ev.get("kind", "other"),
    }
    mem.setdefault("calendar", []).append(event)
    save(mem)
    return mem


def delete_event(event_id: str) -> dict:
    mem = load()
    mem["calendar"] = [e for e in mem.get("calendar", []) if e.get("id") != event_id]
    save(mem)
    return mem


def set_workout_schedule(days: list[str], time: str) -> dict:
    mem = load()
    mem["workout_schedule"] = {"days": [d.lower()[:3] for d in days], "time": time}
    save(mem)
    return mem


def set_monthly_payments(payments: list[dict]) -> dict:
    mem = load()
    mem.setdefault("finances", {})["monthly_payments"] = [
        {"name": p.get("name", ""), "amount": p.get("amount", 0), "due_day": int(p.get("due_day", 1))}
        for p in payments
    ]
    save(mem)
    return mem


def upsert_goal(goal: dict) -> dict:
    """Insert or update a goal by id (new id assigned when missing/unknown)."""
    mem = load()
    goals = mem.setdefault("goals", [])
    gid = goal.get("id")
    merged = {
        "id": gid or _new_id(),
        "title": goal.get("title", ""),
        "target_amount": goal.get("target_amount", 0),
        "saved": goal.get("saved", 0),
        "deadline": goal.get("deadline", _months_out(12)),
        "notes": goal.get("notes", ""),
    }
    for i, g in enumerate(goals):
        if gid and g.get("id") == gid:
            goals[i] = {**g, **{k: v for k, v in goal.items() if v is not None}}
            break
    else:
        goals.append(merged)
    save(mem)
    return mem


def set_profile(fields: dict) -> dict:
    """Merge fields into user_profile; income_monthly routes to finances."""
    mem = load()
    fields = dict(fields)
    if "income_monthly" in fields:
        mem.setdefault("finances", {})["income_monthly"] = fields.pop("income_monthly")
    profile = mem.setdefault("user_profile", {})
    profile.update(fields)
    profile["name"] = f"{profile.get('first_name', '')} {profile.get('last_name', '')}".strip() or profile.get("name", "")
    save(mem)
    return mem


def delete_goal(goal_id: str) -> dict:
    mem = load()
    mem["goals"] = [g for g in mem.get("goals", []) if g.get("id") != goal_id]
    save(mem)
    return mem


# Sections the frontend may replace wholesale; value = path into mem.
_EDITABLE_SECTIONS = {
    "meals": ("meals",),
    "workouts": ("workouts",),
    "calendar": ("calendar",),
    "subscriptions": ("finances", "subscriptions"),
    "monthly_payments": ("finances", "monthly_payments"),
}


def set_section(section: str, items: list) -> dict:
    """Replace a whitelisted list section wholesale (manual memory editing)."""
    path = _EDITABLE_SECTIONS.get(section)
    if path is None:
        raise ValueError(f"unknown section: {section!r}")
    if not isinstance(items, list):
        raise ValueError("items must be a list")
    mem = load()
    target = mem
    for key in path[:-1]:
        target = target.setdefault(key, {})
    target[path[-1]] = items
    save(mem)
    return mem


def events_in_window(days: int = 7, mem: dict | None = None) -> list[dict]:
    """Calendar events from today through today+days, sorted."""
    mem = mem or load()
    start = date.today().isoformat()
    end = (date.today() + timedelta(days=days)).isoformat()
    return sorted(
        (e for e in mem.get("calendar", []) if start <= e.get("date", "") <= end),
        key=lambda e: (e.get("date", ""), e.get("start", "")),
    )


def recent_meals(days: int = 7, mem: dict | None = None) -> list[dict]:
    mem = mem or load()
    cutoff = (date.today() - timedelta(days=days)).isoformat()
    return [m for m in mem["meals"] if m["date"] >= cutoff]


def workouts_in_window(days: int = 14, mem: dict | None = None) -> list[dict]:
    mem = mem or load()
    cutoff = (date.today() - timedelta(days=days)).isoformat()
    return sorted((w for w in mem["workouts"] if w["date"] >= cutoff), key=lambda w: w["date"])