"""
Analisi sistematica dei file parquet in MyICR_Suite/local_db/parquet/
Produce PARQUET_ANALYSIS.md con schema, statistiche, campioni.
"""
import duckdb
import os
from pathlib import Path

PARQUET_DIR = r"I:\SOFTWARE\PYTHON_SRV\PYTHON_LOCAL\MyICR_Suite\local_db\parquet"
OUTPUT = r"C:\PYTHON\PARQUET_ANALYSIS.md"

files = sorted([f for f in Path(PARQUET_DIR).glob("*.parquet")])
con = duckdb.connect()

lines = []
lines.append("# PARQUET_ANALYSIS.md")
lines.append("**Generato automaticamente** — da completare con contesto di business.\n")
lines.append(f"**Sorgente:** `{PARQUET_DIR}`\n")
lines.append("---\n")

for f in files:
    name = f.stem
    path = str(f).replace("\\", "/")
    size_mb = round(f.stat().st_size / 1024 / 1024, 1)

    lines.append(f"## {name}.parquet ({size_mb} MB)\n")

    try:
        # Schema
        schema = con.execute(f"DESCRIBE SELECT * FROM read_parquet('{path}')").fetchall()
        row_count = con.execute(f"SELECT COUNT(*) FROM read_parquet('{path}')").fetchone()[0]

        lines.append(f"**Righe:** {row_count:,}\n")
        lines.append("**Colonne:**\n")
        lines.append("| Campo | Tipo | Note |")
        lines.append("|---|---|---|")
        for col in schema:
            lines.append(f"| `{col[0]}` | {col[1]} | |")

        # Campione valori per colonne categoriche (poche distinct)
        lines.append("\n**Valori distinti (colonne categoriche):**\n")
        for col in schema:
            col_name = col[0]
            col_type = col[1].upper()
            if any(t in col_type for t in ['VARCHAR', 'CHAR', 'TEXT', 'BOOL']):
                try:
                    n_distinct = con.execute(
                        f'SELECT COUNT(DISTINCT "{col_name}") FROM read_parquet(\'{path}\')'
                    ).fetchone()[0]
                    if n_distinct <= 30:
                        vals = con.execute(
                            f'SELECT DISTINCT "{col_name}" FROM read_parquet(\'{path}\') '
                            f'WHERE "{col_name}" IS NOT NULL ORDER BY "{col_name}" LIMIT 30'
                        ).fetchall()
                        val_list = ", ".join([f'`{v[0]}`' for v in vals if v[0] is not None])
                        lines.append(f"- **{col_name}** ({n_distinct}): {val_list}")
                    else:
                        lines.append(f"- **{col_name}**: {n_distinct:,} valori distinti")
                except:
                    pass

        # Range date
        lines.append("\n**Range date (colonne temporali):**\n")
        for col in schema:
            col_name = col[0]
            col_type = col[1].upper()
            if any(t in col_type for t in ['DATE', 'TIMESTAMP']):
                try:
                    r = con.execute(
                        f'SELECT MIN("{col_name}"), MAX("{col_name}") FROM read_parquet(\'{path}\')'
                    ).fetchone()
                    lines.append(f"- **{col_name}**: {r[0]} → {r[1]}")
                except:
                    pass

    except Exception as e:
        lines.append(f"⚠️ Errore: {e}")

    lines.append("\n---\n")

with open(OUTPUT, "w", encoding="utf-8") as out:
    out.write("\n".join(lines))

print(f"Done → {OUTPUT}")
con.close()