Upload 5 files

app/bucket_snapshot.py

@@ -52,6 +52,16 @@ def snapshot_status() -> dict[str, Any] | None:
     except (OSError, json.JSONDecodeError):
         return None
 
+def publish_due(bucket_dir: Path, minimum_interval_seconds: int) -> bool:
+    if minimum_interval_seconds <= 0:
+        return True
+    try:
+        manifest = json.loads(manifest_path(bucket_dir).read_text(encoding="utf-8"))
+        created_at = int(manifest.get("created_at", 0))
+    except (OSError, ValueError, TypeError, json.JSONDecodeError):
+        return True
+    return time.time() - created_at >= minimum_interval_seconds
+
 
 def local_repository() -> Path:
     return DATA_ROOT / "es-repositories" / "production"

app/data_loader.py

@@ -1,5 +1,6 @@
 import hashlib
 import json
+import os
 import re
 import shutil
 import subprocess
@@ -12,6 +13,7 @@ from .config import ARCHIVE_END, ARCHIVE_START, PARSED_ROOT, REPO_PREFIX
 CJK_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]+")
 SEARCH_RUN_RE = re.compile(r"[a-z0-9]+|[\u3400-\u4dbf\u4e00-\u9fff]+", re.IGNORECASE)
 DOCUMENT_BUILD_VERSION = "1"
+PARSED_CACHE_ROOT = Path(os.environ["BHA_PARSED_CACHE_ROOT"]) if os.environ.get("BHA_PARSED_CACHE_ROOT") else None
 
 def initialize_search_tokenizer() -> None:
     jieba.initialize()
@@ -19,14 +21,31 @@ def initialize_search_tokenizer() -> None:
 def run(command: list[str], cwd: Path | None = None) -> None:
     subprocess.run(command, cwd=str(cwd) if cwd else None, check=True)
 
+def cache_parsed_repository(source: Path, archive_id: int) -> None:
+    if PARSED_CACHE_ROOT is None:
+        return
+    PARSED_CACHE_ROOT.mkdir(parents=True, exist_ok=True)
+    target = PARSED_CACHE_ROOT / f"archives{archive_id}"
+    if target.is_dir() and (target / ".git").is_dir():
+        source_head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=source, capture_output=True, text=True, check=True).stdout.strip()
+        cached_head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=target, capture_output=True, text=True, check=True).stdout.strip()
+        if source_head == cached_head:
+            return
+    temporary = PARSED_CACHE_ROOT / f".archives{archive_id}.{os.getpid()}.tmp"
+    shutil.rmtree(temporary, ignore_errors=True)
+    shutil.copytree(source, temporary)
+    shutil.rmtree(target, ignore_errors=True)
+    temporary.replace(target)
+
 def ensure_parsed_data(
     progress: Callable[[int, int], None] | None = None,
     expected_commits: dict[int, str] | None = None,
     archive_ids: Iterable[int] | None = None,
-) -> None:
+) -> dict[int, set[str]] | None:
     PARSED_ROOT.mkdir(parents=True, exist_ok=True)
     archive_ids = list(archive_ids) if archive_ids is not None else list(range(ARCHIVE_START, ARCHIVE_END + 1))
     total = len(archive_ids)
+    changed_paths: dict[int, set[str]] | None = {} if expected_commits is not None else None
     for position, archive_id in enumerate(archive_ids, start=1):
         if progress:
             progress(position, total)
@@ -39,8 +58,17 @@ def ensure_parsed_data(
                 ["git", "rev-parse", "HEAD"], cwd=target, check=True, capture_output=True, text=True,
             ).stdout.strip()
             if expected and current == expected:
+                if changed_paths is not None:
+                    changed_paths[archive_id] = set()
+                cache_parsed_repository(target, archive_id)
                 continue
             run(["git", "fetch", "--depth", "1", "origin", expected or "parsed"], target)
+            if expected:
+                diff = subprocess.run(
+                    ["git", "diff", "--name-only", current, "FETCH_HEAD"],
+                    cwd=target, check=True, capture_output=True, text=True,
+                ).stdout.splitlines()
+                changed_paths[archive_id] = {path for path in diff if path}
             run(["git", "clean", "-fdx"], target)
             run(["git", "checkout", "parsed"], target)
             run(["git", "reset", "--hard", "FETCH_HEAD"], target)
@@ -48,16 +76,23 @@ def ensure_parsed_data(
                 ["git", "rev-parse", "HEAD"], cwd=target, check=True, capture_output=True, text=True,
             ).stdout.strip() != expected:
                 raise RuntimeError(f"parsed commit mismatch for archive {archive_id}")
+            cache_parsed_repository(target, archive_id)
             continue
         if target.exists():
             shutil.rmtree(target)
-        run(["git", "clone", "--depth", "1", "--branch", "parsed", repo, str(target)])
+        cached = PARSED_CACHE_ROOT / f"archives{archive_id}" if PARSED_CACHE_ROOT is not None else None
+        if cached is not None and cached.is_dir() and (cached / ".git").exists():
+            shutil.copytree(cached, target)
+        else:
+            run(["git", "clone", "--depth", "1", "--branch", "parsed", repo, str(target)])
         expected = expected_commits.get(archive_id) if expected_commits else None
         if expected and subprocess.run(
             ["git", "rev-parse", "HEAD"], cwd=target, check=True, capture_output=True, text=True,
         ).stdout.strip() != expected:
             run(["git", "fetch", "--depth", "1", "origin", expected], target)
             run(["git", "reset", "--hard", "FETCH_HEAD"], target)
+        cache_parsed_repository(target, archive_id)
+    return changed_paths
 
 def parsed_corpus_fingerprint(
     remote: bool = False, revisions: dict[int, str] | None = None
@@ -220,7 +255,10 @@ def normalize_tags(tags: Any) -> tuple[list[str], list[str]]:
             types.append(str(tag_type).strip())
     return names, types
 
-def iter_documents(archive_ids: Iterable[int] | None = None) -> Iterator[dict[str, Any]]:
+def iter_documents(
+    archive_ids: Iterable[int] | None = None,
+    changed_paths: dict[int, set[str]] | None = None,
+) -> Iterator[dict[str, Any]]:
     selected_archives = (
         sorted(set(archive_ids)) if archive_ids is not None else range(ARCHIVE_START, ARCHIVE_END + 1)
     )
@@ -237,7 +275,14 @@ def iter_documents(archive_ids: Iterable[int] | None = None) -> Iterator[dict[st
             publication_name_words = search_words(publication_name)
             publication_files = publication.get("files") if isinstance(publication.get("files"), list) else []
             book_root = metadata_path.parent
+            metadata_relative = metadata_path.relative_to(archive_root).as_posix()
+            archive_changed = changed_paths.get(archive_id) if changed_paths is not None else None
             for article_path in book_root.glob("*/*.json"):
+                if archive_changed is not None:
+                    article_relative = article_path.relative_to(archive_root).as_posix()
+                    tags_relative = article_path.with_suffix(".tags").relative_to(archive_root).as_posix()
+                    if metadata_relative not in archive_changed and article_relative not in archive_changed and tags_relative not in archive_changed:
+                        continue
                 article_id = article_path.stem
                 article = read_json(article_path)
                 if not isinstance(article, dict):

app/doc_store.py

@@ -101,6 +101,27 @@ def rebuild_years_to_conn(conn: sqlite3.Connection) -> None:
     conn.execute("DELETE FROM years")
     conn.executemany("INSERT INTO years(year) VALUES (?)", [(year,) for year in sorted(years)])
 
+def update_years_to_conn(conn: sqlite3.Connection, affected_docs: list[dict[str, Any]]) -> None:
+    affected: set[int] = set()
+    for doc in affected_docs:
+        for year in doc.get("date_years") or []:
+            if isinstance(year, int):
+                affected.add(year)
+        for value in doc.get("date_display") or []:
+            raw_year = str(value).split("/", 1)[0]
+            if raw_year.isdigit():
+                affected.add(int(raw_year))
+    if not affected:
+        return
+    placeholders = ",".join("?" for _ in affected)
+    conn.execute(f"DELETE FROM years WHERE year IN ({placeholders})", sorted(affected))
+    for year in sorted(affected):
+        if conn.execute(
+            "SELECT 1 FROM docs WHERE date_display_json LIKE ? LIMIT 1",
+            (f'%"{year}/%',),
+        ).fetchone():
+            conn.execute("INSERT INTO years(year) VALUES (?)", (year,))
+
 def years_from_docs(docs: list[dict[str, Any]]) -> set[int]:
     years: set[int] = set()
     for doc in docs:

app/indexer.py

@@ -9,12 +9,12 @@ import time
 import uuid
 from collections import Counter
 from pathlib import Path
-from typing import IO
+from typing import IO, Iterable
 from elasticsearch import Elasticsearch, helpers
 from elasticsearch.exceptions import NotFoundError
 from .config import DATA_ROOT, ARCHIVE_END, ARCHIVE_START, ES_URL, INDEX_NAME, INDEX_VERSION, RESET_INDEX
 from .data_loader import DOCUMENT_BUILD_VERSION, ensure_parsed_data, iter_documents, parsed_archive_commits, parsed_archive_revisions, parsed_corpus_fingerprint
-from .doc_store import DOC_DB, DOC_STORE_SCHEMA_VERSION, connect as connect_doc_db, delete_docs_to_conn, doc_row, init_db as init_doc_db, reset_db, rebuild_years_to_conn, row_to_doc, write_docs_to_conn
+from .doc_store import DOC_DB, DOC_STORE_SCHEMA_VERSION, connect as connect_doc_db, delete_docs_to_conn, doc_row, init_db as init_doc_db, reset_db, row_to_doc, update_years_to_conn, write_docs_to_conn
 from .facet_store import FACET_DB, FACET_STORE_SCHEMA_VERSION, apply_facet_deltas, connect as connect_facet_db, init_db as init_facet_db, reset_db as reset_facet_db, write_facets
 from .storage_lock import serving_lock
 MAX_RESULT_WINDOW = 500000
@@ -202,7 +202,13 @@ def empty_facet_counters() -> dict[str, Counter[str]]:
 def elasticsearch_document(doc: dict) -> dict:
     return {key: value for key, value in doc.items() if key != "article"}
 
-def incremental_build(es: Elasticsearch, source_fingerprint: str, revisions: dict[int, str], old_meta: dict) -> None:
+def incremental_build(
+    es: Elasticsearch,
+    source_fingerprint: str,
+    revisions: dict[int, str],
+    old_meta: dict,
+    changed_paths: dict[int, set[str]] | None = None,
+) -> None:
     source_index = serving_physical_index(es)
     old_revisions = old_meta.get("archive_revisions")
     expected_static = desired_metadata()
@@ -241,7 +247,7 @@ def incremental_build(es: Elasticsearch, source_fingerprint: str, revisions: dic
                 )
             }
             seen_ids: set[str] = set()
-            for doc in iter_documents(changed):
+            for doc in iter_documents(changed, changed_paths=changed_paths):
                 doc_id = doc["doc_id"]
                 if doc_id in seen_ids:
                     raise IncrementalUnsafe(f"duplicate incremental doc_id: {doc_id}")
@@ -263,7 +269,7 @@ def incremental_build(es: Elasticsearch, source_fingerprint: str, revisions: dic
                 )
             delete_docs_to_conn(conn, deleted)
             write_docs_to_conn(conn, upsert_docs)
-            rebuild_years_to_conn(conn)
+            update_years_to_conn(conn, [*removed_docs, *upsert_docs])
             conn.commit()
         removed_facets = empty_facet_counters()
         added_facets = empty_facet_counters()
@@ -499,6 +505,7 @@ def ensure_index(
     lock_file: IO[str] | None = None,
     expected_commits: dict[int, str] | None = None,
     expected_revisions: dict[int, str] | None = None,
+    changed_archives: Iterable[int] | None = None,
 ) -> bool:
     owned_lock = lock_file is None
     if lock_file is None:
@@ -508,7 +515,7 @@ def ensure_index(
     try:
         if expected_commits is None and expected_revisions is None:
             return _ensure_index(reset)
-        return _ensure_index(reset, expected_commits, expected_revisions)
+        return _ensure_index(reset, expected_commits, expected_revisions, changed_archives)
     finally:
         if owned_lock:
             lock_file.close()
@@ -517,6 +524,7 @@ def _ensure_index(
     reset: bool = False,
     expected_commits: dict[int, str] | None = None,
     expected_revisions: dict[int, str] | None = None,
+    changed_archives: Iterable[int] | None = None,
 ) -> bool:
     started_at = time.time()
     write_progress(status="preparing", indexed=0, errors=0, elapsed_seconds=0)
@@ -531,11 +539,20 @@ def _ensure_index(
         and set(old_revisions) == {str(archive_id) for archive_id in expected_revisions}
         and all(old_meta.get(key) == value for key, value in desired_metadata().items())
     ):
+        requested_archives = set(changed_archives) if changed_archives is not None else None
+        derived_archives = [
+            archive_id for archive_id, revision in expected_revisions.items()
+            if old_revisions.get(str(archive_id)) != revision
+        ]
+        if requested_archives is not None and not set(derived_archives).issubset(requested_archives):
+            print("changed_archives_hint_incomplete=using_revision_difference", flush=True)
+            requested_archives = None
         sync_archive_ids = [
             archive_id for archive_id, revision in expected_revisions.items()
             if old_revisions.get(str(archive_id)) != revision
+            and (requested_archives is None or archive_id in requested_archives)
         ]
-    ensure_parsed_data(lambda current, total: write_progress(
+    changed_paths = ensure_parsed_data(lambda current, total: write_progress(
         status="syncing",
         indexed=0,
         errors=0,
@@ -543,6 +560,8 @@ def _ensure_index(
         archive_total=total,
         elapsed_seconds=int(time.time() - started_at),
     ), expected_commits=expected_commits, archive_ids=sync_archive_ids)
+    if not isinstance(changed_paths, dict):
+        changed_paths = None
     if expected_commits is not None:
         local_commits = parsed_archive_commits(archive_ids=sync_archive_ids)
         if local_commits != {
@@ -566,7 +585,10 @@ def _ensure_index(
             return False
     if not force_rebuild and has_serving_data:
         try:
-            incremental_build(es, source_fingerprint, archive_revisions, old_meta)
+            if changed_paths is None:
+                incremental_build(es, source_fingerprint, archive_revisions, old_meta)
+            else:
+                incremental_build(es, source_fingerprint, archive_revisions, old_meta, changed_paths=changed_paths)
             write_progress(
                 status="ready",
                 indexed=es.count(index=INDEX_NAME).get("count", 0),

app/main.py

@@ -31,7 +31,7 @@ from .search_store import SearchStoreError, get_search_docs, literal_match_ids
 from .data_loader import initialize_search_tokenizer, search_words
 from .facet_store import FACET_DB, list_facets, sources_payload
 from .indexer import acquire_index_lock, ensure_index
-from .bucket_snapshot import publish as publish_bucket_snapshot, snapshot_status as bucket_snapshot_status
+from .bucket_snapshot import publish as publish_bucket_snapshot, publish_due as bucket_snapshot_publish_due, snapshot_status as bucket_snapshot_status
 from .storage_lock import serving_lock
 SOURCE_MIRROR_MANIFEST_URL = os.environ.get(
     "BHA_SOURCE_MIRROR_MANIFEST_URL",
@@ -80,6 +80,7 @@ except ValueError:
 INDEX_STATUS_PATH = DATA_ROOT / "index-status"
 INDEX_SWITCH_STATE_PATH = DATA_ROOT / "index-switch.json"
 BUCKET_DIR = Path(os.environ["BHA_BUCKET_DIR"]) if os.environ.get("BHA_BUCKET_DIR") else None
+SNAPSHOT_MIN_INTERVAL_SECONDS = int(os.environ.get("BHA_SNAPSHOT_MIN_INTERVAL_SECONDS", "1800"))
 app = FastAPI(title="BHA Search Lite")
 app.add_middleware(
     CORSMiddleware,
@@ -1842,8 +1843,10 @@ async def download_source_zip(doc_id: str):
 
 def publish_bucket_snapshot_worker() -> None:
     try:
-        if BUCKET_DIR is not None:
+        if BUCKET_DIR is not None and bucket_snapshot_publish_due(BUCKET_DIR, SNAPSHOT_MIN_INTERVAL_SECONDS):
             publish_bucket_snapshot(BUCKET_DIR)
+        elif BUCKET_DIR is not None:
+            print("bucket_snapshot_publish_skipped=minimum_interval", flush=True)
     except Exception as exc:
         print(f"bucket_snapshot_publish_failed={exc}")
 
@@ -1852,6 +1855,7 @@ def run_reindex(
     reset: bool = False,
     expected_commits: dict[int, str] | None = None,
     expected_revisions: dict[int, str] | None = None,
+    changed_archives: list[int] | None = None,
 ) -> None:
     try:
         INDEX_STATUS_PATH.write_text("rebuilding", encoding="utf-8")
@@ -1862,6 +1866,7 @@ def run_reindex(
             changed = ensure_index(
                 reset=reset, lock_file=lock_file,
                 expected_commits=expected_commits, expected_revisions=expected_revisions,
+                changed_archives=changed_archives,
             )
         INDEX_STATUS_PATH.write_text("ready", encoding="utf-8")
         if changed and BUCKET_DIR is not None:
@@ -1879,10 +1884,11 @@ def start_reindex_worker(
     reset: bool = False,
     expected_commits: dict[int, str] | None = None,
     expected_revisions: dict[int, str] | None = None,
+    changed_archives: list[int] | None = None,
 ) -> None:
     threading.Thread(
         target=run_reindex,
-        args=(lock_file, reset, expected_commits, expected_revisions),
+        args=(lock_file, reset, expected_commits, expected_revisions, changed_archives),
         daemon=True,
     ).start()
 
@@ -1911,6 +1917,9 @@ def reindex(request: Request, full: bool = False, payload: ReindexRequest | None
             return JSONResponse({"error": "invalid changed archive"}, status_code=400)
         expected_commits = {int(key): value for key, value in payload.archive_commits.items()}
         expected_revisions = {int(key): value for key, value in payload.archive_revisions.items()}
+        changed_archives = sorted(set(payload.changed_archives))
+    else:
+        changed_archives = None
     try:
         last_start = int(REINDEX_LAST_START.read_text(encoding="utf-8").strip())
     except Exception:
@@ -1932,6 +1941,7 @@ def reindex(request: Request, full: bool = False, payload: ReindexRequest | None
             start_reindex_worker(
                 lock_file, reset=full,
                 expected_commits=expected_commits, expected_revisions=expected_revisions,
+                changed_archives=changed_archives,
             )
     except Exception:
         lock_file.close()