#!/usr/bin/env python3
"""Generate the DDL/spec/ETL backed GWPTD column registry.

The tracked Markdown is produced from a data-free DDL snapshot plus the
ProductSpecs.  ``--apply`` is intentionally separate: it requires explicit
``V3_MYSQL_*`` variables, profiles each physical table with exactly one
single-table aggregate query, and is the *only* writer of
``gwptd_monitoring.dim_column_registry``.  Nothing in this module reads dotenv
files or invents a database connection.
"""
from __future__ import annotations

import argparse
import dataclasses
import datetime as dt
import difflib
import hashlib
import json
import os
import re
import sys
from collections import defaultdict
from pathlib import Path
from typing import Any, Iterable

import yaml

ROOT = Path(__file__).resolve().parents[2]
SPECS_DIR = ROOT / "specs" / "products"
MEANINGS_PATH = ROOT / "specs" / "registry" / "column-meanings.yaml"
OUTPUT_DIR = ROOT / "docs" / "backend" / "generated"
OUTPUT_FILE = OUTPUT_DIR / "COLUMN-REGISTRY.md"
SCHEMAS = frozenset({"gwptd_intake", "gwptd_kernel", "gwptd_monitoring"})
IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


@dataclasses.dataclass(frozen=True)
class Column:
    schema: str
    table: str
    name: str
    ordinal: int
    column_type: str
    nullable: bool
    default: str | None
    collation: str | None
    comment: str | None
    key_roles: tuple[str, ...]


@dataclasses.dataclass(frozen=True)
class Table:
    schema: str
    name: str
    columns: tuple[Column, ...]
    keys: dict[str, tuple[str, ...]]
    unique_keys: dict[str, tuple[str, ...]]

    @property
    def natural_key(self) -> tuple[str | None, tuple[str, ...]]:
        # An auto-increment primary key makes every capture look unique and
        # therefore cannot classify a table as a state snapshot.  The review
        # contract asks for a declared *UNIQUE* business key; without one the
        # nature remains unknown instead of being guessed from ``id``.
        candidates = {name: columns for name, columns in self.unique_keys.items() if name != "PRIMARY"}
        if not candidates:
            return None, ()
        name = min(candidates, key=lambda key: (len(candidates[key]), key))
        return name, candidates[name]


@dataclasses.dataclass(frozen=True)
class SourceMapping:
    endpoint_code: str
    endpoint: str
    source_kind: str
    source_expr: str

    def as_dict(self) -> dict[str, str]:
        return dataclasses.asdict(self)


@dataclasses.dataclass(frozen=True)
class WriterRef:
    file: str
    line: int
    expression: str | None = None

    def as_dict(self) -> dict[str, Any]:
        data = {"file": self.file, "line": self.line}
        if self.expression:
            data["expression"] = self.expression
        return data


@dataclasses.dataclass(frozen=True)
class Profile:
    row_count: int
    empty_count: int
    distinct_count: int
    distinct_key_count: int | None
    captures_per_entity: float | None
    profiled_at: str


def _sql_unquote(value: str) -> str:
    if value.upper() == "NULL":
        return "NULL"
    if value.startswith("'") and value.endswith("'"):
        return value[1:-1].replace("''", "'")
    return value


def _parts_in_parentheses(part: str) -> tuple[str, ...]:
    match = re.search(r"\(([^()]*)\)", part)
    return tuple(re.findall(r"`([^`]+)`", match.group(1) if match else part))


def _column_type(rest: str) -> str:
    marker = re.search(
        r"\s+(?:NOT\s+NULL|NULL|DEFAULT|COLLATE|CHARACTER\s+SET|COMMENT|"
        r"AUTO_INCREMENT|GENERATED|ON\s+UPDATE)\b",
        rest,
        flags=re.I,
    )
    return rest[: marker.start() if marker else len(rest)].strip().rstrip(",")


def _parse_table(schema: str, table_name: str, body: str) -> Table:
    definitions = [line.strip().rstrip(",") for line in body.splitlines() if line.strip()]
    key_roles: dict[str, list[str]] = defaultdict(list)
    keys: dict[str, tuple[str, ...]] = {}
    unique: dict[str, tuple[str, ...]] = {}
    for definition in definitions:
        if definition.startswith("PRIMARY KEY"):
            columns = _parts_in_parentheses(definition)
            keys["PRIMARY"] = columns
            unique["PRIMARY"] = columns
            for name in columns:
                key_roles[name].append("PRIMARY")
            continue
        match = re.match(r"UNIQUE KEY `([^`]+)` \((.*)\)", definition)
        if match:
            key_name, columns_text = match.groups()
            columns = _parts_in_parentheses(columns_text)
            keys[f"UNIQUE:{key_name}"] = columns
            unique[key_name] = columns
            for name in columns:
                key_roles[name].append(f"UNIQUE:{key_name}")
            continue
        match = re.match(r"KEY `([^`]+)` \((.*)\)", definition)
        if match:
            key_name, columns_text = match.groups()
            columns = _parts_in_parentheses(columns_text)
            keys[f"INDEX:{key_name}"] = columns
            for name in columns:
                key_roles[name].append(f"INDEX:{key_name}")
            continue
        match = re.match(r"CONSTRAINT `([^`]+)` FOREIGN KEY \(([^)]*)\)", definition)
        if match:
            key_name, columns_text = match.groups()
            columns = _parts_in_parentheses(columns_text)
            keys[f"FOREIGN:{key_name}"] = columns
            for name in columns:
                key_roles[name].append(f"FOREIGN:{key_name}")

    columns: list[Column] = []
    for definition in definitions:
        match = re.match(r"`([^`]+)`\s+(.+)$", definition)
        if not match:
            continue
        name, rest = match.groups()
        nullable = not bool(re.search(r"\bNOT\s+NULL\b", rest, flags=re.I))
        default_match = re.search(
            r"\bDEFAULT\s+(.+?)(?=\s+(?:ON\s+UPDATE|AUTO_INCREMENT|COMMENT|"
            r"GENERATED)\b|$)",
            rest,
            flags=re.I,
        )
        default = _sql_unquote(default_match.group(1).strip()) if default_match else None
        collation_match = re.search(r"\bCOLLATE\s+([A-Za-z0-9_]+)", rest, flags=re.I)
        comment_match = re.search(r"\bCOMMENT\s+'((?:''|[^'])*)'", rest, flags=re.I)
        columns.append(
            Column(
                schema=schema,
                table=table_name,
                name=name,
                ordinal=len(columns) + 1,
                column_type=_column_type(rest),
                nullable=nullable,
                default=default,
                collation=collation_match.group(1) if collation_match else None,
                comment=comment_match.group(1).replace("''", "'") if comment_match else None,
                key_roles=tuple(sorted(key_roles[name])),
            )
        )
    if not columns:
        raise ValueError(f"{schema}.{table_name}: no columns parsed")
    return Table(schema, table_name, tuple(columns), keys, unique)


def parse_schema_snapshot(path: Path) -> list[Table]:
    """Read the three V3 schemas from a data-free mysqldump snapshot."""
    text = path.read_text(encoding="utf-8")
    token = re.compile(
        r"(?ms)^USE `?([^`;]+)`?;\s*$|^CREATE TABLE `([^`]+)` \(\n(.*?)\n\) ENGINE=.*?;$"
    )
    current_schema: str | None = None
    tables: list[Table] = []
    for match in token.finditer(text):
        if match.group(1):
            current_schema = match.group(1)
            continue
        if current_schema not in SCHEMAS:
            continue
        tables.append(_parse_table(current_schema, match.group(2), match.group(3)))
    if not tables:
        raise ValueError(f"no V3 tables parsed from {path}")
    return sorted(tables, key=lambda item: (item.schema, item.name))


def _source_kind(field: dict[str, Any]) -> tuple[str, str]:
    if "json_path" in field:
        return "json_path", str(field["json_path"])
    if "python_func" in field:
        return "python_func", str(field["python_func"])
    if "var" in field:
        return "injected", str(field["var"])
    if "static" in field:
        return "static", json.dumps(field["static"], ensure_ascii=False)
    if "literal" in field:
        return "static", json.dumps(field["literal"], ensure_ascii=False)
    return "declared", json.dumps(field, ensure_ascii=False, sort_keys=True)


def load_spec_mappings(specs_dir: Path = SPECS_DIR) -> dict[tuple[str, str, str], list[SourceMapping]]:
    """Return exact ProductSpec field-to-intake-column mappings.

    No human text is reconstructed here: ``json_path`` and the supporting
    source-expression are retained byte-for-byte from the parsed YAML scalar.
    """
    mappings: dict[tuple[str, str, str], list[SourceMapping]] = defaultdict(list)
    for path in sorted(specs_dir.glob("*.yaml")):
        raw = yaml.safe_load(path.read_text(encoding="utf-8"))
        if not isinstance(raw, dict):
            raise ValueError(f"{path}: ProductSpec must be a mapping")
        endpoint_code = str((raw.get("metadata") or {}).get("name") or "")
        if not endpoint_code:
            raise ValueError(f"{path}: missing metadata.name")
        request = raw.get("request") or {}
        endpoint = f"{request.get('method', '—')} {request.get('path', '—')}"
        outputs = raw.get("outputs") or {}
        targets: list[dict[str, Any]] = []
        intake = outputs.get("intake")
        if isinstance(intake, dict):
            targets.append(intake)
        by_kind = outputs.get("intakeByKind")
        if isinstance(by_kind, dict):
            for target in (by_kind.get("targets") or {}).values():
                if isinstance(target, dict):
                    targets.append(target)
        for target in targets:
            table = target.get("table")
            fields = target.get("fields")
            if not isinstance(table, str) or not isinstance(fields, dict):
                continue
            for column, source in fields.items():
                if not isinstance(column, str) or not isinstance(source, dict):
                    raise ValueError(f"{path}: outputs.intake.fields has invalid entry {column!r}")
                source_kind, source_expr = _source_kind(source)
                mappings[("gwptd_intake", table, column)].append(
                    SourceMapping(endpoint_code, endpoint, source_kind, source_expr)
                )
    for sources in mappings.values():
        sources.sort(key=lambda item: (item.endpoint_code, item.source_kind, item.source_expr))
    return mappings


def _line_number(text: str, position: int) -> int:
    return text.count("\n", 0, position) + 1


def _split_sql_list(value: str) -> list[str]:
    """Split a SQL list while preserving nested expressions and quoted text."""
    result: list[str] = []
    start = depth = 0
    quote: str | None = None
    index = 0
    while index < len(value):
        char = value[index]
        if quote:
            if char == quote:
                if quote == "'" and index + 1 < len(value) and value[index + 1] == "'":
                    index += 1
                else:
                    quote = None
        elif char in "'\"`":
            quote = char
        elif char == "(":
            depth += 1
        elif char == ")":
            depth = max(depth - 1, 0)
        elif char == "," and depth == 0:
            result.append(value[start:index].strip())
            start = index + 1
        index += 1
    tail = value[start:].strip()
    if tail:
        result.append(tail)
    return result


def _select_projection(sql: str, start: int) -> str | None:
    """Return the top-level SELECT list following an INSERT target list."""
    select_match = re.search(r"\bSELECT\b", sql[start:], flags=re.I)
    if not select_match:
        return None
    begin = start + select_match.end()
    depth = 0
    quote: str | None = None
    index = begin
    while index < len(sql):
        char = sql[index]
        if quote:
            if char == quote:
                if quote == "'" and index + 1 < len(sql) and sql[index + 1] == "'":
                    index += 1
                else:
                    quote = None
        elif char in "'\"`":
            quote = char
        elif char == "(":
            depth += 1
        elif char == ")":
            depth = max(depth - 1, 0)
        elif depth == 0 and re.match(r"\bFROM\b", sql[index:], flags=re.I):
            return sql[begin:index].strip()
        index += 1
    return None


def _insert_targets(text: str, table_expression: str) -> Iterable[tuple[int, list[str], list[str]]]:
    pattern = re.compile(
        rf"INSERT\s+INTO\s+{re.escape(table_expression)}\s*\((.*?)\)\s*SELECT",
        flags=re.I | re.S,
    )
    for match in pattern.finditer(text):
        targets = re.findall(r"\b([A-Za-z_][A-Za-z0-9_]*)\b", match.group(1))
        projection = _select_projection(text, match.start())
        values = _split_sql_list(projection) if projection else []
        if len(targets) == len(values):
            yield _line_number(text, match.start()), targets, values


def _writer_refs(etl_dir: Path) -> dict[tuple[str, str, str], list[WriterRef]]:
    refs: dict[tuple[str, str, str], list[WriterRef]] = defaultdict(list)
    for path in sorted(etl_dir.glob("build_fact_*.py")):
        text = path.read_text(encoding="utf-8")
        relative = str(path.relative_to(ROOT))
        for match in re.finditer(
            r"INSERT\s+INTO\s+\{K\}\.(fact_[A-Za-z0-9_]+)\s*\((.*?)\)\s*SELECT",
            text,
            flags=re.I | re.S,
        ):
            table = match.group(1)
            targets = re.findall(r"\b([A-Za-z_][A-Za-z0-9_]*)\b", match.group(2))
            projection = _select_projection(text, match.start())
            values = _split_sql_list(projection) if projection else []
            line = _line_number(text, match.start())
            for position, column in enumerate(targets):
                expression = values[position] if position < len(values) else None
                refs[("gwptd_kernel", table, column)].append(WriterRef(relative, line, expression))
    return refs


def _wb_returns_projection(
    etl_dir: Path,
    intake_columns: set[str],
) -> dict[tuple[str, str, str], list[SourceMapping]]:
    """Derive the WB return intake-to-fact hop from the writer's SQL itself.

    This is deliberately narrow but not handwritten: it follows the temporary
    projection in ``build_fact_returns.py`` and only emits a mapping when the
    same SQL block explicitly names ``{I}.wb_returns``.
    """
    path = etl_dir / "build_fact_returns.py"
    text = path.read_text(encoding="utf-8")
    output: dict[tuple[str, str, str], list[SourceMapping]] = defaultdict(list)
    wb_function = re.search(
        r"def _wb_returns_plan\(\).*?\nWB_PREPARE_SQLS, WB_ATTEST_SQL, WB_SQL, WB_CLEANUP_SQLS = _wb_returns_plan\(\)",
        text,
        flags=re.S,
    )
    if not wb_function:
        return output
    text = wb_function.group(0)
    temp_values: dict[str, str] = {}
    for _line, targets, values in _insert_targets(text, "{WB_CURRENT}"):
        fragment_start = text.find("INSERT INTO {WB_CURRENT}")
        fragment = text[fragment_start : fragment_start + 12000] if fragment_start >= 0 else ""
        if "{I}.wb_returns" not in fragment:
            continue
        temp_values.update(dict(zip(targets, values)))
    for _line, targets, values in _insert_targets(text, "{K}.fact_returns"):
        for column, expression in zip(targets, values):
            direct = re.fullmatch(r"current_return\.([A-Za-z_][A-Za-z0-9_]*)", expression.strip())
            if not direct:
                continue
            upstream = temp_values.get(direct.group(1))
            if not upstream:
                continue
            fields = sorted(set(re.findall(r"latest\.([A-Za-z_][A-Za-z0-9_]*)", upstream)))
            for field in fields:
                if field not in intake_columns:
                    continue
                normalized = re.sub(r"latest\." + re.escape(field), f"gwptd_intake.wb_returns.{field}", upstream)
                output[("gwptd_kernel", "fact_returns", column)].append(
                    SourceMapping("wb.returns", "GET /api/v1/supplier/sales", "kernel_projection", normalized)
                )
    for mappings in output.values():
        mappings.sort(key=lambda item: (item.source_expr, item.endpoint_code))
    return output


def load_manual(path: Path = MEANINGS_PATH) -> dict[tuple[str, str, str], dict[str, str]]:
    raw = yaml.safe_load(path.read_text(encoding="utf-8"))
    if not isinstance(raw, dict) or raw.get("version") != 1 or not isinstance(raw.get("columns"), dict):
        raise ValueError(f"{path}: expected version: 1 and columns mapping")
    result: dict[tuple[str, str, str], dict[str, str]] = {}
    for identity, values in raw["columns"].items():
        parts = str(identity).split(".")
        if len(parts) != 3 or not all(IDENTIFIER.fullmatch(part) for part in parts):
            raise ValueError(f"{path}: invalid column identity {identity!r}")
        if not isinstance(values, dict):
            raise ValueError(f"{path}: {identity} manual metadata must be a mapping")
        # Все метаданные колонки лежат на одном уровне: и meaning/units/constraints,
        # и уже существующие endpoint/api_call/response_path/origin_kind. Поэтому
        # отдельный список разрешённых ключей неизбежно отстаёт от самого YAML.
        # build_registry ниже явно читает только ручные поля, которые умеет
        # публиковать, так что новый ключ здесь не может затереть порождённые
        # source_mappings, writer_refs или поля DDL.
        result[tuple(parts)] = {key: str(value).strip() for key, value in values.items() if str(value).strip()}
    return result


def _mapping_digest(mappings: list[SourceMapping], writers: list[WriterRef]) -> str:
    payload = {
        "mappings": [item.as_dict() for item in mappings],
        "writers": [item.as_dict() for item in writers],
    }
    return hashlib.sha256(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def build_registry(
    tables: list[Table],
    spec_mappings: dict[tuple[str, str, str], list[SourceMapping]],
    writer_refs: dict[tuple[str, str, str], list[WriterRef]],
    manual: dict[tuple[str, str, str], dict[str, str]],
    profiles: dict[tuple[str, str], dict[str, Profile]] | None = None,
) -> list[dict[str, Any]]:
    profiles = profiles or {}
    wb_projections = _wb_returns_projection(
        ROOT / "marketplace-collector-v3" / "kernel" / "etl",
        {column for (schema, table, column) in spec_mappings if schema == "gwptd_intake" and table == "wb_returns"},
    )
    known = {(column.schema, column.table, column.name) for table in tables for column in table.columns}
    absent = sorted(".".join(identity) for identity in set(manual) - known)
    if absent:
        raise ValueError("declared columns missing from schema: " + ", ".join(absent))
    rows: list[dict[str, Any]] = []
    for table in tables:
        natural_key_name, natural_key_columns = table.natural_key
        for column in table.columns:
            identity = (column.schema, column.table, column.name)
            mappings = [*spec_mappings.get(identity, []), *wb_projections.get(identity, [])]
            mappings.sort(key=lambda item: (item.endpoint_code, item.source_kind, item.source_expr))
            writers = sorted(writer_refs.get(identity, []), key=lambda item: (item.file, item.line, item.expression or ""))
            profile = profiles.get((table.schema, table.name), {}).get(column.name)
            human = manual.get(identity, {})
            rows.append(
                {
                    "schema_name": column.schema,
                    "table_name": column.table,
                    "column_name": column.name,
                    "record_origin": "generated",
                    "ordinal_position": column.ordinal,
                    "column_type": column.column_type,
                    "is_nullable": column.nullable,
                    "column_default": column.default,
                    "collation_name": column.collation,
                    "column_comment": column.comment,
                    "key_roles": list(column.key_roles),
                    "source_mappings": [item.as_dict() for item in mappings],
                    "writer_refs": [item.as_dict() for item in writers],
                    "source_digest": _mapping_digest(mappings, writers),
                    "table_nature": _nature(profile),
                    "natural_key_name": natural_key_name,
                    "natural_key_columns": list(natural_key_columns),
                    "profile": dataclasses.asdict(profile) if profile else None,
                    "meaning": human.get("meaning"),
                    "units": human.get("units"),
                    "constraints": human.get("constraints"),
                }
            )
    return rows


def _nature(profile: Profile | None) -> str:
    if profile is None or profile.captures_per_entity is None:
        return "unknown"
    return "journal" if profile.captures_per_entity > 1 else "snapshot"


def _md(value: object | None) -> str:
    if value is None or value == "" or value == []:
        return "—"
    return str(value).replace("|", "\\|").replace("\n", "<br>")


def _source_text(row: dict[str, Any]) -> str:
    parts = []
    for source in row["source_mappings"]:
        parts.append(
            f"`{source['endpoint_code']}` {source['endpoint']} · "
            f"`{source['source_kind']}` `{source['source_expr']}`"
        )
    return "<br>".join(parts) if parts else "—"


def _writer_text(row: dict[str, Any]) -> str:
    parts = []
    for writer in row["writer_refs"]:
        expression = f" → `{writer['expression']}`" if writer.get("expression") else ""
        parts.append(f"`{writer['file']}:{writer['line']}`{expression}")
    return "<br>".join(parts) if parts else "—"


def _profile_text(row: dict[str, Any]) -> str:
    profile = row["profile"]
    if not profile:
        return "не снят"
    ratio = profile["captures_per_entity"]
    ratio_text = "—" if ratio is None else f"{ratio:.3f}"
    return (
        f"строк {profile['row_count']}; пусто {profile['empty_count']}; "
        f"различных {profile['distinct_count']}; захватов/сущность {ratio_text}"
    )


def render_markdown(rows: list[dict[str, Any]], schema_source: str) -> str:
    generated = len(rows)
    mapped = sum(bool(row["source_mappings"] or row["writer_refs"]) for row in rows)
    meanings = sum(bool(row["meaning"]) for row in rows)
    debt = [f"{row['schema_name']}.{row['table_name']}.{row['column_name']}" for row in rows if not row["meaning"]]
    lines = [
        "# Реестр колонок GWPTD",
        "",
        "> Сгенерировано `scripts/docs/generate_column_registry.py`; не редактировать вручную.",
        f"> DDL: `{schema_source}`; API-источник: `specs/products/*.yaml` → `outputs.intake.fields`; ETL: `kernel/etl/build_fact_*.py`.",
        "",
        "## Покрытие",
        "",
        f"- Колонок из DDL: **{generated}**.",
        f"- С порождённым source/writer evidence: **{mapped}**.",
        f"- С ручным human meaning: **{meanings}**; долг: **{len(debt)}**.",
        "- Долг не скрывается: строгий `--check` завершается ошибкой, пока он не закрыт; `--allow-meaning-debt` перечисляет его явно для общего refresh-контракта.",
        "",
    ]
    grouped: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        grouped[(row["schema_name"], row["table_name"])].append(row)
    for (schema, table), table_rows in sorted(grouped.items()):
        key = table_rows[0]["natural_key_name"] or "нет UNIQUE"
        key_columns = ", ".join(table_rows[0]["natural_key_columns"]) or "—"
        lines += [
            f"## `{schema}.{table}`",
            "",
            f"Ключ различимости из DDL: `{key}` ({_md(key_columns)}). Природа: `{table_rows[0]['table_nature']}`.",
            "",
            "| # | Колонка | DDL | Ключ | Источник (дословно) | Писатель факта | Профиль | Meaning | Units | Constraints |",
            "|---:|---|---|---|---|---|---|---|---|---|",
        ]
        for row in table_rows:
            ddl = f"`{row['column_type']}`; {'NULL' if row['is_nullable'] else 'NOT NULL'}"
            if row["column_default"] is not None:
                ddl += f"; DEFAULT `{row['column_default']}`"
            lines.append(
                f"| {row['ordinal_position']} | `{row['column_name']}` | {ddl} | "
                f"{_md(', '.join(row['key_roles']))} | {_source_text(row)} | {_writer_text(row)} | "
                f"{_profile_text(row)} | {_md(row['meaning'])} | {_md(row['units'])} | {_md(row['constraints'])} |"
            )
        lines.append("")
    lines += [
        "## Долг human meaning",
        "",
        "Следующие колонки существуют в DDL, но их смысл не был выдуман из имени и потому пока пуст:",
        "",
    ]
    lines.extend(f"- `{identity}`" for identity in debt)
    return "\n".join(lines) + "\n"


def _latest_schema() -> Path:
    snapshots = sorted((ROOT / "docs" / "collectors").glob("SCHEMA-SNAPSHOT-*.sql"))
    if not snapshots:
        raise FileNotFoundError("no data-free schema snapshot found")
    return snapshots[-1]


def _mysql_connection() -> Any:
    required = ("V3_MYSQL_HOST", "V3_MYSQL_USER", "V3_MYSQL_PASSWORD")
    missing = [key for key in required if not os.getenv(key)]
    if missing:
        raise RuntimeError("--database/--apply requires explicit " + ", ".join(missing))
    try:
        import pymysql
        import pymysql.cursors
    except ImportError as exc:
        raise RuntimeError("PyMySQL is required only for --database/--apply") from exc
    return pymysql.connect(
        host=os.environ["V3_MYSQL_HOST"],
        port=int(os.getenv("V3_MYSQL_PORT", "3406")),
        user=os.environ["V3_MYSQL_USER"],
        password=os.environ["V3_MYSQL_PASSWORD"],
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=False,
    )


def tables_from_database(connection: Any) -> list[Table]:
    """Read current DDL metadata from information_schema without mutation."""
    column_sql = """
        SELECT table_schema, table_name, column_name, ordinal_position,
               column_type, is_nullable, column_default, collation_name,
               column_comment
        FROM information_schema.columns
        WHERE table_schema IN ('gwptd_intake', 'gwptd_kernel', 'gwptd_monitoring')
        ORDER BY table_schema, table_name, ordinal_position
    """
    index_sql = """
        SELECT table_schema, table_name, index_name, non_unique, seq_in_index,
               column_name
        FROM information_schema.statistics
        WHERE table_schema IN ('gwptd_intake', 'gwptd_kernel', 'gwptd_monitoring')
        ORDER BY table_schema, table_name, index_name, seq_in_index
    """
    foreign_sql = """
        SELECT kcu.table_schema, kcu.table_name, kcu.column_name,
               kcu.constraint_name
        FROM information_schema.key_column_usage AS kcu
        JOIN information_schema.table_constraints AS tc
          ON tc.constraint_schema = kcu.constraint_schema
         AND tc.table_schema = kcu.table_schema
         AND tc.table_name = kcu.table_name
         AND tc.constraint_name = kcu.constraint_name
        WHERE kcu.table_schema IN ('gwptd_intake', 'gwptd_kernel', 'gwptd_monitoring')
          AND tc.constraint_type = 'FOREIGN KEY'
        ORDER BY kcu.table_schema, kcu.table_name, kcu.constraint_name, kcu.ordinal_position
    """
    with connection.cursor() as cursor:
        cursor.execute(column_sql)
        column_rows = cursor.fetchall()
        cursor.execute(index_sql)
        index_rows = cursor.fetchall()
        cursor.execute(foreign_sql)
        foreign_rows = cursor.fetchall()
    key_columns: dict[tuple[str, str], dict[str, list[str]]] = defaultdict(lambda: defaultdict(list))
    key_roles: dict[tuple[str, str, str], list[str]] = defaultdict(list)
    unique_keys: dict[tuple[str, str], dict[str, tuple[str, ...]]] = defaultdict(dict)
    for index in index_rows:
        schema, table, name = index["table_schema"], index["table_name"], index["index_name"]
        if name == "PRIMARY":
            role, key_name = "PRIMARY", "PRIMARY"
        elif int(index["non_unique"]) == 0:
            role, key_name = f"UNIQUE:{name}", name
        else:
            role, key_name = f"INDEX:{name}", f"INDEX:{name}"
        key_columns[(schema, table)][key_name].append(index["column_name"])
        key_roles[(schema, table, index["column_name"])].append(role)
    for foreign in foreign_rows:
        key_roles[(foreign["table_schema"], foreign["table_name"], foreign["column_name"])].append(
            f"FOREIGN:{foreign['constraint_name']}"
        )
    for identity, keys in key_columns.items():
        for key_name, columns in keys.items():
            if key_name == "PRIMARY":
                unique_keys[identity]["PRIMARY"] = tuple(columns)
            elif key_name.startswith("INDEX:"):
                continue
            else:
                unique_keys[identity][key_name] = tuple(columns)
    grouped: dict[tuple[str, str], list[Column]] = defaultdict(list)
    for item in column_rows:
        identity = (item["table_schema"], item["table_name"])
        grouped[identity].append(
            Column(
                schema=item["table_schema"], table=item["table_name"], name=item["column_name"],
                ordinal=int(item["ordinal_position"]), column_type=item["column_type"],
                nullable=item["is_nullable"] == "YES", default=item["column_default"],
                collation=item["collation_name"], comment=item["column_comment"] or None,
                key_roles=tuple(sorted(key_roles[(item["table_schema"], item["table_name"], item["column_name"])])),
            )
        )
    tables = []
    for (schema, name), columns in grouped.items():
        keys = {key: tuple(values) for key, values in key_columns[(schema, name)].items()}
        tables.append(Table(schema, name, tuple(columns), keys, unique_keys[(schema, name)]))
    return sorted(tables, key=lambda item: (item.schema, item.name))


def _quote_identifier(value: str) -> str:
    if not IDENTIFIER.fullmatch(value):
        raise ValueError(f"unsafe SQL identifier {value!r}")
    return f"`{value}`"


def _empty_expr(column: Column) -> str:
    quoted = _quote_identifier(column.name)
    if re.match(r"(?:var)?char|text|enum|set", column.column_type, flags=re.I):
        return f"SUM({quoted} IS NULL OR TRIM({quoted}) = '')"
    return f"SUM({quoted} IS NULL)"


def _distinct_key_expr(columns: tuple[str, ...]) -> str:
    """Encode a nullable multi-column UNIQUE key in one aggregate expression.

    ``COUNT(DISTINCT col_a, col_b)`` drops NULL-containing tuples and a
    subquery ``GROUP BY`` would force a second read of the same table.  The
    length-prefixed representation preserves NULL versus an empty string and
    keeps the profile to one single-table SELECT.
    """
    parts = []
    for column in columns:
        quoted = _quote_identifier(column)
        parts.append(
            "CASE WHEN {q} IS NULL THEN 'N;' ELSE "
            "CONCAT('V', CHAR_LENGTH(CAST({q} AS CHAR)), ':', CAST({q} AS CHAR), ';') END".format(q=quoted)
        )
    return "CONCAT(" + ", ".join(parts) + ")"


def profile_tables(connection: Any, tables: list[Table]) -> dict[tuple[str, str], dict[str, Profile]]:
    """Profile each table with one aggregate SELECT and no joins.

    The query is intentionally one table at a time.  It never combines tables,
    joins, or re-reads a table through a profiling subquery.
    """
    output: dict[tuple[str, str], dict[str, Profile]] = {}
    observed_at = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M:%S")
    with connection.cursor() as cursor:
        cursor.execute("SET SESSION MAX_EXECUTION_TIME = 120000")
        for table in tables:
            table_name = f"{_quote_identifier(table.schema)}.{_quote_identifier(table.name)}"
            select_parts = ["COUNT(*) AS row_count"]
            for column in table.columns:
                escaped = column.name.replace("`", "``")
                select_parts.append(f"COALESCE({_empty_expr(column)}, 0) AS `empty__{escaped}`")
                select_parts.append(f"COUNT(DISTINCT {_quote_identifier(column.name)}) AS `distinct__{escaped}`")
            natural_name, natural_columns = table.natural_key
            if natural_columns:
                select_parts.append(
                    f"COUNT(DISTINCT {_distinct_key_expr(natural_columns)}) AS distinct_key_count"
                )
            else:
                select_parts.append("NULL AS distinct_key_count")
            cursor.execute("SELECT " + ", ".join(select_parts) + f" FROM {table_name}")
            row = cursor.fetchone()
            if row is None:
                raise RuntimeError(f"profile query returned no row for {table.schema}.{table.name}")
            row_count = int(row["row_count"])
            key_count = int(row["distinct_key_count"]) if row["distinct_key_count"] is not None else None
            captures = (row_count / key_count) if key_count else None
            output[(table.schema, table.name)] = {
                column.name: Profile(
                    row_count=row_count,
                    empty_count=int(row[f"empty__{column.name}"]),
                    distinct_count=int(row[f"distinct__{column.name}"]),
                    distinct_key_count=key_count,
                    captures_per_entity=captures,
                    profiled_at=observed_at,
                )
                for column in table.columns
            }
    return output


def _database_rows(connection: Any) -> dict[tuple[str, str, str], dict[str, Any]]:
    with connection.cursor() as cursor:
        cursor.execute(
            "SELECT schema_name, table_name, column_name, source_digest, source_mappings, writer_refs "
            "FROM gwptd_monitoring.dim_column_registry"
        )
        return {
            (item["schema_name"], item["table_name"], item["column_name"]): item
            for item in cursor.fetchall()
        }


def _upsert_rows(connection: Any, rows: list[dict[str, Any]]) -> None:
    columns = (
        "schema_name", "table_name", "column_name", "record_origin", "ordinal_position", "column_type",
        "is_nullable", "column_default", "collation_name", "column_comment", "key_roles", "source_mappings",
        "writer_refs", "source_digest", "table_nature", "natural_key_name", "natural_key_columns",
        "profile_row_count", "profile_empty_count", "profile_distinct_count", "profile_distinct_key_count",
        "captures_per_entity", "profiled_at", "meaning", "units", "semantic_constraints", "observed_at",
    )
    placeholders = ", ".join(["%s"] * len(columns))
    update = ", ".join(f"{name}=VALUES({name})" for name in columns[3:])
    sql = (
        "INSERT INTO gwptd_monitoring.dim_column_registry (" + ", ".join(columns) + ") VALUES (" + placeholders + ") "
        "ON DUPLICATE KEY UPDATE " + update
    )
    observed_at = dt.datetime.now(dt.timezone.utc).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M:%S")
    values = []
    for row in rows:
        profile = row["profile"] or {}
        values.append(
            (
                row["schema_name"], row["table_name"], row["column_name"], row["record_origin"],
                row["ordinal_position"], row["column_type"], int(row["is_nullable"]), row["column_default"],
                row["collation_name"], row["column_comment"], json.dumps(row["key_roles"], ensure_ascii=False),
                json.dumps(row["source_mappings"], ensure_ascii=False, sort_keys=True),
                json.dumps(row["writer_refs"], ensure_ascii=False, sort_keys=True), row["source_digest"],
                row["table_nature"], row["natural_key_name"], json.dumps(row["natural_key_columns"], ensure_ascii=False),
                profile.get("row_count"), profile.get("empty_count"), profile.get("distinct_count"),
                profile.get("distinct_key_count"), profile.get("captures_per_entity"), profile.get("profiled_at"),
                row["meaning"], row["units"], row["constraints"], observed_at,
            )
        )
    with connection.cursor() as cursor:
        cursor.executemany(sql, values)
    connection.commit()


def check_registry(rows: list[dict[str, Any]], database_rows: dict[tuple[str, str, str], dict[str, Any]] | None) -> list[str]:
    failures = []
    missing_meaning = [row for row in rows if not row["meaning"]]
    if missing_meaning:
        failures.append(
            "missing human meaning (" + str(len(missing_meaning))
            + "); full generated list: docs/backend/generated/COLUMN-REGISTRY.md#долг-human-meaning"
        )
    if database_rows is not None:
        for row in rows:
            identity = (row["schema_name"], row["table_name"], row["column_name"])
            stored = database_rows.get(identity)
            if stored is None:
                failures.append("registry row missing: " + ".".join(identity))
            elif stored["source_digest"] != row["source_digest"]:
                failures.append("spec/code source mapping diverged: " + ".".join(identity))
    return failures


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--schema", type=Path, default=None, help="data-free mysqldump snapshot")
    parser.add_argument("--output-dir", type=Path, default=None, help="directory for COLUMN-REGISTRY.md")
    parser.add_argument("--check", action="store_true", help="fail for stale Markdown, missing meanings or stored mapping drift")
    parser.add_argument("--allow-meaning-debt", action="store_true", help="list meaning debt but do not make it non-zero")
    parser.add_argument("--database", action="store_true", help="compare with persistent registry using explicit V3_MYSQL_* variables")
    parser.add_argument("--apply", action="store_true", help="profile physical tables and upsert registry; never implicit")
    args = parser.parse_args(argv)
    if args.apply and not args.database:
        parser.error("--apply requires --database")
    if args.apply and not args.output_dir:
        parser.error("--apply requires --output-dir so a deployed checkout is never written implicitly")
    if args.apply and args.check:
        parser.error("choose --apply or --check, not both")

    schema_path = args.schema or _latest_schema()
    try:
        connection = _mysql_connection() if args.database else None
        tables = tables_from_database(connection) if connection else parse_schema_snapshot(schema_path)
        manual = load_manual()
        spec_mappings = load_spec_mappings()
        writer_refs = _writer_refs(ROOT / "marketplace-collector-v3" / "kernel" / "etl")
        profiles = profile_tables(connection, tables) if args.apply and connection else None
        rows = build_registry(tables, spec_mappings, writer_refs, manual, profiles)
        schema_source = (
            "live information_schema via explicit V3_MYSQL_*"
            if connection else str(schema_path.relative_to(ROOT))
        )
        output = render_markdown(rows, schema_source)
        if args.apply:
            assert connection is not None
            output_path = args.output_dir / OUTPUT_FILE.name
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_path.write_text(output, encoding="utf-8")
            _upsert_rows(connection, rows)
            print(
                f"APPLIED {len(rows)} registry rows; one single-table profile query per table; "
                f"wrote {output_path}"
            )
            return 0
        if args.check:
            stale = not OUTPUT_FILE.exists() or OUTPUT_FILE.read_text(encoding="utf-8") != output
            if stale:
                diff = difflib.unified_diff(
                    OUTPUT_FILE.read_text(encoding="utf-8").splitlines(keepends=True) if OUTPUT_FILE.exists() else [],
                    output.splitlines(keepends=True), fromfile=str(OUTPUT_FILE), tofile="(generated)"
                )
                sys.stderr.writelines(diff)
            failures = check_registry(rows, _database_rows(connection) if connection else None)
            if args.allow_meaning_debt:
                failures = [item for item in failures if not item.startswith("missing human meaning")]
                debt = [row for row in rows if not row["meaning"]]
                print(
                    "KNOWN MEANING DEBT (" + str(len(debt))
                    + "); full generated list: docs/backend/generated/COLUMN-REGISTRY.md#долг-human-meaning"
                )
            if stale:
                failures.insert(0, "generated Markdown is stale")
            if failures:
                print("Column registry check failed:", file=sys.stderr)
                print("\n".join("- " + item for item in failures), file=sys.stderr)
                return 1
            print(f"OK  {OUTPUT_FILE.relative_to(ROOT)} ({len(rows)} generated columns)")
            return 0
        output_path = (args.output_dir / OUTPUT_FILE.name) if args.output_dir else OUTPUT_FILE
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(output, encoding="utf-8")
        print(f"Wrote {output_path.relative_to(ROOT) if output_path.is_relative_to(ROOT) else output_path} ({len(rows)} columns)")
        return 0
    except (OSError, ValueError, RuntimeError, yaml.YAMLError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    finally:
        if 'connection' in locals() and connection is not None:
            connection.close()


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