diff --git a/README.md b/README.md index 706bb07..f7dff03 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,25 @@ uv run python scripts/update_daily_asmr_data.py --dry-run --diff-limit 100 新規作品がなくてもフィルタと `web/data` を作り直す場合は `--force` を付けます。 +## pull後のベクトル差分更新 + +Actions が更新した `asmr_works.jsonl` / `asmr_works.filtered.jsonl` を pull した後、ローカルの `web/data` だけを差分更新します。 + +```bash +git pull --rebase --autostash +uv run --extra remote python scripts/update_pulled_vector_data.py +``` + +このスクリプトは `asmr_works.filtered.jsonl` と `web/data/works.json` を比較し、既存作品のembeddingは `web/data/embeddings.f32` からコピーして再利用します。新規作品、またはタイトル・タグ・声優が変わった作品だけembedding APIへ送ります。 + +ファイルを書かずに確認する場合: + +```bash +uv run --extra remote python scripts/update_pulled_vector_data.py --dry-run +``` + +APIキーが必要な場合は `EMBEDDING_API_KEY` に設定します。接続先やモデルは `web/data/manifest.json` の `remote-openai-compatible` 設定を使います。 + ## 翻訳重複の修復 日本語原作と翻訳版が両方 `asmr_works.filtered.jsonl` / `web/data` に入ってしまった場合は、自動検出して翻訳版だけ除外できます。 diff --git a/scripts/update_pulled_vector_data.py b/scripts/update_pulled_vector_data.py new file mode 100644 index 0000000..5e3ad1e --- /dev/null +++ b/scripts/update_pulled_vector_data.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +"""Update web/data after pulling JSONL data generated by actions.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path +from typing import Any, Iterable + + +ROOT_DIR = Path(__file__).resolve().parents[1] +if str(ROOT_DIR) not in sys.path: + sys.path.insert(0, str(ROOT_DIR)) + +from build_vector_data import normalize_work # noqa: E402 +from update_daily_asmr_data import ( # noqa: E402 + build_incremental_web_data, + dry_run_web_stats, + format_work, + identity_keys, + read_json, + read_jsonl, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Incrementally update web/data from the pulled filtered JSONL dataset." + ) + parser.add_argument( + "--filtered-input", + default="asmr_works.filtered.jsonl", + help="Pulled filtered ASMR JSONL file.", + ) + parser.add_argument("--data-dir", default="web/data", help="Directory containing web data assets.") + parser.add_argument("--dry-run", action="store_true", help="Show diffs without writing web/data.") + parser.add_argument("--force", action="store_true", help="Rewrite web/data even when no diff is detected.") + parser.add_argument( + "--diff-limit", + type=int, + default=50, + help="Maximum rows to print per diff section. Use 0 for all rows.", + ) + parser.add_argument( + "--api-key-env", + default="EMBEDDING_API_KEY", + help="Environment variable that contains the remote embedding API key.", + ) + parser.add_argument("--embedding-timeout", type=float, default=120.0, help="Embedding request timeout seconds.") + parser.add_argument("--embedding-retries", type=int, default=3, help="Embedding retries per failed batch.") + parser.add_argument("--embedding-retry-wait", type=float, default=2.0, help="Initial embedding retry wait seconds.") + return parser.parse_args() + + +def validate_args(args: argparse.Namespace) -> None: + if args.diff_limit < 0: + raise SystemExit("--diff-limit must be greater than or equal to 0") + if args.embedding_timeout <= 0: + raise SystemExit("--embedding-timeout must be greater than 0") + if args.embedding_retries < 0: + raise SystemExit("--embedding-retries must be greater than or equal to 0") + if args.embedding_retry_wait < 0: + raise SystemExit("--embedding-retry-wait must be greater than or equal to 0") + + +def normalized_filtered_works(filtered_works: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + works: list[dict[str, Any]] = [] + for raw_work in filtered_works: + works.append(normalize_work(raw_work, len(works))) + return works + + +def key_map(works: Iterable[dict[str, Any]]) -> dict[str, dict[str, Any]]: + result: dict[str, dict[str, Any]] = {} + for work in works: + for key in identity_keys(work): + result.setdefault(key, work) + return result + + +def has_matching_key(work: dict[str, Any], works_by_key: dict[str, dict[str, Any]]) -> bool: + return any(key in works_by_key for key in identity_keys(work)) + + +def matching_work(work: dict[str, Any], works_by_key: dict[str, dict[str, Any]]) -> dict[str, Any] | None: + for key in identity_keys(work): + match = works_by_key.get(key) + if match is not None: + return match + return None + + +def changed_fields(old: dict[str, Any], new: dict[str, Any]) -> list[str]: + fields = sorted((old.keys() | new.keys()) - {"embeddingIndex"}) + return [field for field in fields if old.get(field) != new.get(field)] + + +def data_diffs( + *, + old_works: list[dict[str, Any]], + new_works: list[dict[str, Any]], +) -> dict[str, list[Any]]: + old_by_key = key_map(old_works) + new_by_key = key_map(new_works) + added: list[dict[str, Any]] = [] + removed: list[dict[str, Any]] = [] + changed: list[dict[str, Any]] = [] + reordered: list[dict[str, Any]] = [] + + for work in new_works: + old_work = matching_work(work, old_by_key) + if old_work is None: + added.append(work) + elif old_work != work: + reasons = changed_fields(old_work, work) + if reasons: + changed.append({"work": work, "reasons": reasons}) + else: + reordered.append(work) + + for work in old_works: + if not has_matching_key(work, new_by_key): + removed.append(work) + + return {"added": added, "removed": removed, "changed": changed, "reordered": reordered} + + +def limited_rows(rows: list[Any], limit: int) -> list[Any]: + return rows if limit == 0 else rows[:limit] + + +def remaining_count(rows: list[Any], limit: int) -> int: + return 0 if limit == 0 else max(0, len(rows) - limit) + + +def print_work_section(title: str, prefix: str, works: list[dict[str, Any]], limit: int) -> None: + print(f"\n{title}: {len(works)}") + for work in limited_rows(works, limit): + print(f" {prefix} {format_work(work)}") + remaining = remaining_count(works, limit) + if remaining: + print(f" ... {remaining} more") + + +def print_changed_section(changes: list[dict[str, Any]], limit: int) -> None: + print(f"\nChanged web works: {len(changes)}") + for change in limited_rows(changes, limit): + print(f" ~ {format_work(change['work'])} / {', '.join(change['reasons'])}") + remaining = remaining_count(changes, limit) + if remaining: + print(f" ... {remaining} more") + + +def print_embedding_section(diffs: list[dict[str, Any]], limit: int) -> None: + print(f"\nEmbedding requests: {len(diffs)}") + for diff in limited_rows(diffs, limit): + marker = "+" if diff["reason"] == "new work" else "~" + print(f" {marker} {format_work(diff['work'])} / {diff['reason']}") + remaining = remaining_count(diffs, limit) + if remaining: + print(f" ... {remaining} more") + + +def embeddings_file_in_sync(data_dir: Path, manifest: dict[str, Any], old_works: list[dict[str, Any]]) -> bool: + embeddings_path = data_dir / str(manifest.get("embeddingFile", "embeddings.f32")) + if not embeddings_path.exists(): + return False + try: + dimensions = int(manifest["dimensions"]) + except (KeyError, TypeError, ValueError): + return False + return embeddings_path.stat().st_size == len(old_works) * dimensions * 4 + + +def needs_update( + *, + old_works: list[dict[str, Any]], + new_works: list[dict[str, Any]], + manifest: dict[str, Any], + data_dir: Path, +) -> bool: + if old_works != new_works: + return True + if int(manifest.get("count", -1)) != len(new_works): + return True + return not embeddings_file_in_sync(data_dir, manifest, old_works) + + +def main() -> int: + args = parse_args() + validate_args(args) + filtered_path = Path(args.filtered_input) + data_dir = Path(args.data_dir) + manifest_path = data_dir / "manifest.json" + works_path = data_dir / "works.json" + + if not filtered_path.exists(): + raise SystemExit(f"filtered input file not found: {filtered_path}") + if not manifest_path.exists(): + raise SystemExit(f"manifest file not found: {manifest_path}") + if not works_path.exists(): + raise SystemExit(f"works file not found: {works_path}") + + filtered_works = read_jsonl(filtered_path) + new_works = normalized_filtered_works(filtered_works) + old_works = read_json(works_path) + if not isinstance(old_works, list): + raise SystemExit(f"{works_path} must contain a JSON array") + manifest = read_json(manifest_path) + if not isinstance(manifest, dict): + raise SystemExit(f"{manifest_path} must contain a JSON object") + + diffs = data_diffs(old_works=old_works, new_works=new_works) + web_stats = dry_run_web_stats(args, filtered_works) + should_update = needs_update( + old_works=old_works, + new_works=new_works, + manifest=manifest, + data_dir=data_dir, + ) + + print(f"Filtered works: {len(new_works)}") + print(f"Existing web/data works: {len(old_works)}") + print(f"web/data would reuse {web_stats['reused']} embeddings and request {web_stats['embedded']} embeddings") + + if args.dry_run: + print_work_section("Web data additions", "+", diffs["added"], args.diff_limit) + print_work_section("Web data removals", "-", diffs["removed"], args.diff_limit) + print_changed_section(diffs["changed"], args.diff_limit) + print(f"\nReordered existing works: {len(diffs['reordered'])}") + print_embedding_section(web_stats.get("embeddingDiffs", []), args.diff_limit) + print("\nDry run: no files were written.") + return 0 + + if not should_update and not args.force: + print("web/data is already in sync. update skipped.") + return 0 + + try: + stats = build_incremental_web_data(args, filtered_works) + except RuntimeError as exc: + raise SystemExit(f"web/data update failed: {exc}") from exc + + print( + f"Updated {data_dir}: {stats['count']} works, " + f"reused {stats['reused']} embeddings, requested {stats['embedded']} embeddings" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())