Feat: /capture endpoint core (Phase A) — single-post extraction

Refactors _run_url_pipeline's extract core into shared _places_from_caption()
(analyze_batch → split_multi_venue → location-tag coords). New POST /capture
accepts embed_html (the iOS Shortcut fetches /embed/captioned/ on-device so the
server never touches Instagram), or a ready caption, or a url (local fetch),
plus optional lat/lng for a free location-tag pin. Returns {places, count}.
BYOK api_key. tests/test_capture.py added to CI; e2e still green (refactor safe).

Inbox delivery (Phase B) and the Shortcut itself (Phase C) are next.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.github/workflows/benchmark.yml

@@ -39,6 +39,9 @@ jobs:
       - name: Merge — living-library merge keeps curation
         run: python3 tests/test_merge.py
 
+      - name: Capture — single-post extraction (Shortcut path)
+        run: python3 tests/test_capture.py
+
   benchmark:
     name: Benchmark smoke test (Ollama)
     runs-on: ubuntu-latest

CLAUDE.md

@@ -49,6 +49,12 @@ web/app.py         FastAPI: pipeline, SSE progress, results browser, resolve_pro
                    POST /import — KML or CSV re-import (no extraction, skip to tabs)
                    POST /extract-url — single Instagram URL → caption fetch → LLM + geocode
                    POST /url-chatbot-prepare — same but returns chatbot export package
+                   POST /capture — single shared post → place(s) JSON (iOS Shortcut path).
+                     Accepts embed_html (Shortcut fetches /embed/captioned/ ON-DEVICE so the
+                     server never touches Instagram) / caption / url, + optional lat,lng
+                     (IG location tag → free coords). Core extraction is the shared
+                     _places_from_caption() (refactored out of _run_url_pipeline; both use it).
+                     BYOK api_key. Inbox delivery (token) lands in Phase B.
                    CAPTION_PROXY_URL env var: reserved for a future residential proxy
 web/static/        served at /static/ via FastAPI StaticFiles mount
   app.css          all styles (~1300 lines, no Jinja)

README.md

@@ -426,6 +426,7 @@ tests/
   test_prefilter.py         Guardrail: prefilter zero-false-negatives against oracle (CI)
   test_persistence.py       Guardrail: cache/restore CSV round-trip is lossless (CI)
   test_merge.py             Guardrail: living-library merge keeps curation (CI)
+  test_capture.py           Guardrail: /capture single-post extraction (CI)
   benchmark.py              Evaluate any model against the oracle
   create_fixture.py         Sample 50 posts for the benchmark fixture
   generate_ground_truth.py  Label posts with Opus 4.8 (the oracle)

tests/test_capture.py

@@ -0,0 +1,104 @@
+"""
+POST /capture — single-post extraction for the iOS Shortcut path (Phase A).
+
+The Shortcut fetches /embed/captioned/ on-device and POSTs the HTML (or a ready
+caption, or a URL for local server-fetch). /capture runs the shared
+_places_from_caption and returns the extracted place(s). Optional lat/lng carry
+an Instagram location tag → coordinates without geocoding.
+
+Only the LLM is mocked. Run: python3 tests/test_capture.py
+"""
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).parent.parent
+sys.path.insert(0, str(ROOT))
+
+try:
+    from starlette.testclient import TestClient
+except Exception as exc:
+    print(f"SKIP: TestClient unavailable ({exc}). `pip install httpx` to run.")
+    sys.exit(0)
+
+from pipeline import extract as extract_mod
+from web import app as webapp
+
+failures: list[str] = []
+
+
+def check(label: str, cond: bool) -> None:
+    print(("PASS" if cond else "FAIL"), "-", label)
+    if not cond:
+        failures.append(label)
+
+
+def _place(name, city="Tokyo", country="Japan", category="Cafe"):
+    d = {k: "UNKNOWN" for k in ("name", "city", "state", "country", "address",
+                                "cuisine", "price_range", "highlight", "occasion")}
+    d.update(name=name, city=city, country=country, category=category)
+    return d
+
+
+def fake_analyze_batch(client, posts, model=None, provider="anthropic", ollama_url=""):
+    """Deterministic stand-in: a place when the caption looks place-y, else None."""
+    out = []
+    for p in posts:
+        cap = (p.get("caption") or "").lower()
+        if "ichiran" in cap or "ramen" in cap or "cafe" in cap or "koffee" in cap:
+            out.append(_place("Koffee Mameya" if "koffee" in cap else "Ichiran"))
+        else:
+            out.append(None)
+    return out
+
+
+_orig_ab = extract_mod.analyze_batch
+extract_mod.analyze_batch = fake_analyze_batch
+try:
+    with TestClient(webapp.app) as client:
+
+        # ── caption path ──────────────────────────────────────────────────────
+        r = client.post("/capture", data={"caption": "Amazing @koffee_mameya cafe in Tokyo",
+                                           "url": "https://instagram.com/p/aaa",
+                                           "api_key": "sk-test-unused"})
+        check("caption path: 200", r.status_code == 200)
+        body = r.json()
+        check("caption path: 1 place", body.get("count") == 1)
+        check("caption path: place name extracted", body["places"][0]["name"] == "Koffee Mameya")
+        check("caption path: instagram_url carried", body["places"][0]["instagram_url"] == "https://instagram.com/p/aaa")
+
+        # ── location tag → coords without geocoding ─────────────────────────────
+        r2 = client.post("/capture", data={"caption": "Ichiran ramen", "url": "https://instagram.com/p/bbb",
+                                            "lat": "35.6595", "lng": "139.7005", "api_key": "x"})
+        p2 = r2.json()["places"][0]
+        check("location tag: lat applied", p2["lat"].startswith("35.6595"))
+        check("location tag: lng applied", p2["lng"].startswith("139.7005"))
+
+        # ── embed_html path (server parses the Caption block) ───────────────────
+        embed = ('<html><body><div class="Caption">'
+                 '<a class="CaptionUsername" href="#">timeouttokyo_</a>'
+                 'Best ramen — Ichiran in Tokyo</div></body></html>')
+        r3 = client.post("/capture", data={"embed_html": embed,
+                                           "url": "https://instagram.com/p/ccc", "api_key": "x"})
+        check("embed_html path: 200", r3.status_code == 200)
+        check("embed_html path: place extracted", r3.json()["places"][0]["name"] == "Ichiran")
+
+        # ── error: no caption ───────────────────────────────────────────────────
+        r4 = client.post("/capture", data={"api_key": "x"})
+        check("no caption → 422", r4.status_code == 422)
+
+        # ── error: caption with no place ────────────────────────────────────────
+        r5 = client.post("/capture", data={"caption": "my gym workout routine today", "api_key": "x"})
+        check("non-place caption → 422", r5.status_code == 422)
+
+finally:
+    extract_mod.analyze_batch = _orig_ab
+
+print()
+if failures:
+    print(f"FAIL — {len(failures)} capture check(s) failed:")
+    for f in failures:
+        print(f"  ✗ {f}")
+    sys.exit(1)
+print("PASS — /capture extracts place(s) from a single shared post.")
+sys.exit(0)

web/app.py

@@ -285,6 +285,39 @@ def _run_pipeline(job_id: str, json_bytes: bytes, model: str,
 
 
 
+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:
@@ -319,29 +352,13 @@ def _run_url_pipeline(job_id: str, url: str, model: str,
 
         _update(job_id, step="extract", progress=40, message="Extracting place with AI…")
 
-        client = None
-        if provider == "anthropic":
-            import anthropic as _anthropic
-            client = _anthropic.Anthropic(api_key=api_key) if api_key else _anthropic.Anthropic()
-
-        post = {"caption": meta["caption"], "hashtags": []}
-        results = extract_mod.analyze_batch(client, [post],
-                                            model=model, provider=provider, ollama_url=ollama_url)
-        # analyze_batch returns [None] for non-place; multi-venue posts return
-        # multiple results for the single input post.
-        infos = [r for r in results if r is not None]
-        if not infos:
+        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
 
-        rows = extract_mod.split_multi_venue([
-            {**info, "creator": meta["creator"], "instagram_url": url, "lat": "", "lng": ""}
-            for info in infos
-        ])
-        if meta["location"] and len(rows) == 1:
-            rows[0]["lat"] = f"{meta['location']['lat']:.7f}"
-            rows[0]["lng"] = f"{meta['location']['lng']:.7f}"
-
         with open(csv_path, "w", newline="", encoding="utf-8") as f:
             writer = csv.DictWriter(f, fieldnames=extract_mod.FIELDNAMES)
             writer.writeheader()
@@ -618,6 +635,79 @@ async def extract_url_route(
     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(""),
+    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)
+
+    return {"places": rows, "count": len(rows)}
+
+
 @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."""