| |
| """Remove workstation identity and absolute prefixes from text metadata.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import os |
| import sys |
| import tempfile |
| from pathlib import Path |
|
|
|
|
| PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| if str(PROJECT_ROOT) not in sys.path: |
| sys.path.insert(0, str(PROJECT_ROOT)) |
|
|
| from accesspath3r.privacy import sanitize_text |
|
|
|
|
| TEXT_SUFFIXES = {".err", ".json", ".jsonl", ".log", ".md", ".out", ".txt"} |
|
|
|
|
| def sanitize_file( |
| path: Path, |
| *, |
| project_root: Path, |
| output_root: Path, |
| sensitive_paths: list[str], |
| ) -> bool: |
| try: |
| original = path.read_text(encoding="utf-8") |
| except (OSError, UnicodeDecodeError): |
| return False |
| sanitized = sanitize_text( |
| original, |
| project_root=project_root, |
| output_root=output_root, |
| sensitive_paths=sensitive_paths, |
| ) |
| if sanitized == original: |
| return False |
|
|
| original_mode = path.stat().st_mode & 0o7777 |
| file_descriptor, temporary_name = tempfile.mkstemp( |
| prefix=f".{path.name}.", |
| suffix=".tmp", |
| dir=path.parent, |
| ) |
| try: |
| os.fchmod(file_descriptor, original_mode) |
| with os.fdopen(file_descriptor, "w", encoding="utf-8") as handle: |
| handle.write(sanitized) |
| handle.flush() |
| os.fsync(handle.fileno()) |
| os.replace(temporary_name, path) |
| except BaseException: |
| try: |
| os.unlink(temporary_name) |
| except OSError: |
| pass |
| raise |
| return True |
|
|
|
|
| def build_parser() -> argparse.ArgumentParser: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--root", type=Path, required=True) |
| parser.add_argument("--project-root", type=Path, required=True) |
| parser.add_argument("--sensitive-path", action="append", default=[]) |
| return parser |
|
|
|
|
| def main() -> int: |
| args = build_parser().parse_args() |
| if args.root.is_file(): |
| paths = [args.root] |
| output_root = args.root.parent |
| elif args.root.is_dir(): |
| paths = args.root.rglob("*") |
| output_root = args.root |
| else: |
| return 0 |
| for path in paths: |
| if path.is_file() and path.suffix.lower() in TEXT_SUFFIXES: |
| sanitize_file( |
| path, |
| project_root=args.project_root, |
| output_root=output_root, |
| sensitive_paths=args.sensitive_path, |
| ) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
