#!/usr/bin/env python3
"""Audit duplicate rows in hk_songs by name, lyricist, and composer.

The audit is read-only by default. Pass --fix-merge to process pending
merge-direction records from the import staging table.
"""

from __future__ import annotations

import argparse
import csv
import os
import re
from pathlib import Path
from typing import Any, Iterable

import pymysql
from dotenv import load_dotenv
from tqdm import tqdm


load_dotenv()

TARGET_DB_CONFIG = {
    "host": os.getenv("TARGET_DB_HOST"),
    "port": int(os.getenv("TARGET_DB_PORT", 3306)),
    "user": os.getenv("TARGET_DB_USER"),
    "password": os.getenv("TARGET_DB_PASSWORD"),
    "database": os.getenv("TARGET_DB_NAME"),
    "charset": "utf8mb4",
    "cursorclass": pymysql.cursors.DictCursor,
}

SOURCE_DB_CONFIG = {
    "host": os.getenv("SOURCE_DB_HOST"),
    "port": int(os.getenv("SOURCE_DB_PORT", 3306)),
    "user": os.getenv("SOURCE_DB_USER"),
    "password": os.getenv("SOURCE_DB_PASSWORD"),
    "database": os.getenv("SOURCE_DB_NAME"),
    "charset": "utf8mb4",
    "cursorclass": pymysql.cursors.DictCursor,
}

DEFAULT_TARGET_TABLE = os.getenv("TARGET_TABLE_NAME", "hk_songs")
DEFAULT_STAGING_TABLE = os.getenv("TARGET_TABLE_NAME_TMP", "hk_songs_import_staging")
OUTPUT_DIR = Path(__file__).resolve().parent / "output" / "reports"

_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_]+$")


def quote_identifier(identifier: str) -> str:
    """Return a backtick-quoted SQL identifier after strict validation."""
    if not _IDENTIFIER_RE.fullmatch(identifier):
        raise ValueError(f"Unsafe SQL identifier: {identifier!r}")
    return f"`{identifier}`"


def _key_expr(column: str, *, case_sensitive: bool = True) -> str:
    expr = f"COALESCE(NULLIF(TRIM({column}), ''), '')"
    if case_sensitive:
        return f"{expr} COLLATE utf8mb4_bin"
    return expr


def _where_clause(include_deleted: bool) -> str:
    return "" if include_deleted else "WHERE deleted = '0'"


def _limit_clause(limit: int | None) -> str:
    if limit is None:
        return ""
    if limit <= 0:
        raise ValueError("--limit must be greater than 0")
    return f"LIMIT {limit}"


def build_duplicate_group_sql(
    table_name: str,
    *,
    include_deleted: bool = False,
    limit: int | None = None,
    case_sensitive: bool = True,
) -> str:
    """Build SQL that returns duplicate metadata groups."""
    table = quote_identifier(table_name)
    where = _where_clause(include_deleted)
    limit_sql = _limit_clause(limit)
    return f"""
SELECT
  {_key_expr('name', case_sensitive=case_sensitive)} AS name_key,
  {_key_expr('lyricist', case_sensitive=case_sensitive)} AS lyricist_key,
  {_key_expr('composer', case_sensitive=case_sensitive)} AS composer_key,
  COUNT(*) AS duplicate_count,
  GROUP_CONCAT(id ORDER BY id SEPARATOR ',') AS song_ids,
  GROUP_CONCAT(source_song_id ORDER BY id SEPARATOR ',') AS source_song_ids
FROM {table}
{where}
GROUP BY name_key, lyricist_key, composer_key
HAVING COUNT(*) > 1
ORDER BY duplicate_count DESC, name_key, lyricist_key, composer_key
{limit_sql}
""".strip()


def build_duplicate_detail_sql(
    table_name: str,
    *,
    include_deleted: bool = False,
    limit: int | None = None,
    case_sensitive: bool = True,
) -> str:
    """Build SQL that returns all rows belonging to duplicate metadata groups."""
    table = quote_identifier(table_name)
    where = _where_clause(include_deleted)
    limit_sql = _limit_clause(limit)
    key_select = f"""
    {_key_expr('name', case_sensitive=case_sensitive)} AS name_key,
    {_key_expr('lyricist', case_sensitive=case_sensitive)} AS lyricist_key,
    {_key_expr('composer', case_sensitive=case_sensitive)} AS composer_key
    """.strip()
    return f"""
SELECT
  s.id,
  s.name,
  s.lyricist,
  s.composer,
  s.singer,
  s.source_table_name,
  s.source_song_id,
  s.deleted,
  s.create_time,
  s.modify_time,
  dup.duplicate_count
FROM (
  SELECT
    id, name, lyricist, composer, singer, source_table_name, source_song_id,
    deleted, create_time, modify_time,
    {key_select}
  FROM {table}
  {where}
) AS s
JOIN (
  SELECT
    {_key_expr('name', case_sensitive=case_sensitive)} AS name_key,
    {_key_expr('lyricist', case_sensitive=case_sensitive)} AS lyricist_key,
    {_key_expr('composer', case_sensitive=case_sensitive)} AS composer_key,
    COUNT(*) AS duplicate_count
  FROM {table}
  {where}
  GROUP BY name_key, lyricist_key, composer_key
  HAVING COUNT(*) > 1
) AS dup
  ON dup.name_key = s.name_key
 AND dup.lyricist_key = s.lyricist_key
 AND dup.composer_key = s.composer_key
ORDER BY dup.duplicate_count DESC, s.name_key, s.lyricist_key, s.composer_key, s.id
{limit_sql}
""".strip()


def build_duplicate_detail_with_record_count_sql(
    table_name: str,
    staging_table_name: str,
    *,
    include_deleted: bool = False,
    limit: int | None = None,
    case_sensitive: bool = True,
) -> str:
    """Build SQL returning duplicate rows with staging record_count attached."""
    detail_sql = build_duplicate_detail_sql(
        table_name,
        include_deleted=include_deleted,
        limit=limit,
        case_sensitive=case_sensitive,
    )
    staging_table = quote_identifier(staging_table_name)
    return f"""
SELECT
  d.*,
  COALESCE(rc.record_count, 0) AS record_count
FROM (
  {detail_sql}
) AS d
LEFT JOIN (
  SELECT source_song_id, MAX(COALESCE(record_count, 0)) AS record_count
  FROM {staging_table}
  GROUP BY source_song_id
) AS rc
  ON rc.source_song_id = d.source_song_id
ORDER BY d.duplicate_count DESC, d.name, d.lyricist, d.composer, d.id
""".strip()


def _row_group_key(row: dict[str, Any], *, case_sensitive: bool = True) -> tuple[str, str, str]:
    name = (row.get("name") or "").strip()
    lyricist = (row.get("lyricist") or "").strip()
    composer = (row.get("composer") or "").strip()
    if not case_sensitive:
        name, lyricist, composer = name.lower(), lyricist.lower(), composer.lower()
    return (name, lyricist, composer)


def build_merge_plan(rows: Iterable[dict[str, Any]], *, case_sensitive: bool = True) -> list[dict[str, Any]]:
    """Choose one survivor per duplicate group and return loser merge actions."""
    grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
    for row in rows:
        grouped.setdefault(_row_group_key(row, case_sensitive=case_sensitive), []).append(row)

    plan: list[dict[str, Any]] = []
    for key, group_rows in grouped.items():
        if len(group_rows) < 2:
            continue
        sorted_rows = sorted(
            group_rows,
            key=lambda row: (-(int(row.get("record_count") or 0)), int(row["id"])),
        )
        survivor = sorted_rows[0]
        for loser in sorted_rows[1:]:
            plan.append(
                {
                    "name": key[0],
                    "lyricist": key[1],
                    "composer": key[2],
                    "survivor_id": survivor["id"],
                    "survivor_source_song_id": survivor.get("source_song_id"),
                    "survivor_record_count": int(survivor.get("record_count") or 0),
                    "survivor_lyricist": survivor.get("lyricist"),
                    "survivor_composer": survivor.get("composer"),
                    "loser_id": loser["id"],
                    "loser_source_song_id": loser.get("source_song_id"),
                    "loser_record_count": int(loser.get("record_count") or 0),
                    "loser_lyricist": loser.get("lyricist"),
                    "loser_composer": loser.get("composer"),
                    "reason": "record_count smaller; tie keeps lower id",
                }
            )
    return plan


def build_soft_delete_sql(table_name: str, loser_count: int) -> str:
    """Build a soft-delete SQL statement for planned loser rows."""
    if loser_count <= 0:
        raise ValueError("loser_count must be greater than 0")
    table = quote_identifier(table_name)
    placeholders = ",".join(["%s"] * loser_count)
    return f"""
UPDATE {table}
SET deleted = '1',
    modify_time = NOW(),
    off_shelf_remark = 'metadata duplicate merged by audit script'
WHERE deleted = '0'
  AND id IN ({placeholders})
""".strip()


def apply_soft_delete_plan(conn, table_name: str, plan: list[dict[str, Any]]) -> int:
    loser_ids = [item["loser_id"] for item in plan]
    if not loser_ids:
        return 0
    sql = build_soft_delete_sql(table_name, len(loser_ids))
    with conn.cursor() as cursor:
        cursor.execute(sql, loser_ids)
        return cursor.rowcount


def apply_merge_plan(conn, table_name: str, plan: list[dict[str, Any]]) -> dict[str, int]:
    """软删除 loser 并将其独有作者增量合并到幸存者，在同一事务内完成。"""
    from import_hk_songs import _merge_author_field

    if not plan:
        return {"soft_deleted": 0, "author_merged": 0}

    table = quote_identifier(table_name)

    # 按幸存者聚合，把同一幸存者的所有 loser 作者依次合并进来
    survivors: dict[int, dict[str, Any]] = {}
    for item in plan:
        sid = int(item["survivor_id"])
        if sid not in survivors:
            survivors[sid] = {
                "orig_lyricist": item.get("survivor_lyricist") or "",
                "orig_composer": item.get("survivor_composer") or "",
                "lyricist": item.get("survivor_lyricist") or "",
                "composer": item.get("survivor_composer") or "",
            }
        s = survivors[sid]
        result_l = _merge_author_field(s["lyricist"], item.get("loser_lyricist"))
        if result_l:
            s["lyricist"] = result_l
        result_c = _merge_author_field(s["composer"], item.get("loser_composer"))
        if result_c:
            s["composer"] = result_c

    with conn.cursor() as cursor:
        # 1. 批量软删除 loser
        loser_ids = [int(item["loser_id"]) for item in plan]
        cursor.execute(build_soft_delete_sql(table_name, len(loser_ids)), loser_ids)
        soft_deleted = cursor.rowcount

        # 2. 作者增量合并（每个幸存者最多两条 UPDATE）
        author_merged = 0
        for sid, s in survivors.items():
            if s["lyricist"] != s["orig_lyricist"]:
                cursor.execute(
                    f"UPDATE {table} SET {quote_identifier('lyricist')} = %s WHERE id = %s",
                    (s["lyricist"], sid),
                )
                author_merged += 1
            if s["composer"] != s["orig_composer"]:
                cursor.execute(
                    f"UPDATE {table} SET {quote_identifier('composer')} = %s WHERE id = %s",
                    (s["composer"], sid),
                )
                author_merged += 1

    return {"soft_deleted": soft_deleted, "author_merged": author_merged}


def summarize_duplicate_groups(groups: Iterable[dict[str, Any]]) -> dict[str, int]:
    rows = list(groups)
    duplicate_rows = sum(int(row["duplicate_count"]) for row in rows)
    return {
        "duplicate_groups": len(rows),
        "duplicate_rows": duplicate_rows,
        "extra_duplicate_rows": duplicate_rows - len(rows),
    }


def fetch_rows(conn, sql: str) -> list[dict[str, Any]]:
    with conn.cursor() as cursor:
        cursor.execute(sql)
        return list(cursor.fetchall())


def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = list(rows[0].keys()) if rows else []
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def fix_pending_merge_directions(
    source_conn,
    target_conn,
    staging_table: str,
    target_table: str,
    *,
    dry_run: bool = False,
    skip_oss: bool = False,
) -> None:
    """补处理暂存表中 merge+skipped 记录，执行 existing_into_new 的物理合并。

    - existing_into_new（新记录录音多）：歌词上传 OSS → INSERT 主表 →
      软删旧记录 → 作者合并 → 更新暂存状态为 imported
    - new_into_existing（旧记录录音多）：仅补执行作者合并（幂等）
    """
    from import_hk_songs import (
        INSERT_SQL_TEMPLATE,
        _SOFT_DELETE_SQL,
        _merge_author_field,
        execute_author_merges,
        execute_soft_deletes,
        get_oss_bucket,
        load_source_record_counts,
        process_lyrics,
    )

    staging_tbl = quote_identifier(staging_table)
    target_tbl = quote_identifier(target_table)

    # 1. 拉取所有 merge+skipped 暂存记录
    with target_conn.cursor() as cursor:
        cursor.execute(
            f"SELECT * FROM {staging_tbl} WHERE dedup_action='merge' AND staging_status='skipped' ORDER BY staging_id"
        )
        staging_rows: list[dict] = list(cursor.fetchall())

    if not staging_rows:
        print("暂存表中没有待处理的 merge+skipped 记录")
        return

    print(f"找到 {len(staging_rows)} 条 merge+skipped 暂存记录")

    # 2. 批量拉取 matched_song_id → source_song_id 及现有作者（从目标主表）
    matched_ids = {int(r["matched_song_id"]) for r in staging_rows if r.get("matched_song_id")}
    matched_to_source_sid: dict[int, str] = {}
    matched_to_authors: dict[int, dict[str, str]] = {}
    if matched_ids:
        ph = ",".join(["%s"] * len(matched_ids))
        with target_conn.cursor() as cursor:
            cursor.execute(
                f"SELECT id, source_song_id, lyricist, composer FROM {target_tbl} WHERE id IN ({ph})",
                list(matched_ids),
            )
            for row in cursor.fetchall():
                matched_to_source_sid[row["id"]] = str(row["source_song_id"] or "")
                matched_to_authors[row["id"]] = {
                    "lyricist": row.get("lyricist") or "",
                    "composer": row.get("composer") or "",
                }

    # 3. 批量查询源库录音数（用于方向判断）
    source_sids = set(matched_to_source_sid.values()) - {""}
    source_record_counts: dict[str, int] = {}
    if source_sids and source_conn:
        try:
            source_record_counts = load_source_record_counts(source_conn, source_sids)
        except Exception as e:
            print(f"警告: 查询源库录音数失败: {e}，将仅处理 merge_authors，不做 existing_into_new 分流")

    # 4. 初始化 OSS（只有真正需要上传时才初始化）
    bucket = None
    if not skip_oss and not dry_run:
        try:
            bucket = get_oss_bucket()
        except Exception as e:
            print(f"警告: OSS 初始化失败: {e}，歌词将保留原始文本（等同 --skip-oss）")
            skip_oss = True

    insert_sql = INSERT_SQL_TEMPLATE.format(table=target_table)
    update_staging_sql = (
        f"UPDATE {staging_tbl} SET staging_status=%s, imported_song_id=%s WHERE staging_id=%s"
    )

    stats = {"existing_into_new": 0, "author_merged": 0, "soft_deleted": 0, "no_op": 0, "errors": 0}

    for srow in tqdm(staging_rows, desc="修复 merge 记录", unit="条"):
        staging_id = srow["staging_id"]
        new_record_id = srow["id"]
        matched_id_raw = srow.get("matched_song_id")
        matched_id: int | None = int(matched_id_raw) if matched_id_raw is not None else None
        new_count = int(srow.get("record_count") or 0)
        merge_authors_flag = bool(srow.get("merge_authors"))

        # 方向判断
        existing_count: int | None = None
        if matched_id and source_record_counts:
            src_sid = matched_to_source_sid.get(matched_id)
            if src_sid:
                existing_count = source_record_counts.get(src_sid)
        merge_direction = (
            "existing_into_new"
            if existing_count is not None and new_count > existing_count
            else "new_into_existing"
        )

        try:
            with target_conn.cursor() as cursor:
                if merge_direction == "existing_into_new":
                    if dry_run:
                        print(
                            f"[dry-run] existing_into_new staging_id={staging_id} "
                            f"new_id={new_record_id} matched_id={matched_id} "
                            f"new_count={new_count} existing_count={existing_count}"
                        )
                        stats["existing_into_new"] += 1
                        continue

                    # 歌词上传 OSS（staging.lyrics_url 在 skip_oss=True 时存的是原始文本）
                    raw_lyrics = srow.get("lyrics_url")
                    if raw_lyrics and str(raw_lyrics).startswith("http"):
                        lyrics_url, lrc_url = raw_lyrics, srow.get("lrc_url")
                    else:
                        lyrics_url, lrc_url = process_lyrics(bucket, raw_lyrics, str(new_record_id), skip_oss)

                    # INSERT 主表（45 列，与 INSERT_SQL_TEMPLATE 同构）
                    row_tuple = (
                        new_record_id, srow["name"], srow["lyricist"], srow["composer"],
                        srow["issue_status"], srow["intro"],
                        srow["audio_url"], srow["accompany_url"], lyrics_url, lrc_url,
                        srow["song_time"], srow["song_start"], srow["song_end"],
                        srow["creation_url"], srow["opern_url"], srow["cover_version"], srow["issue_time"],
                        srow["cover_url"], srow["animation_type"], srow["bpm_class"], srow["review_status"],
                        srow["in_status"], srow["song_status"], srow["commit_time"], srow["review_time"],
                        srow["shelf_time"], srow["review_remark"], srow["create_time"], srow["creator"],
                        srow["modify_time"], srow["modifier"], srow["deleted"], srow["cooperate_type"], srow["singer"],
                        srow["off_shelf_remark"], srow["musician_id"], srow["commit_id"], srow["sheet_music"],
                        srow["commit_desc"], srow["price"], srow["source_table_name"], srow["source_song_id"],
                        srow["lyric_archive_element_id"], srow["melody_archive_element_id"], srow["audio_fingerprint"],
                    )
                    cursor.execute(insert_sql, row_tuple)
                    stats["existing_into_new"] += 1

                    # 软删旧记录
                    if matched_id:
                        cursor.execute(_SOFT_DELETE_SQL, (matched_id,))
                        stats["soft_deleted"] += 1

                    # 作者合并（反向：把旧记录独有作者增量到新记录）
                    if merge_authors_flag and matched_id:
                        old = matched_to_authors.get(matched_id, {})
                        merged_lyricist = _merge_author_field(srow.get("lyricist") or "", old.get("lyricist") or "")
                        merged_composer = _merge_author_field(srow.get("composer") or "", old.get("composer") or "")
                        if merged_lyricist or merged_composer:
                            n = execute_author_merges(cursor, [{"target_id": new_record_id, "lyricist": merged_lyricist, "composer": merged_composer}])
                            if n == 0:
                                raise RuntimeError(f"作者合并失败 target_id={new_record_id}")
                            stats["author_merged"] += 1

                    cursor.execute(update_staging_sql, ("imported", new_record_id, staging_id))
                    target_conn.commit()

                else:  # new_into_existing：仅补作者合并
                    if not matched_id:
                        stats["no_op"] += 1
                        continue
                    if dry_run:
                        stats["author_merged" if merge_authors_flag else "no_op"] += 1
                        continue
                    if merge_authors_flag:
                        old = matched_to_authors.get(matched_id, {})
                        merged_lyricist = _merge_author_field(old.get("lyricist") or "", srow.get("lyricist") or "")
                        merged_composer = _merge_author_field(old.get("composer") or "", srow.get("composer") or "")
                        if merged_lyricist or merged_composer:
                            n = execute_author_merges(cursor, [{"target_id": matched_id, "lyricist": merged_lyricist, "composer": merged_composer}])
                            if n == 0:
                                raise RuntimeError(f"作者合并失败 target_id={matched_id}")
                            stats["author_merged"] += 1
                        else:
                            stats["no_op"] += 1
                    else:
                        stats["no_op"] += 1
                    cursor.execute(update_staging_sql, ("imported", matched_id, staging_id))
                    target_conn.commit()

        except Exception as e:
            target_conn.rollback()
            stats["errors"] += 1
            print(f"错误: staging_id={staging_id} 处理失败: {e}")

    print(
        f"修复完成: existing_into_new入库={stats['existing_into_new']} | "
        f"旧记录软删={stats['soft_deleted']} | 作者合并={stats['author_merged']} | "
        f"无操作={stats['no_op']} | 错误={stats['errors']}"
    )


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Audit duplicate hk_songs rows by exact name + lyricist + composer."
    )
    parser.add_argument("--limit", type=int, default=None, help="Limit duplicate groups/details returned")
    parser.add_argument("--include-deleted", action="store_true", help="Include deleted rows")
    parser.add_argument("--details", action="store_true", help="Print duplicate row details instead of group summary rows")
    parser.add_argument(
        "--case-insensitive",
        action="store_true",
        help="Use the table's default case-insensitive collation instead of exact utf8mb4_bin comparison",
    )
    parser.add_argument("--csv", type=Path, help="Optional CSV output path")
    parser.add_argument("--merge-plan-csv", type=Path, help="Optional CSV output path for merge plan")
    parser.add_argument("--apply-merge", action="store_true", help="Soft-delete duplicate loser rows")
    parser.add_argument(
        "--fix-merge",
        action="store_true",
        help="补处理暂存表中 merge+skipped 记录：existing_into_new 入库+软删旧记录，new_into_existing 补作者合并",
    )
    parser.add_argument("--skip-oss", action="store_true", help="--fix-merge 时跳过歌词 OSS 上传，保留原始文本")
    parser.add_argument("--dry-run", action="store_true", help="仅打印计划，不写库（配合 --fix-merge 使用）")
    args = parser.parse_args()

    # --fix-merge 模式：连双库，执行合并方向后处理
    if args.fix_merge:
        target_conn = pymysql.connect(**TARGET_DB_CONFIG)
        source_conn = None
        try:
            try:
                source_conn = pymysql.connect(**SOURCE_DB_CONFIG)
            except Exception as e:
                print(f"警告: 源库连接失败: {e}，将跳过录音数查询（只处理 merge_authors）")
            fix_pending_merge_directions(
                source_conn, target_conn,
                staging_table=DEFAULT_STAGING_TABLE,
                target_table=DEFAULT_TARGET_TABLE,
                dry_run=args.dry_run,
                skip_oss=args.skip_oss,
            )
        finally:
            target_conn.close()
            if source_conn:
                source_conn.close()
        return 0

    case_sensitive = not args.case_insensitive
    group_sql = build_duplicate_group_sql(
        DEFAULT_TARGET_TABLE,
        include_deleted=args.include_deleted,
        limit=args.limit,
        case_sensitive=case_sensitive,
    )
    detail_sql = build_duplicate_detail_sql(
        DEFAULT_TARGET_TABLE,
        include_deleted=args.include_deleted,
        limit=args.limit,
        case_sensitive=case_sensitive,
    )
    detail_with_count_sql = build_duplicate_detail_with_record_count_sql(
        DEFAULT_TARGET_TABLE,
        DEFAULT_STAGING_TABLE,
        include_deleted=args.include_deleted,
        limit=args.limit,
        case_sensitive=case_sensitive,
    )

    conn = pymysql.connect(**TARGET_DB_CONFIG)
    try:
        groups = fetch_rows(conn, group_sql)
        summary = summarize_duplicate_groups(groups)
        print(
            f"table={DEFAULT_TARGET_TABLE} duplicate_groups={summary['duplicate_groups']} "
            f"duplicate_rows={summary['duplicate_rows']} "
            f"extra_duplicate_rows={summary['extra_duplicate_rows']} "
            f"case_sensitive={case_sensitive}"
        )

        rows = fetch_rows(conn, detail_sql) if args.details or args.csv else groups
        merge_plan: list[dict[str, Any]] = []
        if args.merge_plan_csv or args.apply_merge:
            if args.limit is not None and args.apply_merge:
                print(f"警告: --limit 限制的是明细行数而非重复组数，--apply-merge 可能只处理部分重复组")
            duplicate_rows_with_counts = fetch_rows(conn, detail_with_count_sql)
            merge_plan = build_merge_plan(duplicate_rows_with_counts, case_sensitive=case_sensitive)
            print(f"merge_plan_losers={len(merge_plan)}")
        for row in rows[:20]:
            print(row)
        if len(rows) > 20:
            print(f"... {len(rows) - 20} more rows")

        if args.csv:
            output_path = args.csv
            if not output_path.is_absolute():
                output_path = OUTPUT_DIR / output_path
            write_csv(output_path, rows)
            print(f"csv={output_path}")
        if args.merge_plan_csv:
            output_path = args.merge_plan_csv
            if not output_path.is_absolute():
                output_path = OUTPUT_DIR / output_path
            write_csv(output_path, merge_plan)
            print(f"merge_plan_csv={output_path}")
        if args.apply_merge:
            stats = apply_merge_plan(conn, DEFAULT_TARGET_TABLE, merge_plan)
            conn.commit()
            print(f"soft_deleted_rows={stats['soft_deleted']} author_merged_fields={stats['author_merged']}")
    finally:
        conn.close()

    return 1 if groups else 0


if __name__ == "__main__":
    raise SystemExit(main())