DEV-MERGE-LEGACY-BACKEND-2026/internal/academic/dualrun/fixture/batch3_run.go

package fixture

import (
	"context"
	"database/sql"
	"fmt"
	"io"
	"sort"
)

// 本文件是第三批的流程编排与报告排版。

// OrgReport 是一次「搬 org 域」的结果。
type OrgReport struct {
	Options  OrgOptions
	Schools  int
	Campuses int
	Warnings []string
	InstIDs  []int64
	CampIDs  []int64
}

// PrepareOrg 把选中的机构与其校区搬进目标库（D1 原值插入）。
func PrepareOrg(ctx context.Context, cfg Config, opts OrgOptions) (*OrgReport, error) {
	src, err := cfg.OpenOrgSource()
	if err != nil {
		return nil, err
	}
	defer src.Close()
	target, err := cfg.OpenTarget()
	if err != nil {
		return nil, err
	}
	defer target.Close()

	rep := &OrgReport{Options: opts}

	schools, err := extractSchools(ctx, src, opts.InstitutionIDs)
	if err != nil {
		return nil, err
	}
	if len(schools) == 0 {
		return nil, fmt.Errorf("t_school 里找不到任何一个指定机构 id（%v）", opts.InstitutionIDs)
	}
	// 请求了但库里没有的 id 必须显式报出来（不允许"少迁一个却当成功"）
	found := map[int64]bool{}
	for _, s := range schools {
		found[s.ID] = true
	}
	var missing []int64
	for _, id := range opts.InstitutionIDs {
		if !found[id] {
			missing = append(missing, id)
		}
	}
	if len(missing) > 0 {
		return nil, fmt.Errorf("指定的机构 id 在 t_school 里不存在：%v（拒绝静默跳过）", missing)
	}

	campuses, err := extractCampuses(ctx, src, opts.InstitutionIDs)
	if err != nil {
		return nil, err
	}
	if !opts.WithAllCampuses {
		legacy, err := cfg.OpenLegacy()
		if err != nil {
			return nil, err
		}
		refd, err := extractReferencedCampuses(ctx, legacy)
		legacy.Close()
		if err != nil {
			return nil, err
		}
		var kept []LegacyCampus
		for _, c := range campuses {
			if refd[c.ID] {
				kept = append(kept, c)
			}
		}
		campuses = kept
	}

	// 机构
	for _, s := range schools {
		ti, w, err := MapInstitution(s)
		if err != nil {
			return nil, err
		}
		rep.Warnings = append(rep.Warnings, w...)
		if err := upsertInstitution(ctx, target, ti); err != nil {
			return nil, err
		}
		rep.InstIDs = append(rep.InstIDs, ti.InstitutionID)
	}
	rep.Schools = len(schools)

	// 校区
	for _, c := range campuses {
		tc, w, err := MapCampus(c)
		if err != nil {
			return nil, err
		}
		rep.Warnings = append(rep.Warnings, w...)
		if err := upsertCampus(ctx, target, tc); err != nil {
			return nil, err
		}
		rep.CampIDs = append(rep.CampIDs, tc.CampusID)
	}
	rep.Campuses = len(campuses)
	sort.Slice(rep.CampIDs, func(i, j int) bool { return rep.CampIDs[i] < rep.CampIDs[j] })
	return rep, nil
}

// OrgIdemResult 是 org 域「跑两次」的比对结论。
type OrgIdemResult struct {
	InstBefore, InstAfter map[int64]string
	CampBefore, CampAfter map[int64]string
}

// AllOK 表示两张表都完全幂等（行数与 id 集合不变）。
func (r *OrgIdemResult) AllOK() bool {
	return sameIDSet(r.InstBefore, r.InstAfter) && sameIDSet(r.CampBefore, r.CampAfter)
}

func sameIDSet(a, b map[int64]string) bool {
	if len(a) != len(b) {
		return false
	}
	for k := range a {
		if _, ok := b[k]; !ok {
			return false
		}
	}
	return true
}

// IdempotencyCheckOrg 先拍快照 → 再跑一次完全相同的映射 → 比对。
func IdempotencyCheckOrg(ctx context.Context, cfg Config, opts OrgOptions) (*OrgIdemResult, error) {
	ib, cb, err := SnapshotOrg(ctx, cfg)
	if err != nil {
		return nil, err
	}
	if _, err := PrepareOrg(ctx, cfg, opts); err != nil {
		return nil, fmt.Errorf("第二次映射失败（幂等被破坏）：%w", err)
	}
	ia, ca, err := SnapshotOrg(ctx, cfg)
	if err != nil {
		return nil, err
	}
	return &OrgIdemResult{InstBefore: ib, InstAfter: ia, CampBefore: cb, CampAfter: ca}, nil
}

// OrgCleanReport 是清理结果。
type OrgCleanReport struct {
	DeletedInst int64
	DeletedCamp int64
	Remaining   map[string]int64
}

// CleanupOrg 删除第三批造的数据（按 D1 id 精确删）。
func CleanupOrg(ctx context.Context, cfg Config, opts OrgOptions) (*OrgCleanReport, error) {
	src, err := cfg.OpenOrgSource()
	if err != nil {
		return nil, err
	}
	defer src.Close()
	target, err := cfg.OpenTarget()
	if err != nil {
		return nil, err
	}
	defer target.Close()

	schools, err := extractSchools(ctx, src, opts.InstitutionIDs)
	if err != nil {
		return nil, err
	}
	campuses, err := extractCampuses(ctx, src, opts.InstitutionIDs)
	if err != nil {
		return nil, err
	}

	rep := &OrgCleanReport{Remaining: map[string]int64{}}
	// 先删校区（引用方），再删机构
	for _, c := range campuses {
		res, err := target.ExecContext(ctx, `DELETE FROM t_campus WHERE campus_id = ?`, c.ID)
		if err != nil {
			return nil, err
		}
		n, _ := res.RowsAffected()
		rep.DeletedCamp += n
	}
	for _, s := range schools {
		res, err := target.ExecContext(ctx, `DELETE FROM t_institution WHERE institution_id = ?`, s.ID)
		if err != nil {
			return nil, err
		}
		n, _ := res.RowsAffected()
		rep.DeletedInst += n
	}
	for _, t := range []string{"t_institution", "t_campus"} {
		var n int64
		if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&n); err != nil {
			return nil, err
		}
		rep.Remaining[t] = n
	}
	return rep, nil
}

// CountOrgTables 统计 org 两张表行数。
func CountOrgTables(ctx context.Context, cfg Config) (map[string]int64, error) {
	db, err := cfg.OpenTarget()
	if err != nil {
		return nil, err
	}
	defer db.Close()
	inst, camp, err := countTargetOrg(ctx, db)
	if err != nil {
		return nil, err
	}
	return map[string]int64{"t_institution": inst, "t_campus": camp}, nil
}

// PrintOrgReport 打印搬数据结果。
func PrintOrgReport(w io.Writer, r *OrgReport) {
	fmt.Fprintf(w, "== 第三批 org 域映射 ==\n")
	fmt.Fprintf(w, "机构 id 集合：%v（%d 个）\n", r.Options.InstitutionIDs, len(r.Options.InstitutionIDs))
	fmt.Fprintf(w, "迁移校区策略：WithAllCampuses=%v\n", r.Options.WithAllCampuses)
	fmt.Fprintf(w, "写入：t_institution=%d 行，t_campus=%d 行\n", r.Schools, r.Campuses)
	fmt.Fprintf(w, "institution_id：%v\n", r.InstIDs)
	fmt.Fprintf(w, "campus_id：%v\n", r.CampIDs)
	if len(r.Warnings) > 0 {
		fmt.Fprintf(w, "映射告警（%d）：\n", len(r.Warnings))
		for _, x := range r.Warnings {
			fmt.Fprintf(w, "  - %s\n", x)
		}
	} else {
		fmt.Fprintf(w, "映射告警：无\n")
	}
}

// PrintOrgVerify 打印核对结果。
func PrintOrgVerify(w io.Writer, r *OrgVerifyReport) {
	fmt.Fprintf(w, "== 第三批逐列核对（列 × 行 聚合）==\n")
	last := ""
	for _, c := range r.Cols {
		if c.Table != last {
			cnt := r.TargetInst
			if c.Table == "t_campus" {
				cnt = r.TargetCamp
			}
			fmt.Fprintf(w, "\n-- %s（target %d 行）--\n", c.Table, cnt)
			last = c.Table
		}
		st := markOK
		if !c.OK() {
			st = markFail
		}
		fmt.Fprintf(w, "  %s %-18s %d/%d 相等   %s\n", st, c.Column, c.Equal, c.Total, c.Note)
		for _, s := range c.Samples {
			fmt.Fprintf(w, "        ✘ %s\n", s)
		}
	}

	fmt.Fprintf(w, "\n== 引用完整性（逐个来源）==\n")
	fmt.Fprintf(w, "  %-28s %-8s %-10s %-10s %-10s %s\n", "来源", "行数", "机构数", "校区数", "孤儿机构", "孤儿校区")
	for _, s := range r.RefStats {
		st := markOK
		if s.OrphanInst > 0 || s.OrphanCampus > 0 {
			st = markFail
		}
		fmt.Fprintf(w, "  %s %-26s %-8d %-10d %-10d %-10d %d\n",
			st, s.Source, s.Rows, s.DistinctInst, s.DistinctCampus, s.OrphanInst, s.OrphanCampus)
		if len(s.OrphanInstList) > 0 || len(s.OrphanCampusList) > 0 {
			fmt.Fprintf(w, "        孤儿机构 id：%v   孤儿校区 id：%v\n", s.OrphanInstList, s.OrphanCampusList)
		}
	}
	fmt.Fprintf(w, "  （campus_id=0 表示机构级、不建行，已按来源单独统计，不计入孤儿）\n")
	for _, s := range r.RefStats {
		if s.CampusZeroRows > 0 {
			fmt.Fprintf(w, "    %-26s campus_id=0 的行数：%d\n", s.Source, s.CampusZeroRows)
		}
	}

	if len(r.CrossOrg) > 0 {
		fmt.Fprintf(w, "\n== R17 标注：cross-org-campus（只标注，不修正）==\n")
		for _, c := range r.CrossOrg {
			fmt.Fprintf(w, "  · %s\n", c)
		}
	}

	fmt.Fprintf(w, "\n== 语义断言 ==\n")
	for _, a := range r.Assertions {
		st := markOK
		if !a.Pass {
			st = markFail
		}
		fmt.Fprintf(w, "  %s %s\n     %s\n", st, a.Name, a.Detail)
	}
	fmt.Fprintf(w, "\n核对结论：%s（不一致列 + 失败断言 = %d）\n", verdict(r.OK()), r.FailCount())
}

// PrintOrgIdem 打印幂等结论。
func PrintOrgIdem(w io.Writer, r *OrgIdemResult) {
	fmt.Fprintf(w, "== 幂等实测（同一批跑两次）==\n")
	st := markOK
	if !r.AllOK() {
		st = markFail
	}
	fmt.Fprintf(w, "  %s t_institution：%d 行 → %d 行\n", st, len(r.InstBefore), len(r.InstAfter))
	fmt.Fprintf(w, "  %s t_campus：%d 行 → %d 行\n", st, len(r.CampBefore), len(r.CampAfter))
	fmt.Fprintf(w, "  id 集合不变：institution=%v campus=%v\n",
		sameIDSet(r.InstBefore, r.InstAfter), sameIDSet(r.CampBefore, r.CampAfter))
	fmt.Fprintf(w, "\n幂等结论：%s\n", verdict(r.AllOK()))
}

// PrintOrgClean 打印清理结果。
func PrintOrgClean(w io.Writer, r *OrgCleanReport) {
	fmt.Fprintf(w, "== 第三批清理 ==\n")
	fmt.Fprintf(w, "  t_campus      删除 %d 行，剩余全表 %d 行\n", r.DeletedCamp, r.Remaining["t_campus"])
	fmt.Fprintf(w, "  t_institution 删除 %d 行，剩余全表 %d 行\n", r.DeletedInst, r.Remaining["t_institution"])
	ok := r.Remaining["t_campus"] == 0 && r.Remaining["t_institution"] == 0
	fmt.Fprintf(w, "两表回 0 行：%s\n", verdict(ok))
}

var _ = sql.ErrNoRows