diff --git a/tests/test_scaffold_clean.py b/tests/test_scaffold_clean.py index 9c8bc3d..580cba4 100644 --- a/tests/test_scaffold_clean.py +++ b/tests/test_scaffold_clean.py @@ -10,12 +10,12 @@ import pytest +from toolkit.profile.raw import suggest_dateformat as _suggest_dateformat from toolkit.scaffold.clean import ( _columns_spec, _find_anno_raw_column, _has_anno_column, _select_expr, - _suggest_dateformat, generate_clean_sql, ) diff --git a/tests/test_scaffold_full.py b/tests/test_scaffold_full.py new file mode 100644 index 0000000..be29ef2 --- /dev/null +++ b/tests/test_scaffold_full.py @@ -0,0 +1,105 @@ +"""Tests for refactored scaffold/full.py — dict-based YAML generation.""" + +from __future__ import annotations +import pytest +import yaml +from toolkit.scaffold.full import generate_full_scaffold, suggest_mart_sql, suggest_validation + +pytestmark = pytest.mark.pure_unit + + +class TestSuggestValidation: + def test_none(self) -> None: + assert suggest_validation(None) == {} + + def test_with_columns(self) -> None: + v = suggest_validation({"columns_norm": ["Nome", "Importo"], "row_count": 100}) + assert v["required_columns"] == ["nome", "importo"] + assert v["min_rows"] == 50 + + +class TestSuggestMartSql: + def test_empty(self) -> None: + assert "placeholder" in suggest_mart_sql([], {}) + + def test_with_columns(self) -> None: + p = {"mapping_suggestions": {"nome": {"type": "VARCHAR"}}} + sql = suggest_mart_sql(["nome"], p) + assert "nome: VARCHAR" in sql + + +@pytest.mark.policy +class TestGenerateFullScaffold: + def test_file_no_profile(self) -> None: + probe = {"final_url": "https://example.com/data.csv", "source_type": "file"} + files = generate_full_scaffold("my-ds", probe) + assert "dataset.yml" in files + cfg = yaml.safe_load(files["dataset.yml"]) + assert cfg["raw"]["sources"][0]["type"] == "http_file" + + def test_ckan(self) -> None: + probe = { + "final_url": "https://data.example.org/dataset/x", + "source_type": "ckan", + "ckan_resources": [ + { + "name": "r1", + "url": "https://data.example.org/x.csv", + "format": "CSV", + "id": "123", + } + ], + } + files = generate_full_scaffold("my-ds", probe) + cfg = yaml.safe_load(files["dataset.yml"]) + assert cfg["raw"]["sources"][0]["type"] == "ckan" + + def test_sdmx_estat(self) -> None: + probe = { + "final_url": "https://ec.europa.eu/.../NAMA", + "source_type": "sdmx", + "sdmx_info": {"flow_id": "NAMA", "agency": "ESTAT"}, + } + files = generate_full_scaffold("my-ds", probe, inferred_years=list(range(2010, 2024))) + cfg = yaml.safe_load(files["dataset.yml"]) + assert cfg["raw"]["sources"][0]["args"]["agency"] == "ESTAT" + + def test_sparql(self) -> None: + probe = { + "final_url": "https://ld.istat.it/sparql", + "source_type": "sparql", + "sparql_info": {"endpoint": "https://ld.istat.it/sparql"}, + } + files = generate_full_scaffold("my-ds", probe) + cfg = yaml.safe_load(files["dataset.yml"]) + assert cfg["raw"]["sources"][0]["type"] == "sparql" + + def test_with_profile(self) -> None: + profile = { + "columns_norm": ["nome", "importo"], + "mapping_suggestions": {"nome": {"type": "VARCHAR"}, "importo": {"type": "DOUBLE"}}, + "row_count": 200, + "encoding_suggested": "utf-8", + "delim_suggested": ";", + } + probe = { + "final_url": "https://example.com/data.csv", + "source_type": "file", + "encoding_suggested": "utf-8", + "delim_suggested": ";", + } + files = generate_full_scaffold("my-ds", probe, profile=profile, inferred_years=[2020]) + cfg = yaml.safe_load(files["dataset.yml"]) + assert "read" in cfg["clean"] + assert "normalize_string" in files["sql/clean.sql"] + + def test_all_files(self) -> None: + probe = {"final_url": "https://example.com/data.csv", "source_type": "file"} + files = generate_full_scaffold("my-ds", probe) + assert set(files.keys()) == { + "dataset.yml", + "sql/clean.sql", + "sql/mart.sql", + "README.md", + "notes.md", + } diff --git a/tests/test_scout_infer.py b/tests/test_scout_infer.py index 779d058..d068519 100644 --- a/tests/test_scout_infer.py +++ b/tests/test_scout_infer.py @@ -269,7 +269,7 @@ def test_empty_fallback(self) -> None: cols: list[str] = [] profile: dict[str, Any] = {"mapping_suggestions": {}} sql = suggest_mart_sql(cols, profile) - assert "Sostituisci" in sql + assert "sostituisci" in sql assert "SELECT * FROM clean_input" in sql @pytest.mark.pure_unit @@ -303,9 +303,9 @@ def test_with_columns_and_rows(self) -> None: "row_count": 100, } val = suggest_validation(profile) - assert "clean" in val - assert val["clean"]["validate"]["required_columns"] == ["a", "b", "c"] - assert val["clean"]["validate"]["min_rows"] <= 100 + assert "required_columns" in val + assert val["required_columns"] == ["a", "b", "c"] + assert val["min_rows"] <= 100 @pytest.mark.pure_unit def test_empty_profile(self) -> None: diff --git a/toolkit/cli/cmd_scout.py b/toolkit/cli/cmd_scout.py index bb70d39..acb8dc8 100644 --- a/toolkit/cli/cmd_scout.py +++ b/toolkit/cli/cmd_scout.py @@ -21,12 +21,9 @@ import typer from toolkit.cli.cmd_run import run_init as _run_init -from toolkit.scout.http import DEFAULT_TIMEOUT, fetch_content -from toolkit.scaffold.full import ( - generate_full_scaffold, - suggest_validation, -) +from toolkit.scaffold.full import generate_full_scaffold from toolkit.scaffold.sources import infer_ext, slugify +from toolkit.scout.http import DEFAULT_TIMEOUT, fetch_content from toolkit.scout.infer import ( infer_granularity_from_name_and_columns, infer_topics, @@ -186,41 +183,11 @@ def scout_url( # --------------------------------------------------------------------------- -def _scaffold_file( +def _profile_sample( + sample_path: Path, url: str, - probe_result: dict[str, Any], - *, - run_raw: bool = False, - slug: str | None = None, -) -> None: - """Scarica sample, profila, inferisce, genera scaffold. - - Args: - url: URL del file da scaricare. - probe_result: Risultato del probe (dict). - run_raw: Se True, esegue raw run dopo scaffold. - slug: Slug personalizzato. Se None, auto-generato da url. - """ - slug = slug or slugify(url) - tmp_dir = Path(tempfile.gettempdir()) - tmp_name = f"scout_{slug}_{uuid.uuid4().hex[:8]}" - - # 1. Download sample - typer.echo("Downloading sample...") - try: - fetched = fetch_content(url, max_bytes=_SAMPLE_SIZE, timeout=30) - except RuntimeError as exc: - typer.echo(f"error: failed to fetch {url}: {exc}", err=True) - raise typer.Exit(code=1) - - content = fetched["content"] - ct = fetched.get("content_type") or probe_result.get("content_type", "") - ext = infer_ext(url, ct) - sample_path = tmp_dir / f"{tmp_name}{ext}" - sample_path.write_bytes(content) - typer.echo(f" Saved {len(content)} bytes to {sample_path}") - - # 2. Sniff + Profile +) -> tuple[dict[str, Any], dict[str, Any]]: + """Sniff + profile a downloaded sample file. Returns (profile, sniff_hints).""" from toolkit.profile.raw import profile_with_read_cfg, sniff_source_file sniff_hints = sniff_source_file(sample_path) @@ -228,7 +195,6 @@ def _scaffold_file( typer.echo(f" Delimiter: {sniff_hints.get('delim_suggested')}") typer.echo(f" Columns: {sniff_hints.get('columns_preview')}") - # XLSX → profiling via openpyxl (stesso reader del runtime clean) binary_fmt = sniff_hints.get("is_binary_file") if binary_fmt in ("xlsx", "xls"): from toolkit.profile.raw import profile_excel @@ -250,49 +216,65 @@ def _scaffold_file( profile = profile_with_read_cfg(sample_path, sniff_hints, read_cfg) - # 3. Retry skip se 0 colonne retry_skip = _resolve_columns(profile, sniff_hints, read_cfg, sample_path) if retry_skip is not None and retry_skip != sniff_hints.get("skip_suggested"): sniff_hints["skip_suggested"] = retry_skip read_cfg["skip"] = retry_skip profile = profile_with_read_cfg(sample_path, sniff_hints, read_cfg) - # 4. Propaga robust_read_suggested a profile per generate_full_scaffold - # Il flag puo' venire da sniff_hints (sniff_source_file) o da profile - # (profile_with_read_cfg, specie dopo retry con robust_preset). - if sniff_hints.get("robust_read_suggested") or profile.get("robust_read_suggested"): - profile["_robust_read_suggested"] = True + return profile, sniff_hints - # 5. Clean read via scaffold canonico - from toolkit.scaffold.clean import propose_clean_read - enriched = dict(profile) - for k in ( - "encoding_suggested", - "delim_suggested", - "decimal_suggested", - "skip_suggested", - "header_line", - "true_header_line", - "robust_read_suggested", - ): - if sniff_hints.get(k) is not None: - enriched[k] = sniff_hints[k] +def _scaffold_file( + url: str, + probe_result: dict[str, Any], + *, + run_raw: bool = False, + slug: str | None = None, +) -> None: + """Download sample, profile, generate scaffold via orchestrator.""" + slug = slug or slugify(url) + tmp_dir = Path(tempfile.gettempdir()) + tmp_name = f"scout_{slug}_{uuid.uuid4().hex[:8]}" + + # 1. Download sample + typer.echo("Downloading sample...") + try: + fetched = fetch_content(url, max_bytes=_SAMPLE_SIZE, timeout=30) + except RuntimeError as exc: + typer.echo(f"error: failed to fetch {url}: {exc}", err=True) + raise typer.Exit(code=1) + + content = fetched["content"] + ct = fetched.get("content_type") or probe_result.get("content_type", "") + ext = infer_ext(url, ct) + sample_path = tmp_dir / f"{tmp_name}{ext}" + sample_path.write_bytes(content) + typer.echo(f" Saved {len(content)} bytes to {sample_path}") - clean_read = propose_clean_read(enriched) + # 2. Sniff + Profile + profile, sniff_hints = _profile_sample(sample_path, url) - # 5. Legge i valori anno dal sample (se colonna Anno presente) + # 3. Propagate robust_read_suggested + if sniff_hints.get("robust_read_suggested") or profile.get("robust_read_suggested"): + profile["_robust_read_suggested"] = True + probe_result["robust_read_suggested"] = True + + # 4. Infer years + granularity + topics year_values = _read_year_values_from_sample(sample_path, sniff_hints, profile) if year_values: typer.echo(f" Year values in data: {sorted(year_values)}") - # 6. Inferenze norm_cols = ( profile.get("columns_norm") or profile.get("columns_raw") or profile.get("columns") or [] ) col_names = [str(c) for c in norm_cols] - inferred_years = suggest_years(column_names=col_names, profile=profile, year_values=year_values) + inferred_years = suggest_years( + column_names=col_names, + profile=profile, + year_values=year_values, + ) typer.echo(f" Suggested years: {inferred_years}") granularity = infer_granularity_from_name_and_columns(slug, col_names) @@ -300,54 +282,48 @@ def _scaffold_file( topics = infer_topics(f"{slug} {' '.join(col_names)}") if topics: - top_topics = [t["topic"] for t in topics[:3]] - typer.echo(f" Topics: {', '.join(top_topics)}") - - validation = suggest_validation(profile) - if validation: - typer.echo(" Validation rules: suggested") - - # 6. Genera scaffold - out_dir = Path(slug) - out_dir.mkdir(parents=True, exist_ok=True) + typer.echo(f" Topics: {', '.join(t['topic'] for t in topics[:3])}") + # 5. Enrich probe_result for orchestrator probe_result["inferred_granularity"] = granularity probe_result["inferred_topics"] = topics + for key in ( + "encoding_suggested", + "delim_suggested", + "decimal_suggested", + "skip_suggested", + "header_line", + "true_header_line", + ): + if sniff_hints.get(key) is not None: + probe_result[key] = sniff_hints[key] + # 6. Generate scaffold (one call) files = generate_full_scaffold( slug, probe_result, - clean_read=clean_read, profile=profile, inferred_years=inferred_years, - validation_suggestions=validation, ) - for rel_path, content in files.items(): + # 7. Write files + out_dir = Path(slug) + for rel_path, file_content in files.items(): full_path = out_dir / rel_path full_path.parent.mkdir(parents=True, exist_ok=True) - full_path.write_text(content, encoding="utf-8") - + full_path.write_text(file_content, encoding="utf-8") (out_dir / "notebooks").mkdir(exist_ok=True) - columns_count = len( - clean_read.get("columns") or profile.get("columns_raw") or profile.get("columns_norm") or [] - ) typer.echo(f"\nDataset YAML generated: {out_dir / 'dataset.yml'}") - typer.echo(f" clean.read.columns: {columns_count} columns") typer.echo(f" years: {inferred_years}") typer.echo(f" source_type: {probe_result.get('source_type', 'file')}") - typer.echo(" sql/clean.sql: generated (with type casts)") - typer.echo(" sql/mart.sql: generated (skeleton)") - typer.echo(" README.md, notes.md, notebooks/: created") + typer.echo(" sql/clean.sql, sql/mart.sql: generated") - # 7. Opzionalmente raw run + # 8. Optional raw run if run_raw: _run_bootstrap(str(out_dir / "dataset.yml")) - # 9. Cleanup sample_path.unlink(missing_ok=True) - if not run_raw: typer.echo(f"\nNext: toolkit run all --config {out_dir / 'dataset.yml'}") @@ -373,69 +349,43 @@ def _scaffold_ckan( run_raw: bool = False, slug: str | None = None, ) -> None: - """Scaffold per risorsa CKAN. - - Tenta nell'ordine: - 1. **DataStore schema** (se ``datastore_active=True``) — colonne e tipi - via ``datastore_search?limit=0``, nessun download. - 2. **Download + profiling CSV** (comportamento attuale). - 3. **Scaffold minimale** via ``generate_full_scaffold`` (fallback). - """ + """Scaffold per risorsa CKAN. Tenta DataStore → CSV profiling → minimal.""" resources = probe_result.get("ckan_resources") or [] if not resources: typer.echo("error: no CKAN resources available", err=True) raise typer.Exit(code=1) - # Prova DataStore se la prima risorsa lo supporta + slug = slug or slugify(url) + + # Try DataStore schema first if resources[0].get("datastore_active"): - first = resources[0] - slug = slug or slugify(url) try: from toolkit.scout.http import fetch_ckan_datastore_schema from toolkit.scaffold.clean import profile_from_datastore - fields = fetch_ckan_datastore_schema(url, first["id"]) + fields = fetch_ckan_datastore_schema(url, resources[0]["id"]) if fields: profile = profile_from_datastore(fields) - scaffold_files = generate_full_scaffold( + files = generate_full_scaffold( slug, probe_result, profile=profile, - inferred_years=[2024], ) - out_dir = _write_scaffold_files(slug, scaffold_files) - - typer.echo(f"\nCKAN DataStore scaffold generated: {out_dir / 'dataset.yml'}") - typer.echo(f" clean.sql with {len(fields)} columns from DataStore schema") + out_dir = _write_scaffold_files(slug, files) + typer.echo(f"\nCKAN DataStore scaffold: {out_dir / 'dataset.yml'}") + typer.echo(f" clean.sql with {len(fields)} columns") return except Exception: typer.echo(" DataStore schema fetch failed, trying CSV profiling...") - first_url = resources[0]["url"] + # Fallback: download + profile first resource try: - _scaffold_file(first_url, probe_result, run_raw=run_raw, slug=slug) - return + _scaffold_file(resources[0]["url"], probe_result, run_raw=run_raw, slug=slug) except (typer.Exit, Exception): - typer.echo(" Warning: profiling failed for resource, generating minimal scaffold") - _scaffold_minimal_ckan(url, probe_result, slug=slug) - - -def _scaffold_minimal_ckan( - url: str, - probe_result: dict[str, Any], - *, - slug: str | None = None, -) -> None: - """Genera scaffold minimale CKAN quando il profiling fallisce. - - Delega a ``generate_full_scaffold`` senza profile per ottenere - dataset.yml, clean.sql placeholder e mart.sql skeleton. - """ - slug = slug or slugify(url) - scaffold_files = generate_full_scaffold(slug, probe_result) - out_dir = _write_scaffold_files(slug, scaffold_files) - typer.echo(f"\nMinimal dataset YAML generated: {out_dir / 'dataset.yml'}") - typer.echo(" clean.read, clean.sql, mart.sql: manual editing required") + typer.echo(" Warning: profiling failed, generating minimal scaffold") + files = generate_full_scaffold(slug, probe_result) + out_dir = _write_scaffold_files(slug, files) + typer.echo(f"\nMinimal scaffold: {out_dir / 'dataset.yml'}") def _scaffold_html( @@ -465,76 +415,11 @@ def _scaffold_sparql( run_raw: bool = False, slug: str | None = None, ) -> None: - """Scaffold per endpoint SPARQL.""" + """Scaffold per endpoint SPARQL — uses orchestrator.""" slug = slug or slugify(url) - from toolkit.scaffold.sources import block_sparql as _generate_raw_sources_block_sparql - - sparql_info = probe_result.get("sparql_info") or {} - datasets = sparql_info.get("datasets") or [] - - lines = [ - "# Auto-generated by toolkit scout --scaffold", - "# Review and adjust the SPARQL query before running", - "", - 'root: "../../out"', - "schema_version: 1", - "", - "dataset:", - f' name: "{slug}"', - " years: [2024]", - "", - "raw:", - " output_policy: overwrite", - " sources:", - ] - # Usa la prima query disponibile o una SELECT base - sample_query = "SELECT * WHERE { ?s ?p ?o } LIMIT 1000" - lines.extend(_generate_raw_sources_block_sparql(url, sample_query)) - lines.append("") - lines.append("clean:") - lines.append(' sql: "sql/clean.sql"') - lines.append("") - lines.append("mart:") - lines.append(" tables:") - lines.append(f' - name: "{slug}"') - lines.append(' sql: "sql/mart.sql"') - - out_dir = Path(slug) - out_dir.mkdir(parents=True, exist_ok=True) - (out_dir / "dataset.yml").write_text("\n".join(lines) + "\n", encoding="utf-8") - - clean_sql_path = out_dir / "sql" / "clean.sql" - clean_sql_path.parent.mkdir(parents=True, exist_ok=True) - clean_sql_path.write_text( - "-- SPARQL source: transform raw CSV to clean tabular.\nSELECT * FROM raw_input\n", - encoding="utf-8", - ) - - mart_sql_path = out_dir / "sql" / "mart.sql" - if not mart_sql_path.exists(): - mart_sql_path.parent.mkdir(parents=True, exist_ok=True) - mart_sql_path.write_text( - "-- Default mart: SELECT * FROM clean_input.\nSELECT * FROM clean_input\n", - encoding="utf-8", - ) - - (out_dir / "README.md").write_text( - f"# {slug}\n\nFonte SPARQL: {url}\n\n" - "## Domanda\n\n-\n\n" - "## Dataset\n\n-\n\n" - "## Stato\n\n- intake\n", - encoding="utf-8", - ) - (out_dir / "notes.md").write_text( - "## Tecnico\n\n- Fonte SPARQL\n" - f"- Endpoint: {url}\n" - f"- Dataset trovati: {len(datasets)}\n\n" - "## Analitico\n\n-\n\n" - "## Cautele\n\n- Verificare rate limiting dell'endpoint SPARQL\n", - encoding="utf-8", - ) - - # run_raw non supportato per SPARQL — usa: toolkit run all -c dataset.yml + files = generate_full_scaffold(slug, probe_result) + _write_scaffold_files(slug, files) + typer.echo(f"\nSPARQL scaffold generated: {Path(slug) / 'dataset.yml'}") def _scaffold_sdmx( @@ -544,85 +429,20 @@ def _scaffold_sdmx( run_raw: bool = False, slug: str | None = None, ) -> None: - """Scaffold per endpoint SDMX.""" + """Scaffold per endpoint SDMX — uses orchestrator.""" slug = slug or slugify(url) - from toolkit.scaffold.sources import block_sdmx as _generate_raw_sources_block_sdmx - sdmx_info = probe_result.get("sdmx_info") or {} year_min = sdmx_info.get("year_min") year_max = sdmx_info.get("year_max") + years = list(range(year_min, year_max + 1)) if year_min and year_max else None - if year_min and year_max: - inferred_years = list(range(year_min, year_max + 1)) - else: - inferred_years = [2024] - - # Scaffold minimo per SDMX (no profiling CSV) - # generate_full_scaffold non usato perché SDMX ha template diverso - # (dataset.yml sovrascritto qui sotto) - - # Sovrascrivi dataset.yml con configurazione SDMX - lines = [ - "# Auto-generated by toolkit scout --scaffold", - "# Review and adjust before running", - "", - 'root: "../../out"', - "schema_version: 1", - "", - "dataset:", - f' name: "{slug}"', - " years: " + _fmt_years(inferred_years), - "", - "raw:", - " output_policy: overwrite", - " sources:", - ] - lines.extend(_generate_raw_sources_block_sdmx(sdmx_info, url)) - lines.append("") - lines.append("clean:") - lines.append(' sql: "sql/clean.sql"') - lines.append("") - lines.append("mart:") - lines.append(" tables:") - lines.append(f' - name: "{slug}"') - lines.append(' sql: "sql/mart.sql"') - - out_dir = Path(slug) - out_dir.mkdir(parents=True, exist_ok=True) - (out_dir / "dataset.yml").write_text("\n".join(lines) + "\n", encoding="utf-8") - - # clean.sql generico per SDMX - clean_sql_path = out_dir / "sql" / "clean.sql" - clean_sql_path.parent.mkdir(parents=True, exist_ok=True) - clean_sql_path.write_text( - "-- SDMX flow: transform raw SDMX to clean tabular.\n" - "-- Personalizza estrazione delle dimensioni e misure.\n" - "SELECT * FROM raw_input\n", - encoding="utf-8", - ) - - mart_sql_path = out_dir / "sql" / "mart.sql" - if not mart_sql_path.exists(): - mart_sql_path.parent.mkdir(parents=True, exist_ok=True) - mart_sql_path.write_text( - "-- Default mart: SELECT * FROM clean_input.\nSELECT * FROM clean_input\n", - encoding="utf-8", - ) - - (out_dir / "README.md").write_text( - f"# {slug}\n\nFonte: {url}\n\n## Domanda\n\n-\n\n## Dataset\n\n-\n\n## Stato\n\n- intake\n", - encoding="utf-8", - ) - (out_dir / "notes.md").write_text( - "## Tecnico\n\n- Fonte SDMX\n\n" - "## Analitico\n\n-\n\n" - "## Cautele\n\n- Verificare completezza serie storica\n", - encoding="utf-8", + files = generate_full_scaffold( + slug, + probe_result, + inferred_years=years, ) - (out_dir / "notebooks").mkdir(exist_ok=True) - - typer.echo(f"\nDataset YAML generated: {out_dir / 'dataset.yml'}") - typer.echo(" source_type: sdmx") + out_dir = _write_scaffold_files(slug, files) + typer.echo(f"\nSDMX scaffold generated: {out_dir / 'dataset.yml'}") typer.echo(f" flow: {sdmx_info.get('flow_id', '?')}") if year_min and year_max: typer.echo(f" years: {year_min}-{year_max}") @@ -722,12 +542,6 @@ def _run_bootstrap(config_path: str) -> None: typer.echo(f" toolkit run mart --config {config_path}") -def _fmt_years(years: list[int]) -> str: - if len(years) <= 4: - return "[" + ", ".join(str(y) for y in years) + "]" - return f"[{years[0]}..{years[-1]}]" - - # --------------------------------------------------------------------------- # CLI command # --------------------------------------------------------------------------- diff --git a/toolkit/profile/raw.py b/toolkit/profile/raw.py index 4a1ef93..cd12b6d 100644 --- a/toolkit/profile/raw.py +++ b/toolkit/profile/raw.py @@ -29,6 +29,84 @@ from toolkit.profile._column_profile import _build_mapping_suggestions, _normalize_colname +# --------------------------------------------------------------------------- +# Date format detection (moved from scaffold/clean.py to fix dependency inversion) +# --------------------------------------------------------------------------- + +_DATE_FORMAT_GROUPS: list[tuple[str, ...]] = [ + ("%d/%m/%Y", "%m/%d/%Y"), # slash 4-digit + ("%d-%m-%Y", "%m-%d-%Y"), # dash 4-digit + ("%d/%m/%y", "%m/%d/%y"), # slash 2-digit + ("%d-%m-%y", "%m-%d-%y"), # dash 2-digit + ("%Y/%m/%d",), # ISO con slash +] + + +def _try_strptime(value: str, fmt: str) -> bool: + """Try to parse a date string with the given strptime format.""" + import datetime + + try: + datetime.datetime.strptime(value, fmt) + return True + except ValueError: + return False + + +def suggest_dateformat(profile: dict[str, Any]) -> str | None: + """Detect non-ISO date format from raw date values in the profile. + + Uses ``date_raw_values`` (extracted from raw CSV *before* DuckDB + converts dates) and ``datetime.strptime`` for validation. + + Each column picks its best format (>=60% of its non-empty values). + Only suggests a ``dateformat`` if EVERY column picks the SAME format. + """ + date_raw = profile.get("date_raw_values", {}) + if not date_raw: + return None + + col_formats: dict[str, str] = {} + for col, values in date_raw.items(): + non_empty = [v for v in values if v] + total = len(non_empty) + if total == 0: + continue + + best_fmt: str | None = None + best_score = 0 + + for fmt_group in _DATE_FORMAT_GROUPS: + if len(fmt_group) == 1: + fmt = fmt_group[0] + count = sum(1 for v in non_empty if _try_strptime(v, fmt)) + if count > best_score: + best_score = count + best_fmt = fmt + else: + dmy_fmt, mdy_fmt = fmt_group + dmy_count = sum(1 for v in non_empty if _try_strptime(v, dmy_fmt)) + mdy_count = sum(1 for v in non_empty if _try_strptime(v, mdy_fmt)) + if dmy_count > mdy_count and dmy_count > best_score: + best_score = dmy_count + best_fmt = dmy_fmt + elif mdy_count > dmy_count and mdy_count > best_score: + best_score = mdy_count + best_fmt = mdy_fmt + + if best_fmt is not None and best_score >= total * 0.6: + col_formats[col] = best_fmt + + if not col_formats: + return None + + unique = set(col_formats.values()) + if len(unique) == 1: + return unique.pop() + + return None + + def _safe_mkdir(p: Path) -> None: p.mkdir(parents=True, exist_ok=True) @@ -171,9 +249,7 @@ def build_suggested_read_cfg( # Propaga dateformat se non gia' in source_cfg if "dateformat" not in cfg and data.get("date_raw_values"): - from toolkit.scaffold.clean import _suggest_dateformat - - fmt = _suggest_dateformat(data) + fmt = suggest_dateformat(data) if fmt: cfg["dateformat"] = fmt diff --git a/toolkit/scaffold/clean.py b/toolkit/scaffold/clean.py index ec8c8e3..2b82318 100644 --- a/toolkit/scaffold/clean.py +++ b/toolkit/scaffold/clean.py @@ -1,6 +1,5 @@ from __future__ import annotations -import datetime import re from pathlib import Path from typing import Any @@ -100,90 +99,8 @@ def profile_from_datastore(fields: list[dict[str, Any]]) -> dict[str, Any]: } -# Formati data organizzati per gruppo con stesso separatore. -# I gruppi con due formati (DMY/MDY) richiedono disambiguazione: -# valori che parsano in UN SOLO formato votano quello; valori che -# parsano in entrambi (es. "03/04/2024") sono ambigui e non contano. -# I gruppi con un solo formato (YYYY/mm/dd) sono non-ambigu. -_DATE_FORMAT_GROUPS: list[tuple[str, ...]] = [ - ("%d/%m/%Y", "%m/%d/%Y"), # slash 4-digit - ("%d-%m-%Y", "%m-%d-%Y"), # dash 4-digit - ("%d/%m/%y", "%m/%d/%y"), # slash 2-digit - ("%d-%m-%y", "%m-%d-%y"), # dash 2-digit - ("%Y/%m/%d",), # ISO con slash -] - - -def _try_strptime(value: str, fmt: str) -> bool: - """Try to parse a date string with the given strptime format.""" - try: - datetime.datetime.strptime(value, fmt) - return True - except ValueError: - return False - - -def _suggest_dateformat(profile: dict[str, Any]) -> str | None: - """Detect non-ISO date format from raw date values in the profile. - - Uses ``date_raw_values`` (extracted from raw CSV *before* DuckDB - converts dates) and ``datetime.strptime`` for validation. - - For ambiguous separators (``/``, ``-``), counts how many values - successfully parse in each format (DMY and MDY). Values that parse - in both formats are counted for both — the ambiguity means neither - format is *excluded*, but the format with more total parses wins. - - Each column picks its best format (>=60% of its non-empty values). - Only suggests a ``dateformat`` if EVERY column picks the SAME format. - - Returns the ``dateformat`` string or ``None``. - """ - date_raw = profile.get("date_raw_values", {}) - if not date_raw: - return None - - col_formats: dict[str, str] = {} - for col, values in date_raw.items(): - non_empty = [v for v in values if v] - total = len(non_empty) - if total == 0: - continue - - best_fmt: str | None = None - best_score = 0 - - for fmt_group in _DATE_FORMAT_GROUPS: - if len(fmt_group) == 1: - fmt = fmt_group[0] - count = sum(1 for v in non_empty if _try_strptime(v, fmt)) - if count > best_score: - best_score = count - best_fmt = fmt - else: - dmy_fmt, mdy_fmt = fmt_group - dmy_count = sum(1 for v in non_empty if _try_strptime(v, dmy_fmt)) - mdy_count = sum(1 for v in non_empty if _try_strptime(v, mdy_fmt)) - - # Only consider when one format clearly wins over the other - if dmy_count > mdy_count and dmy_count > best_score: - best_score = dmy_count - best_fmt = dmy_fmt - elif mdy_count > dmy_count and mdy_count > best_score: - best_score = mdy_count - best_fmt = mdy_fmt - - if best_fmt is not None and best_score >= total * 0.6: - col_formats[col] = best_fmt - - if not col_formats: - return None - - unique = set(col_formats.values()) - if len(unique) == 1: - return unique.pop() - - return None +# _suggest_dateformat, _try_strptime, _DATE_FORMAT_GROUPS +# moved to profile/raw.py (fixes dependency inversion: profile → scaffold) def _find_anno_raw_column(profile: dict[str, Any]) -> str | None: @@ -457,7 +374,9 @@ def propose_clean_read(profile: dict[str, Any]) -> dict[str, Any]: read["decimal"] = decimal # --- dateformat: auto-detect non-ISO date formats (es. dd/mm/YYYY) --- - date_fmt = _suggest_dateformat(profile) + from toolkit.profile.raw import suggest_dateformat + + date_fmt = suggest_dateformat(profile) if date_fmt: read["dateformat"] = date_fmt diff --git a/toolkit/scaffold/full.py b/toolkit/scaffold/full.py index 3707415..ef866d0 100644 --- a/toolkit/scaffold/full.py +++ b/toolkit/scaffold/full.py @@ -1,195 +1,15 @@ -"""Generazione completa di un candidate dataset: YAML, SQL, README, notes. +"""Generazione di un candidate dataset: YAML, SQL, README, notes. -Dipende da: - scaffold/clean.py → propose_clean_read (serializza clean.read) - scaffold/sources.py → slugify, infer_filename, block_* +Thin orchestrator: builds config dict → yaml_dumps(), +delegates SQL to clean.generate_clean_sql(). """ from __future__ import annotations from typing import Any -from urllib.parse import urlparse -from toolkit.scaffold.clean import generate_clean_sql -from toolkit.scaffold.sources import ( - block_ckan, - block_http_file, - block_links, - block_sdmx, - infer_filename, -) - - -def _format_years(years: list[int]) -> str: - if len(years) <= 4: - return "[" + ", ".join(str(y) for y in years) + "]" - return f"[{years[0]}..{years[-1]}]" - - -def _serialize_clean_read(clean_read: dict[str, Any]) -> list[str]: - """Serializza clean.read come righe YAML (senza intestazione clean:). - - Opzioni emesse solo se esplicitamente presenti in ``clean_read``: - delim, encoding, decimal, header, skip, columns. - Opzioni robuste (strict_mode, null_padding, ignore_errors) sono emesse - solo se il profilo le richiede — NON per default. - """ - lines: list[str] = [] - lines.append(" read:") - if "delim" in clean_read: - lines.append(f' delim: "{clean_read["delim"]}"') - if "encoding" in clean_read: - lines.append(f' encoding: "{clean_read["encoding"]}"') - if "decimal" in clean_read: - lines.append(f' decimal: "{clean_read["decimal"]}"') - if "header" in clean_read: - lines.append(f" header: {str(clean_read['header']).lower()}") - if clean_read.get("skip", 0) > 0: - lines.append(f" skip: {clean_read['skip']}") - # Opzioni robuste: emesse solo se esplicitamente richieste dal profilo - if clean_read.get("strict_mode") is False: - lines.append(" strict_mode: false") - if clean_read.get("null_padding") is True: - lines.append(" null_padding: true") - if clean_read.get("ignore_errors") is True: - lines.append(" ignore_errors: true") - columns = clean_read.get("columns") - if columns: - lines.append(" columns:") - for col_name, col_type in columns.items(): - lines.append(f' "{col_name}": "{col_type}"') - else: - lines.append(" # columns: auto-detected from header") - return lines - - -def _generate_readme(slug: str, url: str) -> str: - return ( - f"# {slug}\n\n" - f"Fonte: {url}\n\n" - "## Domanda\n\n-\n\n" - "## Dataset\n\n-\n\n" - "## Perche vale la pena testarlo\n\n-\n\n" - "## Output minimo atteso\n\n-\n\n" - "## Criterio di promozione\n\n-\n\n" - "## Stato\n\n- intake\n\n" - "## Prossimo passo\n\n- scout URL poi run all\n" - ) - - -def _generate_notes(granularity: str | None, topics: list[dict[str, Any]] | None) -> str: - lines: list[str] = ["## Tecnico\n\n-\n"] - if granularity: - lines.append(f"- Granularita rilevata: {granularity}\n") - if topics: - topic_names = ", ".join(t["topic"] for t in topics[:3]) - lines.append(f"- Topic suggeriti: {topic_names}\n") - lines.append("\n## Analitico\n\n-\n\n") - lines.append("## Cautele\n\n") - lines.append("- La serie storica e omogenea su tutti gli anni?\n") - lines.append("- Ci sono discontinuita dichiarate dalla fonte?\n") - lines.append("- I valori nulli sono zero reale o dato mancante?\n") - return "".join(lines) - - -# --------------------------------------------------------------------------- -# SQL generation -# --------------------------------------------------------------------------- - - -def suggest_validation(profile: dict[str, Any]) -> dict[str, Any]: - """Suggerisce validation rules da inserire in dataset.yml.""" - from toolkit.scaffold.clean import _snake_case - - norm_cols = profile.get("columns_norm") or profile.get("columns_raw") or [] - row_count = profile.get("row_count", 0) - validation: dict[str, Any] = {} - clean_val: dict[str, Any] = {} - if row_count: - clean_val["min_rows"] = max(1, int(row_count * 0.5)) - if norm_cols: - # required_columns va in snake_case (come la clean output) - clean_val["required_columns"] = [_snake_case(c) for c in norm_cols[:5]] - if clean_val: - validation["clean"] = {"validate": clean_val} - return validation - - -def _has_year_column(columns: list[dict[str, Any]] | list[str]) -> bool: - year_keywords = ["anno", "year", "periodo", "period", "data", "date", "mese", "month"] - for col in columns: - name = col if isinstance(col, str) else col.get("name", "") - if any(kw in name.lower() for kw in year_keywords): - return True - return False - - -def _has_region_column(columns: list[dict[str, Any]] | list[str]) -> bool: - region_keywords = [ - "regione", - "region", - "provincia", - "province", - "comune", - "municip", - "area", - "territorio", - ] - for col in columns: - name = col if isinstance(col, str) else col.get("name", "") - if any(kw in name.lower() for kw in region_keywords): - return True - return False - - -def _has_numeric_column(columns: list[dict[str, Any]] | list[str], profile: dict[str, Any]) -> bool: - mapping = profile.get("mapping_suggestions") or {} - for col in columns: - name = col if isinstance(col, str) else col.get("name", "") - spec = mapping.get(name) or {} - if spec.get("type") in ("integer", "float", "double", "bigint", "decimal", "int"): - return True - if isinstance(col, dict) and col.get("type") in ("integer", "float", "double", "int"): - return True - return False - - -def _find_matching_column(col_names: list[str], keywords: list[str]) -> str | None: - """Trova la prima colonna in col_names che contiene uno dei keywords (case-insensitive).""" - for col in col_names: - if any(kw in col.lower() for kw in keywords): - return col - return None - - -def suggest_mart_sql(columns: list[dict[str, Any]] | list[str], profile: dict[str, Any]) -> str: - """Genera mart.sql come scheletro commentato. - - Non tenta aggregazioni automatiche — produrrebbero GROUP BY rumorosi - o SUM su chiavi. Lascia all'utente la decisione su come aggregare. - """ - if columns and isinstance(columns[0], dict): - col_names = [c.get("name", f"col{i}") for i, c in enumerate(columns)] - else: - col_names = list(columns) if columns else [] - if not col_names: - return ( - "-- mart placeholder. Sostituisci con la tua aggregazione.\nSELECT * FROM clean_input\n" - ) - - mapping = profile.get("mapping_suggestions") or {} - type_hints = {name: (mapping.get(name) or {}).get("type", "?") for name in col_names} - hint = ", ".join(f"{n}: {t}" for n, t in type_hints.items()) - return ( - f"-- Colonne clean_input: {hint}\n" - f"-- Sostituisci con la tua aggregazione (es. SUM, COUNT, AVG).\n" - f"SELECT * FROM clean_input\n" - ) - - -# --------------------------------------------------------------------------- -# Full scaffold orchestration -# --------------------------------------------------------------------------- +from toolkit.core.io import yaml_dumps +from toolkit.scaffold.clean import generate_clean_sql, propose_clean_read def generate_full_scaffold( @@ -201,108 +21,47 @@ def generate_full_scaffold( inferred_years: list[int] | None = None, validation_suggestions: dict[str, Any] | None = None, ) -> dict[str, str]: - """Genera tutti i file di un candidate dataset. + """Generate all files for a candidate dataset. - Returns dict: {filename: content} con dataset.yml, sql/clean.sql, - sql/mart.sql, README.md, notes.md. + Returns {filename: content} with dataset.yml, sql/clean.sql, + sql/mart.sql, README.md, notes.md. """ - years = inferred_years or [2024] - source_type = probe_result.get("source_type", "file") - final_url = probe_result["final_url"] - - # Nome SQL-safe per dataset.name e mart.tables[].name: - # lo slug (directory) usa trattini, ma SQL/YAML require underscore. safe_name = slug.replace("-", "_") - - yml_lines: list[str] = [ - "# Auto-generated by toolkit", - "# Review and adjust before running", - "", - 'root: "../../out"', - "schema_version: 1", - "", - "dataset:", - f' name: "{safe_name}"', - " years: " + _format_years(years), - "", - "", - "raw:", - " output_policy: overwrite", - " sources:", - ] - - if source_type == "ckan" and probe_result.get("ckan_resources"): - parsed = urlparse(final_url) - portal_base = f"{parsed.scheme}://{parsed.netloc}" - yml_lines.extend(block_ckan(probe_result["ckan_resources"], portal_base)) - elif source_type == "sdmx": - yml_lines.extend(block_sdmx(probe_result.get("sdmx_info"), final_url)) - elif source_type == "html" and probe_result.get("candidate_links"): - yml_lines.extend(block_links(probe_result["candidate_links"])) - else: - fname = infer_filename(final_url, slug) - yml_lines.extend(block_http_file(final_url, slug, fname)) - - # clean section (read + sql + validate) - yml_lines.append("") - yml_lines.append("clean:") - if clean_read: - yml_lines.extend(_serialize_clean_read(clean_read)) - # read_mode: robust top-level (fuori da clean.read) se il profilo lo richiede. - # Controlla entrambi i nomi: _robust_read_suggested (propagato dal wrapper - # CLI) e robust_read_suggested (direct profile), cosi' il contratto sta - # nel generatore e non solo nel chiamante. - if profile and (profile.get("_robust_read_suggested") or profile.get("robust_read_suggested")): - yml_lines.append(" read_mode: robust") - # required_columns a livello clean: (fuori da validate:) - if validation_suggestions: - cv = validation_suggestions.get("clean", {}) - req = cv.get("required_columns") or cv.get("validate", {}).get("required_columns") - if req: - items = ", ".join(f'"{c}"' for c in req) - yml_lines.append(f" required_columns: [{items}]") - yml_lines.append(' sql: "sql/clean.sql"') - - if validation_suggestions: - cv = validation_suggestions.get("clean", {}) - vblock = cv.get("validate", {}) - # required_columns non va dentro validate: — gia' emesso sopra. - vblock.pop("required_columns", None) - if vblock: - yml_lines.append(" validate:") - for key, val in vblock.items(): - if isinstance(val, list): - items = ", ".join(f'"{v}"' for v in val) - yml_lines.append(f" {key}: [{items}]") - elif isinstance(val, bool): - yml_lines.append(f" {key}: {str(val).lower()}") - else: - yml_lines.append(f" {key}: {val}") - - yml_lines.append("") - yml_lines.append("mart:") - yml_lines.append(" tables:") - yml_lines.append(f' - name: "{safe_name}"') - yml_lines.append(' sql: "sql/mart.sql"') - - if validation_suggestions: - mv = validation_suggestions.get("mart") - if mv: - validate_block = mv.get("validate", mv) - if validate_block: - yml_lines.append(" validate:") - for key, val in validate_block.items(): - if isinstance(val, list): - items = ", ".join(f'"{v}"' for v in val) - yml_lines.append(f" {key}: [{items}]") - elif isinstance(val, bool): - yml_lines.append(f" {key}: {str(val).lower()}") - else: - yml_lines.append(f" {key}: {val}") + source_type = probe_result.get("source_type", "file") + years = inferred_years or [2024] + final_url = probe_result.get("final_url", "") + + # Infer years from profile's anno column if available + if inferred_years is None and profile: + anno_col = _find_anno_col(profile) + if anno_col: + anno_values = profile.get("date_raw_values", {}).get(anno_col, []) + if not anno_values: + # Try reading from column values in mapping + anno_values = profile.get("column_values", {}).get(anno_col, []) + if anno_values: + unique_years = sorted( + { + int(v) + for v in anno_values + if v and str(v).isdigit() and 1900 <= int(v) <= 2100 + } + ) + if unique_years: + years = unique_years + + config = _build_config_dict( + safe_name, + years, + source_type, + probe_result, + profile, + clean_read=clean_read, + validation_suggestions=validation_suggestions, + ) if profile: - first_year = years[0] - clean_sql = generate_clean_sql(profile, slug, first_year) + clean_sql = generate_clean_sql(profile, slug, years[0]) norm_cols = ( profile.get("columns_norm") or profile.get("columns_raw") @@ -311,22 +70,248 @@ def generate_full_scaffold( ) mart_sql = suggest_mart_sql(norm_cols, profile) else: - mart_sql = "-- Nessuna colonna rilevata dal profiling. Sostituisci con la tua aggregazione.\nSELECT * FROM clean_input\n" - clean_sql = "-- ATTENZIONE: profiling non ha rilevato colonne.\nSELECT 1 AS placeholder FROM raw_input\n" + clean_sql = ( + "-- ATTENZIONE: profiling non ha rilevato colonne.\n" + "SELECT 1 AS placeholder FROM raw_input\n" + ) + mart_sql = ( + "-- mart placeholder — sostituisci con la tua aggregazione.\n" + "SELECT * FROM clean_input\n" + ) - if profile: - topics = probe_result.get("inferred_topics") - granularity = probe_result.get("inferred_granularity") - notes = _generate_notes(granularity, topics) - else: - notes = _generate_notes(None, None) + topics = probe_result.get("inferred_topics") + granularity = probe_result.get("inferred_granularity") - result: dict[str, str] = { - "dataset.yml": "\n".join(yml_lines) + "\n", + return { + "dataset.yml": yaml_dumps(config), "sql/clean.sql": clean_sql, "sql/mart.sql": mart_sql, "README.md": _generate_readme(slug, final_url), - "notes.md": notes, + "notes.md": _generate_notes(granularity, topics), + } + + +def _build_config_dict( + name: str, + years: list[int], + source_type: str, + probe_result: dict[str, Any], + profile: dict[str, Any] | None, + *, + clean_read: dict[str, Any] | None = None, + validation_suggestions: dict[str, Any] | None = None, +) -> dict[str, Any]: + config: dict[str, Any] = { + "root": "../../out", + "schema_version": 1, + "dataset": {"name": name, "years": years}, + "raw": { + "output_policy": "overwrite", + "sources": _build_sources(source_type, probe_result, name), + }, + "clean": {"sql": "sql/clean.sql"}, + "mart": {"tables": [{"name": name, "sql": "sql/mart.sql"}]}, + } + + if clean_read: + config["clean"]["read"] = clean_read + elif profile: + enriched = _enrich_profile(profile, probe_result) + suggested = propose_clean_read(enriched) + if suggested: + config["clean"]["read"] = suggested + # Propose robust mode when profiling detected CSV errors + if profile.get("robust_read_suggested") or profile.get("_robust_read_suggested"): + config["clean"]["read_mode"] = "robust" + + vs = validation_suggestions or suggest_validation(profile) + if vs: + req = vs.get("required_columns") + if req: + config["clean"]["required_columns"] = req + min_rows = vs.get("min_rows") + if min_rows is not None: + config["clean"].setdefault("validate", {})["min_rows"] = min_rows + mart_min = vs.get("mart_min_rows") + if mart_min is not None: + config["mart"]["validate"] = { + "table_rules": {name: {"min_rows": mart_min}}, + } + + return config + + +def _build_sources( + source_type: str, + probe_result: dict[str, Any], + slug: str, +) -> list[dict[str, Any]]: + final_url = probe_result.get("final_url", "") + if source_type == "ckan" and probe_result.get("ckan_resources"): + return _ckan_sources(probe_result) + if source_type == "sdmx": + return _sdmx_sources(probe_result, final_url) + if source_type == "sparql": + info = probe_result.get("sparql_info") or {} + return [ + { + "name": "sparql", + "type": "sparql", + "args": { + "endpoint": info.get("endpoint", final_url), + "query": "SELECT * WHERE { ?s ?p ?o } LIMIT 1000", + }, + "primary": True, + } + ] + if source_type == "html": + links = probe_result.get("candidate_links") or [] + return [_http_file_dict(links[0] if links else final_url, slug)] + return [_http_file_dict(final_url, slug)] + + +def _ckan_sources(probe_result: dict[str, Any]) -> list[dict[str, Any]]: + import re + from pathlib import Path as _P + from urllib.parse import urlparse as _up + + resources = probe_result.get("ckan_resources") or [] + parsed = _up(probe_result.get("final_url", "")) + portal = f"{parsed.scheme}://{parsed.netloc}" + out = [] + for r in resources: + name = re.sub(r"[^a-z0-9_]", "_", (r.get("name") or "resource").lower()) + url = r.get("url", "") + fmt = r.get("format", "csv") + fname = _P(_up(url).path).name or f"{name}.{fmt}" + out.append( + { + "name": name, + "type": "ckan", + "args": {"portal_url": portal, "resource_id": r.get("id") or "", "filename": fname}, + "primary": True, + } + ) + return out + + +def _sdmx_sources(probe_result: dict[str, Any], url: str) -> list[dict[str, Any]]: + info = probe_result.get("sdmx_info") or {} + flow = info.get("flow_id") + if not flow: + return [_http_file_dict(url, "sdmx")] + s: dict[str, Any] = { + "name": f"sdmx_{flow}", + "type": "sdmx", + "args": {"flow": flow}, + "primary": True, + } + agency = info.get("agency") + if agency and str(agency).upper() == "ESTAT": + s["args"]["agency"] = "ESTAT" + else: + s["args"]["endpoint"] = url + return [s] + + +def _http_file_dict(url: str, slug: str) -> dict[str, Any]: + from pathlib import Path as _P + from urllib.parse import urlparse as _up + + fname = _P(_up(url).path).name or f"{slug}.csv" + return { + "name": f"{slug}_source", + "type": "http_file", + "args": {"url": url, "filename": fname}, + "primary": True, } - return result + +def _find_anno_col(profile: dict[str, Any]) -> str | None: + """Find a year/anno column name in the profile.""" + from toolkit.scaffold.clean import _find_anno_raw_column + + return _find_anno_raw_column(profile) + + +def _enrich_profile(profile: dict[str, Any], probe: dict[str, Any]) -> dict[str, Any]: + enriched = dict(profile) + for k in ( + "encoding_suggested", + "delim_suggested", + "decimal_suggested", + "skip_suggested", + "header_line", + "true_header_line", + "robust_read_suggested", + ): + if probe.get(k) is not None: + enriched[k] = probe[k] + return enriched + + +def suggest_validation(profile: dict[str, Any] | None) -> dict[str, Any]: + """Suggest validation rules from profiling results.""" + if not profile: + return {} + rules: dict[str, Any] = {} + cols = profile.get("columns_norm") or profile.get("columns_raw") or [] + rc = profile.get("row_count", 0) + if cols: + from toolkit.scaffold.clean import _snake_case + + rules["required_columns"] = [_snake_case(c) for c in cols] + if rc: + rules["min_rows"] = max(1, int(rc * 0.5)) + rules["mart_min_rows"] = 1 + return rules + + +def suggest_mart_sql( + columns: list[dict[str, Any]] | list[str], + profile: dict[str, Any], +) -> str: + """Generate mart.sql skeleton with column type hints.""" + if columns and isinstance(columns[0], dict): + names = [c.get("name", f"col{i}") for i, c in enumerate(columns)] + else: + names = list(columns) if columns else [] + if not names: + return ( + "-- mart placeholder — sostituisci con la tua aggregazione.\n" + "SELECT * FROM clean_input\n" + ) + mapping = profile.get("mapping_suggestions") or {} + hints = [f"{n}: {(mapping.get(n) or {}).get('type', '?')}" for n in names] + return ( + f"-- Colonne clean_input ({len(names)}): {', '.join(hints[:8])}" + f"{' ...' if len(hints) > 8 else ''}\n" + "-- Sostituisci con la tua aggregazione (es. SUM, COUNT, AVG).\n" + "SELECT * FROM clean_input\n" + ) + + +def _generate_readme(slug: str, url: str) -> str: + return ( + f"# {slug}\n\nFonte: {url}\n\n" + "## Domanda\n\n-\n\n## Dataset\n\n-\n\n" + "## Perche vale la pena testarlo\n\n-\n\n" + "## Output minimo atteso\n\n-\n\n" + "## Criterio di promozione\n\n-\n\n" + "## Stato\n\n- intake\n\n" + "## Prossimo passo\n\n- scout URL poi run all\n" + ) + + +def _generate_notes(granularity: str | None, topics: list[dict[str, Any]] | None) -> str: + lines: list[str] = ["## Tecnico\n\n-\n"] + if granularity: + lines.append(f"- Granularita rilevata: {granularity}\n") + if topics: + names = ", ".join(t["topic"] for t in topics[:3]) + lines.append(f"- Topic suggeriti: {names}\n") + lines.append("\n## Analitico\n\n-\n\n## Cautele\n\n") + lines.append("- La serie storica e omogenea su tutti gli anni?\n") + lines.append("- Ci sono discontinuita dichiarate dalla fonte?\n") + lines.append("- I valori nulli sono zero reale o dato mancante?\n") + return "".join(lines)