mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-10-14 13:30:55 +02:00
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
"""Shared deterministic corpus walks and append-safe JSONL checkpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Collection
|
|
from pathlib import Path
|
|
|
|
|
|
def corpus_day(path: Path, corpus: Path) -> datetime.date | None:
|
|
"""Read the leading YYYY-MM-DD corpus segment, if present."""
|
|
try:
|
|
return datetime.date.fromisoformat(path.relative_to(corpus).parts[0])
|
|
except (ValueError, IndexError):
|
|
return None
|
|
|
|
|
|
def corpus_files(corpus: Path, since: datetime.date | None, *, suffixes: Collection[str] | None = None) -> list[Path]:
|
|
"""Return a deterministic recursive walk, optionally date- and suffix-bounded."""
|
|
if since is None:
|
|
candidates = corpus.rglob("*")
|
|
else:
|
|
roots = [
|
|
path
|
|
for path in corpus.iterdir()
|
|
if path.is_dir() and (day := corpus_day(path, corpus)) is not None and day >= since
|
|
]
|
|
candidates = (path for root in roots for path in root.rglob("*"))
|
|
return sorted(
|
|
path for path in candidates if path.is_file() and (suffixes is None or path.suffix.lower() in suffixes)
|
|
)
|
|
|
|
|
|
def repair_jsonl_tail(path: Path) -> None:
|
|
"""Truncate an interrupted final JSONL fragment before another append."""
|
|
if not path.exists():
|
|
return
|
|
with path.open("rb+") as stream:
|
|
data = stream.read()
|
|
if data and not data.endswith(b"\n"):
|
|
stream.truncate(data.rfind(b"\n") + 1)
|