"""Probe: file_scanner.metadata.auto_dispatch — recursive scanner that detects and dispatches.
The meta-probe: crawls a directory, classifies files, detects partition patterns,
and dispatches to the appropriate sub-probe (Excel, Power BI, Tableau, CSV, JSON).
Also detects:
- yyyy-mm, yyyy/mm/dd, date=YYYYMMDD partition patterns
- Hive-style partitions (key=value directories)
- Glob patterns in directory structures
"""
from __future__ import annotations
import re
from collections import defaultdict
from pathlib import Path
from core.base_probe import BaseProbe
from core.registry import register
# File extension → probe source type
DISPATCH_MAP = {
".xlsx": "excel", ".xlsm": "excel", ".xls": "excel",
".pbit": "powerbi", ".pbix": "powerbi",
".twb": "tableau", ".twbx": "tableau",
".csv": "csv", ".tsv": "csv", ".txt": "csv",
".json": "json", ".jsonl": "json",
".parquet": "parquet", ".pq": "parquet",
".avro": "avro",
".sql": "sql", ".ddl": "sql", ".pls": "sql",
".xml": "xml",
".yaml": "yaml", ".yml": "yaml",
}
# Partition patterns
PARTITION_PATTERNS = [
(r"\b(\d{4})-(\d{2})\b", "yyyy-mm"),
(r"\b(\d{4})/(\d{2})/(\d{2})\b", "yyyy/mm/dd"),
(r"\b(\d{4})(\d{2})(\d{2})\b", "yyyymmdd"),
(r"\bdate=(\d{4}-\d{2}-\d{2})\b", "hive_date"),
(r"\byear=(\d{4})\b", "hive_year"),
(r"\bmonth=(\d{1,2})\b", "hive_month"),
(r"\bregion=(\w+)\b", "hive_region"),
(r"\bcountry=(\w+)\b", "hive_country"),
(r"\bpartition[_-]?(\d+)\b", "numeric_partition"),
]
[docs]
@register
class AutoDispatchProbe(BaseProbe):
[docs]
@staticmethod
def probe_name() -> str:
return "file_scanner.metadata.auto_dispatch"
[docs]
@staticmethod
def description() -> str:
return "Recursive scanner: discovers files, detects partitions, dispatches to sub-probes"
[docs]
def execute(self, connection, schema: str, **kwargs) -> list[dict]:
root = connection if isinstance(connection, Path) else Path(str(connection))
results = []
# Phase 1: Inventory all files
file_groups: dict[str, list[Path]] = defaultdict(list) # type → [paths]
partition_groups: dict[str, dict] = {} # pattern → {values, file_count, sample}
for f in sorted(root.rglob("*")):
if not f.is_file() or f.name.startswith(".") or f.name.startswith("~$"):
continue
ext = f.suffix.lower()
if ext not in DISPATCH_MAP:
continue
probe_type = DISPATCH_MAP[ext]
file_groups[probe_type].append(f)
# Detect partition patterns in path
rel_path = str(f.relative_to(root))
for pattern, pattern_name in PARTITION_PATTERNS:
match = re.search(pattern, rel_path)
if match:
key = f"{pattern_name}:{ext}"
if key not in partition_groups:
partition_groups[key] = {
"pattern_type": pattern_name,
"file_extension": ext,
"values": set(),
"file_count": 0,
"sample_paths": [],
"glob_pattern": "",
}
partition_groups[key]["values"].add(match.group(0))
partition_groups[key]["file_count"] += 1
if len(partition_groups[key]["sample_paths"]) < 3:
partition_groups[key]["sample_paths"].append(rel_path)
# Phase 2: Build dispatch plan
dispatch_plan = []
for probe_type, files in sorted(file_groups.items()):
dispatch_plan.append({
"probe_type": probe_type,
"file_count": len(files),
"total_size_bytes": sum(f.stat().st_size for f in files),
"extensions": sorted({f.suffix.lower() for f in files}),
"sample_files": [str(f.relative_to(root)) for f in files[:5]],
"recommended_probe": _recommend_probe(probe_type),
})
results.append({
"type": "dispatch_plan",
"root": str(root),
"total_files": sum(len(fs) for fs in file_groups.values()),
"file_types": {k: len(v) for k, v in file_groups.items()},
"dispatch": dispatch_plan,
})
# Phase 3: Report partition patterns
for key, info in partition_groups.items():
# Generate glob pattern
glob = _generate_glob(info["sample_paths"], info["pattern_type"])
results.append({
"type": "partition_pattern",
"pattern_type": info["pattern_type"],
"file_extension": info["file_extension"],
"distinct_values": len(info["values"]),
"file_count": info["file_count"],
"sample_values": sorted(info["values"])[:10],
"sample_paths": info["sample_paths"],
"suggested_glob": glob,
})
return results
def _recommend_probe(probe_type: str) -> str:
return {
"excel": "excel.*",
"powerbi": "powerbi.*",
"tableau": "tableau.*",
"csv": "file_scanner.metadata.csv_schema",
"json": "file_scanner.metadata.json_schema",
"parquet": "file_scanner.metadata.parquet_schema",
"sql": "oracle.metadata.source_code (or manual upload to Metastore)",
"xml": "file_scanner.metadata.inventory",
"yaml": "file_scanner.metadata.inventory",
"avro": "file_scanner.metadata.inventory",
}.get(probe_type, "file_scanner.metadata.inventory")
def _generate_glob(sample_paths: list[str], pattern_type: str) -> str:
"""Generate a glob pattern from sample paths."""
if not sample_paths:
return "*"
path = sample_paths[0]
# Replace partition values with wildcards
if pattern_type == "yyyy-mm":
return re.sub(r"\d{4}-\d{2}", "*", path)
if pattern_type == "yyyy/mm/dd":
return re.sub(r"\d{4}/\d{2}/\d{2}", "*/*/*", path)
if pattern_type == "yyyymmdd":
return re.sub(r"\d{8}", "*", path)
if pattern_type.startswith("hive_"):
return re.sub(r"(\w+)=\w+", r"\1=*", path)
return re.sub(r"\d+", "*", path)