"""Web UI for Instagram Restaurant Mapper. Run with: uvicorn app:app --reload Or: python app.py """ import asyncio import contextlib import csv import io import json import math import os import re import sys import threading import uuid from pathlib import Path from fastapi import FastAPI, File, Form, Request, UploadFile from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from pydantic import BaseModel sys.path.insert(0, str(Path(__file__).parent.parent)) from pipeline import extract as extract_mod from pipeline import geocode as geocode_mod from pipeline import export as export_mod from pipeline import override as override_mod from pipeline.extract import DEFAULT_MODEL, DEFAULT_OLLAMA_MODEL, FIELDNAMES, MODELS, MODELS_OLLAMA, estimate_cost, load_posts JOBS_DIR = Path(__file__).parent.parent / "jobs" JOBS_DIR.mkdir(exist_ok=True) OLLAMA_ENABLED = os.environ.get("OLLAMA_ENABLED", "true").lower() not in ("0", "false", "no") OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434") JOB_TTL_HOURS = float(os.environ.get("JOB_TTL_HOURS", "24")) # Bound concurrent pipelines on a shared host so simultaneous uploads don't # exhaust RAM. Extra jobs wait in "queued" state until a slot frees. MAX_CONCURRENT_JOBS = max(1, int(os.environ.get("MAX_CONCURRENT_JOBS", "2"))) _pipeline_slots = threading.Semaphore(MAX_CONCURRENT_JOBS) # Public-URL guardrails: reject oversized uploads / libraries before doing work. MAX_UPLOAD_MB = float(os.environ.get("MAX_UPLOAD_MB", "25")) MAX_POSTS = int(os.environ.get("MAX_POSTS", "5000")) # Redact anything that looks like an API key before surfacing an error message. _SECRET_RE = re.compile(r"sk-ant-[A-Za-z0-9_\-]+") def safe_err(exc) -> str: """Stringify an exception with any API-key-looking substring redacted.""" return _SECRET_RE.sub("sk-ant-***", str(exc)) def _too_large(contents: bytes) -> bool: return len(contents) > MAX_UPLOAD_MB * 1024 * 1024 # Hosted/shared deploy can't bulk-geocode against public OSM (one server IP → # banned). When that guard is active, the server skips geocoding and the browser # does it from each visitor's own IP (see DEPLOY.md, the geocodeClientSide() JS). CLIENT_GEOCODE = geocode_mod.public_bulk_blocked(os.environ.get("HOSTED")) # Optional CORS proxy for Instagram fetching. Not normally needed — the app # uses Instagram's /embed/captioned/ endpoint which works from any server IP. # Set only if you have a custom proxy and want to override the default path. CAPTION_PROXY_URL: str | None = os.environ.get("CAPTION_PROXY_URL") or None # Categories treated as "food" for the food-only filter in Browse and Roulette. FOOD_CATEGORIES = {"Restaurant", "Bar", "Cafe", "Bakery", "Market"} _BENCHMARK_RESULTS_PATH = Path(__file__).parent.parent / "tests" / "benchmark_results.json" def _load_benchmark_scores() -> dict: try: with open(_BENCHMARK_RESULTS_PATH) as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {} app = FastAPI(title="Instagram Places Mapper") app.mount("/static", StaticFiles(directory=str(Path(__file__).parent / "static")), name="static") templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates")) @app.on_event("startup") def _on_startup() -> None: _cleanup_old_jobs() # In-memory job state, mirrored to /state.json so it survives a restart. _jobs: dict[str, dict] = {} _lock = threading.Lock() # Ephemeral capture inbox: token -> {"places": [row, ...], "ts": epoch}. Captures # from the iOS Shortcut land here keyed by an opaque token the user holds; the web # app drains + merges them on load. Intentionally in-memory and TTL'd — a transient # queue, not a database (a lost capture can just be re-shared). Capped per token. _capture_inbox: dict[str, dict] = {} _CAPTURE_INBOX_TTL = 7 * 24 * 3600 # 7 days _CAPTURE_INBOX_MAX = 500 # rows kept per token def _inbox_put(token: str, rows: list[dict]) -> None: import time as _time with _lock: entry = _capture_inbox.setdefault(token, {"places": [], "ts": _time.time()}) entry["places"].extend(rows) entry["places"] = entry["places"][-_CAPTURE_INBOX_MAX:] entry["ts"] = _time.time() def _inbox_drain(token: str) -> list[dict]: """Return and clear a token's pending captures (dropping stale tokens).""" import time as _time cutoff = _time.time() - _CAPTURE_INBOX_TTL with _lock: for t in [t for t, e in _capture_inbox.items() if e["ts"] < cutoff]: _capture_inbox.pop(t, None) entry = _capture_inbox.pop(token, None) return entry["places"] if entry else [] def _state_path(job_id: str) -> Path: return JOBS_DIR / job_id / "state.json" def _persist_state(job_id: str, state: dict) -> None: """Mirror a job's state to disk so the progress endpoint survives a restart.""" try: job_dir = JOBS_DIR / job_id job_dir.mkdir(exist_ok=True) _state_path(job_id).write_text(json.dumps(state), encoding="utf-8") except OSError: pass # disk mirror is best-effort; in-memory state is authoritative def _get_state(job_id: str) -> dict: """Return a job's state from memory, falling back to the on-disk mirror.""" with _lock: if job_id in _jobs: return dict(_jobs[job_id]) try: return json.loads(_state_path(job_id).read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {"step": "unknown", "message": "Job not found"} def _update(job_id: str, **kwargs) -> None: with _lock: _jobs[job_id].update(kwargs) snapshot = dict(_jobs[job_id]) _persist_state(job_id, snapshot) def _cleanup_old_jobs() -> None: """Delete job directories older than JOB_TTL_HOURS (best-effort).""" import shutil import time as _time cutoff = _time.time() - JOB_TTL_HOURS * 3600 for d in JOBS_DIR.glob("*"): if not d.is_dir(): continue try: if d.stat().st_mtime < cutoff: shutil.rmtree(d, ignore_errors=True) with _lock: _jobs.pop(d.name, None) except OSError: pass def _haversine_km(lat1, lon1, lat2, lon2) -> float: R = 6371 dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2 return R * 2 * math.asin(math.sqrt(a)) def _acquire_slot(job_id: str) -> None: """Block until a pipeline worker slot is free (bounded concurrency).""" if not _pipeline_slots.acquire(blocking=False): _update(job_id, step="queued", progress=5, message="Waiting for a free processing slot…") _pipeline_slots.acquire() def _run_pipeline(job_id: str, json_bytes: bytes, model: str, provider: str = "anthropic", ollama_url: str = "http://localhost:11434", api_key: str | None = None, merge_into: str = "") -> None: job_dir = JOBS_DIR / job_id job_dir.mkdir(exist_ok=True) json_path = job_dir / "saved_posts.json" csv_path = job_dir / "places_full.csv" kml_path = job_dir / "places_map.kml" override_path = job_dir / "places_override.csv" _acquire_slot(job_id) try: json_path.write_bytes(json_bytes) # ── Living-library merge: seed the job CSV with the caller's existing # library so extract's warm-start skips the LLM on posts already known # (keyed on instagram_url). Results are merged back below so nothing in # the library is lost if a slimmer export omits it. existing_rows: list[dict] = [] existing_urls: set[str] = set() if merge_into: try: existing_rows = _parse_import_csv(merge_into.encode()) except ValueError: existing_rows = [] if existing_rows: _write_csv(job_id, existing_rows) # seed for warm-start existing_urls = {r.get("instagram_url") for r in existing_rows if r.get("instagram_url")} # ── Extract ──────────────────────────────────────────────────────── _update(job_id, step="extract", progress=10, message="Reading your saved posts…") posts = load_posts(str(json_path)) if len(posts) > MAX_POSTS: _update(job_id, step="error", progress=0, message=(f"Library too large for the hosted demo " f"({len(posts)} posts, max {MAX_POSTS}). Run the Docker " f"or CLI version locally for libraries this big.")) return # Only posts not already in the library need the LLM. new_urls = {p["url"] for p in posts if p["url"] not in existing_urls} known = len(posts) - len(new_urls) est = estimate_cost(posts, new_urls, model, provider=provider) provider_label = f"Ollama/{model}" if provider == "ollama" else f"Claude/{model}" pre_filtered = est.get("prefiltered", 0) llm_posts = est.get("llm_posts", len(new_urls)) pre_note = f" ({pre_filtered} pre-filtered, no LLM call)" if pre_filtered else "" known_note = f" — {known} already in your library, skipped" if known else "" _update(job_id, step="extract", progress=15, message=f"Extracting {llm_posts} new posts with {provider_label}{pre_note}{known_note}…", total_posts=len(posts), prefiltered=pre_filtered) with contextlib.redirect_stdout(io.StringIO()): extract_mod.run(str(json_path), str(csv_path), model=model, provider=provider, ollama_url=ollama_url, resume=bool(existing_rows), batch_size=10, api_key=api_key) with open(csv_path) as f: extracted = list(csv.DictReader(f)) # Merge the fresh extraction back into the existing library # (keep-existing-wins): preserves status/coords on known posts and # re-adds any library places the new export happened to omit. if existing_rows: extracted, _added = _merge_rows(existing_rows, extracted) _write_csv(job_id, extracted) # ── Geocode ──────────────────────────────────────────────────────── if CLIENT_GEOCODE: # Hosted/shared deploy: skip server-side geocoding (public OSM bulk # use is banned from one IP). The browser geocodes each row from the # visitor's own IP and persists coords via POST /set-coords. with open(csv_path) as f: geocoded_rows = list(csv.DictReader(f)) pinned = sum(1 for r in geocoded_rows if r.get("lat") and r.get("lng")) _update(job_id, step="export", progress=85, message=f"Extracted {len(geocoded_rows)} places — pinning in your browser…", pinned=pinned, client_geocode=True) else: _update(job_id, step="geocode", progress=55, message=f"Found {len(extracted)} places — geocoding…", extracted=len(extracted)) with contextlib.redirect_stdout(io.StringIO()): geocode_mod.run(str(csv_path)) with open(csv_path) as f: geocoded_rows = list(csv.DictReader(f)) pinned = sum(1 for r in geocoded_rows if r.get("lat") and r.get("lng")) _update(job_id, step="export", progress=85, message=f"Geocoded {pinned}/{len(geocoded_rows)} — building your map…", pinned=pinned) # ── Override (generate template) ─────────────────────────────────── with contextlib.redirect_stdout(io.StringIO()): override_mod.apply(geocoded_rows, override_path) # ── Export KML ───────────────────────────────────────────────────── with contextlib.redirect_stdout(io.StringIO()): export_mod.run(str(csv_path), str(kml_path)) done_msg = ( f"Your map is ready! {len(geocoded_rows)} places — pinning in your browser…" if CLIENT_GEOCODE else f"Your map is ready! {len(geocoded_rows)} places, {pinned} pinned." ) _update(job_id, step="done", progress=100, message=done_msg, extracted=len(geocoded_rows), pinned=pinned, prefiltered=pre_filtered, client_geocode=CLIENT_GEOCODE, cost=round(est["total_cost"], 3)) except Exception as exc: _update(job_id, step="error", progress=0, message=safe_err(exc)) finally: _pipeline_slots.release() def _places_from_caption(caption: str, url: str, creator: str = "", location: dict | None = None, *, provider: str = "anthropic", model: str = DEFAULT_MODEL, api_key: str | None = None, ollama_url: str = "http://localhost:11434") -> list[dict]: """Extract place row(s) from one post's caption — shared by the single-URL pipeline and POST /capture. Returns FIELDNAMES-shaped rows (empty if no place found). Multi-venue posts yield one row per venue (split_multi_venue). When the post carries an Instagram location tag and resolves to a single venue, its coords are used directly (no geocoding needed). """ client = None if provider == "anthropic": import anthropic as _anthropic client = _anthropic.Anthropic(api_key=api_key) if api_key else _anthropic.Anthropic() results = extract_mod.analyze_batch(client, [{"caption": caption, "hashtags": []}], model=model, provider=provider, ollama_url=ollama_url) infos = [r for r in results if r is not None] if not infos: return [] rows = extract_mod.split_multi_venue([ {**info, "creator": creator, "instagram_url": url, "lat": "", "lng": ""} for info in infos ]) if location and len(rows) == 1: rows[0]["lat"] = f"{location['lat']:.7f}" rows[0]["lng"] = f"{location['lng']:.7f}" return rows def _run_url_pipeline(job_id: str, url: str, model: str, provider: str = "anthropic", ollama_url: str = "http://localhost:11434", api_key: str | None = None) -> None: """Extract a single place from an Instagram post URL and build a map.""" job_dir = JOBS_DIR / job_id job_dir.mkdir(exist_ok=True) csv_path = job_dir / "places_full.csv" kml_path = job_dir / "places_map.kml" override_path = job_dir / "places_override.csv" _acquire_slot(job_id) try: _update(job_id, step="extract", progress=20, message="Fetching post…") if CLIENT_GEOCODE and not CAPTION_PROXY_URL: # Hosted build without a proxy: Instagram blocks all cloud server IPs. _update(job_id, step="error", progress=0, message="URL extraction isn't available on the hosted build — " "Instagram blocks requests from cloud servers. " "Use the free chatbot mode instead, or run the local Docker build.") return from pipeline.transcribe import fetch_post_metadata try: meta = fetch_post_metadata(url, proxy_url=CAPTION_PROXY_URL) except ValueError as exc: _update(job_id, step="error", progress=0, message=str(exc)) return if not meta["caption"]: _update(job_id, step="error", progress=0, message="Post has no caption — video-only posts aren't supported on the hosted build yet.") return _update(job_id, step="extract", progress=40, message="Extracting place with AI…") rows = _places_from_caption(meta["caption"], url, meta["creator"], meta["location"], provider=provider, model=model, api_key=api_key, ollama_url=ollama_url) if not rows: _update(job_id, step="error", progress=0, message="No place found in this post.") return with open(csv_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=extract_mod.FIELDNAMES) writer.writeheader() for row in rows: writer.writerow({k: row.get(k, "") for k in extract_mod.FIELDNAMES}) pinned_from_tag = sum(1 for r in rows if r.get("lat") and r.get("lng")) if not (CLIENT_GEOCODE or pinned_from_tag == len(rows)): _update(job_id, step="geocode", progress=60, message="Geocoding…") with contextlib.redirect_stdout(io.StringIO()): geocode_mod.run(str(csv_path)) with open(csv_path) as f: geocoded_rows = list(csv.DictReader(f)) pinned = sum(1 for r in geocoded_rows if r.get("lat") and r.get("lng")) _update(job_id, step="export", progress=85, message="Building map…") with contextlib.redirect_stdout(io.StringIO()): override_mod.apply(geocoded_rows, override_path) with contextlib.redirect_stdout(io.StringIO()): export_mod.run(str(csv_path), str(kml_path)) place_name = geocoded_rows[0].get("name", "Place") if geocoded_rows else "Place" done_msg = ( f"Done! {place_name} — pinning in your browser…" if CLIENT_GEOCODE else f"Done! {place_name}" + (" 📍" if pinned else " — no geocode match") ) _update(job_id, step="done", progress=100, message=done_msg, extracted=len(geocoded_rows), pinned=pinned, client_geocode=CLIENT_GEOCODE, cost=0) except Exception as exc: _update(job_id, step="error", progress=0, message=safe_err(exc)) finally: _pipeline_slots.release() def _run_geocode_export(job_id: str) -> None: """Geocode + export an already-written CSV (no extraction step).""" job_dir = JOBS_DIR / job_id csv_path = job_dir / "places_full.csv" kml_path = job_dir / "places_map.kml" override_path = job_dir / "places_override.csv" _acquire_slot(job_id) try: if CLIENT_GEOCODE: # Hosted mode: skip server geocoding; the browser pins each row. with open(csv_path) as f: rows = list(csv.DictReader(f)) pinned = sum(1 for r in rows if r.get("lat") and r.get("lng")) _update(job_id, step="export", progress=80, message=f"Parsed {len(rows)} places — pinning in your browser…", pinned=pinned) else: _update(job_id, step="geocode", progress=30, message="Geocoding places…") with contextlib.redirect_stdout(io.StringIO()): geocode_mod.run(str(csv_path)) with open(csv_path) as f: rows = list(csv.DictReader(f)) pinned = sum(1 for r in rows if r.get("lat") and r.get("lng")) _update(job_id, step="export", progress=80, message=f"Geocoded {pinned}/{len(rows)} — building map…", pinned=pinned) with contextlib.redirect_stdout(io.StringIO()): override_mod.apply(rows, override_path) with contextlib.redirect_stdout(io.StringIO()): export_mod.run(str(csv_path), str(kml_path)) done_msg = ( f"Your map is ready! {len(rows)} places — pinning in your browser…" if CLIENT_GEOCODE else f"Your map is ready! {len(rows)} places, {pinned} pinned." ) _update(job_id, step="done", progress=100, message=done_msg, extracted=len(rows), pinned=pinned, client_geocode=CLIENT_GEOCODE, cost=0) except Exception as exc: _update(job_id, step="error", progress=0, message=safe_err(exc)) finally: _pipeline_slots.release() # ── Results browser helpers ─────────────────────────────────────────────────── _EDITABLE_FIELDS = frozenset({ "name", "city", "state", "country", "address", "category", "cuisine", "price_range", "highlight", "occasion", "status", }) _VALID_STATUSES = frozenset({"unvisited", "visited", "want_to_go", "closed"}) class UpdateRequest(BaseModel): url: str updates: dict[str, str] class RegeocodeRequest(BaseModel): url: str class SetCoordsRequest(BaseModel): url: str lat: str lng: str def _csv_path(job_id: str) -> Path: return JOBS_DIR / job_id / "places_full.csv" def _read_csv(job_id: str) -> list[dict] | None: p = _csv_path(job_id) if not p.exists(): return None with open(p, encoding="utf-8") as f: return list(csv.DictReader(f)) def _write_csv(job_id: str, rows: list[dict]) -> None: p = _csv_path(job_id) with open(p, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=FIELDNAMES) writer.writeheader() for row in rows: writer.writerow({k: row.get(k, "") for k in FIELDNAMES}) def resolve_provider(provider: str, model: str, ollama_model: str, *, ollama_enabled: bool) -> tuple[str, str]: """Deployment-mode guardrail for the web UI's provider selection. Returns the effective (provider, model) for a request: - Local Docker deploy (ollama_enabled=True): the Ollama provider is honored; an unknown ollama_model falls back to the default. - Hosted free-web deploy (ollama_enabled=False): Ollama is never used — there is no local model server on a shared host — so any request is served by Claude (BYOK / chatbot), with an unknown model defaulting. Keep this pure; tests/test_deployment_modes.py pins this contract so a change for one deployment can't silently break the other. """ if provider == "ollama" and ollama_enabled: return "ollama", (ollama_model if ollama_model in MODELS_OLLAMA else DEFAULT_OLLAMA_MODEL) return "anthropic", (model if model in MODELS else DEFAULT_MODEL) def valid_latlng(lat, lng) -> tuple[float, float] | None: """Validate browser-supplied coordinates (client-geocode /set-coords). Returns (lat, lng) floats if both parse and are in range, else None. Pure; pinned by tests/test_deployment_modes.py. """ try: latf, lngf = float(lat), float(lng) except (TypeError, ValueError): return None if -90.0 <= latf <= 90.0 and -180.0 <= lngf <= 180.0: return (latf, lngf) return None def _row_to_result(r: dict) -> dict: return { "name": r.get("name", ""), "city": r.get("city", ""), "state": r.get("state", ""), "country": r.get("country", ""), "address": r.get("address", ""), "category": r.get("category", ""), "cuisine": r.get("cuisine", ""), "price_range": r.get("price_range", ""), "highlight": r.get("highlight", ""), "occasion": r.get("occasion", ""), "lat": r.get("lat", ""), "lng": r.get("lng", ""), "creator": r.get("creator", ""), "saved_at": r.get("saved_at", ""), "instagram_url": r.get("instagram_url", ""), "geocoded": bool(r.get("lat") and r.get("lng")), "status": r.get("status") or "unvisited", } # ── Routes ──────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def index(request: Request): # New Starlette signature: request first (the old name-first form is removed # in recent Starlette and 500s with "unhashable type: dict"). return templates.TemplateResponse(request, "index.html", { "models": MODELS, "default_model": DEFAULT_MODEL, "models_ollama": MODELS_OLLAMA, "default_ollama_model": DEFAULT_OLLAMA_MODEL, "benchmark_scores": _load_benchmark_scores(), "ollama_enabled": OLLAMA_ENABLED, "client_geocode": CLIENT_GEOCODE, "has_server_key": bool(os.environ.get("ANTHROPIC_API_KEY")), }) @app.post("/upload") async def upload( file: UploadFile = File(...), model: str = Form(DEFAULT_MODEL), provider: str = Form("anthropic"), ollama_model: str = Form(DEFAULT_OLLAMA_MODEL), ollama_url: str = Form(None), api_key: str = Form(""), merge_into: str = Form(""), ): provider, active_model = resolve_provider( provider, model, ollama_model, ollama_enabled=OLLAMA_ENABLED ) contents = await file.read() if _too_large(contents): return JSONResponse({"error": f"File too large (max {MAX_UPLOAD_MB:.0f} MB)."}, status_code=413) _cleanup_old_jobs() job_id = uuid.uuid4().hex initial = {"step": "queued", "progress": 0, "message": "Starting…"} with _lock: _jobs[job_id] = initial _persist_state(job_id, initial) thread = threading.Thread( target=_run_pipeline, args=(job_id, contents, active_model, provider, ollama_url or OLLAMA_URL, api_key.strip() or None, merge_into), daemon=True, ) thread.start() return {"job_id": job_id} @app.post("/extract-url") async def extract_url_route( url: str = Form(...), model: str = Form(DEFAULT_MODEL), provider: str = Form("anthropic"), ollama_model: str = Form(DEFAULT_OLLAMA_MODEL), ollama_url: str = Form(None), api_key: str = Form(""), ): url = url.strip() if not re.search(r"https?://(www\.)?instagram\.com/", url): return JSONResponse( {"error": "Please enter an Instagram post URL (https://www.instagram.com/p/…)"}, status_code=422, ) provider, active_model = resolve_provider( provider, model, ollama_model, ollama_enabled=OLLAMA_ENABLED ) _cleanup_old_jobs() job_id = uuid.uuid4().hex initial = {"step": "queued", "progress": 0, "message": "Starting…"} with _lock: _jobs[job_id] = initial _persist_state(job_id, initial) threading.Thread( target=_run_url_pipeline, args=(job_id, url, active_model, provider, ollama_url or OLLAMA_URL, api_key.strip() or None), daemon=True, ).start() return {"job_id": job_id} @app.post("/capture") async def capture( embed_html: str = Form(""), caption: str = Form(""), url: str = Form(""), creator: str = Form(""), lat: str = Form(""), lng: str = Form(""), token: str = Form(""), model: str = Form(DEFAULT_MODEL), provider: str = Form("anthropic"), ollama_model: str = Form(DEFAULT_OLLAMA_MODEL), ollama_url: str = Form(None), api_key: str = Form(""), ): """Extract place(s) from a single shared post — the iOS Shortcut path. The caller fetches /embed/captioned/ ON-DEVICE (residential IP, no CORS) and POSTs `embed_html`; alternatively send a ready `caption`, or a `url` for the server to fetch (local builds only — the hosted server is IP-walled). Optional `lat`/`lng` carry an Instagram location tag for free coordinates. Returns {places, count}. Inbox delivery (token) is added in Phase B. """ provider, active_model = resolve_provider( provider, model, ollama_model, ollama_enabled=OLLAMA_ENABLED ) location = None if lat and lng: try: location = {"lat": float(lat), "lng": float(lng)} except ValueError: location = None if embed_html: if _too_large(embed_html.encode()): return JSONResponse({"error": f"Payload too large (max {MAX_UPLOAD_MB:.0f} MB)."}, status_code=413) from pipeline.transcribe import _parse_embed_caption parsed = _parse_embed_caption(embed_html) caption = caption or parsed["caption"] creator = creator or parsed["creator"] elif url and not caption: if CLIENT_GEOCODE and not CAPTION_PROXY_URL: return JSONResponse( {"error": "The server can't fetch Instagram on the hosted build — " "the Shortcut should POST embed_html fetched on your device."}, status_code=422) from pipeline.transcribe import fetch_post_metadata try: meta = await asyncio.to_thread(fetch_post_metadata, url, None, CAPTION_PROXY_URL) except ValueError as exc: return JSONResponse({"error": str(exc)}, status_code=422) caption = meta["caption"] creator = creator or meta["creator"] location = location or meta["location"] if not caption.strip(): return JSONResponse({"error": "No caption found in the post."}, status_code=422) try: rows = await asyncio.to_thread( _places_from_caption, caption, url, creator, location, provider=provider, model=active_model, api_key=api_key.strip() or None, ollama_url=ollama_url or OLLAMA_URL, ) except Exception as exc: return JSONResponse({"error": safe_err(exc)}, status_code=502) if not rows: return JSONResponse({"error": "No place found in this post."}, status_code=422) # If a token is supplied, queue the place(s) for the web app to merge on load. if token: _inbox_put(token, rows) return {"places": rows, "count": len(rows), "queued": bool(token)} @app.post("/capture-inbox/{token}") async def capture_inbox(token: str, merge_into: str = Form("")): """Drain a token's pending captures and merge them into the caller's library. Reuses the living-library merge: returns a fresh job (like /import) with the captured places folded into merge_into (keep-existing-wins), or {empty:true} when nothing is queued. The web app calls this on load with its cached library and shows "N places captured while you were away". """ places = _inbox_drain(token) if not places: return {"empty": True} added = len(places) added_urls = [r.get("instagram_url", "") for r in places if r.get("instagram_url")] rows = places if merge_into: try: existing = _parse_import_csv(merge_into.encode()) except ValueError: existing = [] if existing: rows, added_rows = _merge_rows(existing, places) added = len(added_rows) added_urls = [r.get("instagram_url", "") for r in added_rows if r.get("instagram_url")] _cleanup_old_jobs() job_id = uuid.uuid4().hex (JOBS_DIR / job_id).mkdir(exist_ok=True) _write_csv(job_id, rows) kml = export_mod.build_kml(rows) (JOBS_DIR / job_id / "places_map.kml").write_text(kml, encoding="utf-8") pinned = sum(1 for r in rows if r.get("lat") and r.get("lng")) state = {"step": "done", "progress": 100, "message": f"Merged {added} captured places.", "extracted": len(rows), "pinned": pinned, "cost": 0, "client_geocode": False} with _lock: _jobs[job_id] = state _persist_state(job_id, state) return {"job_id": job_id, "rows": len(rows), "added": added, "added_urls": added_urls, "merged": bool(merge_into)} @app.post("/url-chatbot-prepare") async def url_chatbot_prepare(url: str = Form(...)): """Fetch caption via proxy/yt-dlp and return a chatbot export package.""" url = url.strip() if not re.search(r"https?://(www\.)?instagram\.com/", url): return JSONResponse( {"error": "Please enter an Instagram post URL (https://www.instagram.com/p/…)"}, status_code=422, ) if CLIENT_GEOCODE and not CAPTION_PROXY_URL: return JSONResponse( {"error": "URL extraction isn't available on the hosted build — " "Instagram blocks requests from cloud servers. " "Use the free chatbot mode instead, or run the local Docker build."}, status_code=422, ) from pipeline.transcribe import fetch_post_metadata try: meta = await asyncio.to_thread(fetch_post_metadata, url, None, CAPTION_PROXY_URL) except ValueError as exc: return JSONResponse({"error": str(exc)}, status_code=422) if not meta["caption"]: return JSONResponse( {"error": "Post has no caption — video-only posts aren't supported on the hosted build yet."}, status_code=422, ) _cleanup_old_jobs() job_id = uuid.uuid4().hex job_dir = JOBS_DIR / job_id job_dir.mkdir(exist_ok=True) post = {"url": url, "caption": meta["caption"], "creator": meta["creator"], "hashtags": [], "saved_at": ""} from pipeline import chatbot as chatbot_mod result = chatbot_mod.prepare_export([post]) (job_dir / "post_mapping.json").write_text( json.dumps(result["post_mapping"]), encoding="utf-8" ) pending = {"step": "chatbot_pending", "progress": 0, "message": "Waiting for chatbot response…"} with _lock: _jobs[job_id] = pending _persist_state(job_id, pending) return { "job_id": job_id, "export_posts": result["export_posts"], "post_count": len(result["export_posts"]), "total": result["total"], "skipped": result["skipped"], "prompt": chatbot_mod.get_prompt(len(result["export_posts"])), "warn_large": False, "chunk_size": chatbot_mod.CHUNK_SIZE, "chunk_count": 1, } @app.get("/progress/{job_id}") async def progress(job_id: str): async def _generate(): import asyncio while True: state = _get_state(job_id) yield f"data: {json.dumps(state)}\n\n" if state.get("step") in ("done", "error"): break await asyncio.sleep(0.6) return StreamingResponse(_generate(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) @app.get("/download/{job_id}") async def download(job_id: str): kml = JOBS_DIR / job_id / "places_map.kml" if not kml.exists(): return HTMLResponse("Not found", status_code=404) return FileResponse(str(kml), media_type="application/vnd.google-earth.kml+xml", filename="places_map.kml") @app.get("/download/{job_id}/geojson") async def download_geojson(job_id: str): csv_p = _csv_path(job_id) if not csv_p.exists(): return HTMLResponse("Not found", status_code=404) geojson_str = await asyncio.to_thread(export_mod.to_geojson, str(csv_p)) return Response( content=geojson_str, media_type="application/geo+json", headers={"Content-Disposition": 'attachment; filename="places_map.geojson"'}, ) @app.get("/download/{job_id}/csv") async def download_csv(job_id: str): csv_p = _csv_path(job_id) if not csv_p.exists(): return HTMLResponse("Not found", status_code=404) return FileResponse(str(csv_p), media_type="text/csv", filename="places_full.csv") @app.get("/results/{job_id}") async def results(job_id: str): rows = _read_csv(job_id) if rows is None: return JSONResponse({"error": "Not found"}, status_code=404) rows.sort(key=lambda r: r.get("saved_at") or "", reverse=True) return {"rows": [_row_to_result(r) for r in rows], "total": len(rows)} @app.patch("/results/{job_id}") async def update_result(job_id: str, body: UpdateRequest): rows = _read_csv(job_id) if rows is None: return JSONResponse({"error": "Job not found"}, status_code=404) bad = [f for f in body.updates if f not in _EDITABLE_FIELDS] if bad: return JSONResponse({"error": f"Non-editable fields: {bad}"}, status_code=422) if "status" in body.updates and body.updates["status"] not in _VALID_STATUSES: return JSONResponse({"error": f"Invalid status. Must be one of: {sorted(_VALID_STATUSES)}"}, status_code=422) for row in rows: if row.get("instagram_url") == body.url: for field, value in body.updates.items(): row[field] = value.strip() _write_csv(job_id, rows) return _row_to_result(row) return JSONResponse({"error": "Row not found"}, status_code=404) @app.post("/regeocode/{job_id}") async def regeocode(job_id: str, body: RegeocodeRequest): rows = _read_csv(job_id) if rows is None: return JSONResponse({"error": "Job not found"}, status_code=404) for row in rows: if row.get("instagram_url") == body.url: lat, lng = geocode_mod.geocode_one(row) row["lat"] = lat row["lng"] = lng _write_csv(job_id, rows) return {"url": body.url, "lat": lat, "lng": lng, "geocoded": bool(lat and lng)} return JSONResponse({"error": "Row not found"}, status_code=404) @app.post("/set-coords/{job_id}") async def set_coords(job_id: str, body: SetCoordsRequest): """Persist coordinates the BROWSER geocoded (hosted client-geocode mode). The server is geocode-blocked on a shared deploy, so the page geocodes each row from the visitor's own IP and saves the result here. """ rows = _read_csv(job_id) if rows is None: return JSONResponse({"error": "Job not found"}, status_code=404) coords = valid_latlng(body.lat, body.lng) if coords is None: return JSONResponse({"error": "lat/lng must be numbers in range"}, status_code=422) latf, lngf = coords for row in rows: if row.get("instagram_url") == body.url: row["lat"] = f"{latf:.7f}" row["lng"] = f"{lngf:.7f}" _write_csv(job_id, rows) return {"url": body.url, "lat": row["lat"], "lng": row["lng"], "geocoded": True} return JSONResponse({"error": "Row not found"}, status_code=404) @app.post("/export/{job_id}") async def reexport(job_id: str): csv_p = _csv_path(job_id) if not csv_p.exists(): return JSONResponse({"error": "Job not found"}, status_code=404) kml_p = JOBS_DIR / job_id / "places_map.kml" def _run(): with contextlib.redirect_stdout(io.StringIO()): export_mod.run(str(csv_p), str(kml_p)) await asyncio.to_thread(_run) return {"kml_url": f"/download/{job_id}"} def _merge_key(row: dict) -> str: """Stable merge key: instagram_url when present, else normalized name|city. instagram_url (its /p/) is the canonical per-post key. The name|city fallback dedups rows that lack a URL (some KML / manual entries). """ url = (row.get("instagram_url") or "").strip() if url: return "url:" + url name = (row.get("name") or "").strip().lower() city = (row.get("city") or "").strip().lower() return "nc:" + name + "|" + city def _merge_rows(existing: list[dict], incoming: list[dict]) -> tuple[list[dict], list[dict]]: """Fold incoming rows into existing, keyed on _merge_key — keep-existing-wins. Existing rows are never dropped or overwritten, so user-owned status and manual edits are preserved (the status-is-user-owned invariant). Only genuinely new keys are appended. Returns (merged_rows, added_rows) where added_rows is the list of incoming rows that were new (use len() for the count, instagram_url for badging). """ seen: set[str] = set() merged: list[dict] = [] for row in existing: k = _merge_key(row) if k not in seen: seen.add(k) merged.append(row) added: list[dict] = [] for row in incoming: k = _merge_key(row) if k not in seen: seen.add(k) merged.append(row) added.append(row) return merged, added def _parse_import_csv(contents: bytes) -> list[dict]: """Parse a places_full.csv from a previous run into row dicts.""" import io text = contents.decode("utf-8", errors="replace") reader = csv.DictReader(io.StringIO(text)) rows = list(reader) if not rows: raise ValueError("CSV is empty") headers = set(reader.fieldnames or []) if not {"name", "country"}.issubset(headers): raise ValueError( "File doesn't look like a places_full.csv (missing 'name' or 'country' column)" ) result = [] for r in rows: clean = {k: r.get(k, "") for k in FIELDNAMES} clean["status"] = clean.get("status") or "unvisited" result.append(clean) return result def _parse_import_kml(contents: bytes) -> list[dict]: """Parse a places_map.kml from a previous run into row dicts. Reads the two-level folder hierarchy (country → city), extracts name + coordinates from each Placemark, and parses the CDATA description for the remaining fields. KML coordinates are in lng,lat,alt order. """ import re import xml.etree.ElementTree as ET try: root = ET.fromstring(contents.decode("utf-8", errors="replace")) except ET.ParseError as exc: raise ValueError(f"Invalid KML: {exc}") KML_NS = "http://www.opengis.net/kml/2.2" def _t(tag): return f"{{{KML_NS}}}{tag}" def _find_text(elem, tag): el = elem.find(_t(tag)) return (el.text or "").strip() if el is not None else "" def _parse_placemark(pm, country, city) -> dict | None: name = _find_text(pm, "name") # Coordinates: lng,lat,alt coord_el = pm.find(f".//{_t('coordinates')}") lat = lng = "" if coord_el is not None and coord_el.text: parts = coord_el.text.strip().split(",") if len(parts) >= 2: try: lng = f"{float(parts[0]):.7f}" lat = f"{float(parts[1]):.7f}" except ValueError: pass if not name and not lat: return None # Description CDATA parsing desc_el = pm.find(_t("description")) desc = (desc_el.text or "") if desc_el is not None else "" # Normalise CDATA: replace
with newlines, strip HTML tags desc = re.sub(r"", "\n", desc, flags=re.I) desc = re.sub(r"<[^>]+>", " ", desc) category = cuisine = price_range = highlight = address = "" creator = saved_at = instagram_url = "" desc_city = desc_state = "" for line in desc.split("\n"): line = line.strip() if not line: continue if line.startswith("City:"): desc_city = line[len("City:"):].strip() elif line.startswith("State:"): desc_state = line[len("State:"):].strip() elif line.startswith("Highlight:"): highlight = line[len("Highlight:"):].strip() elif line.startswith("via @"): creator = line[len("via @"):].strip() elif line.startswith("Saved "): saved_at = line[len("Saved "):].strip() elif "instagram.com" in line: m = re.search(r"https?://[^\s\"'<>]+instagram\.com/[^\s\"'<>]+", line) if m: instagram_url = m.group(0).rstrip(".,;)") elif " · " in line: # meta line: category · cuisine · price meta_parts = [p.strip() for p in line.split(" · ")] for i, p in enumerate(meta_parts): if p in ("$", "$$", "$$$", "$$$$"): price_range = p elif i == 0 and not category: category = p elif i == 1 and not cuisine and p not in ("$", "$$", "$$$", "$$$$"): cuisine = p # Description-embedded city/state take priority over folder-derived values resolved_city = desc_city or city resolved_state = desc_state def _u(v): return v if v else "UNKNOWN" return { "name": _u(name), "city": _u(resolved_city), "state": _u(resolved_state), "country": _u(country), "address": _u(address), "category": _u(category), "cuisine": _u(cuisine), "price_range": _u(price_range), "highlight": _u(highlight), "occasion": "UNKNOWN", "lat": lat, "lng": lng, "creator": creator, "saved_at": saved_at, "instagram_url": instagram_url, "status": "unvisited", } rows: list[dict] = [] def _walk(elem, country="", city=""): tag = elem.tag if tag == _t("Folder"): folder_name = _find_text(elem, "name") if not country: new_country, new_city = folder_name, city else: new_country, new_city = country, folder_name for child in elem: _walk(child, new_country, new_city) elif tag == _t("Placemark"): row = _parse_placemark(elem, country, city) if row: rows.append(row) else: for child in elem: _walk(child, country, city) _walk(root) return rows class ChatbotProcessRequest(BaseModel): response_json: str = "" responses: list[str] | None = None # one per export chunk (preferred) @app.post("/chatbot-prepare") async def chatbot_prepare(file: UploadFile = File(...)): """Receive an Instagram JSON/zip, run prefilter, return the chatbot export package.""" import zipfile as _zipfile contents = await file.read() if _too_large(contents): return JSONResponse({"error": f"File too large (max {MAX_UPLOAD_MB:.0f} MB)."}, status_code=413) _cleanup_old_jobs() if file.filename and file.filename.lower().endswith(".zip"): try: with _zipfile.ZipFile(io.BytesIO(contents)) as zf: match = next((n for n in zf.namelist() if n.endswith("saved_posts.json")), None) if not match: return JSONResponse({"error": "saved_posts.json not found in zip"}, status_code=422) contents = zf.read(match) except Exception as exc: return JSONResponse({"error": f"Could not read zip: {exc}"}, status_code=422) if _too_large(contents): # guard against a small zip expanding hugely return JSONResponse({"error": f"Extracted file too large (max {MAX_UPLOAD_MB:.0f} MB)."}, status_code=413) job_id = uuid.uuid4().hex job_dir = JOBS_DIR / job_id job_dir.mkdir(exist_ok=True) json_path = job_dir / "saved_posts.json" json_path.write_bytes(contents) try: posts = load_posts(str(json_path)) except Exception as exc: return JSONResponse({"error": f"Could not parse saved_posts.json: {exc}"}, status_code=422) if len(posts) > MAX_POSTS: return JSONResponse( {"error": f"Library too large for the hosted demo ({len(posts)} posts, " f"max {MAX_POSTS}). Use the Docker or CLI version locally."}, status_code=413, ) from pipeline import chatbot as chatbot_mod result = chatbot_mod.prepare_export(posts) # Persist mapping so /chatbot-process can match results back to posts (job_dir / "post_mapping.json").write_text( json.dumps(result["post_mapping"]), encoding="utf-8" ) pending = {"step": "chatbot_pending", "progress": 0, "message": "Waiting for chatbot response…"} with _lock: _jobs[job_id] = pending _persist_state(job_id, pending) return { "job_id": job_id, "export_posts": result["export_posts"], "post_count": len(result["export_posts"]), "total": result["total"], "skipped": result["skipped"], "prompt": chatbot_mod.get_prompt(len(result["export_posts"])), "warn_large": len(result["export_posts"]) > chatbot_mod.CHUNK_WARN_THRESHOLD, "chunk_size": chatbot_mod.CHUNK_SIZE, "chunk_count": len(chatbot_mod.chunk_export(result["export_posts"])), } @app.post("/chatbot-process/{job_id}") async def chatbot_process(job_id: str, body: ChatbotProcessRequest): """Parse the chatbot's JSON response and kick off geocoding + export.""" job_dir = JOBS_DIR / job_id mapping_path = job_dir / "post_mapping.json" if not mapping_path.exists(): return JSONResponse({"error": "Job not found or expired"}, status_code=404) with open(mapping_path, encoding="utf-8") as f: post_mapping = json.load(f) from pipeline import chatbot as chatbot_mod texts = body.responses if body.responses else [body.response_json] try: rows = chatbot_mod.parse_responses(texts, post_mapping) except ValueError as exc: return JSONResponse({"error": str(exc)}, status_code=422) if not rows: return JSONResponse( {"error": "No place posts found in the response. Check that you included all rows."}, status_code=422, ) csv_path = job_dir / "places_full.csv" with open(csv_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=FIELDNAMES) writer.writeheader() for row in rows: writer.writerow({k: row.get(k, "") for k in FIELDNAMES}) _update(job_id, step="geocode", progress=20, message=f"Parsed {len(rows)} places from chatbot — geocoding…", extracted=len(rows)) threading.Thread(target=_run_geocode_export, args=(job_id,), daemon=True).start() return {"job_id": job_id, "rows": len(rows)} @app.get("/roulette") async def roulette(job_id: str = "", lat: float = None, lng: float = None, radius_km: float = 40, city: str = "", all_statuses: bool = False, category: str = "", food_only: bool = False): """Pick a random place from a specific job's CSV. A job_id is required — there is deliberately no "newest job wins" fallback, which would leak one user's places to another on a shared deployment. """ import random if not job_id: return {"error": "No job selected. Run a pipeline first."} csv_path = JOBS_DIR / job_id / "places_full.csv" if not csv_path.exists(): return {"error": "No place data for this job yet. Run a pipeline first."} with open(csv_path, encoding="utf-8") as f: rows = [r for r in csv.DictReader(f) if r.get("name") and r["name"].upper() != "UNKNOWN"] city = city.strip() if city: q = city.lower() rows = [r for r in rows if r.get("city", "").strip().lower() == q or r.get("country", "").strip().lower() == q] if not rows: return {"error": f"No places found in {city}."} elif lat is not None and lng is not None: def _ok(r): try: return _haversine_km(lat, lng, float(r["lat"]), float(r["lng"])) <= radius_km except (ValueError, TypeError): return False rows = [r for r in rows if _ok(r)] if not rows: return {"error": "No places match your criteria."} # Food-only filter (applied before explicit category filter) if food_only: rows = [r for r in rows if r.get("category", "").strip() in FOOD_CATEGORIES] if not rows: return {"error": "No food places found. Turn off 'Food only' to see all categories."} # Category filter category = category.strip() if category: rows = [r for r in rows if r.get("category", "").strip().lower() == category.lower()] if not rows: return {"error": f"No {category} places found. Try a different category or location."} # Status filter: default to unvisited + want_to_go only if not all_statuses: _ACTIVE = {"unvisited", "want_to_go", ""} filtered = [r for r in rows if (r.get("status") or "") in _ACTIVE] if not filtered: return {"error": "No unvisited places here — try the 'All places' toggle."} rows = filtered pick = random.choice(rows) return { "name": pick.get("name", ""), "city": pick.get("city", ""), "country": pick.get("country", ""), "category": pick.get("category", ""), "cuisine": pick.get("cuisine", ""), "price_range": pick.get("price_range", ""), "highlight": pick.get("highlight", ""), "occasion": pick.get("occasion", ""), "creator": pick.get("creator", ""), "saved_at": pick.get("saved_at", ""), "lat": pick.get("lat", ""), "lng": pick.get("lng", ""), "url": pick.get("instagram_url", ""), "job_id": csv_path.parent.name, } @app.post("/import") async def import_places(file: UploadFile = File(...), merge_into: str | None = Form(None)): """Import a KML or CSV from a previous run, skip extraction entirely. Creates a job dir, writes places_full.csv + places_map.kml, and returns {job_id, rows, pinned, added, merged, no_overlap} with step already set to "done" — no SSE stream needed, the client can call GET /results/{job_id}. When merge_into (the caller's existing library CSV) is supplied, the parsed file is folded into it keyed on instagram_url (keep-existing-wins), so a re-import or capture augments the library instead of replacing it. """ contents = await file.read() if _too_large(contents): return JSONResponse({"error": f"File too large (max {MAX_UPLOAD_MB:.0f} MB)."}, status_code=413) filename = (file.filename or "").lower() try: if filename.endswith(".csv"): rows = _parse_import_csv(contents) elif filename.endswith(".kml"): rows = _parse_import_kml(contents) else: return JSONResponse( {"error": "Unsupported format. Upload a .kml or .csv file from a previous run."}, status_code=422, ) except ValueError as exc: return JSONResponse({"error": str(exc)}, status_code=422) if not rows: return JSONResponse({"error": "No places found in the file."}, status_code=422) # Living-library merge: fold the incoming file into the caller's existing # library (keep-existing-wins) when one is supplied. incoming_count = len(rows) added = incoming_count added_urls: list[str] = [] kept = 0 merged = False no_overlap = False if merge_into: try: existing = _parse_import_csv(merge_into.encode()) except ValueError: existing = [] if existing: rows, added_rows = _merge_rows(existing, rows) added = len(added_rows) added_urls = [r["instagram_url"] for r in added_rows if r.get("instagram_url")] kept = incoming_count - added # posts the file shared with the library merged = True no_overlap = added == incoming_count # nothing matched → maybe wrong file _cleanup_old_jobs() job_id = uuid.uuid4().hex job_dir = JOBS_DIR / job_id job_dir.mkdir(exist_ok=True) _write_csv(job_id, rows) kml = export_mod.build_kml(rows) (job_dir / "places_map.kml").write_text(kml, encoding="utf-8") pinned = sum(1 for r in rows if r.get("lat") and r.get("lng")) state = { "step": "done", "progress": 100, "message": f"Imported {len(rows)} places.", "extracted": len(rows), "pinned": pinned, "cost": 0, "client_geocode": False, } with _lock: _jobs[job_id] = state _persist_state(job_id, state) return {"job_id": job_id, "rows": len(rows), "pinned": pinned, "added": added, "added_urls": added_urls, "kept": kept, "merged": merged, "no_overlap": no_overlap} if __name__ == "__main__": import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)