mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-10-19 15:37:16 +02:00
Victor KuznetsovandClaude Opus 5.5
cf695de57f
The upstream registry gained be.immostory.videoseal.1 at revision d5960248, so `sync_c2pa_soft_bindings.py --check` failed lint on main and on every pull request. That step runs ahead of Ruff, which hid an ICN001 error in scripts/_corpus_scan.py; import `date` directly, as the sibling scripts do. Document the check's live-upstream behavior next to the refresh command. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
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
|
|
|
|
from datetime import date
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Collection
|
|
from pathlib import Path
|
|
|
|
|
|
def corpus_day(path: Path, corpus: Path) -> date | None:
|
|
"""Read the leading YYYY-MM-DD corpus segment, if present."""
|
|
try:
|
|
return date.fromisoformat(path.relative_to(corpus).parts[0])
|
|
except (ValueError, IndexError):
|
|
return None
|
|
|
|
|
|
def corpus_files(corpus: Path, since: 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)
|