Spaces:
Sleeping
Sleeping
| """ | |
| 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 shutil | |
| 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) | |
| # ── inbox (Phase B): /capture?token enqueues, /capture-inbox drains+merges ─ | |
| created: list[str] = [] | |
| client.post("/capture", data={"caption": "@koffee cafe Tokyo", | |
| "url": "https://instagram.com/p/zzz", | |
| "token": "tok-test", "api_key": "x"}) | |
| drain = client.post("/capture-inbox/tok-test").json() | |
| if drain.get("job_id"): | |
| created.append(drain["job_id"]) | |
| check("inbox drain returns a job", bool(drain.get("job_id"))) | |
| check("inbox drain added == 1", drain.get("added") == 1) | |
| check("inbox drain added_urls correct", | |
| drain.get("added_urls") == ["https://instagram.com/p/zzz"]) | |
| rows = client.get(f"/results/{drain['job_id']}").json()["rows"] | |
| check("inbox job contains the captured place", | |
| any(r["instagram_url"] == "https://instagram.com/p/zzz" for r in rows)) | |
| # idempotent: a second drain is empty | |
| again = client.post("/capture-inbox/tok-test").json() | |
| check("second drain is empty", again.get("empty") is True) | |
| # drain with merge_into folds into an existing library (keep-existing-wins) | |
| client.post("/capture", data={"caption": "Ichiran ramen", | |
| "url": "https://instagram.com/p/aaa", | |
| "token": "tok-m", "api_key": "x"}) | |
| lib = ("name,city,state,country,address,category,cuisine,price_range,highlight," | |
| "occasion,lat,lng,creator,saved_at,instagram_url,status\n" | |
| "Old Spot,Paris,,France,,Cafe,,,,,48.85,2.35,,,https://instagram.com/p/old,visited\n") | |
| merged = client.post("/capture-inbox/tok-m", data={"merge_into": lib}).json() | |
| if merged.get("job_id"): | |
| created.append(merged["job_id"]) | |
| mrows = {r["instagram_url"]: r for r in client.get(f"/results/{merged['job_id']}").json()["rows"]} | |
| check("inbox merge: existing place kept", "https://instagram.com/p/old" in mrows) | |
| check("inbox merge: existing status preserved", | |
| mrows["https://instagram.com/p/old"]["status"] == "visited") | |
| check("inbox merge: captured place added", "https://instagram.com/p/aaa" in mrows) | |
| for j in created: | |
| shutil.rmtree(webapp.JOBS_DIR / j, ignore_errors=True) | |
| 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) | |