package ingest import ( "context" "crypto/sha256" "database/sql" "encoding/hex" "encoding/json" "errors" "fmt" "io" "log" "os" "path/filepath" "sort" "time" "rond-with-you/internal/geo" "rond-with-you/internal/rond" "rond-with-you/internal/stats" ) type Stats struct { Visits int Places int RawVisits int Skipped int Identical bool // 内容与已有数据集完全一致,本次没有新建快照 FirstVisit *time.Time LastVisit *time.Time } // File 计算文件摘要,用于去重。 func File(path string) (string, int64, error) { f, err := os.Open(path) if err != nil { return "", 0, err } defer f.Close() h := sha256.New() n, err := io.Copy(h, f) if err != nil { return "", 0, err } return hex.EncodeToString(h.Sum(nil)), n, nil } // insertEntityRaw 把「实体 -> src_pk -> 完整原始行」写进 entity_raw。 // 单条失败不阻断导入(留档只影响导出保真度),但会记日志提示。 func insertEntityRaw(ctx context.Context, tx *sql.Tx, datasetID int64, rows map[string]map[int]map[string]any, skipped map[string]map[int]bool) error { for entity, byPK := range rows { for pk, raw := range byPK { blob, err := json.Marshal(raw) if err != nil { log.Printf("留档 %s#%d 序列化失败: %v", entity, pk, err) continue } // 传 string 而不是 []byte:pgx 会把 []byte 当 bytea,jsonb 列要文本 if _, err := tx.ExecContext(ctx, `INSERT INTO entity_raw (dataset_id, entity, src_pk, raw, skipped) VALUES ($1,$2,$3,$4,$5) ON CONFLICT (dataset_id, entity, src_pk) DO UPDATE SET raw=EXCLUDED.raw, skipped=EXCLUDED.skipped`, datasetID, entity, pk, string(blob), skipped[entity][pk]); err != nil { return fmt.Errorf("留档 %s 原始行: %w", entity, err) } } } return nil } // contentHash 给备份内容算指纹,用来识别「内容没变、只是重新导出了一次」的上传。 // rond 每次导出连 WAL 一起打包,字节必然不同,文件摘要挡不住这种重复。 // // 指纹按**到访/活动/地点的稳定标识**算,绝不能掺 Z_PK:rond 每次导出/同步都会重排 // 编号(实测两份备份里 822 条到访的 UUID 完全一致、Z_PK 100% 不同),拿编号当指纹会把 // 「同一批数据重导一次」判成新内容 —— 于是新建数据集,而备注与投票是按 dataset_id // 隔离的,换了数据集就再也看不见。 // // 没有 UUID 的老备份退回「到达时间 + 地点坐标」,仍比编号可靠。 func contentHash(b *rond.Backup) string { locCoord := make(map[int]string, len(b.Locations)) keys := make([]string, 0, len(b.Visits)+len(b.Locations)+len(b.Activities)) for _, l := range b.Locations { locCoord[l.SrcPK] = fmt.Sprintf("%.6f/%.6f", l.Lat, l.Lon) keys = append(keys, "L"+l.Name+"/"+locCoord[l.SrcPK]) } for _, v := range b.Visits { if v.Arrival == nil { continue // 导入时就被跳过的行,不该影响指纹 } if u := b.VisitUUID[v.SrcPK]; u != "" { keys = append(keys, "V"+u) continue } loc := "" if v.LocationSrcPK != nil { loc = locCoord[*v.LocationSrcPK] } keys = append(keys, fmt.Sprintf("T%d/%s", v.Arrival.Unix(), loc)) } for _, a := range b.Activities { if u := b.ActivityUUID[a.SrcPK]; u != "" { keys = append(keys, "A"+u) continue } keys = append(keys, "A"+a.Name) } sort.Strings(keys) h := sha256.New() for _, k := range keys { h.Write([]byte(k)) h.Write([]byte{'\n'}) } return hex.EncodeToString(h.Sum(nil)) } // Import 把解析后的备份写入 PostgreSQL,整个过程在一个事务内完成。 func Import(ctx context.Context, db *sql.DB, userID int64, originalName, sha string, size int64, b *rond.Backup) (int64, *Stats, error) { hash := contentHash(b) // 留档 Core Data 结构快照(实体编号 / DDL / 关联表)。失败不致命,留 NULL。 metaJSON, _ := json.Marshal(b.Meta) hasMeta := len(metaJSON) > 0 && string(metaJSON) != "null" // 内容与已有数据集完全一致时直接复用,不新建快照,也不改动当前展示的那一份。 var prevID int64 var prev Stats err := db.QueryRowContext(ctx, `SELECT id, visit_count, place_count, raw_visit_count FROM datasets WHERE user_id=$1 AND content_hash=$2 AND status='done' ORDER BY uploaded_at DESC LIMIT 1`, userID, hash). Scan(&prevID, &prev.Visits, &prev.Places, &prev.RawVisits) switch { case err == nil: if hasMeta { if _, e := db.ExecContext(ctx, `UPDATE datasets SET coredata_meta=$1 WHERE id=$2`, metaJSON, prevID); e != nil { log.Printf("补抓 Core Data 元信息失败: %v", e) } } prev.Identical = true return prevID, &prev, nil case !errors.Is(err, sql.ErrNoRows): return 0, nil, err } tx, err := db.BeginTx(ctx, nil) if err != nil { return 0, nil, err } defer tx.Rollback() var datasetID int64 err = tx.QueryRowContext(ctx, `INSERT INTO datasets (user_id, original_name, sha256, size_bytes, status) VALUES ($1,$2,$3,$4,'importing') RETURNING id`, userID, originalName, sha, size).Scan(&datasetID) if err != nil { return 0, nil, fmt.Errorf("创建数据集: %w", err) } // 缓存 Core Data 元信息,导出 rondbackup 时据此重建库结构。 if hasMeta { if _, err := tx.ExecContext(ctx, `UPDATE datasets SET coredata_meta=$1 WHERE id=$2`, metaJSON, datasetID); err != nil { return 0, nil, fmt.Errorf("写入 Core Data 元信息: %w", err) } } // 站点模型装不下的行(缺必填字段)记在这里,导出时按留档原样写回。 rawSkipped := map[string]map[int]bool{} skip := func(entity string, pk int) { if rawSkipped[entity] == nil { rawSkipped[entity] = map[int]bool{} } rawSkipped[entity][pk] = true } // 活动类型 actID := map[int]int64{} homeAct, workAct := map[int]bool{}, map[int]bool{} for _, a := range b.Activities { var id int64 if err := tx.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`, datasetID, a.SrcPK, a.Name, a.Color, a.Icon, a.IsHome, a.IsWork, a.IsExcluded, a.IsArchived).Scan(&id); err != nil { return 0, nil, fmt.Errorf("写入活动类型: %w", err) } actID[a.SrcPK] = id homeAct[a.SrcPK] = a.IsHome workAct[a.SrcPK] = a.IsWork } // 标签 tagID := map[int]int64{} for _, t := range b.Tags { var id int64 if err := tx.QueryRowContext(ctx, `INSERT INTO tags (dataset_id, src_pk, name, color, group_name) VALUES ($1,$2,$3,$4,NULLIF($5,'')) RETURNING id`, datasetID, t.SrcPK, t.Name, t.Color, t.GroupName).Scan(&id); err != nil { return 0, nil, fmt.Errorf("写入标签: %w", err) } tagID[t.SrcPK] = id } // 交通方式 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 } // 地点 placeID := map[int]int64{} locCoord := map[int][2]float64{} for _, l := range b.Locations { var id int64 if err := tx.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`, datasetID, l.SrcPK, 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 0, nil, fmt.Errorf("写入地点: %w", err) } placeID[l.SrcPK] = id locCoord[l.SrcPK] = [2]float64{l.Lat, l.Lon} } // 访问记录 visitID := map[int]int64{} var first, last *time.Time var vcount, skipped int visitPlace := map[int]int{} for _, v := range b.Visits { if v.Arrival == nil { skip("ZVISIT", v.SrcPK) continue } var pid any if v.LocationSrcPK != nil { if id, ok := placeID[*v.LocationSrcPK]; ok { pid = id } } var aid any isHome, isWork := false, false if v.ActivitySrcPK != nil { if id, ok := actID[*v.ActivitySrcPK]; ok { aid = id } isHome, isWork = homeAct[*v.ActivitySrcPK], workAct[*v.ActivitySrcPK] } var dur any if v.Departure != nil { m := int(v.Departure.Sub(*v.Arrival).Minutes()) if m < 0 { m = 0 } dur = m } var id int64 err := tx.QueryRowContext(ctx, `INSERT INTO visits (dataset_id, user_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,$12,NULLIF($13,''),NULLIF($14,''),NULLIF($15,'')) ON CONFLICT DO NOTHING RETURNING id`, datasetID, userID, v.SrcPK, pid, aid, v.Arrival, v.Departure, dur, isHome, isWork, v.Bookmarked, v.UserAdded, v.Remark, v.Emoji, v.WeatherSymbol).Scan(&id) if errors.Is(err, sql.ErrNoRows) { // 该到访已随更早的备份导入过,跳过即可,连同它的标签一起跳过 skipped++ continue } if err != nil { return 0, nil, fmt.Errorf("写入访问记录: %w", err) } visitID[v.SrcPK] = id vcount++ if v.LocationSrcPK != nil { visitPlace[v.SrcPK] = *v.LocationSrcPK } if first == nil || v.Arrival.Before(*first) { first = v.Arrival } if last == nil || v.Arrival.After(*last) { last = v.Arrival } for _, tpk := range v.TagSrcPKs { if tid, ok := tagID[tpk]; ok { if _, err := tx.ExecContext(ctx, `INSERT INTO visit_tags (visit_id, tag_id) VALUES ($1,$2) ON CONFLICT DO NOTHING`, id, tid); err != nil { return 0, nil, fmt.Errorf("写入访问标签: %w", err) } } } } // 位移 for _, m := range b.Movements { var dist any var fromPID, toPID any if m.FromVisitSrc != nil && m.ToVisitSrc != nil { fp, okf := visitPlace[*m.FromVisitSrc] tp, okt := visitPlace[*m.ToVisitSrc] if okf && okt { if id, ok := placeID[fp]; ok { fromPID = id } if id, ok := placeID[tp]; ok { toPID = id } c1, c2 := locCoord[fp], locCoord[tp] if c1[0] != 0 && c2[0] != 0 { d := geo.Haversine(c1[0], c1[1], c2[0], c2[1]) dist = d } } } var tname, tcolor, ticon any if m.TransportSrc != nil { tname, tcolor, ticon = nullStr(transName[*m.TransportSrc]), nullStr(transColor[*m.TransportSrc]), 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 } } if _, err := tx.ExecContext(ctx, `INSERT INTO movements (dataset_id, user_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,$15) ON CONFLICT DO NOTHING`, datasetID, userID, m.SrcPK, m.TransportSrc, tname, tcolor, ticon, m.FromVisitSrc, m.ToVisitSrc, fromPID, toPID, m.Start, m.End, dur, dist); err != nil { return 0, nil, fmt.Errorf("写入位移记录: %w", err) } } // 小时级天气(去重,仅保留整点) for _, w := range b.Weather { if w.At == nil { skip("ZHOURLYWEATHER", w.SrcPK) continue } if _, err := tx.ExecContext(ctx, `INSERT INTO weather (dataset_id, user_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,$14,NULLIF($15,''),NULLIF($16,''),NULLIF($17,''),NULLIF($18,'')) ON CONFLICT DO NOTHING`, datasetID, userID, w.SrcPK, w.VisitSrc, 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 0, nil, fmt.Errorf("写入天气记录: %w", err) } } // 汇总 st := &Stats{Visits: vcount, Skipped: skipped, Places: len(b.Locations), RawVisits: b.RawVisitNum, FirstVisit: first, LastVisit: last} if _, err := tx.ExecContext(ctx, `UPDATE datasets SET status='done', visit_count=$1, place_count=$2, raw_visit_count=$3, first_visit_at=$4, last_visit_at=$5, source_created_at=$6, content_hash=$7 WHERE id=$8`, st.Visits, st.Places, st.RawVisits, first, last, time.Now(), hash, datasetID); err != nil { return 0, nil, fmt.Errorf("更新数据集汇总: %w", err) } // 快照式语义:最新导入的那一份接管展示,其余自动退居后台(后台仍可手动切回) if _, err := tx.ExecContext(ctx, `UPDATE datasets SET is_active = (id=$1) WHERE user_id=$2`, datasetID, userID); err != nil { return 0, nil, fmt.Errorf("切换生效数据集: %w", err) } // 原始行留档:站点只映射了一部分字段,把 Core Data 里的整行存下来, // 导出时逐列还原——这样既不受映射范围限制,也不依赖基线库去补剩下的列。 if err := insertEntityRaw(ctx, tx, datasetID, b.RawRows, rawSkipped); err != nil { return 0, nil, err } // 派生列:从留档里认「已并入 / 子段」的到访(读侧只看这一列,见 stats.SoloVisitSQL)。 // ⚠️ 同样必须排在 insertEntityRaw 之后——留档还没写盘时一条都认不出来。 if _, err := tx.ExecContext(ctx, stats.RefreshMergedFlagsSQL, datasetID); err != nil { return 0, nil, fmt.Errorf("回填合并标记: %w", err) } // 回填地点聚合:走与刷新路径(refreshPlaceAgg / recountPlaces)同一条 SQL, // 口径才不会在两处漂移——到访数与停留按访客口径(stats.SoloVisitSQL,不含已并入 / 子段), // 首末到访保持全量。原来这里在 Go 里另算一遍,还漏了停留上限(LEAST(...,10080))。 // // ⚠️ 必须排在 insertEntityRaw **之后**:判据读的是 entity_raw,留档还没写盘时 // 每一行都会被当成独立到访(实测会整整多算一个「已并入」的量,26 条)。 if _, err := tx.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`, datasetID); err != nil { return 0, nil, fmt.Errorf("回填地点统计: %w", err) } // 坐标来源留档要等 entity_raw 写完才能做(原始 GPS 只在留档行里)。 if _, err := BackfillCoordSource(ctx, tx, datasetID); err != nil { return 0, nil, err } if err := tx.Commit(); err != nil { return 0, nil, err } return datasetID, st, nil } func nullStr(s string) any { if s == "" { return nil } return s } // Cached 记录已有上传的摘要,供重复上传时直接复用。 type Cached struct { DatasetID int64 Stats Stats } func SortLocations(list []rond.Location) { sort.Slice(list, func(i, j int) bool { return list[i].SrcPK < list[j].SrcPK }) } func TempDir(base, name string) (string, error) { d := filepath.Join(base, fmt.Sprintf("import-%d", time.Now().UnixNano())) if err := os.MkdirAll(d, 0o700); err != nil { return "", err } return d, nil }