"""Probe: file_scanner.metadata.csv_schema — detect CSV/TSV column schemas by header + sampling."""
import csv
import io
from pathlib import Path
from core.base_probe import BaseProbe
from core.registry import register
[docs]
@register
class CsvSchemaProbe(BaseProbe):
[docs]
@staticmethod
def probe_name() -> str:
return "file_scanner.metadata.csv_schema"
[docs]
@staticmethod
def description() -> str:
return "Detect CSV/TSV schemas: columns, types (inferred from sample), delimiter, row count"
[docs]
def execute(self, connection, schema: str, **kwargs) -> list[dict]:
root = connection if isinstance(connection, Path) else Path(str(connection))
results = []
for f in sorted(root.rglob("*")):
if not f.is_file() or f.suffix.lower() not in (".csv", ".tsv", ".txt"):
continue
if f.stat().st_size == 0 or f.stat().st_size > 100_000_000: # Skip empty or >100MB
continue
try:
result = self._analyze_csv(f, root)
if result:
results.append(result)
except Exception as e:
results.append({"path": str(f.relative_to(root)), "error": str(e)})
return results
def _analyze_csv(self, filepath: Path, root: Path) -> dict | None:
# Read first 50 lines for analysis
with open(filepath, "r", encoding="utf-8", errors="replace") as fh:
sample = fh.read(100_000)
# Detect delimiter
sniffer = csv.Sniffer()
try:
dialect = sniffer.sniff(sample[:5000])
delimiter = dialect.delimiter
except csv.Error:
delimiter = "," if ".csv" in filepath.suffix.lower() else "\t"
reader = csv.reader(io.StringIO(sample), delimiter=delimiter)
rows = []
for i, row in enumerate(reader):
rows.append(row)
if i >= 50:
break
if len(rows) < 2:
return None
headers = rows[0]
data_rows = rows[1:]
# Infer column types from sample
columns = []
for col_idx, header in enumerate(headers):
values = [r[col_idx] for r in data_rows if col_idx < len(r) and r[col_idx].strip()]
col_type = _infer_type(values[:20])
null_count = sum(1 for r in data_rows if col_idx >= len(r) or not r[col_idx].strip())
columns.append({
"name": header.strip(),
"position": col_idx,
"inferred_type": col_type,
"null_count_sample": null_count,
"sample_values": [v[:50] for v in values[:3]],
})
total_lines = sample.count("\n")
return {
"path": str(filepath.relative_to(root)),
"delimiter": repr(delimiter),
"column_count": len(headers),
"row_count_estimate": total_lines - 1,
"columns": columns,
}
def _infer_type(values: list[str]) -> str:
"""Infer column type from sample values."""
if not values:
return "UNKNOWN"
int_count = 0
float_count = 0
date_count = 0
for v in values:
v = v.strip()
try:
int(v)
int_count += 1
continue
except ValueError:
pass
try:
float(v.replace(",", ""))
float_count += 1
continue
except ValueError:
pass
if len(v) == 10 and v[4] == "-" and v[7] == "-":
date_count += 1
n = len(values)
if int_count > n * 0.8:
return "INTEGER"
if (int_count + float_count) > n * 0.8:
return "DECIMAL"
if date_count > n * 0.5:
return "DATE"
return "VARCHAR"