File size: 1,949 Bytes
f619a9b edc8d43 f619a9b edc8d43 f619a9b e4f211a f619a9b e4f211a f619a9b edc8d43 f619a9b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """Validate privacy-safe trace JSONL files."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
TRACE_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from traces.runtime import validate_trace
def validate_file(path: Path) -> tuple[int, list[str]]:
count = 0
errors: list[str] = []
trace_ids: set[str] = set()
for line_number, line in enumerate(
path.read_text(encoding="utf-8").splitlines(),
start=1,
):
if not line.strip():
continue
count += 1
try:
record = json.loads(line)
except json.JSONDecodeError as exc:
errors.append(f"{path}:{line_number}: invalid JSON: {exc}")
continue
trace_id = record.get("trace_id") if isinstance(record, dict) else None
if trace_id in trace_ids:
errors.append(f"{path}:{line_number}: Duplicate trace ID: {trace_id}")
elif isinstance(trace_id, str):
trace_ids.add(trace_id)
for error in validate_trace(record):
errors.append(f"{path}:{line_number}: {error}")
return count, errors
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("paths", nargs="*", type=Path)
args = parser.parse_args()
paths = args.paths or [TRACE_DIR / "data" / "trace_samples.jsonl"]
total = 0
all_errors: list[str] = []
for path in paths:
if not path.exists():
all_errors.append(f"{path}: file does not exist")
continue
count, errors = validate_file(path)
total += count
all_errors.extend(errors)
if all_errors:
print("\n".join(all_errors), file=sys.stderr)
return 1
print(f"Validated {total} trace records.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|