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_=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 }