#!/usr/bin/env python3 """Push tomorrow's ChatPlan to Habit Journal (Dar es Salaam date). Auth from env / .env: AGENT_TOKEN preferred, else APP_PASSWORD session. Never prints secrets. EMBASSY_DAY=1 switches to embassy variant (not default). """ from __future__ import annotations import json import os import sys import uuid from datetime import datetime, timedelta from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.request import Request, build_opener, HTTPCookieProcessor from http.cookiejar import CookieJar try: from zoneinfo import ZoneInfo except ImportError: # pragma: no cover ZoneInfo = None # type: ignore ROOT = Path(__file__).resolve().parents[1] def load_dotenv() -> None: path = ROOT / ".env" if not path.is_file(): return for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, val = line.split("=", 1) key = key.strip() val = val.strip().strip('"').strip("'") os.environ.setdefault(key, val) def dar_tomorrow() -> str: if ZoneInfo is not None: try: tz = ZoneInfo("Africa/Dar_es_Salaam") return (datetime.now(tz).date() + timedelta(days=1)).isoformat() except Exception: # noqa: BLE001 pass # Fallback: UTC+3 approx (EAT) return (datetime.utcnow().date() + timedelta(days=1)).isoformat() def minutes_between(start: str, end: str) -> int: sh, sm = map(int, start.split(":")) eh, em = map(int, end.split(":")) return max(1, (eh * 60 + em) - (sh * 60 + sm)) def pack_chat_plan(day: str) -> dict[str, Any]: blocks_spec = [ ("07:30", "08:00", "Wake - water - leave bed", "P0", "body_care", "duty", False, "No shame scroll. Phone after standing."), ("08:00", "08:25", "Interrupt walk (no headphones)", "P0", "body_care", "measure", False, "Different path if possible. Eyes outside. Wallet home. Preview ≠ proof."), ("08:25", "08:50", "Breakfast + hygiene", "P0", "food_out", "duty", False, "Eat before admin spiral."), ("08:50", "09:05", "App Daily open + movement fields", "P1", "stabilize", "measure", False, "Mark left_home / interrupt minutes. Court closed."), ("09:05", "10:35", "P0 Fix AXA insurance PDF", "P0", "admin_spain", "duty", True, "BLOCKER. Residence=Tanzania. Coverage 26 Aug 2026–5 Sep 2026. Schengen min €30k. Replace SCHBE000291367 if still wrong country. Save PDF to embassy folder."), ("10:35", "10:50", "Break - water - no status war", "P2", "stabilize", "restore_fun", False, "No her thread. No mockers."), ("10:50", "12:20", "P0 Refundable stay proof MAD (+ BCN)", "P0", "admin_spain", "duty", True, "Flexible/refundable only while visa pending. Budget ~€110/night MAD, ~€130 BCN, ~€960 total band. Save confirmations PDF. Do not over-research."), ("12:20", "13:15", "Lunch + rest (not bed freeze)", "P0", "food_out", "duty", False, "Leave bedroom if freeze risk."), ("13:15", "14:00", "P0 Qatar hold PDF + print pack list", "P0", "admin_spain", "duty", True, "PNR 9QBP66 hold PDF only — DO NOT PAY. Cover letter + itinerary print-ready. Checklist for next walk-in (form, photo, fee, residence, AXA, stays, funds)."), ("14:00", "14:20", "Buffer / compound walk", "P2", "body_care", "measure", False, "Still no cinema headphones if FSE high."), ("14:20", "15:20", "Earn: offer + 3 payment asks", "P1", "earn_ship", "duty", False, "One sentence offer + price. Send ≥3 asks. Price once then silence. Pay-first or 50% deposit. Metric=TZS not likes. Solo no partners."), ("15:20", "15:40", "Break", "P2", "stabilize", "restore_fun", False, ""), ("15:40", "16:15", "Embassy folder: bank + bonds proof", "P1", "admin_spain", "duty", False, "One physical/digital folder. Statements + investment/bonds as discussed. No UTT drama if bonds path chosen."), ("16:15", "16:40", "Restore — music max 25m AFTER proof only", "P2", "restore_fun", "restore_fun", False, "Only if P0 AXA or stays moved. Else skip music; short silent walk. Leisure not medicine."), ("16:40", "17:05", "Seal: Log + plan checkoffs + day_review notes", "P1", "stabilize", "measure", False, "Log AXA/stays/asks. FSE in feedback notes if spiked. List gaps for next embassy day."), ("22:00", "22:30", "Sleep wind-down", "P0", "sleep_window", "duty", False, "No multi-escape spiral. Phone out of bed if possible."), ] empty_review = { "did": None, "minutes": None, "comment": "", "emotions": [], "fse_event": "", "intensity": None, } blocks = [] for start, end, title, priority, kind, intent, locked, notes in blocks_spec: blocks.append( { "id": str(uuid.uuid4()), "start": start, "end": end, "title": title, "priority": priority, "kind": kind, "intent": intent, "notes": notes, "locked": locked, "status": "planned", "review": dict(empty_review), } ) return { "schema_version": 1, "date": day, "title": "Spain pack + first asks", "intention": ( "Close AXA + stay proof + hold PDF. Three customer asks. " "No embassy unless pack 100% and day is Mon/Wed/Fri. No flight pay." ), "constraints": [ "do not pay Qatar until visa stamped", "AXA residence must be Tanzania not Spain/Belgium", "interrupt walk: no headphones first 20 min", "Weeknd/music only after P0 proof", "wallet home on morning walk", "no WhatsApp Spain flex", "solo earn - no partners", ], "blocks": blocks, "day_review": { "comment": "", "emotions": [], "fse_events": "", "what_moved": "", "what_avoided": "", "tomorrow_change": "", }, } def to_day_plan_put(chat: dict[str, Any], day: str) -> dict[str, Any]: notes = ( "INTENTION: Close AXA + stay proof + hold PDF. Three customer asks. " "No flight pay. Embassy only if Mon/Wed/Fri AND pack complete.\n" "CONSTRAINTS: no pay Qatar pre-stamp; AXA residence Tanzania; walk no HP 20m; " "Weeknd after P0; wallet home morning; no Spain status flex; solo earn\n" "AXA: fix/replace if residence Spain/Belgium; dates 26 Aug–5 Sep 2026; PNR hold 9QBP66 PDF only" ) blocks = [] for raw in chat["blocks"]: start, end = raw["start"], raw["end"] blocks.append( { "id": raw.get("id") or str(uuid.uuid4()), "date": day, "start": start, "end": end, "title": raw["title"], "kind": raw["kind"], "intent": raw["intent"], "priority": raw["priority"], "planned_min": minutes_between(start, end), "status": "planned", "source": "cursor", "locked": bool(raw.get("locked")), "notes": raw.get("notes") or "", "version_added": 1, } ) return { "source": "cursor", "capacity_hint": 0.75, "notes": notes, "title": chat.get("title") or "", "intention": chat.get("intention") or "", "constraints": chat.get("constraints") or [], "day_review": chat.get("day_review") or {}, "force_p0_move": True, "blocks": blocks, } class Client: def __init__(self, base: str) -> None: self.base = base.rstrip("/") self.jar = CookieJar() self.opener = build_opener(HTTPCookieProcessor(self.jar)) self.endpoint = "" def request( self, method: str, path: str, *, body: dict | None = None, bearer: str | None = None, raw: bool = False, ) -> Any: data = None if body is None else json.dumps(body).encode("utf-8") headers = {"Accept": "application/json"} if body is not None: headers["Content-Type"] = "application/json" if bearer: headers["Authorization"] = f"Bearer {bearer}" req = Request(f"{self.base}{path}", data=data, headers=headers, method=method) try: with self.opener.open(req, timeout=60) as resp: text = resp.read().decode("utf-8") if raw: return text return json.loads(text) if text else {} except HTTPError as exc: err_body = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {exc.code} {path}: {err_body[:800]}") from exc except URLError as exc: raise RuntimeError(f"Network error {path}: {exc}") from exc def push(day: str) -> int: load_dotenv() base = os.environ.get("HF_SPACE_URL", "https://Mbonea-Fastwhisper.hf.space").rstrip("/") agent = (os.environ.get("AGENT_TOKEN") or "").strip() password = (os.environ.get("APP_PASSWORD") or "").strip() client = Client(base) chat = pack_chat_plan(day) put = to_day_plan_put(chat, day) # Peek existing prev_count = None try: if agent: prev = client.request("GET", f"/api/agent/context?day={day}", bearer=agent) if prev.get("ok") and isinstance(prev.get("data"), dict): plan = prev["data"].get("current_plan") or {} prev_count = len(plan.get("blocks") or []) except Exception: # noqa: BLE001 pass last_err = None # 1) Import ChatPlan (session or agent session login) # Prefer agent PUT DayPlanPut if agent: try: env = client.request( "PUT", f"/api/agent/plan/{day}", body=put, bearer=agent, ) if env.get("ok"): client.endpoint = f"PUT /api/agent/plan/{day} (bearer)" else: raise RuntimeError(str(env.get("error"))) except Exception as exc: # noqa: BLE001 last_err = exc agent = agent # keep for later GET # fall through env = None else: return report(client, day, env, prev_count, agent_token=agent) if password: login = client.request("POST", "/api/auth/login", body={"password": password}) if not login.get("ok"): print(f"LOGIN_FAIL: {login.get('error')}", file=sys.stderr) return 1 # Try import first (ChatPlan native) try: env = client.request( "POST", f"/api/plan/{day}/import", body={"plan": chat, "mode": "replace"}, ) if env.get("ok"): client.endpoint = f"POST /api/plan/{day}/import (session)" return report(client, day, env, prev_count) last_err = RuntimeError(str(env.get("error"))) except Exception as exc: # noqa: BLE001 last_err = exc try: env = client.request("PUT", f"/api/plan/{day}", body=put) if env.get("ok"): client.endpoint = f"PUT /api/plan/{day} (session)" return report(client, day, env, prev_count) last_err = RuntimeError(str(env.get("error"))) except Exception as exc: # noqa: BLE001 last_err = exc print(f"PUSH_FAILED: {last_err}", file=sys.stderr) print("Set AGENT_TOKEN or APP_PASSWORD in env/.env (not git).", file=sys.stderr) return 1 def report( client: Client, day: str, env: dict[str, Any], prev_count: int | None, agent_token: str | None = None, ) -> int: # Prefer GET plan with session; agent context as fallback data = env.get("data") if isinstance(env.get("data"), dict) else {} blocks = data.get("blocks") if not isinstance(blocks, list): try: got = client.request("GET", f"/api/plan/{day}") if got.get("ok"): data = got["data"] blocks = data.get("blocks") or [] except Exception: # noqa: BLE001 if agent_token: got = client.request( "GET", f"/api/agent/context?day={day}", bearer=agent_token ) plan = (got.get("data") or {}).get("current_plan") or {} blocks = plan.get("blocks") or [] data = plan else: blocks = [] n = len(blocks) titles = [str(b.get("title") or "") for b in blocks] print(f"DAY={day}") print(f"ENDPOINT={client.endpoint}") if prev_count is not None: print(f"PREV_BLOCKS={prev_count}") print(f"BLOCKS={n}") print("FIRST_THREE=" + " | ".join(titles[:3])) if data.get("title"): print(f"TITLE={data.get('title')}") if n < 14: print(f"VERIFY_FAIL: expected >=14 blocks, got {n}", file=sys.stderr) return 1 print("OK") return 0 def main() -> int: load_dotenv() day = os.environ.get("DAY") or dar_tomorrow() if os.environ.get("EMBASSY_DAY", "0") == "1": print("EMBASSY_DAY=1 not implemented in this script; use pack day.", file=sys.stderr) return 2 return push(day) if __name__ == "__main__": raise SystemExit(main())