diff --git a/tools/dedup_site_visits.py b/tools/dedup_site_visits.py new file mode 100644 index 0000000..1f54ee2 --- /dev/null +++ b/tools/dedup_site_visits.py @@ -0,0 +1,160 @@ +#!/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()