Files
rond-with-you/internal/web/backup_merge.go
T
tamakyi 9c1413bb39 导出/合并:交通方式编号空间串了,导出会吐出两条同名方式(真机整包拒收)
继上一轮修「活动颜色为 NULL」之后,生产重新导出 `data/20260929.rondbackup` 依旧导不回 rond。
这次拿**最后一次被真机接受的包**(20260920-fixed-A-数据修复)当参照逐项比对:SQLite 头、31 表 /
47 索引 / 建表语句 / 列定义、`Z_METADATA.Z_PLIST` 与 `Z_MODELCACHE`(逐字节)、值类型分布、文本合法
字节、BLOB 长度、UUID 唯一性、关系完整性、`Z_OPT`、行级时间不变量——全部一致;唯一差异是
**`ZTRANSPORT` 里出现了两条同名「公交」(Z_PK 2 与 4),而真机与可导入的 A 包都只有一条**。

根因:`mergeVisits` 把实机的 `ZTRANSPORT` 编号原样写进 `movements.transport_src`。站点没有交通方式表,
方式只活在位移的 transport_* 四列里,两套编号空间同一个数字含义不同(实测 src=4:站点=电🐔、实机=公交,
src 3/5/6 同样撞车)。于是同一个 src 带上两种含义:导出按 src 聚合只能取一个名字 → 另一种含义的位移挂错
方式;更糟的是吐出两条同名方式,真机那边是模型唯一约束级别的违规 → 整包拒收。

修复两处:
1. 合并侧 `mergeVisits`:按名称把实机方式归一到站点编号(新增 `(*Server).transportsByName`:同名沿用
   站点已有的编号,没有才 `nextSrcPK(transportSeq)` 新发一个)。
2. 导出侧 `queryTransports`:改为按**名称**归组(名称就是身份),用量最多的名称优先认领编号,其余编号的
   位移改挂到它上面;并清掉「基线里同名但编号已被归并掉」的旧行——`keepMissing` 会把它原样带回,于是又是
   两条同名。这段清理必须排在所有托管表写完之后,否则兜底判据看到的还是基线里未重写的 ZMOVEMENT。

真实数据效果(本地库,与生产同一形状):方式恢复成站点原本那套 1火车/2公交/3高铁/4电🐔/5地铁/6步行、
无重名;位移各挂回自己的方式(电🐔 336、地铁 25、高铁 23、公交 19、火车 9、步行 3——修复前只有 6 条能对上)。

用例:`TestExportMergesDuplicateTransportNames`(同名归并 + 位移挂回自己那条 + 无悬空引用)、
`TestMergeKeepsTransportNumberingConsistent`(合并后同一 transport_src 不许有两种名称,附「位移确实被补
回来了」的哨兵断言避免写成空用例)。三处都做了变异验证:去掉改挂 / 去掉旧行清理 / 合并照抄实机编号 →
对应用例分别变红。

全量验收 295 PASS / 0 FAIL / 0 SKIP。
2026-09-29 13:25:50 +08:00

811 lines
27 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package web
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"strconv"
"strings"
"rond-with-you/internal/geo"
"rond-with-you/internal/ingest"
"rond-with-you/internal/rond"
)
// 差异页确认后的合并执行。
//
// 全部改动在一个事务里完成:半合并的数据集比没合并更难收拾(比如地点建了、
// 到访没建,地图上就多一个永远没人去的点)。删除不在这里——删地点会级联删掉
// 它的全部到访,那件事单独再议。
//
// 编号是这里最容易出错的地方:**库里那侧的编号空间才是唯一有效的**。
// 新插入的行拿 nextSrcPK 发的新号,于是位移的 from_visit_src、天气的 visit_src
// 都必须写「库里的新号」,而不是备份里的原号。
type mergeStats struct {
Places, Activities, Tags int
Visits, Movements, Weather int
UpdPlaces, UpdActs, UpdVisits int
DelPlaces, DelActs, DelVisits int
}
func (s *Server) adminBackupMerge(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
ds, err := s.datasetFor(ctx)
if err != nil {
s.serverError(w, r, err)
return
}
if ds == nil {
s.adminFlash(w, r, "err", "没有可合并的数据集")
return
}
token := r.PathValue("token")
path, ok := stashPath(s.backupStashDir(), backupStashExt, token)
if !ok {
s.adminFlash(w, r, "err", "合并链接无效,请重新上传备份")
return
}
if _, err := os.Stat(path); err != nil {
s.adminFlash(w, r, "err", "这次预览已过期,请重新上传备份")
return
}
if err := r.ParseForm(); err != nil {
s.serverError(w, r, err)
return
}
backup, err := s.parseBackupFile(path)
if err != nil {
s.adminFlash(w, r, "err", err.Error())
return
}
visits, places, acts, err := s.loadStoredAll(ctx, ds.ID)
if err != nil {
s.serverError(w, r, err)
return
}
// 重新算一遍差异再执行:从渲染差异页到点下按钮之间,库可能已经被别处改过
vd := rond.DiffVisits(visits, backup)
pd := rond.DiffPlaces(places, backup)
ad := rond.DiffActivities(acts, backup)
addNew := r.PostFormValue("add_new") == "1"
plan := rond.BuildMergePlan(vd, pd, ad, rond.Choices{
AddVisits: addNew, AddPlaces: addNew, AddActs: addNew,
VisitClash: pickForm(r, "vc_"),
PlaceClash: pickForm(r, "pc_"),
ActClash: pickForm(r, "ac_"),
DelVisits: pickIntForm(r, "dv_"),
DelPlaces: pickIntForm(r, "dp_"),
DelActs: pickIntForm(r, "da_"),
})
if plan.Empty() {
s.adminFlash(w, r, "ok", "按当前的选择没有要写入的改动,数据未变")
return
}
st, err := s.applyMergePlan(ctx, ds, backup, plan, visits)
if err != nil {
log.Printf("合并失败: %v", err)
s.adminFlash(w, r, "err", "合并失败,已整体回滚、数据未变:"+err.Error())
return
}
stashDrop(s.backupStashDir(), backupStashExt, token)
msg := fmt.Sprintf("合并完成:新增 %d 个地点 / %d 个活动 / %d 条到访;更新 %d 个地点 / %d 个活动 / %d 条到访",
st.Places, st.Activities, st.Visits, st.UpdPlaces, st.UpdActs, st.UpdVisits)
if st.DelVisits+st.DelActs+st.DelPlaces > 0 {
msg += fmt.Sprintf(";删除 %d 条到访 / %d 个活动 / %d 个地点", st.DelVisits, st.DelActs, st.DelPlaces)
}
if st.Tags > 0 {
msg += fmt.Sprintf(";顺带建了 %d 个标签", st.Tags)
}
if st.Movements+st.Weather > 0 {
msg += fmt.Sprintf(";随到访带入 %d 条位移、%d 条天气", st.Movements, st.Weather)
}
s.adminFlash(w, r, "ok", msg)
}
// pickIntForm 收集「前缀 + 整数」形式的勾选,比如 dv_825=1(要删除的到访)。
func pickIntForm(r *http.Request, prefix string) map[int]bool {
out := map[int]bool{}
for k, vs := range r.PostForm {
if len(vs) == 0 || vs[0] == "" || !strings.HasPrefix(k, prefix) {
continue
}
if n, err := strconv.Atoi(strings.TrimPrefix(k, prefix)); err == nil {
out[n] = true
}
}
return out
}
// pickForm 收集「前缀 + 键」形式的逐条选择,比如 vc_<uuid>=backup。
func pickForm(r *http.Request, prefix string) map[string]string {
out := map[string]string{}
for k, vs := range r.PostForm {
if len(vs) == 0 || !strings.HasPrefix(k, prefix) {
continue
}
out[strings.TrimPrefix(k, prefix)] = vs[0]
}
return out
}
func (s *Server) loadStoredAll(ctx context.Context, dsID int64) ([]rond.StoredVisit, []rond.StoredPlace, []rond.StoredActivity, error) {
visits, err := s.loadStoredVisits(ctx, dsID)
if err != nil {
return nil, nil, nil, err
}
places, err := s.loadStoredPlaces(ctx, dsID)
if err != nil {
return nil, nil, nil, err
}
acts, err := s.loadStoredActivities(ctx, dsID)
if err != nil {
return nil, nil, nil, err
}
return visits, places, acts, nil
}
func (s *Server) parseBackupFile(path string) (*rond.Backup, error) {
tmp, err := ingest.TempDir(s.cfg.DataDir, "merge")
if err != nil {
return nil, err
}
defer os.RemoveAll(tmp)
db, err := rond.Open(path, tmp)
if err != nil {
return nil, fmt.Errorf("解析备份失败:%w", err)
}
defer db.Close()
b, err := rond.Parse(db)
if err != nil {
return nil, fmt.Errorf("读取备份数据失败:%w", err)
}
return b, nil
}
func (s *Server) applyMergePlan(ctx context.Context, ds *Dataset, b *rond.Backup, plan rond.MergePlan,
storedVisits []rond.StoredVisit) (*mergeStats, error) {
st := &mergeStats{}
err := s.withTx(ctx, func(ex execer) error {
placeID, err := s.mergePlaces(ctx, ex, ds.ID, b, plan, st)
if err != nil {
return err
}
actID, err := s.mergeActivities(ctx, ex, ds.ID, b, plan, st)
if err != nil {
return err
}
tagID, err := s.mergeTags(ctx, ex, ds.ID, b, st)
if err != nil {
return err
}
if err := s.mergeVisits(ctx, ex, ds.ID, b, plan, placeID, actID, tagID, storedVisits, st); err != nil {
return err
}
// 删除放在最后:先让新增/更新落定,再按最终状态删
if err := s.applyDeletions(ctx, ex, ds.ID, plan, st); err != nil {
return err
}
return s.recountPlaces(ctx, ex, ds.ID)
})
if err != nil {
return nil, err
}
return st, nil
}
// mergePlaces 先把备份地点映射到库里的 places.id,再处理冲突更新与新增,
// 返回「备份地点编号 -> 库 id」。
//
// 配对直接沿用比对阶段那一份(rond.MatchPlaces):**这里绝不能自己再配一遍**。原先用
// 单槽位 map 配,同坐标/同名的多条备份地点会全部映射到同一条站内地点上,到访于是被挂到
// 错的分店(实测「茶颜悦色」两家分店相隔 8.3km,到访却指向同一条)。
func (s *Server) mergePlaces(ctx context.Context, ex execer, dsID int64, b *rond.Backup,
plan rond.MergePlan, st *mergeStats) (map[int]int64, error) {
// 顺序必须稳定(按 id)且与 loadStoredPlaces 一致,否则候选多于一个时配对会飘
stored := make([]rond.StoredPlace, 0, 512)
bySrcPK := map[int]int64{}
rows, err := ex.QueryContext(ctx, `SELECT id, src_pk, COALESCE(name,''), COALESCE(lat,0), COALESCE(lon,0)
FROM places WHERE dataset_id=$1 ORDER BY id`, dsID)
if err != nil {
return nil, err
}
for rows.Next() {
var id int64
var sp rond.StoredPlace
if err := rows.Scan(&id, &sp.SrcPK, &sp.Name, &sp.Lat, &sp.Lon); err != nil {
rows.Close()
return nil, err
}
if sp.Lat != 0 || sp.Lon != 0 {
sp.Key = rond.PlaceCoordKey(sp.Lat, sp.Lon)
}
stored = append(stored, sp)
bySrcPK[sp.SrcPK] = id
}
rows.Close()
if err := rows.Err(); err != nil {
return nil, err
}
placeID := map[int]int64{}
for srcPK, sp := range rond.MatchPlaces(stored, b.Locations) {
if id, ok := bySrcPK[sp.SrcPK]; ok {
placeID[srcPK] = id
}
}
for _, up := range plan.UpdPlaces {
id := placeID[up.L.SrcPK]
if id == 0 {
return nil, fmt.Errorf("找不到要更新的地点(%s)", up.L.Name)
}
if _, err := ex.ExecContext(ctx, `UPDATE places SET name=$1, poi_category=$2, lat=$3, lon=$4,
country_code=$5, province=$6, city=$7, district=$8, sublocality=$9, thoroughfare=$10, timezone=$11
WHERE id=$12`,
nullStr(up.L.Name), nullStr(up.L.POICategory), up.L.Lat, up.L.Lon,
nullStr(up.L.CountryCode), nullStr(up.L.Province), nullStr(up.L.City), nullStr(up.L.District),
nullStr(up.L.Sublocality), nullStr(up.L.Thoroughfare), nullStr(up.L.Timezone), id); err != nil {
return nil, fmt.Errorf("更新地点: %w", err)
}
st.UpdPlaces++
}
for _, l := range plan.AddPlaces {
pk, err := s.nextSrcPK(ctx, ex, dsID, placeSeq)
if err != nil {
return nil, err
}
var id int64
if err := ex.QueryRowContext(ctx, `INSERT INTO places
(dataset_id, src_pk, name, poi_category, lat, lon, country_code, province, city,
district, sublocality, thoroughfare, timezone)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) RETURNING id`,
dsID, pk, nullStr(l.Name), nullStr(l.POICategory), l.Lat, l.Lon, nullStr(l.CountryCode),
nullStr(l.Province), nullStr(l.City), nullStr(l.District), nullStr(l.Sublocality),
nullStr(l.Thoroughfare), nullStr(l.Timezone)).Scan(&id); err != nil {
return nil, fmt.Errorf("新增地点: %w", err)
}
placeID[l.SrcPK] = id
st.Places++
if err := writeRawRow(ctx, ex, dsID, "ZLOCATION", pk, b.RawRows["ZLOCATION"][l.SrcPK]); err != nil {
return nil, err
}
}
return placeID, nil
}
func (s *Server) mergeActivities(ctx context.Context, ex execer, dsID int64, b *rond.Backup,
plan rond.MergePlan, st *mergeStats) (map[int]int64, error) {
byUUID, byName := map[string]int64{}, map[string]int64{}
rows, err := ex.QueryContext(ctx, `SELECT a.id, a.name, COALESCE(er.raw #>> '{ZUID_,$blob}','')
FROM activities a
LEFT JOIN entity_raw er ON er.dataset_id = a.dataset_id
AND er.entity = 'ZACTIVITY' AND er.src_pk = a.src_pk
WHERE a.dataset_id=$1`, dsID)
if err != nil {
return nil, err
}
for rows.Next() {
var id int64
var name, b64 string
if err := rows.Scan(&id, &name, &b64); err != nil {
rows.Close()
return nil, err
}
if u := uuidFromBase64(b64); u != "" {
byUUID[u] = id
}
if _, dup := byName[name]; !dup && name != "" {
byName[name] = id
}
}
rows.Close()
if err := rows.Err(); err != nil {
return nil, err
}
actID := map[int]int64{}
for _, a := range b.Activities {
if u := b.ActivityUUID[a.SrcPK]; u != "" {
if id, ok := byUUID[u]; ok {
actID[a.SrcPK] = id
continue
}
}
if id, ok := byName[a.Name]; ok {
actID[a.SrcPK] = id
}
}
for _, up := range plan.UpdActs {
id := actID[up.A.SrcPK]
if id == 0 {
id = byName[up.A.Name]
}
if id == 0 {
return nil, fmt.Errorf("找不到要更新的活动(%s)", up.A.Name)
}
if _, err := ex.ExecContext(ctx, `UPDATE activities SET name=$1, color=$2, icon=$3,
is_home=$4, is_work=$5, is_excluded=$6, is_archived=$7 WHERE id=$8`,
up.A.Name, nullStr(up.A.Color), nullStr(up.A.Icon), up.A.IsHome, up.A.IsWork,
up.A.IsExcluded, up.A.IsArchived, id); err != nil {
return nil, fmt.Errorf("更新活动: %w", err)
}
st.UpdActs++
}
for _, a := range plan.AddActs {
pk, err := s.nextSrcPK(ctx, ex, dsID, activitySeq)
if err != nil {
return nil, err
}
var id int64
if err := ex.QueryRowContext(ctx, `INSERT INTO activities
(dataset_id, src_pk, name, color, icon, is_home, is_work, is_excluded, is_archived)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING id`,
dsID, pk, a.Name, nullStr(a.Color), nullStr(a.Icon),
a.IsHome, a.IsWork, a.IsExcluded, a.IsArchived).Scan(&id); err != nil {
return nil, fmt.Errorf("新增活动: %w", err)
}
actID[a.SrcPK] = id
st.Activities++
if err := writeRawRow(ctx, ex, dsID, "ZACTIVITY", pk, b.RawRows["ZACTIVITY"][a.SrcPK]); err != nil {
return nil, err
}
}
return actID, nil
}
// mergeTags 只做「库里有同名的就复用、没有就建」——标签是纯名称实体,没有别的标识。
func (s *Server) mergeTags(ctx context.Context, ex execer, dsID int64, b *rond.Backup, st *mergeStats) (map[int]int64, error) {
byName := map[string]int64{}
rows, err := ex.QueryContext(ctx, `SELECT id, name FROM tags WHERE dataset_id=$1`, dsID)
if err != nil {
return nil, err
}
for rows.Next() {
var id int64
var name string
if err := rows.Scan(&id, &name); err != nil {
rows.Close()
return nil, err
}
byName[name] = id
}
rows.Close()
if err := rows.Err(); err != nil {
return nil, err
}
tagID := map[int]int64{}
for _, t := range b.Tags {
if id, ok := byName[t.Name]; ok {
tagID[t.SrcPK] = id
continue
}
pk, err := s.nextSrcPK(ctx, ex, dsID, tagSeq)
if err != nil {
return nil, err
}
var id int64
if err := ex.QueryRowContext(ctx, `INSERT INTO tags (dataset_id, src_pk, name, color, group_name)
VALUES ($1,$2,$3,$4,NULLIF($5,'')) RETURNING id`,
dsID, pk, t.Name, nullStr(t.Color), t.GroupName).Scan(&id); err != nil {
return nil, fmt.Errorf("新增标签: %w", err)
}
byName[t.Name] = id
tagID[t.SrcPK] = id
st.Tags++
if err := writeRawRow(ctx, ex, dsID, "ZTAG", pk, b.RawRows["ZTAG"][t.SrcPK]); err != nil {
return nil, err
}
}
return tagID, nil
}
func (s *Server) mergeVisits(ctx context.Context, ex execer, dsID int64, b *rond.Backup, plan rond.MergePlan,
placeID, actID, tagID map[int]int64, storedVisits []rond.StoredVisit, st *mergeStats) error {
// 备份到访编号 -> 库里的 visits.src_pk:位移的 from/to_visit_src 与天气的
// visit_src 写的都是**库里的编号**,端点在备份里的原号在这里换成库里的号。
visSrc := map[int]int{}
storedByUUID := map[string]rond.StoredVisit{}
for _, sv := range storedVisits {
if sv.UUID != "" {
storedByUUID[sv.UUID] = sv
}
}
for _, v := range b.Visits {
if u := b.VisitUUID[v.SrcPK]; u != "" {
if sv, ok := storedByUUID[u]; ok {
visSrc[v.SrcPK] = sv.SrcPK
}
}
}
activityByPK := map[int]rond.Activity{}
for _, a := range b.Activities {
activityByPK[a.SrcPK] = a
}
visitPlace := map[int]int{}
for _, v := range b.Visits {
if v.LocationSrcPK != nil {
visitPlace[v.SrcPK] = *v.LocationSrcPK
}
}
resolve := func(v rond.Visit) (pid, aid any, isHome, isWork bool) {
if v.LocationSrcPK != nil {
if x, ok := placeID[*v.LocationSrcPK]; ok {
pid = x
}
}
if v.ActivitySrcPK != nil {
if x, ok := actID[*v.ActivitySrcPK]; ok {
aid = x
}
if a, ok := activityByPK[*v.ActivitySrcPK]; ok {
isHome, isWork = a.IsHome, a.IsWork
}
}
return
}
for _, up := range plan.UpdVisits {
sv, ok := storedByUUID[up.UUID]
if !ok {
return fmt.Errorf("找不到要更新的到访(%s)", up.UUID)
}
pid, aid, isHome, isWork := resolve(up.V)
if _, err := ex.ExecContext(ctx, `UPDATE visits SET place_id=$1, activity_id=$2, arrival=$3, departure=$4,
duration_min=$5, is_home=$6, is_work=$7, bookmarked=$8, remark=NULLIF($9,''), emoji=NULLIF($10,'')
WHERE dataset_id=$11 AND src_pk=$12`,
pid, aid, up.V.Arrival, up.V.Departure, visitDuration(up.V), isHome, isWork,
up.V.Bookmarked, up.V.Remark, up.V.Emoji, dsID, sv.SrcPK); err != nil {
return fmt.Errorf("更新到访: %w", err)
}
st.UpdVisits++
}
addedVisit := map[int]bool{}
for _, v := range plan.AddVisits {
pid, aid, isHome, isWork := resolve(v)
pk, err := s.nextSrcPK(ctx, ex, dsID, visitSeq)
if err != nil {
return err
}
var id int64
if err := ex.QueryRowContext(ctx, `INSERT INTO visits
(dataset_id, src_pk, place_id, activity_id, arrival, departure, duration_min,
is_home, is_work, bookmarked, is_user_added, remark, emoji, weather_symbol)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,NULLIF($12,''),NULLIF($13,''),NULLIF($14,''))
ON CONFLICT DO NOTHING RETURNING id`,
dsID, pk, pid, aid, v.Arrival, v.Departure, visitDuration(v), isHome, isWork,
v.Bookmarked, v.UserAdded, v.Remark, v.Emoji, v.WeatherSymbol).Scan(&id); err != nil {
if err == sql.ErrNoRows {
continue
}
return fmt.Errorf("新增到访: %w", err)
}
visSrc[v.SrcPK] = pk
addedVisit[v.SrcPK] = true
st.Visits++
// 留档必须写:下次再比对时库侧就是靠这里的 UUID 认出这一条,
// 少了它只能退到「到达时间 + 坐标」兜底。
if err := writeRawRow(ctx, ex, dsID, "ZVISIT", pk, b.RawRows["ZVISIT"][v.SrcPK]); err != nil {
return err
}
for _, tpk := range v.TagSrcPKs {
if tid, ok := tagID[tpk]; ok {
if _, err := ex.ExecContext(ctx, `INSERT INTO visit_tags (visit_id, tag_id)
VALUES ($1,$2) ON CONFLICT DO NOTHING`, id, tid); err != nil {
return fmt.Errorf("写入到访标签: %w", err)
}
}
}
}
transName, transColor, transIcon := map[int]string{}, map[int]string{}, map[int]string{}
for _, t := range b.Transports {
transName[t.SrcPK], transColor[t.SrcPK], transIcon[t.SrcPK] = t.Name, t.Color, t.Icon
}
// 交通方式是**两套编号空间**:站点的 movements.transport_src 用自己的编号(站点没有交通方式表,
// 方式只活在位移的 transport_* 四列里),而 m.TransportSrc 是实机的编号。同一个数字两边含义
// 常常不同(实测实机 4=公交、站点 4=电🐔),照抄会把一个 src 变成两种含义——导出按 src 聚合时
// 会吐出两条同名方式,真机那边(以及 Core Data 的唯一约束)会判成损坏数据、整包拒收。
// 按**名称**归一:站点已有同名方式就沿用它的编号,没有才新发一个。
siteTrans, transNext, err := s.transportsByName(ctx, ex, dsID)
if err != nil {
return err
}
transSrc := map[int]int{}
for _, t := range b.Transports {
if t.Name == "" {
continue
}
if src, ok := siteTrans[t.Name]; ok {
transSrc[t.SrcPK] = src
continue
}
siteTrans[t.Name] = transNext
transSrc[t.SrcPK] = transNext
transNext++
}
coordOf := map[int][2]float64{}
for _, l := range b.Locations {
coordOf[l.SrcPK] = [2]float64{l.Lat, l.Lon}
}
// 位移与天气跟着**新插入的到访**走:只插这一轮真正补进来的那些,
// 否则库里已有的位移会被整份复制一遍。另一端可以是老到访(库里已经有)。
for _, m := range b.Movements {
if m.FromVisitSrc == nil || m.ToVisitSrc == nil {
continue
}
if !addedVisit[*m.FromVisitSrc] && !addedVisit[*m.ToVisitSrc] {
continue
}
fsrc, ok1 := visSrc[*m.FromVisitSrc]
tsrc, ok2 := visSrc[*m.ToVisitSrc]
if !ok1 || !ok2 {
continue
}
var fromPID, toPID any
if fp, ok := visitPlace[*m.FromVisitSrc]; ok {
if x, ok := placeID[fp]; ok {
fromPID = x
}
}
if tp, ok := visitPlace[*m.ToVisitSrc]; ok {
if x, ok := placeID[tp]; ok {
toPID = x
}
}
var dist any
if fp, ok1 := visitPlace[*m.FromVisitSrc]; ok1 {
if tp, ok2 := visitPlace[*m.ToVisitSrc]; ok2 {
c1, c2 := coordOf[fp], coordOf[tp]
if c1[0] != 0 && c2[0] != 0 {
dist = geo.Haversine(c1[0], c1[1], c2[0], c2[1])
}
}
}
var tname, tcolor, ticon any
if m.TransportSrc != nil {
tname = nullStr(transName[*m.TransportSrc])
tcolor = nullStr(transColor[*m.TransportSrc])
ticon = nullStr(transIcon[*m.TransportSrc])
}
var dur any
if m.Start != nil && m.End != nil {
if d := int(m.End.Sub(*m.Start).Minutes()); d >= 0 {
dur = d
}
}
pk, err := s.nextSrcPK(ctx, ex, dsID, movementSeq)
if err != nil {
return err
}
// transport_src 写**站点编号空间**里的号(见上面 transSrc 的说明):照抄实机编号会让
// 同一个 src 出现两种含义,导出时吐出两条同名方式、整包被真机拒收。
var transFK any
if m.TransportSrc != nil {
if x, ok := transSrc[*m.TransportSrc]; ok {
transFK = x
} else {
transFK = *m.TransportSrc
}
}
if _, err := ex.ExecContext(ctx, `INSERT INTO movements
(dataset_id, src_pk, transport_src, transport_name, transport_color, transport_icon,
from_visit_src, to_visit_src, from_place_id, to_place_id, started_at, ended_at, duration_min, distance_km)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT DO NOTHING`,
dsID, pk, transFK, tname, tcolor, ticon, fsrc, tsrc,
fromPID, toPID, m.Start, m.End, dur, dist); err != nil {
return fmt.Errorf("新增位移: %w", err)
}
st.Movements++
if err := writeRawRow(ctx, ex, dsID, "ZMOVEMENT", pk, b.RawRows["ZMOVEMENT"][m.SrcPK]); err != nil {
return err
}
}
for _, w := range b.Weather {
if w.VisitSrc == nil || w.At == nil {
continue
}
if !addedVisit[*w.VisitSrc] {
continue
}
vsrc, ok := visSrc[*w.VisitSrc]
if !ok {
continue
}
// 站内删到访时会把天气的归属置空而保留记录(见 editVisitDelete)。
// 这类「同一时刻、没有归属」的天气先认领回来,否则同一条到访找回来之后
// 会多出一份重复的天气。
var claimed int64
if err := ex.QueryRowContext(ctx, `UPDATE weather SET visit_src=$1
WHERE dataset_id=$2 AND at=$3 AND visit_src IS NULL RETURNING id`,
vsrc, dsID, w.At).Scan(&claimed); err == nil {
continue
}
pk, err := s.nextSrcPK(ctx, ex, dsID, weatherSeq)
if err != nil {
return err
}
if _, err := ex.ExecContext(ctx, `INSERT INTO weather
(dataset_id, src_pk, visit_src, at, is_daylight, temperature_c, apparent_c, humidity,
precipitation_amount, precipitation_chance, wind_speed, visibility, uv_index,
condition, symbol, uv_category, wind_direction)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,NULLIF($14,''),NULLIF($15,''),NULLIF($16,''),NULLIF($17,''))
ON CONFLICT DO NOTHING`,
dsID, pk, vsrc, w.At, w.IsDaylight, w.TempC, w.ApparentC, w.Humidity,
w.PrecipAmt, w.PrecipPct, w.WindSpeed, w.Visibility, w.UVIndex,
w.Condition, w.Symbol, w.UVCategory, w.WindDir); err != nil {
return fmt.Errorf("新增天气: %w", err)
}
st.Weather++
if err := writeRawRow(ctx, ex, dsID, "ZHOURLYWEATHER", pk, b.RawRows["ZHOURLYWEATHER"][w.SrcPK]); err != nil {
return err
}
}
return nil
}
// transportsByName 读站点现有的交通方式,按**名称**归并成一个「名称 -> 编号」表(同名取用得多
// 的那个编号,并列取小),并给出下一个可用编号。站点没有交通方式表,方式只活在位移的
// transport_src / transport_name 里,所以这里从 movements 里聚合。
func (s *Server) transportsByName(ctx context.Context, ex execer, dsID int64) (map[string]int, int, error) {
rows, err := ex.QueryContext(ctx, `SELECT transport_src, transport_name, count(*)
FROM movements
WHERE dataset_id=$1 AND transport_src IS NOT NULL AND COALESCE(transport_name,'') <> ''
GROUP BY 1,2`, dsID)
if err != nil {
return nil, 0, err
}
defer rows.Close()
count, best := map[string]int{}, map[string]int{}
for rows.Next() {
var src, n int
var name string
if err := rows.Scan(&src, &name, &n); err != nil {
return nil, 0, err
}
cur, seen := count[name]
if !seen || n > cur || (n == cur && src < best[name]) {
count[name], best[name] = n, src
}
}
if err := rows.Err(); err != nil {
return nil, 0, err
}
next, err := s.nextSrcPK(ctx, ex, dsID, transportSeq)
if err != nil {
return nil, 0, err
}
return best, next, nil
}
// applyDeletions 处理「仅站内有」里被勾选删除的记录。顺序是到访 → 活动 → 地点:
// 删地点会级联删掉它名下的全部到访(外键 ON DELETE CASCADE),所以反过来会误删;
// 而且**有到访引用时直接拒绝**,让站长先处理到访,而不是悄悄带走一批。
func (s *Server) applyDeletions(ctx context.Context, ex execer, dsID int64, plan rond.MergePlan, st *mergeStats) error {
for _, sv := range plan.DelVisits {
if _, err := ex.ExecContext(ctx, `DELETE FROM movements WHERE dataset_id=$1
AND (from_visit_src=$2 OR to_visit_src=$2)`, dsID, sv.SrcPK); err != nil {
return fmt.Errorf("删除到访的位移: %w", err)
}
// 天气保留、只解除归属——与站点 editVisitDelete 的处理一致
if _, err := ex.ExecContext(ctx, `UPDATE weather SET visit_src=NULL
WHERE dataset_id=$1 AND visit_src=$2`, dsID, sv.SrcPK); err != nil {
return fmt.Errorf("解除天气归属: %w", err)
}
if _, err := ex.ExecContext(ctx, `DELETE FROM visits WHERE dataset_id=$1 AND src_pk=$2`,
dsID, sv.SrcPK); err != nil {
return fmt.Errorf("删除到访: %w", err)
}
if _, err := ex.ExecContext(ctx, `DELETE FROM entity_raw WHERE dataset_id=$1
AND entity='ZVISIT' AND src_pk=$2`, dsID, sv.SrcPK); err != nil {
return err
}
st.DelVisits++
}
for _, sa := range plan.DelActs {
// 到访的 activity_id 是 ON DELETE SET NULL,删活动不会带走到访
if _, err := ex.ExecContext(ctx, `DELETE FROM activities WHERE dataset_id=$1 AND src_pk=$2`,
dsID, sa.SrcPK); err != nil {
return fmt.Errorf("删除活动: %w", err)
}
if _, err := ex.ExecContext(ctx, `DELETE FROM entity_raw WHERE dataset_id=$1
AND entity='ZACTIVITY' AND src_pk=$2`, dsID, sa.SrcPK); err != nil {
return err
}
st.DelActs++
}
for _, sp := range plan.DelPlaces {
var n int
if err := ex.QueryRowContext(ctx, `SELECT count(*) FROM visits v
JOIN places p ON p.id = v.place_id
WHERE p.dataset_id=$1 AND p.src_pk=$2`, dsID, sp.SrcPK).Scan(&n); err != nil {
return err
}
if n > 0 {
return fmt.Errorf("地点「%s」名下还有 %d 条到访,删它会连带删掉那些到访——请先处理这些到访",
sp.Name, n)
}
if _, err := ex.ExecContext(ctx, `DELETE FROM places WHERE dataset_id=$1 AND src_pk=$2`,
dsID, sp.SrcPK); err != nil {
return fmt.Errorf("删除地点: %w", err)
}
if _, err := ex.ExecContext(ctx, `DELETE FROM entity_raw WHERE dataset_id=$1
AND entity='ZLOCATION' AND src_pk=$2`, dsID, sp.SrcPK); err != nil {
return err
}
st.DelPlaces++
}
return nil
}
// recountPlaces 重算地点聚合与数据集汇总。全量重算而不是只动受影响的行:
// 367 个地点两条 SQL 就够,省得漏掉任何一条派生列。
func (s *Server) recountPlaces(ctx context.Context, ex execer, dsID int64) error {
if _, err := ex.ExecContext(ctx, `UPDATE places SET visit_count=0, dwell_minutes=0,
first_visit_at=NULL, last_visit_at=NULL WHERE dataset_id=$1`, dsID); err != nil {
return err
}
if _, err := ex.ExecContext(ctx, `UPDATE places p SET visit_count=sub.c, dwell_minutes=sub.d,
first_visit_at=sub.f, last_visit_at=sub.l
FROM (SELECT place_id, count(*) c,
COALESCE(sum(LEAST(COALESCE(duration_min,0),10080)),0) d,
min(arrival) f, max(arrival) l
FROM visits WHERE dataset_id=$1 AND place_id IS NOT NULL GROUP BY place_id) sub
WHERE p.id = sub.place_id`, dsID); err != nil {
return err
}
_, err := ex.ExecContext(ctx, `UPDATE datasets SET
visit_count=(SELECT count(*) FROM visits WHERE dataset_id=$1),
place_count=(SELECT count(*) FROM places WHERE dataset_id=$1),
first_visit_at=(SELECT min(arrival) FROM visits WHERE dataset_id=$1),
last_visit_at=(SELECT max(arrival) FROM visits WHERE dataset_id=$1)
WHERE id=$1`, dsID)
return err
}
// writeRawRow 给新插入的行留一份原始行:导出时要靠它逐列还原,
// 再次比对时要靠它认出 UUID(否则只能退到「时间 + 坐标」兜底)。
func writeRawRow(ctx context.Context, ex execer, dsID int64, entity string, pk int, raw map[string]any) error {
if len(raw) == 0 {
return nil
}
blob, err := json.Marshal(raw)
if err != nil {
// 这里以前只 log 后 return nil:调用方会以为留档成功,合并照常提交,
// 于是这行的原始列永久缺失(下次比对认不出 UUID、导出也还原不回去)。
// 必须上抛,让整个合并事务回滚。
return fmt.Errorf("留档 %s#%d 序列化失败: %w", entity, pk, err)
}
_, err = ex.ExecContext(ctx, `INSERT INTO entity_raw (dataset_id, entity, src_pk, raw, skipped)
VALUES ($1,$2,$3,$4,false)
ON CONFLICT (dataset_id, entity, src_pk) DO UPDATE SET raw=EXCLUDED.raw`,
dsID, entity, pk, string(blob))
return err
}
func visitDuration(v rond.Visit) any {
if v.Arrival == nil || v.Departure == nil {
return nil
}
m := int(v.Departure.Sub(*v.Arrival).Minutes())
if m < 0 {
m = 0
}
return m
}