Audit dei campi vuoti CSV per la ricerca

import argparse import csv import hashlib import json import re from collections import Counter from pathlib import Path from urllib.parse import urlparse MODE = "campi-vuoti-csv" CONTEXT = "la ricerca" REQUIRED_FIELDS = ["id","fonte","data"] parser = argparse.ArgumentParser(description="Audit dei campi vuoti CSV: segnalare righe con campi obbligatori vuoti.") parser.add_argument("percorso", type=Path, help="File o cartella da analizzare in sola lettura") args = parser.parse_args() path = args.percorso def print_csv_report(file_path): with file_path.open(newline="", encoding="utf-8-sig", errors="replace") as handle: rows = list(csv.DictReader(handle)) headers = list(rows[0].keys()) if rows else [] print(f"Contesto: {CONTEXT}; righe: {len(rows)}; colonne: {len(headers)}") if MODE == "duplicati-csv": key = REQUIRED_FIELDS[0] counts = Counter((row.get(key) or "").strip() for row in rows) for value, count in counts.most_common(): if value and count > 1: print(f"duplicato {key}={value!r}: {count}") elif MODE == "campi-vuoti-csv": for line, row in enumerate(rows, 2): missing = [field for field in REQUIRED_FIELDS if not (row.get(field) or "").strip()] if missing: print(f"riga {line}: campi vuoti {', '.join(missing)}") elif MODE == "righe-incoerenti-csv": for line, row in enumerate(rows, 2): if None in row: print(f"riga {line}: celle oltre le intestazioni") elif MODE == "intestazioni-csv": for header in headers: preview = re.sub(r"[^a-z0-9]+", "_", header.lower()).strip("_") print(f"{header!r} -> {preview!r}") elif MODE == "valori-numerici-csv": for header in headers: values = [] for row in rows: try: values.append(float((row.get(header) or "").replace(",", "."))) except ValueError: pass if values: print(f"{header}: n={len(values)} min={min(values)} max={max(values)} media={sum(values)/len(values):.2f}") elif MODE == "date-csv": pattern = re.compile(r"^\d{4}-\d{2}-\d{2}$") for header in headers: values = [(row.get(header) or "").strip() for row in rows] valid = sum(bool(pattern.match(value)) for value in values if value) if valid: print(f"{header}: date ISO valide={valid}; altri valori={len([v for v in values if v])-valid}") elif MODE == "categorie-csv": for header in headers: counts = Counter((row.get(header) or "").strip() for row in rows) if 1 < len(counts) <= 30: print(f"{header}: {counts.most_common(10)}") else: for header in headers: lengths = [len((row.get(header) or "").strip()) for row in rows] if lengths: print(f"{header}: min={min(lengths)} max={max(lengths)} media={sum(lengths)/len(lengths):.1f}") def print_json_report(file_path): data = json.loads(file_path.read_text(encoding="utf-8")) items = data if isinstance(data, list) else [data] print(f"Contesto: {CONTEXT}; elementi: {len(items)}") if MODE == "campi-json": for index, item in enumerate(items): missing = [field for field in REQUIRED_FIELDS if not isinstance(item, dict) or item.get(field) in (None, "")] if missing: print(f"elemento {index}: mancanti {', '.join(missing)}") elif MODE == "id-json": values = [item.get(REQUIRED_FIELDS[0]) for item in items if isinstance(item, dict)] for value, count in Counter(values).items(): if value in (None, "") or count > 1: print(f"id={value!r}: occorrenze={count}") else: types = {} for item in items: if isinstance(item, dict): for key, value in item.items(): types.setdefault(key, Counter())[type(value).__name__] += 1 for key in sorted(types): print(f"{key}: {dict(types[key])}") def print_text_report(file_path): text = file_path.read_text(encoding="utf-8", errors="replace") if MODE == "struttura-markdown": for line_number, line in enumerate(text.splitlines(), 1): if re.match(r"^#{1,6}\s", line): print(f"{line_number}: {line}") elif MODE == "link-markdown": for target in re.findall(r"(?<!!)\[[^]]*\]\(([^)]+)\)", text): print(f"link: {target}") elif MODE == "errori-log": levels = Counter(match.upper() for match in re.findall(r"\b(DEBUG|INFO|WARN|ERROR|CRITICAL)\b", text, re.I)) print(dict(levels)) else: urls = [line.strip() for line in text.splitlines() if line.strip()] for value, count in Counter(urls).items(): parsed = urlparse(value) status = "valido" if parsed.scheme in {"http", "https"} and parsed.netloc else "da verificare" print(f"{value}: {status}; occorrenze={count}") if MODE == "inventario-cartella": files = [item for item in path.rglob("*") if item.is_file()] by_suffix = Counter(item.suffix.lower() or "[senza estensione]" for item in files) print(f"Contesto: {CONTEXT}; file: {len(files)}; byte: {sum(item.stat().st_size for item in files)}") for suffix, count in by_suffix.most_common(): print(f"{suffix}: {count}") elif MODE == "hash-file": files = [path] if path.is_file() else [item for item in path.rglob("*") if item.is_file()] for item in sorted(files): digest = hashlib.sha256(item.read_bytes()).hexdigest() print(f"{digest} {item}") elif MODE.endswith("-json") or MODE in {"campi-json", "id-json", "tipi-json"}: print_json_report(path) elif MODE in {"struttura-markdown", "link-markdown", "errori-log", "url-testo"}: print_text_report(path) else: print_csv_report(path) print("Analisi completata in sola lettura: verificare i risultati sulla fonte prima di correggere i dati.")