工具：tools/dedup_site_visits.py —— 以手机包为参照清掉站点侧的历史重复到访
站点数据里有一批「到达时间 + 地点与手机包里某条完全相同、但 UUID 不同」的到访副本（站点侧生成的 UUID）。差异页/合并按 UUID 认人，于是它们被当成「备份新增」，导出到 rond 后就是成对的两条重复记录 （2026-09-29 站长实测：导入后 app 里出现成对的重复条目）。 工具以手机导出的包为参照（数据集的基线）： * 站点到访的 UUID 在参照里能找到 → 保留（手机自己也有成对条目，那是合法的，不能动） * UUID 找不到、且 (到达时间, 地点) 与参照里某条一致 → 判为副本 副本的附属记录（位移/天气/留档）一并删除；若有别的到访的留档指向副本（ZMERGEDTO / ZPARENT）， 改指到保留的那条，避免导出时留下悬空引用。**默认只预演，--apply 才动手。** 踩坑记录（都在此提交里修掉）： * `raw->>` 取出的是 text，与数字比较会报「text = integer 操作符不存在」，导致工具在动手前整个退出； * 同一个 UPDATE 里不能给 raw 赋两次值（「对同一列 raw 进行了多次分配」），jsonb_set 要嵌套。 演练验证（本地真实数据：导入含 72 条副本的包 → 执行 → 核对）： 到访 1026 → 954、位移 833 → 809、天气 621 → 589（与手机包一致）； 「同时间同地点」重复组 101 → **31**（= 手机包自身）、悬空引用 0/0。
This commit is contained in:
tamakyi
@@ -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()
|
||||
Reference in New Issue
Block a user