"""Probe: file_scanner.metadata.json_schema — detect JSON/JSONL structure and key schemas."""
import json
from pathlib import Path
from core.base_probe import BaseProbe
from core.registry import register
[docs]
@register
class JsonSchemaProbe(BaseProbe):
[docs]
@staticmethod
def probe_name() -> str:
return "file_scanner.metadata.json_schema"
[docs]
@staticmethod
def description() -> str:
return "Detect JSON/JSONL schemas: root type, keys, nested structure, record 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 (".json", ".jsonl"):
continue
if f.stat().st_size == 0 or f.stat().st_size > 100_000_000:
continue
try:
result = self._analyze_json(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_json(self, filepath: Path, root: Path) -> dict | None:
content = filepath.read_text(encoding="utf-8", errors="replace")[:500_000]
# Try JSONL first (newline-delimited)
if filepath.suffix.lower() == ".jsonl" or "\n{" in content[:1000]:
return self._analyze_jsonl(filepath, root, content)
# Standard JSON
try:
data = json.loads(content)
except json.JSONDecodeError:
return {"path": str(filepath.relative_to(root)), "error": "Invalid JSON"}
if isinstance(data, list):
schema = _extract_schema(data[0]) if data else {}
return {
"path": str(filepath.relative_to(root)),
"format": "JSON_ARRAY",
"record_count": len(data),
"schema": schema,
"column_count": len(schema),
}
elif isinstance(data, dict):
schema = _extract_schema(data)
return {
"path": str(filepath.relative_to(root)),
"format": "JSON_OBJECT",
"record_count": 1,
"schema": schema,
"column_count": len(schema),
}
return None
def _analyze_jsonl(self, filepath: Path, root: Path, content: str) -> dict:
lines = [l for l in content.split("\n") if l.strip()]
records = 0
schema: dict = {}
for line in lines[:100]:
try:
obj = json.loads(line)
if isinstance(obj, dict):
records += 1
if not schema:
schema = _extract_schema(obj)
except json.JSONDecodeError:
pass
return {
"path": str(filepath.relative_to(root)),
"format": "JSONL",
"record_count_sample": records,
"total_lines": len(lines),
"schema": schema,
"column_count": len(schema),
}
def _extract_schema(obj: dict, prefix: str = "") -> dict:
"""Extract a flat schema from a JSON object (key → type)."""
schema: dict = {}
for key, val in obj.items():
full_key = f"{prefix}.{key}" if prefix else key
if isinstance(val, dict):
schema[full_key] = "OBJECT"
schema.update(_extract_schema(val, full_key))
elif isinstance(val, list):
schema[full_key] = f"ARRAY[{type(val[0]).__name__}]" if val else "ARRAY"
elif isinstance(val, bool):
schema[full_key] = "BOOLEAN"
elif isinstance(val, int):
schema[full_key] = "INTEGER"
elif isinstance(val, float):
schema[full_key] = "DECIMAL"
elif val is None:
schema[full_key] = "NULL"
else:
schema[full_key] = "STRING"
return schema