rond-with-you/tools/dedup_site_visits.py

#!/usr/bin/env python3
"""清掉「同一次到访在站点库里存了两份」的历史副本。

背景（2026-09-29）：站点数据里存在一批**到达时间 + 地点与手机包里某条完全相同、但 UUID 不同**的
到访副本（站点侧生成的 UUID）。差异页/合并按 UUID 认人，于是它们会被当成「备份新增」，
导出到 rond 后就是成对的两条重复记录。

本工具以**手机导出的包**为参照（也就是这份数据集从手机那儿来的那个 store，即数据集的基线）：
  * 站点到访的 UUID **在参照里能找到** → 保留（手机自己也有成对条目，那是合法的，不能动）；
  * UUID **找不到**、且 (到达时间, 地点) 与参照里某条一致 → 判为副本。
副本的附属记录（位移 / 天气 / 留档）一并删除；若有**别的到访的留档**指向副本（ZMERGEDTO / ZPARENT），
改指到保留下来的那条，避免导出时留下悬空引用。

默认**只预演**（只读，不改库）：列出将删除的行。确认无误后加 --apply 真正执行。

用法：
  python tools/dedup_site_visits.py --dataset 9 --baseline data/20260928_183908.rondbackup
  python tools/dedup_site_visits.py --dataset 9 --baseline data/20260928_183908.rondbackup --apply
"""
import argparse
import os
import sqlite3
import subprocess
import sys
import tempfile
import zipfile


def psql(args, sql, want_rows=False):
    cmd = ["psql", "-U", args.user, "-h", args.host, "-p", str(args.port), "-d", args.db, "-v", "ON_ERROR_STOP=1"]
    cmd += ["-A", "-F", "\t", "-t"] if want_rows else []
    cmd += ["-c", sql]
    r = subprocess.run(cmd, capture_output=True, text=True, env=dict(os.environ, PGPASSWORD=args.password))
    if r.returncode:
        print("SQL 失败:", r.stderr.strip()[:400], file=sys.stderr)
        sys.exit(1)
    return r.stdout


def load_baseline(path):
    """读手机包的到访身份：UUID 集合 + (到达时间, 地点坐标) → UUID。"""
    d = tempfile.mkdtemp(prefix="dedup-base")
    with zipfile.ZipFile(path) as z:
        z.extractall(d)
    main = [os.path.join(r, f) for r, _, fs in os.walk(d) for f in fs if f.endswith(".sqlite")][0]
    con = sqlite3.connect(main)
    con.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    con.commit()
    places = {pk: (name or "", lat or 0.0, lon or 0.0)
              for pk, name, lat, lon in con.execute("SELECT Z_PK, ZNAME_, ZLATITUDE, ZLONGITUDE FROM ZLOCATION")}
    uuids, by_key = set(), {}
    for uuid, arr, loc in con.execute(
            "SELECT ZIDENTIFIER_, ZARRIVALDATE_, ZLOCATION FROM ZVISIT WHERE ZIDENTIFIER_ IS NOT NULL"):
        u = bytes(uuid).hex()
        uuids.add(u)
        if arr is not None and loc in places:
            name, lat, lon = places[loc]
            by_key.setdefault((round(arr, 3), name), u)
            if lat or lon:
                by_key.setdefault((round(arr, 3), "%.6f|%.6f" % (lat, lon)), u)
    con.close()
    return uuids, by_key


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--dataset", type=int, required=True, help="数据集 id")
    ap.add_argument("--baseline", required=True, help="手机导出的 .rondbackup（参照）")
    ap.add_argument("--db", default="rond_with_you")
    ap.add_argument("--user", default="postgres")
    ap.add_argument("--password", default=os.environ.get("PGPASSWORD", ""))
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=5432)
    ap.add_argument("--apply", action="store_true", help="真正删除（默认只预演）")
    args = ap.parse_args()

    uuids, by_key = load_baseline(args.baseline)
    print("参照（手机包）：到访 %d 条（带 UUID）" % len(uuids))

    # 站点侧到访：src_pk / UUID / 到达时间 / 地点名 / 地点坐标
    rows = psql(args, """
        SELECT v.src_pk,
               COALESCE(er.raw #>> '{ZIDENTIFIER_,$blob}',''),
               COALESCE(to_char(v.arrival AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US'),''),
               COALESCE(p.name,''), COALESCE(p.lat,0), COALESCE(p.lon,0),
               (er.raw #>> '{ZMERGEDTO}'), (er.raw #>> '{ZPARENT}')
        FROM visits v
        LEFT JOIN places p ON p.id = v.place_id
        LEFT JOIN entity_raw er ON er.dataset_id = v.dataset_id AND er.entity='ZVISIT' AND er.src_pk = v.src_pk
        WHERE v.dataset_id = %d AND v.arrival IS NOT NULL
        ORDER BY v.src_pk""" % args.dataset, want_rows=True)

    import base64
    from datetime import datetime, timezone

    def apple_secs(iso):
        if not iso:
            return None
        dt = datetime.strptime(iso[:26], "%Y-%m-%dT%H:%M:%S.%f").replace(tzinfo=timezone.utc)
        return (dt - datetime(2001, 1, 1, tzinfo=timezone.utc)).total_seconds()

    visits, victims = [], []
    for line in rows.strip().split("\n"):
        if not line.strip():
            continue
        f = line.split("\t")
        pk, b64, iso, pname, lat, lon = int(f[0]), f[1], f[2], f[3], float(f[4]), float(f[5])
        uuid = base64.b64decode(b64).hex() if b64 else ""
        visits.append((pk, uuid, apple_secs(iso), pname, lat, lon, f[6] if len(f) > 6 else "", f[7] if len(f) > 7 else ""))
    keep_by_key = {}
    for pk, uuid, arr, pname, lat, lon, _, _ in visits:
        if uuid in uuids and arr is not None:
            keep_by_key.setdefault((round(arr, 3), pname), pk)
            if lat or lon:
                keep_by_key.setdefault((round(arr, 3), "%.6f|%.6f" % (lat, lon)), pk)
    for pk, uuid, arr, pname, lat, lon, merged, parent in visits:
        if not uuid or uuid in uuids or arr is None:
            continue
        keep = by_key.get((round(arr, 3), pname)) or by_key.get((round(arr, 3), "%.6f|%.6f" % (lat, lon)))
        keep_pk = keep_by_key.get((round(arr, 3), pname)) or keep_by_key.get((round(arr, 3), "%.6f|%.6f" % (lat, lon)))
        if keep and keep_pk:
            victims.append((pk, arr, pname, keep_pk))

    print("站点侧副本：%d 条" % len(victims))
    for pk, arr, pname, keep_pk in victims[:20]:
        ts = datetime(2001, 1, 1, tzinfo=timezone.utc).timestamp() + arr
        print("   src_pk=%-6d %s  %-22s → 保留 src_pk=%d" % (
            pk, datetime.fromtimestamp(ts, timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M"), pname, keep_pk))
    if len(victims) > 20:
        print("   …… 其余 %d 条" % (len(victims) - 20))
    if not victims:
        print("没有发现副本 ✓")
        return
    if not args.apply:
        print("\n（预演结束，未改动任何数据。确认无误后加 --apply 执行）")
        return

    pks = ",".join(str(v[0]) for v in victims)
    keep = {str(v[0]): str(v[3]) for v in victims}
    # 1) 别的到访的留档若指向副本，改指到保留项（否则导出会留下悬空引用）
    # 注意：raw->> 取出的是 text，必须按字符串比较（拿数字去比会报「text = integer 操作符不存在」，
    # 那样工具会在动手前整个退出 —— 第一版就是这么失败了）
    for victim, survivor in keep.items():
        psql(args, """UPDATE entity_raw
                    SET raw = jsonb_set(jsonb_set(raw, '{ZMERGEDTO}', to_jsonb(%s::int), true),
                                        '{ZPARENT}', to_jsonb(%s::int), true)
                    WHERE dataset_id=%d AND entity='ZVISIT'
                      AND ((raw->>'ZMERGEDTO') = '%s' OR (raw->>'ZPARENT') = '%s')"""
             % (survivor, survivor, args.dataset, victim, victim))
    # 2) 附属记录：位移、天气、留档
    psql(args, "DELETE FROM movements WHERE dataset_id=%d AND (from_visit_src IN (%s) OR to_visit_src IN (%s))"
         % (args.dataset, pks, pks))
    psql(args, "DELETE FROM weather WHERE dataset_id=%d AND visit_src IN (%s)" % (args.dataset, pks))
    psql(args, "DELETE FROM entity_raw WHERE dataset_id=%d AND entity='ZVISIT' AND src_pk IN (%s)" % (args.dataset, pks))
    psql(args, "DELETE FROM visits WHERE dataset_id=%d AND src_pk IN (%s)" % (args.dataset, pks))
    print("已删除 %d 条副本及其附属记录" % len(victims))


if __name__ == "__main__":
    main()