tamakyi
c9ebb3aebd
昨天给「已并入 / 子段」做的收口写成每行去 entity_raw 判 `raw->>'ZMERGEDTO' IS NOT NULL OR raw->>'ZPARENT' IS NOT NULL`，而 entity_raw.raw 是 **TOAST 过的整行 Core Data JSON**：一次探针要取回并解析整行。地点库「每个地点 × 5 个子查询」 的量级下 —— 实测 /places 70.1 秒、首页 43.6 秒才出页面（而库里没有任何一条 SQL 超过 1 秒： 时间全花在成千上万次「取整行再解析」上，进程 CPU 只用了 3s/135s）。 这批「渲染若干页面」的布局用例因此从 20 秒涨到 226~309 秒，整套验收从 6 分钟变 20+ 分钟。 - schema.sql：加派生列 visits.is_merged + 回填 + 给回填用的部分索引 （只索引确实带合并标记的那几十行，不用取回 raw）。 - stats.SoloVisitSQL 改成读这一列（判据仍然只此一份，读侧全部经由它）。 - 写入侧跟着刷：ingest 在 insertEntityRaw 之后、backup_merge 的 recountPlaces 里， 共用 stats.RefreshMergedFlagsSQL。 - 直接改留档里的合并标记（工具脚本 / 用例）之后也要刷一次，schema.sql 与函数注释都写明了。 实测：/places 70084ms → 206ms，/ 43633ms → 698ms（比「关掉判据」的对照还快）。 新增 page_render_budget_test.go：给 7 个关键页卡 10 秒上限的灾难探测器 （功能全对、只是慢的那种退化，之前没有任何用例会红）。
821 lines
28 KiB
Go
821 lines
28 KiB
Go
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"
|
||
"rond-with-you/internal/stats"
|
||
)
|
||
|
||
// 差异页确认后的合并执行。
|
||
//
|
||
// 全部改动在一个事务里完成：半合并的数据集比没合并更难收拾（比如地点建了、
|
||
// 到访没建，地图上就多一个永远没人去的点）。删除不在这里——删地点会级联删掉
|
||
// 它的全部到访，那件事单独再议。
|
||
//
|
||
// 编号是这里最容易出错的地方：**库里那侧的编号空间才是唯一有效的**。
|
||
// 新插入的行拿 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 {
|
||
// 站点补录的行没有留档 UUID，用它的派生 UUID 当键（与差异页同一套对齐，见 rond.VisitMatchUUID）
|
||
if u := rond.VisitMatchUUID(sv); u != "" {
|
||
storedByUUID[u] = 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 就够，省得漏掉任何一条派生列。
|
||
//
|
||
// 到访数与停留按**访客口径**（不含已并入 / 子段，见 stats.SoloVisitSQL），与 stats.Filter 一致；
|
||
// 首末到访与数据集计数保持全量（last_visit_at 会写进导出包，datasets.visit_count 是后台口径）。
|
||
func (s *Server) recountPlaces(ctx context.Context, ex execer, dsID int64) error {
|
||
// 先把派生列刷新（合并刚插进来的行带着留档），下面的聚合读的就是它
|
||
if _, err := ex.ExecContext(ctx, stats.RefreshMergedFlagsSQL, dsID); err != nil {
|
||
return err
|
||
}
|
||
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 vis.place_id,
|
||
count(*) FILTER (WHERE `+stats.SoloVisitSQL("vis")+`) c,
|
||
COALESCE(sum(LEAST(COALESCE(vis.duration_min,0),10080)) FILTER (WHERE `+stats.SoloVisitSQL("vis")+`),0) d,
|
||
min(vis.arrival) f, max(vis.arrival) l
|
||
FROM visits vis WHERE vis.dataset_id=$1 AND vis.place_id IS NOT NULL GROUP BY vis.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
|
||
}
|