备份与迷雾的差异分析、合并执行，附属数据改挂坐标键
备份（.rondbackup）与迷雾（.fwss）各加一层「先看差异再动手」： - 迷雾：上传先出差异页，说清叠加能补多少、整体替换会丢多少，并在真实底图上 用绿/灰/红标出会新增 / 不变 / 会丢的区域 - 备份：与当前数据集按 UUID 逐条比对（rond 每次导出都会重排 Z_PK，编号认不出人）， 新增合入、冲突逐条选、仅站内有逐条勾选删除，全在一个事务里执行； 合入的行写回 entity_raw，保证再传同一份备份时差异归零 - 地点按坐标对齐：站点的「补全地址」会把地点改名，拿名称当键会让一整个地点的 到访全变成假冲突 - 附属数据（备注/结论/访客票/图片）从 dataset_id+src_pk 换成坐标键，换数据集不再失联； 访客票没有 dataset_id，回填按活跃数据集认领 - 修正 contentHash 拿 Z_PK 当指纹的隐患，以及两处「src_pk 跨快照稳定」的错误注释 导出格式（.rondbackup / .fwss）保持原样，往返测试通过。
This commit is contained in:
tamakyi
@@ -0,0 +1,753 @@
|
||||
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」。
|
||||
func (s *Server) mergePlaces(ctx context.Context, ex execer, dsID int64, b *rond.Backup,
|
||||
plan rond.MergePlan, st *mergeStats) (map[int]int64, error) {
|
||||
byCoord, byName := map[string]int64{}, map[string]int64{}
|
||||
rows, err := ex.QueryContext(ctx, `SELECT id, COALESCE(name,''), COALESCE(lat,0), COALESCE(lon,0)
|
||||
FROM places WHERE dataset_id=$1`, dsID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
var name string
|
||||
var lat, lon float64
|
||||
if err := rows.Scan(&id, &name, &lat, &lon); err != nil {
|
||||
rows.Close()
|
||||
return nil, err
|
||||
}
|
||||
if lat != 0 || lon != 0 {
|
||||
byCoord[rond.PlaceCoordKey(lat, lon)] = id
|
||||
}
|
||||
if _, dup := byName[name]; !dup && name != "" {
|
||||
byName[name] = id
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
placeID := map[int]int64{}
|
||||
for _, l := range b.Locations {
|
||||
if l.Lat != 0 || l.Lon != 0 {
|
||||
if id, ok := byCoord[rond.PlaceCoordKey(l.Lat, l.Lon)]; ok {
|
||||
placeID[l.SrcPK] = id
|
||||
continue
|
||||
}
|
||||
}
|
||||
if l.Name != "" {
|
||||
if id, ok := byName[l.Name]; ok {
|
||||
placeID[l.SrcPK] = id
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for _, up := range plan.UpdPlaces {
|
||||
id := placeID[up.L.SrcPK]
|
||||
if id == 0 {
|
||||
id = byCoord[up.CoordKey]
|
||||
}
|
||||
if id == 0 {
|
||||
id = byName[up.L.Name]
|
||||
}
|
||||
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)
|
||||
}
|
||||
placeID[up.L.SrcPK] = id
|
||||
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
|
||||
}
|
||||
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
|
||||
}
|
||||
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, m.TransportSrc, 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
|
||||
}
|
||||
|
||||
// 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.Printf("留档 %s#%d 序列化失败: %v", entity, pk, err)
|
||||
return nil
|
||||
}
|
||||
_, 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
|
||||
}
|
||||
Reference in New Issue
Block a user