lingniu-vehicle-ingest/tmp/g7-mileage-history-20260724/importer.go

package main

import (
	"compress/gzip"
	"context"
	"database/sql"
	"encoding/csv"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	mysql "github.com/go-sql-driver/mysql"
)

const (
	g7Protocol      = "JT808"
	g7SourceID      = int64(661200)
	g7SourceIP      = "manual-g7s-excel"
	g7SourceCode    = "g7s_excel"
	g7PlatformName  = "G7s Excel Override"
	g7QualityReason = "manual_g7s_gps_quarterly_import_20260101_20260630_20260728"
	dateFrom        = "2026-01-01"
	dateTo          = "2026-06-30"
	expectedDays    = 181
)

type mapping struct {
	VIN       string
	Ambiguous bool
}

type sourceStats struct {
	Rows       int64   `json:"rows"`
	Vehicles   int     `json:"vehicles"`
	Dates      int     `json:"dates"`
	Positive   int64   `json:"positive_rows"`
	Zero       int64   `json:"zero_rows"`
	TotalKM    float64 `json:"total_km"`
	MinDate    string  `json:"min_date"`
	MaxDate    string  `json:"max_date"`
	PlateCount int     `json:"plate_count"`
}

type stageStats struct {
	Rows               int64   `json:"rows"`
	Vehicles           int     `json:"vehicles"`
	Dates              int     `json:"dates"`
	Positive           int64   `json:"positive_rows"`
	Zero               int64   `json:"zero_rows"`
	TotalKM            float64 `json:"total_km"`
	MappedPlates       int     `json:"mapped_plates"`
	UnmappedPlates     int     `json:"unmapped_plates"`
	AmbiguousPlates    int     `json:"ambiguous_plates"`
	VINCollisionPlates int     `json:"vin_collision_plates"`
}

type existingStats struct {
	ExistingFinalRows          int64   `json:"existing_final_rows"`
	NewFinalRows               int64   `json:"new_final_rows"`
	ChangedExistingFinalRows   int64   `json:"changed_existing_final_rows"`
	UnchangedExistingFinalRows int64   `json:"unchanged_existing_final_rows"`
	OtherSourceFinalRows       int64   `json:"other_source_final_rows"`
	OtherSelectedCandidateRows int64   `json:"other_selected_candidate_rows"`
	ExistingFinalTotalKM       float64 `json:"existing_final_total_km"`
}

type result struct {
	Mode                 string        `json:"mode"`
	StartedAt            string        `json:"started_at"`
	CompletedAt          string        `json:"completed_at"`
	Source               sourceStats   `json:"source"`
	Stage                stageStats    `json:"stage"`
	Existing             existingStats `json:"existing"`
	UnmappedPlates       []string      `json:"unmapped_plates"`
	AmbiguousPlates      []string      `json:"ambiguous_plates"`
	CollisionPlates      []string      `json:"vin_collision_plates"`
	BackupFinal          string        `json:"backup_final,omitempty"`
	BackupCandidates     string        `json:"backup_candidates,omitempty"`
	FinalMatchedRows     int64         `json:"final_matched_rows"`
	FinalTotalKM         float64       `json:"final_total_km"`
	CandidateMatchedRows int64         `json:"candidate_matched_rows"`
	OtherSelectedRows    int64         `json:"other_selected_rows"`
	SourceID             int64         `json:"source_id"`
}

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

func run() error {
	var inputPath, mappingPath, dsn, backupDir string
	var apply bool
	var timeout time.Duration
	flag.StringVar(&inputPath, "input", "", "normalized CSV or CSV.gz")
	flag.StringVar(&mappingPath, "mapping-csv", "", "optional existing vin,plate mapping CSV")
	flag.StringVar(&dsn, "mysql-dsn", strings.TrimSpace(os.Getenv("MYSQL_DSN")), "MySQL DSN")
	flag.StringVar(&backupDir, "backup-dir", "/tmp/g7-gps-quarterly-backup-20260728", "backup directory for apply")
	flag.BoolVar(&apply, "apply", false, "apply changes; default is dry-run")
	flag.DurationVar(&timeout, "timeout", 30*time.Minute, "overall timeout")
	flag.Parse()

	if strings.TrimSpace(inputPath) == "" {
		return errors.New("-input is required")
	}
	if strings.TrimSpace(dsn) == "" {
		return errors.New("MYSQL_DSN or -mysql-dsn is required")
	}
	normalized, err := normalizedDSN(dsn)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	db, err := sql.Open("mysql", normalized)
	if err != nil {
		return err
	}
	defer db.Close()
	db.SetMaxOpenConns(2)
	db.SetMaxIdleConns(2)
	if err := db.PingContext(ctx); err != nil {
		return err
	}
	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()

	started := time.Now()
	source, inputPlates, err := scanInput(inputPath)
	if err != nil {
		return err
	}
	if source.MinDate != dateFrom || source.MaxDate != dateTo || source.Dates != expectedDays {
		return fmt.Errorf("unexpected input range: %s..%s (%d dates)", source.MinDate, source.MaxDate, source.Dates)
	}
	mappings, err := loadMappings(ctx, conn)
	if err != nil {
		return err
	}
	if mappingPath != "" {
		if err := mergeMappingCSV(mappingPath, mappings); err != nil {
			return err
		}
	}
	resolution := resolveInputPlates(inputPlates, mappings)
	if err := createStage(ctx, conn); err != nil {
		return err
	}
	stage, err := loadStage(ctx, conn, inputPath, resolution)
	if err != nil {
		return err
	}
	stage.MappedPlates = len(resolution.Mapped)
	stage.UnmappedPlates = len(resolution.Unmapped)
	stage.AmbiguousPlates = len(resolution.Ambiguous)
	stage.VINCollisionPlates = len(resolution.Collisions)
	if stage.Rows != int64(stage.MappedPlates*expectedDays) {
		return fmt.Errorf("stage row count mismatch: got %d, expected %d", stage.Rows, stage.MappedPlates*expectedDays)
	}
	existing, err := inspectExisting(ctx, conn)
	if err != nil {
		return err
	}
	out := result{
		Mode:            "dry_run",
		StartedAt:       started.Format(time.RFC3339),
		Source:          source,
		Stage:           stage,
		Existing:        existing,
		UnmappedPlates:  resolution.Unmapped,
		AmbiguousPlates: resolution.Ambiguous,
		CollisionPlates: resolution.Collisions,
		SourceID:        g7SourceID,
	}

	if apply {
		out.Mode = "apply"
		if err := os.MkdirAll(backupDir, 0o700); err != nil {
			return err
		}
		out.BackupFinal = filepath.Join(backupDir, "vehicle_daily_mileage_before.csv.gz")
		out.BackupCandidates = filepath.Join(backupDir, "vehicle_daily_mileage_source_before.csv.gz")
		if err := backupQuery(ctx, conn, out.BackupFinal, `
SELECT m.*
FROM vehicle_daily_mileage m
JOIN tmp_g7_mileage_import s
  ON s.vin = m.vin AND s.stat_date = m.stat_date
WHERE m.protocol = 'JT808'
ORDER BY m.vin, m.stat_date`); err != nil {
			return fmt.Errorf("backup final rows: %w", err)
		}
		if err := backupQuery(ctx, conn, out.BackupCandidates, `
SELECT c.*
FROM vehicle_daily_mileage_source c
JOIN tmp_g7_mileage_import s
  ON s.vin = c.vin AND s.stat_date = c.stat_date
WHERE c.protocol = 'JT808'
ORDER BY c.vin, c.stat_date, c.source_key`); err != nil {
			return fmt.Errorf("backup candidate rows: %w", err)
		}
		finalRows, finalTotal, candidateRows, otherSelected, err := applyImport(ctx, conn)
		if err != nil {
			return err
		}
		out.FinalMatchedRows = finalRows
		out.FinalTotalKM = finalTotal
		out.CandidateMatchedRows = candidateRows
		out.OtherSelectedRows = otherSelected
	}
	out.CompletedAt = time.Now().Format(time.RFC3339)
	encoder := json.NewEncoder(os.Stdout)
	encoder.SetIndent("", "  ")
	return encoder.Encode(out)
}

type plateResolution struct {
	Mapped     map[string]string
	Unmapped   []string
	Ambiguous  []string
	Collisions []string
}

func resolveInputPlates(plates map[string]struct{}, mappings map[string]mapping) plateResolution {
	result := plateResolution{Mapped: make(map[string]string)}
	vinToPlates := make(map[string][]string)
	for plate := range plates {
		value, ok := mappings[plate]
		switch {
		case !ok:
			result.Unmapped = append(result.Unmapped, plate)
		case value.Ambiguous:
			result.Ambiguous = append(result.Ambiguous, plate)
		default:
			result.Mapped[plate] = value.VIN
			vinToPlates[value.VIN] = append(vinToPlates[value.VIN], plate)
		}
	}
	for _, presentPlates := range vinToPlates {
		if len(presentPlates) < 2 {
			continue
		}
		for _, plate := range presentPlates {
			delete(result.Mapped, plate)
			result.Collisions = append(result.Collisions, plate)
		}
	}
	sort.Strings(result.Unmapped)
	sort.Strings(result.Ambiguous)
	sort.Strings(result.Collisions)
	return result
}

func scanInput(inputPath string) (sourceStats, map[string]struct{}, error) {
	reader, closeReader, err := openCSV(inputPath)
	if err != nil {
		return sourceStats{}, nil, err
	}
	defer closeReader()
	csvReader := csv.NewReader(reader)
	header, err := csvReader.Read()
	if err != nil {
		return sourceStats{}, nil, err
	}
	if len(header) < 3 {
		return sourceStats{}, nil, errors.New("input CSV needs plate,date,daily_mileage_km")
	}
	header[0] = strings.TrimPrefix(header[0], "\uFEFF")
	if header[0] != "plate" || header[1] != "date" || header[2] != "daily_mileage_km" {
		return sourceStats{}, nil, fmt.Errorf("unexpected CSV header: %v", header)
	}
	plates := make(map[string]struct{})
	dates := make(map[string]struct{})
	stats := sourceStats{}
	for {
		record, readErr := csvReader.Read()
		if errors.Is(readErr, io.EOF) {
			break
		}
		if readErr != nil {
			return sourceStats{}, nil, readErr
		}
		if len(record) < 3 {
			return sourceStats{}, nil, fmt.Errorf("short CSV row at %d", stats.Rows+2)
		}
		plate := strings.TrimSpace(record[0])
		date := strings.TrimSpace(record[1])
		mileage, parseErr := strconv.ParseFloat(strings.TrimSpace(record[2]), 64)
		if parseErr != nil || mileage < 0 || mileage > 2500 {
			return sourceStats{}, nil, fmt.Errorf("invalid mileage at row %d: %q", stats.Rows+2, record[2])
		}
		if _, parseErr = time.Parse("2006-01-02", date); parseErr != nil {
			return sourceStats{}, nil, fmt.Errorf("invalid date at row %d: %q", stats.Rows+2, date)
		}
		if plate == "" {
			return sourceStats{}, nil, fmt.Errorf("empty plate at row %d", stats.Rows+2)
		}
		plates[plate] = struct{}{}
		dates[date] = struct{}{}
		stats.Rows++
		stats.TotalKM += mileage
		if mileage > 0 {
			stats.Positive++
		} else {
			stats.Zero++
		}
		if stats.MinDate == "" || date < stats.MinDate {
			stats.MinDate = date
		}
		if date > stats.MaxDate {
			stats.MaxDate = date
		}
	}
	stats.PlateCount = len(plates)
	stats.Vehicles = len(plates)
	stats.Dates = len(dates)
	stats.TotalKM = round(stats.TotalKM)
	return stats, plates, nil
}

func loadMappings(ctx context.Context, conn *sql.Conn) (map[string]mapping, error) {
	rows, err := conn.QueryContext(ctx, `
SELECT plate, vin
FROM (
  SELECT TRIM(plate) AS plate, TRIM(vin) AS vin
  FROM vehicle
  WHERE plate IS NOT NULL AND TRIM(plate) <> ''
    AND vin IS NOT NULL AND TRIM(vin) <> ''
  UNION ALL
  SELECT TRIM(plate) AS plate, TRIM(vin) AS vin
  FROM vehicle_identifier
  WHERE enabled = 1 AND plate IS NOT NULL AND TRIM(plate) <> ''
    AND vin IS NOT NULL AND TRIM(vin) <> ''
) mappings`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make(map[string]mapping)
	for rows.Next() {
		var plate, vin string
		if err := rows.Scan(&plate, &vin); err != nil {
			return nil, err
		}
		current, ok := result[plate]
		if !ok {
			result[plate] = mapping{VIN: vin}
		} else if current.VIN != vin {
			current.Ambiguous = true
			result[plate] = current
		}
	}
	return result, rows.Err()
}

func mergeMappingCSV(inputPath string, mappings map[string]mapping) error {
	file, err := os.Open(inputPath)
	if err != nil {
		return err
	}
	defer file.Close()
	reader := csv.NewReader(file)
	header, err := reader.Read()
	if err != nil {
		return err
	}
	if len(header) < 2 {
		return errors.New("mapping CSV must contain vin,plate")
	}
	header[0] = strings.TrimPrefix(header[0], "\uFEFF")
	if header[0] != "vin" || header[1] != "plate" {
		return fmt.Errorf("unexpected mapping CSV header: %v", header[:2])
	}
	for {
		record, readErr := reader.Read()
		if errors.Is(readErr, io.EOF) {
			break
		}
		if readErr != nil {
			return readErr
		}
		if len(record) < 2 {
			continue
		}
		vin := strings.TrimSpace(record[0])
		plate := strings.TrimSpace(record[1])
		if vin == "" || plate == "" {
			continue
		}
		current, ok := mappings[plate]
		if !ok {
			mappings[plate] = mapping{VIN: vin}
		} else if current.VIN != vin {
			current.Ambiguous = true
			mappings[plate] = current
		}
	}
	return nil
}

func createStage(ctx context.Context, conn *sql.Conn) error {
	_, err := conn.ExecContext(ctx, `
DROP TEMPORARY TABLE IF EXISTS tmp_g7_mileage_import`)
	if err != nil {
		return err
	}
	_, err = conn.ExecContext(ctx, `
CREATE TEMPORARY TABLE tmp_g7_mileage_import (
  vin varchar(32) NOT NULL,
  plate varchar(32) NOT NULL,
  stat_date date NOT NULL,
  daily_mileage_km decimal(18,3) NOT NULL,
  PRIMARY KEY (vin, stat_date),
  KEY idx_tmp_plate (plate),
  KEY idx_tmp_date (stat_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`)
	return err
}

func loadStage(ctx context.Context, conn *sql.Conn, inputPath string, resolution plateResolution) (stageStats, error) {
	reader, closeReader, err := openCSV(inputPath)
	if err != nil {
		return stageStats{}, err
	}
	defer closeReader()
	csvReader := csv.NewReader(reader)
	if _, err := csvReader.Read(); err != nil {
		return stageStats{}, err
	}
	tx, err := conn.BeginTx(ctx, nil)
	if err != nil {
		return stageStats{}, err
	}
	defer tx.Rollback()

	type stageRow struct {
		VIN, Plate, Date string
		Mileage          float64
	}
	batch := make([]stageRow, 0, 1000)
	flush := func() error {
		if len(batch) == 0 {
			return nil
		}
		var query strings.Builder
		query.WriteString("INSERT INTO tmp_g7_mileage_import (vin, plate, stat_date, daily_mileage_km) VALUES ")
		args := make([]any, 0, len(batch)*4)
		for index, row := range batch {
			if index > 0 {
				query.WriteByte(',')
			}
			query.WriteString("(?,?,?,?)")
			args = append(args, row.VIN, row.Plate, row.Date, row.Mileage)
		}
		if _, execErr := tx.ExecContext(ctx, query.String(), args...); execErr != nil {
			return execErr
		}
		batch = batch[:0]
		return nil
	}
	for {
		record, readErr := csvReader.Read()
		if errors.Is(readErr, io.EOF) {
			break
		}
		if readErr != nil {
			return stageStats{}, readErr
		}
		plate := strings.TrimSpace(record[0])
		vin, ok := resolution.Mapped[plate]
		if !ok {
			continue
		}
		mileage, _ := strconv.ParseFloat(strings.TrimSpace(record[2]), 64)
		batch = append(batch, stageRow{
			VIN: vin, Plate: plate, Date: strings.TrimSpace(record[1]), Mileage: mileage,
		})
		if len(batch) == cap(batch) {
			if err := flush(); err != nil {
				return stageStats{}, err
			}
		}
	}
	if err := flush(); err != nil {
		return stageStats{}, err
	}
	if err := tx.Commit(); err != nil {
		return stageStats{}, err
	}
	var stats stageStats
	var nullableSum sql.NullFloat64
	err = conn.QueryRowContext(ctx, `
SELECT COUNT(*), COUNT(DISTINCT vin), COUNT(DISTINCT stat_date),
       SUM(daily_mileage_km > 0), SUM(daily_mileage_km = 0),
       SUM(daily_mileage_km)
FROM tmp_g7_mileage_import`).Scan(
		&stats.Rows, &stats.Vehicles, &stats.Dates, &stats.Positive, &stats.Zero, &nullableSum,
	)
	if nullableSum.Valid {
		stats.TotalKM = round(nullableSum.Float64)
	}
	return stats, err
}

func inspectExisting(ctx context.Context, conn *sql.Conn) (existingStats, error) {
	var stats existingStats
	var nullableSum sql.NullFloat64
	err := conn.QueryRowContext(ctx, `
SELECT
  SUM(m.vin IS NOT NULL),
  SUM(m.vin IS NULL),
  SUM(m.vin IS NOT NULL AND ABS(m.daily_mileage_km - s.daily_mileage_km) > 0.0005),
  SUM(m.vin IS NOT NULL AND ABS(m.daily_mileage_km - s.daily_mileage_km) <= 0.0005),
  SUM(m.vin IS NOT NULL AND COALESCE(m.source_id, 0) <> ?),
  SUM(CASE WHEN m.vin IS NOT NULL THEN m.daily_mileage_km ELSE 0 END)
FROM tmp_g7_mileage_import s
LEFT JOIN vehicle_daily_mileage m
  ON m.vin = s.vin AND m.stat_date = s.stat_date AND m.protocol = 'JT808'`, g7SourceID).Scan(
		&stats.ExistingFinalRows,
		&stats.NewFinalRows,
		&stats.ChangedExistingFinalRows,
		&stats.UnchangedExistingFinalRows,
		&stats.OtherSourceFinalRows,
		&nullableSum,
	)
	if err != nil {
		return stats, err
	}
	if nullableSum.Valid {
		stats.ExistingFinalTotalKM = round(nullableSum.Float64)
	}
	err = conn.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM vehicle_daily_mileage_source c
JOIN tmp_g7_mileage_import s
  ON s.vin = c.vin AND s.stat_date = c.stat_date
WHERE c.protocol = 'JT808' AND c.is_selected = 1
  AND c.source_key <> CONCAT('JT808:', c.vin, '@PLATFORM:g7s_excel')`).Scan(
		&stats.OtherSelectedCandidateRows,
	)
	return stats, err
}

func applyImport(ctx context.Context, conn *sql.Conn) (int64, float64, int64, int64, error) {
	tx, err := conn.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return 0, 0, 0, 0, err
	}
	defer tx.Rollback()
	if _, err := tx.ExecContext(ctx, `
UPDATE vehicle_daily_mileage_source c
JOIN tmp_g7_mileage_import s
  ON s.vin = c.vin AND s.stat_date = c.stat_date
SET c.is_selected = 0
WHERE c.protocol = 'JT808'
  AND c.source_key <> CONCAT('JT808:', c.vin, '@PLATFORM:g7s_excel')
  AND c.is_selected <> 0`); err != nil {
		return 0, 0, 0, 0, fmt.Errorf("clear other selected candidates: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
INSERT INTO vehicle_daily_mileage_source (
  vin, stat_date, protocol, source_key, source_ip, source_endpoint,
  platform_name, first_total_mileage_km, latest_total_mileage_km,
  daily_mileage_km, sample_count, first_event_time, latest_event_time,
  quality_status, quality_reason, is_selected
)
SELECT
  s.vin, s.stat_date, 'JT808',
  CONCAT('JT808:', s.vin, '@PLATFORM:g7s_excel'),
  'manual-g7s-excel', 'manual-g7s-excel',
  'G7s Excel Override', NULL, NULL,
  s.daily_mileage_km, 1,
  TIMESTAMP(s.stat_date, '00:00:00'),
  TIMESTAMP(s.stat_date, '23:59:59'),
  'OK', ?, 1
FROM tmp_g7_mileage_import s
ON DUPLICATE KEY UPDATE
  source_ip = VALUES(source_ip),
  source_endpoint = VALUES(source_endpoint),
  platform_name = VALUES(platform_name),
  first_total_mileage_km = NULL,
  latest_total_mileage_km = NULL,
  daily_mileage_km = VALUES(daily_mileage_km),
  sample_count = VALUES(sample_count),
  first_event_time = VALUES(first_event_time),
  latest_event_time = VALUES(latest_event_time),
  quality_status = VALUES(quality_status),
  quality_reason = VALUES(quality_reason),
  is_selected = 1`, g7QualityReason); err != nil {
		return 0, 0, 0, 0, fmt.Errorf("upsert G7 candidates: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
INSERT INTO vehicle_daily_mileage (
  vin, stat_date, protocol, source_id, daily_mileage_km,
  latest_total_mileage_km, updated_at
)
SELECT vin, stat_date, 'JT808', ?, daily_mileage_km, NULL, NOW()
FROM tmp_g7_mileage_import
ON DUPLICATE KEY UPDATE
  source_id = VALUES(source_id),
  daily_mileage_km = VALUES(daily_mileage_km),
  latest_total_mileage_km = NULL,
  updated_at = NOW()`, g7SourceID); err != nil {
		return 0, 0, 0, 0, fmt.Errorf("upsert final mileage: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
UPDATE vehicle_data_source
SET latest_seen_at = GREATEST(COALESCE(latest_seen_at, '2026-06-30 23:59:59'), '2026-06-30 23:59:59'),
    first_seen_at = LEAST(COALESCE(first_seen_at, '2026-01-01 00:00:00'), '2026-01-01 00:00:00'),
    latest_source_endpoint = 'manual-g7s-excel',
    platform_name = 'G7s Excel Override',
    source_code = 'g7s_excel',
    source_kind = 'PLATFORM',
    trust_priority = 0,
    enabled = 1,
    remark = 'Manual GPS override imported from G7s quarterly daily mileage reports for 2026-01-01 through 2026-06-30 on 2026-07-28'
WHERE id = ? AND protocol = 'JT808' AND source_ip = 'manual-g7s-excel'`, g7SourceID); err != nil {
		return 0, 0, 0, 0, fmt.Errorf("update G7 source metadata: %w", err)
	}
	var finalRows, candidateRows, otherSelected int64
	var finalSum sql.NullFloat64
	if err := tx.QueryRowContext(ctx, `
SELECT COUNT(*), SUM(m.daily_mileage_km)
FROM vehicle_daily_mileage m
JOIN tmp_g7_mileage_import s
  ON s.vin = m.vin AND s.stat_date = m.stat_date
WHERE m.protocol = 'JT808' AND m.source_id = ?
  AND ABS(m.daily_mileage_km - s.daily_mileage_km) <= 0.0005`, g7SourceID).Scan(
		&finalRows, &finalSum,
	); err != nil {
		return 0, 0, 0, 0, err
	}
	if err := tx.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM vehicle_daily_mileage_source c
JOIN tmp_g7_mileage_import s
  ON s.vin = c.vin AND s.stat_date = c.stat_date
WHERE c.protocol = 'JT808'
  AND c.source_key = CONCAT('JT808:', c.vin, '@PLATFORM:g7s_excel')
  AND c.is_selected = 1
  AND ABS(c.daily_mileage_km - s.daily_mileage_km) <= 0.0005`).Scan(&candidateRows); err != nil {
		return 0, 0, 0, 0, err
	}
	if err := tx.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM vehicle_daily_mileage_source c
JOIN tmp_g7_mileage_import s
  ON s.vin = c.vin AND s.stat_date = c.stat_date
WHERE c.protocol = 'JT808'
  AND c.source_key <> CONCAT('JT808:', c.vin, '@PLATFORM:g7s_excel')
  AND c.is_selected = 1`).Scan(&otherSelected); err != nil {
		return 0, 0, 0, 0, err
	}
	var stageRows int64
	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM tmp_g7_mileage_import`).Scan(&stageRows); err != nil {
		return 0, 0, 0, 0, err
	}
	if finalRows != stageRows || candidateRows != stageRows || otherSelected != 0 {
		return 0, 0, 0, 0, fmt.Errorf(
			"verification failed before commit: stage=%d final=%d candidates=%d other_selected=%d",
			stageRows, finalRows, candidateRows, otherSelected,
		)
	}
	if err := tx.Commit(); err != nil {
		return 0, 0, 0, 0, err
	}
	return finalRows, round(finalSum.Float64), candidateRows, otherSelected, nil
}

func backupQuery(ctx context.Context, conn *sql.Conn, outputPath, query string) error {
	rows, err := conn.QueryContext(ctx, query)
	if err != nil {
		return err
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		return err
	}
	file, err := os.OpenFile(outputPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
	if err != nil {
		return err
	}
	defer file.Close()
	gzipWriter := gzip.NewWriter(file)
	defer gzipWriter.Close()
	csvWriter := csv.NewWriter(gzipWriter)
	defer csvWriter.Flush()
	if err := csvWriter.Write(columns); err != nil {
		return err
	}
	raw := make([]sql.RawBytes, len(columns))
	pointers := make([]any, len(columns))
	for index := range raw {
		pointers[index] = &raw[index]
	}
	for rows.Next() {
		if err := rows.Scan(pointers...); err != nil {
			return err
		}
		record := make([]string, len(columns))
		for index, value := range raw {
			if value == nil {
				record[index] = "\\N"
			} else {
				record[index] = string(value)
			}
		}
		if err := csvWriter.Write(record); err != nil {
			return err
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	csvWriter.Flush()
	if err := csvWriter.Error(); err != nil {
		return err
	}
	if err := gzipWriter.Close(); err != nil {
		return err
	}
	return file.Sync()
}

func openCSV(inputPath string) (io.Reader, func(), error) {
	file, err := os.Open(inputPath)
	if err != nil {
		return nil, nil, err
	}
	closeReader := func() { _ = file.Close() }
	if !strings.HasSuffix(strings.ToLower(inputPath), ".gz") {
		return file, closeReader, nil
	}
	gzipReader, err := gzip.NewReader(file)
	if err != nil {
		file.Close()
		return nil, nil, err
	}
	return gzipReader, func() {
		_ = gzipReader.Close()
		_ = file.Close()
	}, nil
}

func normalizedDSN(raw string) (string, error) {
	config, err := mysql.ParseDSN(raw)
	if err != nil {
		return "", err
	}
	location, err := time.LoadLocation("Asia/Shanghai")
	if err != nil {
		return "", err
	}
	config.ParseTime = true
	config.Loc = location
	config.Timeout = 10 * time.Second
	config.ReadTimeout = 30 * time.Minute
	config.WriteTimeout = 30 * time.Minute
	config.Params = cloneParams(config.Params)
	config.Params["charset"] = "utf8mb4"
	return config.FormatDSN(), nil
}

func cloneParams(input map[string]string) map[string]string {
	output := make(map[string]string, len(input)+1)
	for key, value := range input {
		output[key] = value
	}
	return output
}

func round(value float64) float64 {
	return float64(int64(value*100+0.5)) / 100
}