"""Shared-state API for the Tokyo Flat Finder viewer.
Stores the couple's likes/dislikes/notes and syncs them across devices.
State is held in memory and persisted to a PRIVATE HF Dataset (state.json) so
it survives Space restarts. Every request must carry the shared X-App-Key.
Env (set as Space secrets):
HF_TOKEN write token for the dataset
DATASET_REPO e.g. ArthurZ/tokyo-flat-finder-state
APP_SECRET shared key the viewer sends as X-App-Key
"""
from __future__ import annotations
import html as _html
import json
import os
import pathlib
import threading
from fastapi import FastAPI, Header, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from huggingface_hub import HfApi, hf_hub_download
from pydantic import BaseModel
# Slim public listing data for the shareable preview page (baked at deploy).
try:
PUBLIC = json.loads(pathlib.Path("public_listings.json").read_text())
except Exception:
PUBLIC = {}
HF_TOKEN = os.environ.get("HF_TOKEN")
DATASET_REPO = os.environ.get("DATASET_REPO", "")
APP_SECRET = os.environ.get("APP_SECRET", "")
STATE_FILE = "state.json"
api = HfApi(token=HF_TOKEN)
_lock = threading.Lock()
_state: dict = {"userStates": {}, "notes": {}}
def _load() -> None:
global _state
try:
path = hf_hub_download(DATASET_REPO, STATE_FILE, repo_type="dataset",
token=HF_TOKEN)
with open(path) as f:
data = json.load(f)
_state = {"userStates": data.get("userStates", {}), "notes": data.get("notes", {})}
except Exception:
_state = {"userStates": {}, "notes": {}}
def _persist() -> None:
buf = json.dumps(_state, ensure_ascii=False).encode("utf-8")
api.upload_file(path_or_fileobj=buf, path_in_repo=STATE_FILE,
repo_id=DATASET_REPO, repo_type="dataset",
commit_message="update shared state")
app = FastAPI()
app.add_middleware(
CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
)
@app.on_event("startup")
def _startup() -> None:
_load()
def _auth(key: str | None) -> None:
if not APP_SECRET or key != APP_SECRET:
raise HTTPException(status_code=403, detail="bad key")
@app.get("/health")
def health() -> dict:
return {"ok": True, "listings": len(_state["userStates"]), "notes": len(_state["notes"])}
@app.get("/state")
def get_state(x_app_key: str | None = Header(default=None)) -> dict:
_auth(x_app_key)
return _state
class UserStateIn(BaseModel):
listingId: str
handle: str # "me" | "partner"
state: str # "unseen" | "like" | "dislike" | "veto"
@app.post("/user_state")
def set_user_state(body: UserStateIn, x_app_key: str | None = Header(default=None)) -> dict:
_auth(x_app_key)
with _lock:
us = _state["userStates"]
cur = us.get(body.listingId, {})
if body.state == "unseen":
cur.pop(body.handle, None)
else:
cur[body.handle] = body.state
if cur:
us[body.listingId] = cur
else:
us.pop(body.listingId, None)
_persist()
return {"ok": True}
class NoteIn(BaseModel):
listingId: str
note: str
@app.post("/note")
def set_note(body: NoteIn, x_app_key: str | None = Header(default=None)) -> dict:
_auth(x_app_key)
with _lock:
if body.note.strip():
_state["notes"][body.listingId] = body.note
else:
_state["notes"].pop(body.listingId, None)
_persist()
return {"ok": True}
# --- Public shareable preview page (no key — anyone with the link) -----------
@app.get("/p/{listing_id}", response_class=HTMLResponse)
def preview(listing_id: str) -> str:
d = PUBLIC.get(listing_id)
if not d:
return HTMLResponse("
Listing not found
", status_code=404)
e = _html.escape
yen = f"¥{d['rent_jpy']:,}" if d.get("rent_jpy") else "—"
eur = f"€{round(d['rent_jpy'] / 186):,}" if d.get("rent_jpy") else "" # ponytail: fixed 186 rate
lat, lon = d.get("lat"), d.get("lon")
ote = (f"{d['ote_min']} min" + (" · " + "→".join(d["ote_lines"]) if d.get("ote_lines") else "")) \
if d.get("ote_min") is not None else "—"
photos = "".join(f'' for u in d.get("photos", []))
gmap = f"https://www.google.com/maps/search/?api=1&query={lat},{lon}"
osm = (f"https://www.openstreetmap.org/export/embed.html?bbox={lon-0.012},{lat-0.01},"
f"{lon+0.012},{lat+0.01}&layer=mapnik&marker={lat},{lon}") if lat and lon else ""
return f"""
{e(d.get('title') or 'Listing')}
{photos}
🏠 {e(d.get('title') or '')}
{e(d.get('layout') or '?')} · {d.get('size_m2') or '?'} m² · {e(d.get('ward') or '')}