ThothII/harness/tht/cli/schema_cmd.py

from pathlib import Path

import typer
from sqlalchemy.exc import OperationalError

from tht.cli.config_cmd import CONFIG_OPT
from tht.config import ConfigError, load_config
from tht.db.connection import make_engine
from tht.db.introspect import IntrospectionError, introspect
from tht.db.sampling import add_examples
from tht.mschema.eligibility import classify_all

schema_app = typer.Typer(help="Gestione mschema (rappresentazione canonica dello schema)")


def _load_config_or_exit(config: Path):
    try:
        return load_config(config)
    except ConfigError as e:
        typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
        raise typer.Exit(code=1)


def physical_path(cfg) -> Path:
    return cfg.paths.artifacts / "mschema" / "physical.yaml"


def annotations_path(cfg) -> Path:
    return cfg.paths.artifacts / "mschema" / "annotations.yaml"


@schema_app.command("introspect")
def introspect_cmd(config: Path = CONFIG_OPT) -> None:
    """Introspeziona lo schema target e genera artifacts/mschema/physical.yaml."""
    cfg = _load_config_or_exit(config)
    if cfg.database.transport == "rest":
        from tht.db.introspect import introspect_rest
        from tht.db.sampling import add_examples_rest
        from tht.rest.client import RestClient, RestError

        client = RestClient(cfg.rest)
        try:
            phys = introspect_rest(
                client, database=cfg.database.database, schema=cfg.database.db_schema
            )
            add_examples_rest(client, phys, cfg.examples)
            classify_all(phys, cfg.eligibility)
        except RestError as e:
            typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
            raise typer.Exit(code=1)
    else:
        engine = make_engine(cfg.database)
        try:
            phys = introspect(
                engine, database=cfg.database.database, schema=cfg.database.db_schema
            )
            add_examples(engine, phys, cfg.examples)
            classify_all(phys, cfg.eligibility)
        except (OperationalError, IntrospectionError) as e:
            typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
            raise typer.Exit(code=1)
    out = physical_path(cfg)
    phys.to_yaml(out)
    n_cols = sum(len(t.columns) for t in phys.tables.values())
    n_ignored = sum(
        1 for t in phys.tables.values() for c in t.columns.values() if not c.eligible
    )
    typer.secho(
        f"OK: {len(phys.tables)} tabelle, {n_cols} colonne "
        f"({n_ignored} ignorate: testo ampio) -> {out}",
        fg=typer.colors.GREEN,
    )


@schema_app.command("check")
def check_cmd(config: Path = CONFIG_OPT) -> None:
    """Confronta physical.yaml e annotations.yaml; segnala annotazioni orfane."""
    from tht.mschema.merge import find_orphans
    from tht.mschema.models import Annotations, PhysicalSchema

    cfg = _load_config_or_exit(config)
    phys_file = physical_path(cfg)
    if not phys_file.exists():
        typer.secho(
            f"ERRORE: {phys_file} non trovato. Esegui prima `tht schema introspect`.",
            fg=typer.colors.RED, err=True,
        )
        raise typer.Exit(code=1)
    physical = PhysicalSchema.from_yaml(phys_file)
    annotations = Annotations.from_yaml(annotations_path(cfg))

    ignored = [
        f"{t}.{c} ({col.eligibility_reason})"
        for t, table in physical.tables.items()
        for c, col in table.columns.items()
        if not col.eligible
    ]
    if ignored:
        typer.secho(
            f"Colonne ignorate (testo ampio, {len(ignored)}):", fg=typer.colors.YELLOW
        )
        for line in ignored:
            typer.echo(f"  - {line}")

    orphans = find_orphans(physical, annotations)
    if orphans:
        typer.secho(f"ATTENZIONE: {len(orphans)} annotazioni orfane:", fg=typer.colors.YELLOW)
        for o in orphans:
            typer.echo(f"  - {o}")
        raise typer.Exit(code=3)
    typer.secho("OK: nessuna annotazione orfana.", fg=typer.colors.GREEN)


@schema_app.command("render")
def render_cmd(
    config: Path = CONFIG_OPT,
    format: str = typer.Option(
        "markdown", "--format", "-f", help="Formato: markdown | mschema-text | schema-dict"
    ),
    tables: list[str] = typer.Option(
        None, "--table", "-t", help="Limita alle tabelle indicate (ripetibile)."
    ),
    output: Path = typer.Option(None, "--output", "-o", help="File di output (default stdout)."),
) -> None:
    """Serializza mschema (physical + annotations) nel formato richiesto."""
    import json

    from tht.mschema.models import Annotations, PhysicalSchema
    from tht.mschema.render import to_markdown, to_mschema_text, to_schema_dict

    cfg = _load_config_or_exit(config)
    phys_file = physical_path(cfg)
    if not phys_file.exists():
        typer.secho(
            f"ERRORE: {phys_file} non trovato. Esegui prima `tht schema introspect`.",
            fg=typer.colors.RED, err=True,
        )
        raise typer.Exit(code=1)
    physical = PhysicalSchema.from_yaml(phys_file)
    annotations = Annotations.from_yaml(annotations_path(cfg))
    table_filter = list(tables) if tables else None

    if format == "markdown":
        out = to_markdown(physical, annotations)
    elif format == "mschema-text":
        out = to_mschema_text(physical, annotations, tables=table_filter)
    elif format == "schema-dict":
        out = json.dumps(to_schema_dict(physical, annotations), ensure_ascii=False, indent=2)
    else:
        typer.secho(f"ERRORE: formato sconosciuto: {format}", fg=typer.colors.RED, err=True)
        raise typer.Exit(code=1)

    if output:
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(out)
        typer.secho(f"OK: scritto {output}", fg=typer.colors.GREEN)
    else:
        typer.echo(out)