rond-with-you/internal/web/backup_merge.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"
)

// 差异页确认后的合并执行。
//
// 全部改动在一个事务里完成：半合并的数据集比没合并更难收拾（比如地点建了、
// 到访没建，地图上就多一个永远没人去的点）。删除不在这里——删地点会级联删掉
// 它的全部到访，那件事单独再议。
//
// 编号是这里最容易出错的地方：**库里那侧的编号空间才是唯一有效的**。
// 新插入的行拿 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
	}
	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 后 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
}