导出是「底座为底 + 只覆盖托管表」,所以底座必须是手机导出的包。底座跟着数据集走, 手机重新导出后想换代,以前只能整包重新导入成新数据集(站点补录的行、主题、备注都得重收拾)。 新动作 /admin/baseline(后台首页加入口)就是给同一份数据的底座换代用的。 三道闸,顺序不能挪: 1. 重合判据:新包里的到访 UUID 与站点侧带 UUID 的到访比。一条都对不上直接拒 (连 force 也不给过)——干跑只验结构,一份不相干的包照样导得出,但换下去站点 每行都在新底座里认不出身份,全按新行插到最大编号之后,真机上就是「这些到访 全变成新的」。重合率不到一半要显式确认;判据没过不当场报错了事:包已经收进 uploads,页面给一个「我确认,强制替换」按钮,不必重传几十 MB。 2. 干跑导出:底座缺表/缺实体编号在这步挡住,不会出现「换完了才发现导不出、 原来那份已经不知去哪」。 3. 旧底座留档 + 记档,换错了可一键回滚。 两个坑: - 留档名不能只用秒级时间戳。「替换后紧接着回滚」落在同一秒时,回滚那次 rename 会把 刚生成的留档覆盖掉,于是回滚回了个寂寞(TestRollbackBaselineRestoresPrevious 当场红)。 改用 baselineBackupName():20060102-150405 + 冲突自增。 - ensureBaseline 必须先看记档。只认 datasets.sha256(= 最初那份包)的话,底座一丢就静默 重建回旧包,上一次换基线白做且毫无提示。抽出 ensureBaselineTo(ctx, dsID, dst) 共用; 「重建」入口写临时文件再 rename,不做破坏性删除。 回归用例 7 条(internal/web/baseline_swap_test.go):成功路径断言「换底座前后导出包里 ZVISIT 的 UUID 集合逐条一致」——编号会换、身份不该换,谁在底座里没对上立刻亮; 记档优先重建带反证(清掉记档后同一条路必须走不通);部分重合需确认、完全对不上 force 也不给过、非包文件第一步就挡住、回滚还原且当前那份也留档、页面渲染到底。 验收:./scripts/verify.sh 全量 PASS=318 FAIL=0 SKIP=0。
469 lines
18 KiB
Go
469 lines
18 KiB
Go
package web
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/base64"
|
||
"fmt"
|
||
"log"
|
||
"math"
|
||
"net/http"
|
||
"os"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"rond-with-you/internal/rond"
|
||
)
|
||
|
||
// ---------- 重复到访:站点侧的历史副本 ----------
|
||
//
|
||
// 背景(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)
|
||
|
||
// 两类重复的成因(卡片上打个小标,站长才好判断该不该删)。
|
||
const (
|
||
// 站点侧副本:手机包里也有「同一时刻同一地点」的一条,但 UUID 不同
|
||
dupSiteCopy = "站点侧副本"
|
||
// 补录行被合并回来:站点补录的行没有留档 UUID,导出时会被补一个派生 UUID,
|
||
// 这份包再合并回站点时对不上,就被当成「备份新增」又插了一条
|
||
dupHintMergedBack = "你补录的那条被合并回来的一份"
|
||
)
|
||
|
||
// EditDupVisit 是「重复到访」卡片里的一对:站点侧多出来的那条,与它该保留的那条。
|
||
type EditDupVisit struct {
|
||
Victim int // 站点侧多出来的 src_pk —— 勾选后删的就是它
|
||
Keep int // 要留下来的 src_pk
|
||
Arrival time.Time
|
||
PlaceName string
|
||
// Reason 是这一对的成因(卡片上打标用)
|
||
Reason string
|
||
// MoveAssets 表示删副本时把它名下的位移 / 天气**改挂**到保留项(而不是删掉):
|
||
// 「补录行的副本」属于这种——那段行程是真实发生过的,而补录行自己名下没有位移。
|
||
MoveAssets bool
|
||
}
|
||
|
||
// 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 := openBaselineRO(path)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer sdb.Close()
|
||
|
||
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) {
|
||
site, err := s.loadDupSiteVisits(ctx, datasetID)
|
||
if err != nil {
|
||
return nil, "", err
|
||
}
|
||
|
||
var out []EditDupVisit
|
||
seen := map[int]bool{}
|
||
// 第二类:站点补录的行(留档里没有 UUID)导出过一次之后,包里那条带的是站点给它补的
|
||
// 派生 UUID;这份包合并回站点时对不上(站点这侧本来是空的),于是又插了一条。
|
||
// 判据全在站点自己这边(不需要基线),所以先判它,基线不可用时也不受影响。
|
||
// 派生 UUID 命中是很强的证据,排在「站点侧副本」之前。
|
||
soloByUUID := map[string]siteDupVisit{}
|
||
for _, sv := range site {
|
||
if sv.uuid == "" && sv.srcPK > 0 {
|
||
soloByUUID[rond.DerivedVisitUUID(sv.srcPK)] = sv
|
||
}
|
||
}
|
||
for _, sv := range site {
|
||
if ssv, ok := soloByUUID[sv.uuid]; sv.uuid != "" && ok && !seen[sv.srcPK] {
|
||
// 派生 UUID 已经很特异了,这里再对一道时间与地点,免得编号被复用的行配到一起
|
||
if ssv.place != sv.place || dupKey(appleSeconds(ssv.arrival), ssv.place) != dupKey(appleSeconds(sv.arrival), sv.place) {
|
||
continue
|
||
}
|
||
out = append(out, EditDupVisit{Victim: sv.srcPK, Keep: ssv.srcPK, Arrival: sv.arrival,
|
||
PlaceName: sv.place, Reason: dupHintMergedBack, MoveAssets: true})
|
||
seen[sv.srcPK] = true
|
||
}
|
||
}
|
||
|
||
// 以下第一类需要基线:缺失 / 读不动 / 一条 UUID 都对不上,就只报第二类 + 一行说明。
|
||
fail := func(msg string) ([]EditDupVisit, string, error) {
|
||
sort.SliceStable(out, func(i, j int) bool { return out[i].Arrival.After(out[j].Arrival) })
|
||
return out, msg, nil
|
||
}
|
||
path := s.baselinePath(datasetID)
|
||
if _, err := os.Stat(path); err != nil {
|
||
return fail("这份数据集还没有可用的基线库(当初上传的原始 .rondbackup 已不在),没法当参照,判定不了「站点侧副本」那一类。")
|
||
}
|
||
base, err := readDupBaseline(ctx, path)
|
||
if err != nil {
|
||
log.Printf("读取基线库失败(数据集 %d,%s): %v", datasetID, path, err)
|
||
return fail("读取基线库失败,判定不了「站点侧副本」那一类。")
|
||
}
|
||
|
||
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 fail("基线里没有带 UUID 的到访,没法当参照,判定不了「站点侧副本」那一类。")
|
||
}
|
||
if withUUID > 0 && matched == 0 {
|
||
return fail("基线跟本站到访一条都对不上,它多半不是这份数据集的手机包,判定不可靠,先不动它。")
|
||
}
|
||
|
||
// 保留项:站点侧 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)
|
||
}
|
||
}
|
||
|
||
// 第一类:站点的 UUID 在基线里找不到、而基线里有一条「同时刻同地点」的——站点侧副本
|
||
for _, sv := range site {
|
||
if sv.uuid == "" || base.uuids[sv.uuid] || seen[sv.srcPK] {
|
||
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, Reason: dupSiteCopy})
|
||
seen[sv.srcPK] = true
|
||
}
|
||
|
||
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))
|
||
var repoint []EditDupVisit
|
||
for _, p := range pairs {
|
||
victims = append(victims, int64(p.Victim))
|
||
if p.MoveAssets {
|
||
repoint = append(repoint, p)
|
||
}
|
||
}
|
||
// 聚合要按地点重算,先把副本所属地点收下来(删完就在 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
|
||
}
|
||
}
|
||
}
|
||
// 「补录行的副本」先把它名下的位移 / 天气改挂到保留项:那段行程真实发生过,
|
||
// 而补录行自己名下没有位移,跟着副本一起删就是把里程白丢了。
|
||
for _, p := range repoint {
|
||
if _, err := ex.ExecContext(ctx, `UPDATE movements SET from_visit_src=$3
|
||
WHERE dataset_id=$1 AND from_visit_src=$2`, datasetID, p.Victim, p.Keep); err != nil {
|
||
return err
|
||
}
|
||
if _, err := ex.ExecContext(ctx, `UPDATE movements SET to_visit_src=$3
|
||
WHERE dataset_id=$1 AND to_visit_src=$2`, datasetID, p.Victim, p.Keep); err != nil {
|
||
return err
|
||
}
|
||
if _, err := ex.ExecContext(ctx, `UPDATE weather SET visit_src=$3
|
||
WHERE dataset_id=$1 AND visit_src=$2`, datasetID, p.Victim, p.Keep); 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()
|
||
}
|