站点库里有一批「到达时间 + 地点与手机包里某条完全相同、但 UUID 不同」的到访副本, 合并/差异按 UUID 认人会把它们当成备份新增,导出到 rond 就成了成对的两条重复记录。 判定与 tools/dedup_site_visits.py 同一套语义:UUID 是前提(只按时间+地点判会误删手机包 自身合法的成对条目);基线跟本站一条都对不上时只给说明、不给按钮。清理在单事务里删 副本 + 其位移/天气/留档,并把指向副本的 ZMERGEDTO/ZPARENT 改指到保留项——只动本来 就有的那个键,工具那版会凭空补出一个 ZMERGEDTO(真机里是「已并入别处」,会白丢记录)。 顺带修一个「一补录天气就整包导出失败」的洞:rond 导出把 weather.is_daylight 扫进裸 bool, 而站点「新增天气」不写这一列 ⇒ NULL ⇒ sql.Scan 报错。改成 COALESCE(is_daylight,false), 与导入侧读真机包同口径。 验收:301 PASS / 0 FAIL / 0 SKIP(新增 4 个用例,含变异验证:去掉 UUID 前提/改回 create_if_missing=true/不删副本位移,三种改法都被用例抓到)。
402 lines
14 KiB
Go
402 lines
14 KiB
Go
package web
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/base64"
|
||
"fmt"
|
||
"log"
|
||
"math"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
// ---------- 重复到访:站点侧的历史副本 ----------
|
||
//
|
||
// 背景(2026-09-29):站点库里有一批**到达时间 + 地点与手机包里某条完全相同、但 UUID 不同**
|
||
// 的到访副本。差异页 / 合并都是按 UUID 认人的,于是它们被判成「备份新增」,
|
||
// 导出到 rond 后就是成对的两条重复记录。
|
||
//
|
||
// 判定与 tools/dedup_site_visits.py 是同一套语义(那里已经演练通过,两边不能各写一份):
|
||
// - 参照是这份数据集的**基线**(导入时留下的手机包底稿,见 baselinePath);
|
||
// - 站点到访的 UUID 在基线里找不到、而基线里有一条「到达时间 + 地点」完全相同的 → 副本。
|
||
//
|
||
// UUID 是前提:只按「时间 + 地点」判,会把手机包自身那些合法的成对条目一起删掉。
|
||
|
||
// appleEpoch 是 Core Data 的时间原点(2001-01-01 UTC)。基线的 ZARRIVALDATE_ 就是它起算的秒数。
|
||
var appleEpoch = time.Date(2001, 1, 1, 0, 0, 0, 0, time.UTC)
|
||
|
||
// EditDupVisit 是「重复到访」卡片里的一对:站点侧的副本,与它该保留的那条(手机包里也有)。
|
||
type EditDupVisit struct {
|
||
Victim int // 站点侧副本的 src_pk —— 勾选后删的就是它
|
||
Keep int // 保留项的 src_pk —— UUID 在基线里能找到的那条
|
||
Arrival time.Time
|
||
PlaceName string
|
||
}
|
||
|
||
// ArrivalStr 是模板里显示的到达时间(本地时区,分钟精度)。
|
||
func (v EditDupVisit) ArrivalStr() string { return v.Arrival.Local().Format("2006-01-02 15:04") }
|
||
|
||
// dupKey 把「到达时间 + 地点」折成一个比较键。时间取到毫秒:副本与原件的到达时间本来
|
||
// 就是同一个瞬间,而基线与站点两侧都用这一个函数,取整方式一致就不会错配。
|
||
func dupKey(secs float64, who string) string {
|
||
return strconv.FormatInt(int64(math.Round(secs*1000)), 10) + "|" + who
|
||
}
|
||
|
||
// dupCoordKey 是地点坐标的比较键(6 位小数,与工具一致)。
|
||
func dupCoordKey(lat, lon float64) string { return fmt.Sprintf("%.6f|%.6f", lat, lon) }
|
||
|
||
// appleSeconds 把时间换算成 Core Data 的秒数(2001-01-01 UTC 起算),与基线的 ZARRIVALDATE_ 同系。
|
||
func appleSeconds(t time.Time) float64 { return t.UTC().Sub(appleEpoch).Seconds() }
|
||
|
||
// dupSet 只在键还没被占时写入。Python 那边用的是 dict.setdefault:同一个「时间 + 地点」
|
||
// 在库里有多条时取先到的那条,顺序照抄,两边结果才一致。
|
||
func dupSet[T any](m map[string]T, k string, v T) {
|
||
if _, ok := m[k]; !ok {
|
||
m[k] = v
|
||
}
|
||
}
|
||
|
||
// dupBaseline 是从基线库读出的判定依据。
|
||
type dupBaseline struct {
|
||
uuids map[string]bool // 基线里出现过的到访 UUID(小写 hex)
|
||
byKey map[string]string // 「到达时间 + 地点」→ 基线里那条的 UUID
|
||
}
|
||
|
||
// dupPlace 是基线里一个地点的名字与坐标。
|
||
type dupPlace struct {
|
||
name string
|
||
lat, lon float64
|
||
}
|
||
|
||
// readDupBaseline 只读打开基线库(Core Data 单文件),取出判定要用的两组信息。
|
||
//
|
||
// 驱动只认 _pragma / _time_format 这几个 DSN 参数,没有 mode=ro;基线库是 rollback journal
|
||
// 模式(VACUUM INTO 的产物),只做 SELECT 不会落下任何旁写文件,再加 query_only 兜住
|
||
// 「误写基线库」这条路——基线是导出的底稿,写坏了整个数据集就导不出去了。
|
||
func readDupBaseline(ctx context.Context, path string) (*dupBaseline, error) {
|
||
sdb, err := sql.Open("sqlite", "file:"+filepath.ToSlash(path)+"?_pragma=query_only(1)")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer sdb.Close()
|
||
sdb.SetMaxOpenConns(1)
|
||
|
||
places, err := readDupPlaces(ctx, sdb)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
b := &dupBaseline{uuids: map[string]bool{}, byKey: map[string]string{}}
|
||
rows, err := sdb.QueryContext(ctx, `SELECT hex(ZIDENTIFIER_), ZARRIVALDATE_, ZLOCATION
|
||
FROM ZVISIT WHERE ZIDENTIFIER_ IS NOT NULL`)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var uuid string
|
||
var arrival sql.NullFloat64
|
||
var loc sql.NullInt64
|
||
if err := rows.Scan(&uuid, &arrival, &loc); err != nil {
|
||
return nil, err
|
||
}
|
||
uuid = strings.ToLower(uuid)
|
||
if uuid == "" {
|
||
continue
|
||
}
|
||
b.uuids[uuid] = true
|
||
if !arrival.Valid || !loc.Valid {
|
||
continue
|
||
}
|
||
pl, ok := places[loc.Int64]
|
||
if !ok {
|
||
continue
|
||
}
|
||
dupSet(b.byKey, dupKey(arrival.Float64, pl.name), uuid)
|
||
if pl.lat != 0 || pl.lon != 0 {
|
||
dupSet(b.byKey, dupKey(arrival.Float64, dupCoordKey(pl.lat, pl.lon)), uuid)
|
||
}
|
||
}
|
||
return b, rows.Err()
|
||
}
|
||
|
||
func readDupPlaces(ctx context.Context, sdb *sql.DB) (map[int64]dupPlace, error) {
|
||
rows, err := sdb.QueryContext(ctx, `SELECT Z_PK, COALESCE(ZNAME_,''),
|
||
COALESCE(ZLATITUDE,0), COALESCE(ZLONGITUDE,0) FROM ZLOCATION`)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := map[int64]dupPlace{}
|
||
for rows.Next() {
|
||
var pk int64
|
||
var pl dupPlace
|
||
if err := rows.Scan(&pk, &pl.name, &pl.lat, &pl.lon); err != nil {
|
||
return nil, err
|
||
}
|
||
out[pk] = pl
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// siteDupVisit 是判定用的站点侧面貌。
|
||
type siteDupVisit struct {
|
||
srcPK int
|
||
uuid string // 留档里的 ZIDENTIFIER_(小写 hex),站点新建的行没有
|
||
arrival time.Time
|
||
place string
|
||
lat, lon float64
|
||
}
|
||
|
||
// loadDupSiteVisits 读出站点侧的到访(含留档里的 UUID 与所属地点的名称/坐标),按 src_pk 排序。
|
||
func (s *Server) loadDupSiteVisits(ctx context.Context, datasetID int64) ([]siteDupVisit, error) {
|
||
rows, err := s.db.QueryContext(ctx, `SELECT v.src_pk,
|
||
COALESCE(er.raw #>> '{ZIDENTIFIER_,$blob}',''),
|
||
v.arrival, COALESCE(p.name,''), COALESCE(p.lat,0), COALESCE(p.lon,0)
|
||
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=$1
|
||
ORDER BY v.src_pk`, datasetID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
var out []siteDupVisit
|
||
for rows.Next() {
|
||
var sv siteDupVisit
|
||
var b64 string
|
||
if err := rows.Scan(&sv.srcPK, &b64, &sv.arrival, &sv.place, &sv.lat, &sv.lon); err != nil {
|
||
return nil, err
|
||
}
|
||
sv.uuid = blobHex(b64)
|
||
out = append(out, sv)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// blobHex 把留档里的 base64 blob 转成小写 hex。基线侧用的是 hex(ZIDENTIFIER_),
|
||
// 两边同一个形式才比得上(大小写也要一致,否则一条都认不出来)。
|
||
func blobHex(s string) string {
|
||
if s == "" {
|
||
return ""
|
||
}
|
||
b, err := base64.StdEncoding.DecodeString(s)
|
||
if err != nil || len(b) == 0 {
|
||
return ""
|
||
}
|
||
return strings.ToLower(fmt.Sprintf("%x", b))
|
||
}
|
||
|
||
// detectDuplicateVisits 检出站点侧的重复到访,并给出每条该保留哪一行。
|
||
//
|
||
// 第二个返回值非空表示**判定不了**(基线不可用 / 跟本站到访对不上),此时页面只给一行
|
||
// 说明、不给按钮——宁可不动,也不能拿一份对不上的参照去删用户的数据。
|
||
func (s *Server) detectDuplicateVisits(ctx context.Context, datasetID int64) ([]EditDupVisit, string, error) {
|
||
path := s.baselinePath(datasetID)
|
||
if _, err := os.Stat(path); err != nil {
|
||
return nil, "这份数据集还没有可用的基线库(当初上传的原始 .rondbackup 已不在),没法当参照,判定不了重复到访。", nil
|
||
}
|
||
base, err := readDupBaseline(ctx, path)
|
||
if err != nil {
|
||
log.Printf("读取基线库失败(数据集 %d,%s): %v", datasetID, path, err)
|
||
return nil, "读取基线库失败,判定不了重复到访。", nil
|
||
}
|
||
site, err := s.loadDupSiteVisits(ctx, datasetID)
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
|
||
var withUUID, matched int
|
||
for _, sv := range site {
|
||
if sv.uuid == "" {
|
||
continue
|
||
}
|
||
withUUID++
|
||
if base.uuids[sv.uuid] {
|
||
matched++
|
||
}
|
||
}
|
||
// 两道闸都防着「拿一份不相干的库当参照」:基线的 UUID 一个都比不上时,
|
||
// 所有站内到访都会被算成副本,照着删就是把整库搬走。
|
||
if len(base.uuids) == 0 {
|
||
return nil, "基线里没有带 UUID 的到访,没法当参照,判定不了重复到访。", nil
|
||
}
|
||
if withUUID > 0 && matched == 0 {
|
||
return nil, "基线跟本站到访一条都对不上,它多半不是这份数据集的手机包,判定不可靠,先不动它。", nil
|
||
}
|
||
|
||
// 保留项:站点侧 UUID 在基线里能找到的那些(手机包里也有这条),按「时间 + 地点」建索引。
|
||
keepByKey := map[string]int{}
|
||
for _, sv := range site {
|
||
if sv.uuid == "" || !base.uuids[sv.uuid] {
|
||
continue
|
||
}
|
||
secs := appleSeconds(sv.arrival)
|
||
dupSet(keepByKey, dupKey(secs, sv.place), sv.srcPK)
|
||
if sv.lat != 0 || sv.lon != 0 {
|
||
dupSet(keepByKey, dupKey(secs, dupCoordKey(sv.lat, sv.lon)), sv.srcPK)
|
||
}
|
||
}
|
||
|
||
var out []EditDupVisit
|
||
for _, sv := range site {
|
||
if sv.uuid == "" || base.uuids[sv.uuid] {
|
||
continue
|
||
}
|
||
secs := appleSeconds(sv.arrival)
|
||
byName, byCoord := dupKey(secs, sv.place), ""
|
||
if sv.lat != 0 || sv.lon != 0 {
|
||
byCoord = dupKey(secs, dupCoordKey(sv.lat, sv.lon))
|
||
}
|
||
// 基线里得有「同一时间同一地点」那条,否则这不是副本,是站点自己新增的到访
|
||
if base.byKey[byName] == "" && (byCoord == "" || base.byKey[byCoord] == "") {
|
||
continue
|
||
}
|
||
// 站点里也得有那条保留项:没有就无从改指,删了就是净丢一条记录
|
||
keep, ok := keepByKey[byName]
|
||
if !ok && byCoord != "" {
|
||
keep, ok = keepByKey[byCoord]
|
||
}
|
||
if !ok {
|
||
continue
|
||
}
|
||
out = append(out, EditDupVisit{Victim: sv.srcPK, Keep: keep, Arrival: sv.arrival, PlaceName: sv.place})
|
||
}
|
||
sort.SliceStable(out, func(i, j int) bool { return out[i].Arrival.After(out[j].Arrival) })
|
||
return out, "", nil
|
||
}
|
||
|
||
// editDuplicateVisitsApply 清理勾选的重复到访副本。
|
||
func (s *Server) editDuplicateVisitsApply(w http.ResponseWriter, r *http.Request) {
|
||
ctx := r.Context()
|
||
ds, err := s.datasetFor(ctx)
|
||
if err != nil || ds == nil {
|
||
redirectEdit(w, r, "err", "还没有可编辑的数据集")
|
||
return
|
||
}
|
||
if err := r.ParseForm(); err != nil {
|
||
s.serverError(w, r, err)
|
||
return
|
||
}
|
||
want := map[int]bool{}
|
||
for _, v := range r.PostForm["victim"] {
|
||
if n, cerr := strconv.Atoi(strings.TrimSpace(v)); cerr == nil && n > 0 {
|
||
want[n] = true
|
||
}
|
||
}
|
||
if len(want) == 0 {
|
||
redirectEdit(w, r, "err", "没有勾选要清理的副本")
|
||
return
|
||
}
|
||
// 编号不照着表单删:渲染到点击之间库可能已经变过,这里重算一遍判定,
|
||
// 只有当下**仍被判为副本**的编号才会动。
|
||
pairs, unavailable, err := s.detectDuplicateVisits(ctx, ds.ID)
|
||
if err != nil {
|
||
s.serverError(w, r, err)
|
||
return
|
||
}
|
||
if unavailable != "" {
|
||
redirectEdit(w, r, "err", unavailable)
|
||
return
|
||
}
|
||
var picked []EditDupVisit
|
||
for _, p := range pairs {
|
||
if want[p.Victim] {
|
||
picked = append(picked, p)
|
||
}
|
||
}
|
||
if len(picked) == 0 {
|
||
redirectEdit(w, r, "err", "选中的记录已经不是副本了(库里可能已经变过),请刷新页面重试")
|
||
return
|
||
}
|
||
if err := s.withTx(ctx, func(ex execer) error {
|
||
return s.cleanDuplicateVisits(ctx, ex, ds.ID, picked)
|
||
}); err != nil {
|
||
s.serverError(w, r, err)
|
||
return
|
||
}
|
||
redirectEdit(w, r, "ok", fmt.Sprintf("已清理 %d 条重复到访", len(picked)))
|
||
}
|
||
|
||
// cleanDuplicateVisits 删掉一批副本,并把指向它们的合并标记改指到保留项。
|
||
//
|
||
// 全程在一个事务里:分开提交会留下「到访没了、留档还指着它」的悬空引用,
|
||
// 导出到 rond 里就是坏数据(到访那条 ZMERGEDTO 指向一个不存在的 Z_PK)。
|
||
func (s *Server) cleanDuplicateVisits(ctx context.Context, ex execer, datasetID int64, pairs []EditDupVisit) error {
|
||
victims := make([]int64, 0, len(pairs))
|
||
for _, p := range pairs {
|
||
victims = append(victims, int64(p.Victim))
|
||
}
|
||
// 聚合要按地点重算,先把副本所属地点收下来(删完就在 visits 里查不到了)
|
||
placeIDs, err := s.dupVictimPlaces(ctx, ex, datasetID, victims)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 改指合并标记时**只动本来就存在的那一个键**(jsonb_set 的 create_if_missing=false),
|
||
// 与 tools/dedup_site_visits.py 有一处刻意的不同:工具把两个键都写成保留项,于是
|
||
// 「只有 ZPARENT 指向副本」的行会凭空长出一个 ZMERGEDTO——那在真机里是「这条已并入
|
||
// 别处」的意思,会白丢一条记录。实测基线里 956 条留档都带着这两个键,但只有 54 / 10
|
||
// 条有值,凭空补键是很容易踩到的。
|
||
//
|
||
// raw->> 取出来是 text,必须按字符串比(拿数字去比会报「text = integer 操作符不存在」)。
|
||
for _, p := range pairs {
|
||
for _, col := range []string{"ZMERGEDTO", "ZPARENT"} {
|
||
if _, err := ex.ExecContext(ctx, `UPDATE entity_raw
|
||
SET raw = jsonb_set(raw, ARRAY[$4::text], to_jsonb($3::int), false)
|
||
WHERE dataset_id=$1 AND entity='ZVISIT' AND raw->>$4 = $2`,
|
||
datasetID, strconv.Itoa(p.Victim), p.Keep, col); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
// 副本的附属记录:位移(两端任一)、天气。到访本身删掉后,它自己的留档与标签关联
|
||
// 分别由 pruneEntityRaw 与 visit_tags 的外键级联清掉。
|
||
if _, err := ex.ExecContext(ctx, `DELETE FROM movements WHERE dataset_id=$1
|
||
AND (from_visit_src = ANY($2::bigint[]) OR to_visit_src = ANY($2::bigint[]))`,
|
||
datasetID, victims); err != nil {
|
||
return err
|
||
}
|
||
if _, err := ex.ExecContext(ctx, `DELETE FROM weather WHERE dataset_id=$1 AND visit_src = ANY($2::bigint[])`,
|
||
datasetID, victims); err != nil {
|
||
return err
|
||
}
|
||
if _, err := ex.ExecContext(ctx, `DELETE FROM visits WHERE dataset_id=$1 AND src_pk = ANY($2::bigint[])`,
|
||
datasetID, victims); err != nil {
|
||
return err
|
||
}
|
||
s.pruneEntityRaw(ctx, ex, datasetID)
|
||
for _, id := range placeIDs {
|
||
if err := s.refreshPlaceAgg(ctx, ex, id); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return s.refreshDatasetAgg(ctx, ex, datasetID)
|
||
}
|
||
|
||
// dupVictimPlaces 是这批副本所属的地点 id(不含 place_id 为空的)。
|
||
func (s *Server) dupVictimPlaces(ctx context.Context, ex execer, datasetID int64, victims []int64) ([]int64, error) {
|
||
rows, err := ex.QueryContext(ctx, `SELECT DISTINCT place_id FROM visits
|
||
WHERE dataset_id=$1 AND src_pk = ANY($2::bigint[]) AND place_id IS NOT NULL`, datasetID, victims)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
var out []int64
|
||
for rows.Next() {
|
||
var id int64
|
||
if err := rows.Scan(&id); err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, id)
|
||
}
|
||
return out, rows.Err()
|
||
}
|