app.py

#!/usr/bin/env python3
"""SetVL Node Graph Viewer —— 只读浏览器。

界面直接复用本地标注台（`tool/annotation`）那套前端：同一套 topographic 配色、
三栏布局、J/K 键盘导航、底图切换、邻接矩阵。区别只有一个——**去掉了全部写入**：
没有判定按钮、没有保存、没有新增边，后端也没有对应的接口。

架构：FastAPI 提供静态页面与只读 API，HF Spaces 用 docker SDK 跑。
曾经在 /gradio 挂过一个 Gradio Blocks 当兜底视图，已移除——它只是一段
带链接的 Markdown，却把 gradio 拖进了镜像依赖。

图片来源两种，按需切换：
- 本地包：SETVL_PACKAGE 指向 prepare.py 的产出目录
- HF 数据集：SETVL_DATASET 指向 "user/repo"，按需从 CDN 取并缓存
"""

from __future__ import annotations

import json
import math
import os
import sqlite3
import threading
from pathlib import Path
from typing import Any, Dict, List, Optional

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse

HERE = Path(__file__).resolve().parent
# 前端与本地复核台共用同一份（tool/web/）。打包进镜像时 prepare/deploy 会把
# 它复制到 /app/web，所以镜像里仍然是 HERE/"web"；本机直接跑则回落到共享目录。
WEB = HERE / "web" if (HERE / "web").is_dir() else HERE.parent / "web"
PACKAGE = Path(os.environ.get("SETVL_PACKAGE", HERE / "package"))
DATASET = os.environ.get("SETVL_DATASET", "").strip()
HF_TOKEN = (os.environ.get("HF_TOKEN")
            or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "").strip()
def _resolve_db() -> Path:
    """按优先级找库：显式环境变量 > 同目录（容器里就在 /app）> space/ > package/。"""
    if os.environ.get("SETVL_DB"):
        return Path(os.environ["SETVL_DB"])
    for cand in (HERE / "graph.sqlite", HERE / "space" / "graph.sqlite",
                 PACKAGE / "graph.sqlite"):
        if cand.exists():
            return cand
    return PACKAGE / "graph.sqlite"          # 供报错信息指向默认位置


DB = _resolve_db()
if not DB.exists():                          # 启动即失败，不要拖到浏览到一半才炸
    raise RuntimeError(
        f"找不到数据库。已尝试 {HERE/'graph.sqlite'}、{HERE/'space'/'graph.sqlite'}、"
        f"{PACKAGE/'graph.sqlite'}；可用 SETVL_DB 指定。")

_TLS = threading.local()


def con() -> sqlite3.Connection:
    """每线程一个只读连接。sqlite3 的连接不能跨线程共用。"""
    c = getattr(_TLS, "con", None)
    if c is None:
        if not DB.exists():
            raise RuntimeError(f"找不到数据库 {DB}")
        c = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
        c.row_factory = sqlite3.Row
        _TLS.con = c
    return c


_HAS_HEADING: Optional[bool] = None


def ground_has_heading(c: sqlite3.Connection) -> bool:
    """旧包的 ground_index 只有 (panoid, rel)，没有 heading 列。

    重新 prepare 过的包才有。这里探一次并记住，免得每个节点都查一次
    pragma；旧包降级成「只有一张街景」而不是 500。
    """
    global _HAS_HEADING
    if _HAS_HEADING is None:
        cols = {r[1] for r in c.execute("PRAGMA table_info(ground_index)")}
        _HAS_HEADING = "heading" in cols
    return _HAS_HEADING


def asset_path(rel: str) -> Optional[Path]:
    """包内相对路径 -> 本地文件。数据集模式下按需从 CDN 拉取并缓存。"""
    if not rel or ".." in rel:
        return None
    local = PACKAGE / rel
    if local.is_file():
        return local
    if not DATASET:
        return None
    try:
        from huggingface_hub import hf_hub_download
        # 私有数据集必须带 token；Space 里要配 HF_TOKEN secret
        return Path(hf_hub_download(DATASET, rel, repo_type="dataset",
                                    token=HF_TOKEN or None))
    except Exception:
        return None


def _loads(raw: Optional[str]) -> Any:
    if not raw:
        return None
    try:
        return json.loads(raw)
    except (ValueError, TypeError):
        return None


def effective(nodes: List[Dict], edges: List[Dict]) -> None:
    """合成 _auto / _human 的生效值，与标注台一致。"""
    for n in nodes:
        n["corridor_id"] = n["corridor_id_human"] or n["corridor_id_auto"]
        n["corridor_name"] = (n["corridor_name_human"] or n["corridor_name_auto"]
                              or n["corridor_id"])
        o = n["is_orphan_human"]
        n["is_orphan"] = int(o) if o is not None else (n["is_orphan_auto"] or 0)
        n["corridor_overridden"] = n["corridor_id_human"] is not None
        n["corridor_ids"] = _loads(n["corridor_ids_auto"]) or []
    for e in edges:
        e["origin"] = e["origin_human"] or e["origin_auto"] or "manual"
        rc = e["road_connected_human"]
        e["road_connected"] = int(rc) if rc is not None else (e["road_connected_auto"] or 0)
        e["verdict"] = e["edge_verdict_human"]
        e["in_graph"] = e["edge_verdict_human"] != "reject"
        e["is_manual"] = e["origin_human"] == "manual" or e["origin_auto"] is None
        e["origin_overridden"] = e["origin_human"] is not None
        e["road_path"] = _loads(e["road_path_auto"])
        e["evidence"] = _loads(e["evidence_auto"])
        e["notes"] = _loads(e["notes_auto"]) or []
        e["review_status"] = e["review_status_human"] or "unreviewed"


api = FastAPI(title="SetVL Node Graph Viewer")


@api.get("/")
def index() -> FileResponse:
    return FileResponse(WEB / "index.html")


@api.get("/app.js")
def appjs() -> FileResponse:
    return FileResponse(WEB / "app.js", media_type="application/javascript")


@api.get("/styles.css")
def css() -> FileResponse:
    return FileResponse(WEB / "styles.css", media_type="text/css")


@api.get("/api/bootstrap")
def bootstrap() -> JSONResponse:
    c = con()
    cities = [r[0] for r in c.execute("SELECT DISTINCT city FROM scene ORDER BY city")]
    return JSONResponse({
        "cities": cities,
        "stats": stats_payload(),
        "dataset": DATASET or "local package",
        "read_only": True,
    })


def stats_payload() -> Dict[str, Any]:
    c = con()
    q = lambda s: c.execute(s).fetchone()[0]
    origins = {r[0] or "unknown": r[1] for r in
               c.execute("SELECT origin_auto, COUNT(*) FROM edge GROUP BY origin_auto")}
    return {
        "scenes": q("SELECT COUNT(*) FROM scene"),
        "nodes": q("SELECT COUNT(*) FROM node"),
        "edges": q("SELECT COUNT(*) FROM edge"),
        "total_items": q("SELECT COUNT(*) FROM node") + q("SELECT COUNT(*) FROM edge"),
        "origins": origins,
        "noted_edges": q("SELECT COUNT(*) FROM edge WHERE notes_auto IS NOT NULL "
                         "AND notes_auto <> '[]'"),
        "orphans": q("SELECT COUNT(*) FROM node WHERE is_orphan_auto=1"),
        "policy": q("SELECT graph_version FROM scene WHERE graph_version IS NOT NULL "
                    "LIMIT 1") or "",
    }


@api.get("/api/stats")
def stats() -> JSONResponse:
    return JSONResponse(stats_payload())


@api.get("/api/scenes")
def scenes(city: str = "", min_nodes: Optional[int] = None,
           orphan: str = "", noted: str = "", origin: str = "",
           q: str = "", order: str = "scene_id",
           limit: int = 60, offset: int = 0) -> JSONResponse:
    where, args = [], []
    if city:
        where.append("s.city = ?")
        args.append(city)
    if min_nodes:
        where.append("s.node_count >= ?")
        args.append(int(min_nodes))
    if q:
        where.append("s.scene_id LIKE ?")
        args.append(f"%{q}%")
    if orphan in ("1", "true"):
        where.append("EXISTS (SELECT 1 FROM node n WHERE n.scene_id=s.scene_id "
                     "AND n.is_orphan_auto=1)")
    if noted in ("1", "true"):
        where.append("EXISTS (SELECT 1 FROM edge e WHERE e.scene_id=s.scene_id "
                     "AND e.notes_auto IS NOT NULL AND e.notes_auto <> '[]')")
    if origin:
        where.append("EXISTS (SELECT 1 FROM edge e WHERE e.scene_id=s.scene_id "
                     "AND e.origin_auto = ?)")
        args.append(origin)
    clause = (" WHERE " + " AND ".join(where)) if where else ""
    order_sql = {"scene_id": "s.scene_id",
                 "nodes_desc": "s.node_count DESC, s.scene_id",
                 "nodes_asc": "s.node_count ASC, s.scene_id"}.get(order, "s.scene_id")
    limit = max(1, min(int(limit), 200))
    rows = con().execute(
        "SELECT s.scene_id, s.city, s.node_count, s.satellite_path, "
        "s.graph_version, s.utm_zone_auto, "
        "(SELECT COUNT(*) FROM edge e WHERE e.scene_id=s.scene_id) AS edge_count, "
        "(SELECT COUNT(*) FROM node n WHERE n.scene_id=s.scene_id AND n.is_orphan_auto=1) "
        " AS orphan_count, "
        "(SELECT COUNT(*) FROM edge e WHERE e.scene_id=s.scene_id "
        " AND e.notes_auto IS NOT NULL AND e.notes_auto <> '[]') AS noted_count, "
        "0 AS revision_count "
        f"FROM scene s{clause} ORDER BY {order_sql} LIMIT ? OFFSET ?",
        args + [limit, max(0, int(offset))])
    total = con().execute(f"SELECT COUNT(*) FROM scene s{clause}", args).fetchone()[0]
    return JSONResponse({"items": [dict(r) for r in rows], "total": total,
                         "limit": limit, "offset": offset})


@api.get("/api/queue")
def queue(city: str = "", note: str = "", kind: str = "",
          limit: int = 60, offset: int = 0) -> JSONResponse:
    """「带标记的」列表。

    merged_policy_0915 没有分诊分层和置信度分数，所以这里不再按分数排序。
    列出的是策略本身认为值得看一眼的：§6 的孤儿节点，和带非阻断标记的边
    （§0 跨水/穿建筑、§4 绕远、无道路通路）。
    """
    parts, args = [], []
    limit = max(1, min(int(limit), 200))
    for tbl in ("node", "edge"):
        if kind and kind != tbl:
            continue
        key = "t.label" if tbl == "node" else "t.source_label || '-' || t.target_label"
        sub: List[Any] = []
        if tbl == "node":
            if note and note != "orphan":
                continue
            cond = ["t.is_orphan_auto = 1"]
            reason = ("json_array(json_object('code','orphan','snap_m',"
                      "t.snap_distance_m_auto))")
        else:
            if note == "orphan":
                continue
            cond = ["t.notes_auto IS NOT NULL", "t.notes_auto <> '[]'"]
            reason = "t.notes_auto"
            if note:
                cond.append("EXISTS (SELECT 1 FROM json_each(t.notes_auto) je "
                            "WHERE json_extract(je.value,'$.code') = ?)")
                sub.append(note)
        if city:
            cond.append("s.city = ?")
            sub.append(city)
        parts.append(
            f"SELECT '{tbl}' AS kind, t.scene_id AS scene_id, s.city AS city, "
            f"{key} AS item_key, {reason} AS notes_json, "
            "'unreviewed' AS review_status "
            f"FROM {tbl} t JOIN scene s ON s.scene_id=t.scene_id "
            "WHERE " + " AND ".join(cond))
        args += sub
    if not parts:
        return JSONResponse({"items": [], "limit": limit, "offset": offset})
    sql = (" UNION ALL ".join(parts) +
           " ORDER BY scene_id, kind, item_key LIMIT ? OFFSET ?")
    items = []
    for r in con().execute(sql, args + [limit, max(0, int(offset))]):
        d = dict(r)
        d["notes"] = _loads(d.pop("notes_json")) or []
        items.append(d)
    return JSONResponse({"items": items, "limit": limit, "offset": offset})


@api.get("/api/scene/{scene_id:path}")
def scene(scene_id: str) -> JSONResponse:
    c = con()
    s = c.execute("SELECT * FROM scene WHERE scene_id=?", (scene_id,)).fetchone()
    if s is None:
        raise HTTPException(404, f"无此场景 {scene_id}")
    nodes = [dict(r) for r in c.execute(
        "SELECT * FROM node WHERE scene_id=? ORDER BY label", (scene_id,))]
    edges = [dict(r) for r in c.execute(
        "SELECT * FROM edge WHERE scene_id=? ORDER BY source_label, target_label",
        (scene_id,))]
    effective(nodes, edges)
    # 街景路径：打包时建的索引，运行时猜不出文件名（含 alt/year/heading）。
    # 每个 pano 有 4 个朝向，全部返回；按 heading 排序让前端的顺序稳定。
    sql = ("SELECT rel, heading FROM ground_index WHERE panoid=?"
           if ground_has_heading(c) else
           "SELECT rel, NULL FROM ground_index WHERE panoid=?")

    def order(row) -> tuple:
        # 新包里 prepare.py 存的已经是修正过的视线方向（[0,360)）；旧包存的是
        # 文件名原值，带负角且差 180。这里的归一化只保证排序稳定，**修不了旧包
        # 的 180 度错误**——旧包必须用 prepare.py 重新打过才是对的。
        h = row[1]
        return (1, 0.0, row[0]) if h is None else (0, (float(h) % 360 + 360) % 360, row[0])

    for n in nodes:
        rows = sorted(c.execute(sql, (n["panoid"],)).fetchall(), key=order)
        n["ground_assets"] = [r[0] for r in rows]
        n["ground_headings"] = [r[1] for r in rows]
        n["ground_asset"] = rows[0][0] if rows else None
    return JSONResponse({"scene": dict(s), "nodes": nodes, "edges": edges,
                         "asset": {"satellite": dict(s)["satellite_path"]}})


@api.get("/api/revisions")
def revisions(table: str = "", key: str = "", limit: int = 50) -> JSONResponse:
    # 只读包不带修订历史，返回空让前端正常渲染
    return JSONResponse({"items": []})


@api.get("/asset")
def serve_asset(path: str = Query(...)) -> FileResponse:
    p = asset_path(path)
    if p is None:
        raise HTTPException(404, "取不到该图片")
    return FileResponse(p)


app = api


if __name__ == "__main__":
    import argparse
    import uvicorn
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=7860)
    ap.add_argument("--host", default="0.0.0.0")
    a = ap.parse_args()
    uvicorn.run(app, host=a.host, port=a.port, log_level="info")