diff --git "a/app.py" "b/app.py"
--- "a/app.py"
+++ "b/app.py"
@@ -1,1732 +1,1580 @@
-import asyncio
+from __future__ import annotations
import gzip
import hashlib
+import heapq
import json
-import posixpath
+import os
import random
import re
+import secrets
+import sqlite3
+import tempfile
+import threading
import time
-import functools
-import io
+import unicodedata
+from functools import lru_cache
+from contextlib import asynccontextmanager, contextmanager
from pathlib import Path
-from typing import Optional
-from urllib.parse import unquote, quote, urljoin, urlparse
-import aiohttp
-import jieba
+from typing import Literal, Optional
+from urllib.parse import parse_qs, quote, unquote
from fastapi import FastAPI, Query, Request
-from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse, Response, StreamingResponse
-from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
+from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse, Response
+from fastapi.staticfiles import StaticFiles
+from starlette.background import BackgroundTask
from pydantic import BaseModel, Field
-from contextlib import asynccontextmanager
-ALLOWED_DOWNLOAD_HOSTS = {
- "huggingface.co",
- "hf-mirror.com",
- "hf.co",
-}
-ALLOWED_DOWNLOAD_HOST_SUFFIXES = (
- ".hf.co",
- ".huggingface.co",
- ".xethub.hf.co",
-)
-MAX_REDIRECTS = 5
-MAX_ARCHIVE_NAME_LENGTH = 240
-DATA_PATH = Path("data/search_data.json.gz")
-NGRAM_PATHS = {2: Path("data/search_ngrams_2.bin.gz"), 3: Path("data/search_ngrams_3.bin.gz")}
-WORD_INDEX_PATH = Path("data/search_words.bin.gz")
-NGRAM_MAGIC = b"VNG3"
-WORD_INDEX_MAGIC = b"VWI2"
-WORD_INDEX_TOKENIZER = "jieba-0.42.1-cjk-char-v1"
-FOLDER_TREE_PATH = Path("data/folder_tree.json.gz")
-FOLDER_BROWSER_PATH = Path("data/folder_browser.json.gz")
-TXT_DIR = Path("txt")
-TXT_SPACE_RAW_BASE = "https://huggingface.co/spaces/VoiceOfML/Search/raw/main/txt"
-TXT_WARM_INTERVAL_SECONDS = 600
-TXT_WARM_IDLE_SECONDS = 600
-READER_ASSETS_URL = "https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/reader_assets.json.gz"
-READER_ASSETS_CACHE_TTL_SECONDS = 300
-READER_ASSETS_MAX_COMPRESSED_BYTES = 5 * 1024 * 1024
-READER_ASSETS_MAX_DECOMPRESSED_BYTES = 50 * 1024 * 1024
+BASE_DIR = Path(__file__).resolve().parent
+DATA_PATH = BASE_DIR / "data/search_data.json.gz"
+FOLDER_TREE_PATH = BASE_DIR / "data/folder_tree.json.gz"
+FOLDER_BROWSER_PATH = BASE_DIR / "data/folder_browser.json.gz"
+FULLTEXT_DIR = BASE_DIR / "data/fulltext"
+INDEX_BUILD_STATUS_PATH = FULLTEXT_DIR / "build-status.json"
+BUCKET_INDEX_STATUS_PATH = FULLTEXT_DIR / "bucket-index-status.json"
records: list[dict] = []
+name_order: list[int] = []
+size_order: list[int] = []
+source_name_order: dict[str, list[int]] = {}
+source_size_order: dict[str, list[int]] = {}
+record_map: dict[str, dict] = {}
+record_map_index: dict[str, int] = {}
+sources: list[dict] = []
+source_counts: dict[str, int] = {}
folder_tree_data: dict[str, list[dict]] = {}
folder_browser_data: dict[str, dict[str, dict]] = {}
-word_index = {}
-folder_index = {}
-extension_counts = {}
-repo_counts = {}
-repo_list = []
-extension_list = []
-repo_extension_counts = {}
-vocab_by_len = {}
-repo_records_map = {}
-sorted_by_name = []
-sorted_by_size = []
-repo_sorted_by_name = {}
-repo_sorted_by_size = {}
-txt_record_indices = []
-repo_txt_record_indices = {}
-READER_EXTENSIONS = frozenset({"pdf", "epub", "txt", "md", "markdown", "jpg", "jpeg", "png", "gif", "bmp", "webp"})
-reader_record_indices = []
-repo_reader_record_indices = {}
-reader_record_key_indices = {}
-initial_payload_global = None
-initial_payload_by_repo = {}
+source_records_map: dict[str, list[int]] = {}
+extension_counts: dict[str, int] = {}
+source_extension_counts: dict[str, dict[str, int]] = {}
+word_index: dict[str, set[int]] = {}
+latin_all_index: dict[str, set[int]] = {}
+latin_file_index: dict[str, set[int]] = {}
+latin_vocabulary_all: set[str] = set()
+TOKEN_RE = re.compile(r"[a-z0-9]+|[\u4e00-\u9fff\u3400-\u4dbf]+")
+LATIN_RE = re.compile(r"[a-z0-9]+")
+TOKENIZER_VERSION = "cjk-bigram-boundary-fts5-v6-snippet-anchors"
+LITERAL_PREFIX = "\0literal:"
API_CACHE_TTL_SECONDS = 120
-api_response_cache = {}
-_data_generation = 0
-SEARCH_CACHE_TTL_SECONDS = 120
-_search_cache: dict[tuple, tuple[float, dict]] = {}
-METADATA_CACHE_HEADERS = {
- "Cache-Control": "public, max-age=300, stale-while-revalidate=1800",
-}
-ngram_postings: dict[int, dict[str, bytes]] = {}
-ngram_posting_buffers: dict[int, bytes] = {}
-injected_html_cache: dict[str | None, str] = {}
-last_user_request_at = time.monotonic()
-reader_assets_cache = {"v": 1, "f": {}}
-reader_assets_cache_at = 0.0
-
-def cached_payload(key, builder):
- now = time.monotonic()
- cached = api_response_cache.get(key)
- if cached and now - cached[0] < API_CACHE_TTL_SECONDS:
- return cached[1]
- value = builder()
- api_response_cache[key] = (now, value)
- while len(api_response_cache) > 500:
- oldest_key = min(api_response_cache, key=lambda item: api_response_cache[item][0])
- api_response_cache.pop(oldest_key, None)
- return value
-
-
-def decode_reader_assets(raw: bytes) -> dict:
- if len(raw) > READER_ASSETS_MAX_COMPRESSED_BYTES:
- raise ValueError("reader assets sidecar is too large")
- with gzip.GzipFile(fileobj=io.BytesIO(raw)) as stream:
- decoded = stream.read(READER_ASSETS_MAX_DECOMPRESSED_BYTES + 1)
- if len(decoded) > READER_ASSETS_MAX_DECOMPRESSED_BYTES:
- raise ValueError("reader assets sidecar expands beyond limit")
- payload = json.loads(decoded)
- files = payload.get("f") if isinstance(payload, dict) and payload.get("v") == 1 else None
- if not isinstance(files, dict):
- raise ValueError("invalid reader assets sidecar")
- clean = {}
- for key, entry in files.items():
- if not isinstance(key, str) or not isinstance(entry, dict) or entry.get("s") not in (2, 4):
- continue
- if entry.get("s") == 2:
- path = entry.get("p")
- mode = entry.get("m")
- if mode not in ("p", "e", "d", "h") or not isinstance(path, str) or not re.fullmatch(r"objects/[0-9a-f]{2}/[0-9a-f]{64}/(?:[a-z0-9-]+/)?(?:document\.pdf|book\.epub|document\.docx|document\.html)", path):
- continue
- clean[key] = {"s": 2, "m": mode, "p": path}
+SEARCH_CACHE_TTL_SECONDS = 300
+API_CACHE_MAX_ENTRIES = 500
+SEARCH_CACHE_MAX_ENTRIES = 500
+PREVIEW_CACHE_MAX_ENTRIES = 64
+PREVIEW_CACHE_MAX_ENTRY_BYTES = 2 * 1024 * 1024
+api_response_cache: dict[tuple, tuple[float, object]] = {}
+api_response_cache_lock = threading.Lock()
+search_response_cache: dict[tuple, tuple[float, object]] = {}
+search_response_cache_lock = threading.Lock()
+preview_text_cache: dict[str, str] = {}
+preview_text_cache_lock = threading.Lock()
+cache_key_locks: dict[tuple, threading.Lock] = {}
+cache_key_locks_guard = threading.Lock()
+ZIP_TOKEN_TTL_SECONDS = 600
+ZIP_TOKEN_MAX_ENTRIES = 32
+zip_download_tokens: dict[str, tuple[float, list[str]]] = {}
+zip_download_lock = threading.Lock()
+FULLTEXT_IN_MEMORY = os.environ.get("CCRD_FULLTEXT_IN_MEMORY", "1").strip().lower() in {"1", "true", "yes", "on"}
+FULLTEXT_REFRESH_INTERVAL_SECONDS = max(1, int(os.environ.get("CCRD_FULLTEXT_REFRESH_INTERVAL_SECONDS", "5")))
+fulltext_refresh_stop = threading.Event()
+
+def normalize_text(text: str) -> str:
+ return unicodedata.normalize("NFKC", text or "").casefold()
+
+def query_terms(text: str) -> list[str]:
+ return list(dict.fromkeys(TOKEN_RE.findall(normalize_text(text))))
+
+def index_tokens(text: str) -> set[str]:
+ tokens = set()
+ for part in query_terms(text):
+ if re.fullmatch(r"[\u4e00-\u9fff\u3400-\u4dbf]+", part):
+ tokens.update(part)
+ tokens.update(part[index:index + 2] for index in range(len(part) - 1))
else:
- clean[key] = {"s": 4}
- return {"v": 1, "f": clean}
+ tokens.add(part)
+ return tokens
+@lru_cache(maxsize=4096)
+def query_tokens(text: str) -> list[str]:
+ tokens = []
+ for part in query_terms(text):
+ if re.fullmatch(r"[\u4e00-\u9fff\u3400-\u4dbf]+", part) and len(part) > 1:
+ tokens.extend(part[index:index + 2] for index in range(len(part) - 1))
+ else:
+ tokens.append(part)
+ return list(dict.fromkeys(tokens))
+
+@lru_cache(maxsize=4096)
+def literal_query_tokens(text: str) -> list[str]:
+ normalized = normalize_text(text)
+ return list(dict.fromkeys(
+ LITERAL_PREFIX + normalized[index:index + 3]
+ for index in range(len(normalized) - 2)
+ if not TOKEN_RE.fullmatch(normalized[index + 1])
+ ))
-async def get_reader_assets() -> dict:
- global reader_assets_cache, reader_assets_cache_at
- now = time.monotonic()
- if now - reader_assets_cache_at < READER_ASSETS_CACHE_TTL_SECONDS:
- return reader_assets_cache
- async with app.state.reader_assets_lock:
- now = time.monotonic()
- if now - reader_assets_cache_at < READER_ASSETS_CACHE_TTL_SECONDS:
- return reader_assets_cache
- response = None
+@lru_cache(maxsize=4096)
+def wildcard_required_tokens(text: str) -> list[str]:
+ tokens = []
+ for fixed_part in re.split(r"[*?]+", normalize_text(text)):
+ for term in TOKEN_RE.findall(fixed_part):
+ if re.fullmatch(r"[\u4e00-\u9fff\u3400-\u4dbf]+", term):
+ tokens.extend(query_tokens(term))
+ return list(dict.fromkeys(tokens))
+
+def decode_doc_numbers(payload: bytes, candidates: set[int] | None = None) -> set[int]:
+ results = set()
+ number = shift = previous = 0
+ for byte in payload:
+ number |= (byte & 0x7F) << shift
+ if byte & 0x80:
+ shift += 7
+ continue
+ previous += number
+ if candidates is None or previous in candidates:
+ results.add(previous)
+ number = shift = 0
+ return results
+
+class FulltextDatabases:
+
+ def __init__(self, directory: Path, in_memory: bool = False, refresh: bool = True):
+ self.directory = directory
+ self.in_memory = in_memory
+ self.lock = threading.RLock()
+ self.refresh_lock = threading.Lock()
+ self.connections = {}
+ self.has_doc_counts = {}
+ self.tokenizer_versions = {}
+ self.has_content_fts = {}
+ self.has_snippet_anchors = {}
+ self.document_counts = {}
+ self.fts_counts = {}
+ self.file_signatures = {}
+ self.connection_readers = {}
+ self.retired_connections = set()
+ self.closed = False
+ self.generation = 0
+ if refresh:
+ self.refresh()
+
+ def refresh(self) -> None:
+ with self.refresh_lock:
+ for path in self.directory.glob("*.sqlite3"):
+ try:
+ stat = path.stat()
+ signature = (stat.st_ino, stat.st_size, stat.st_mtime_ns)
+ except OSError:
+ continue
+ with self.lock:
+ if self.closed or self.file_signatures.get(path.stem) == signature:
+ continue
+ try:
+ candidate = self._open_candidate(path)
+ except (OSError, sqlite3.Error) as exc:
+ print(f"fulltext_open_failed={path.name}:{exc}")
+ continue
+ try:
+ current_stat = path.stat()
+ current_signature = (current_stat.st_ino, current_stat.st_size, current_stat.st_mtime_ns)
+ except OSError:
+ candidate["connection"].close()
+ continue
+ if current_signature != signature:
+ candidate["connection"].close()
+ continue
+ with self.lock:
+ if self.closed or self.file_signatures.get(path.stem) == signature:
+ candidate["connection"].close()
+ continue
+ self._install_candidate(path.stem, signature, candidate)
+
+ def _open_candidate(self, path: Path) -> dict:
+ uri = f"file:{path.resolve()}?mode=ro&immutable=1"
+ source_connection = sqlite3.connect(uri, uri=True, check_same_thread=False)
+ connection = source_connection
try:
- response = await app.state.http_session.get(
- READER_ASSETS_URL,
- timeout=aiohttp.ClientTimeout(total=20, connect=8, sock_read=10),
- )
- if response.status != 200:
- raise RuntimeError(f"reader assets HTTP {response.status}")
- chunks = bytearray()
- async for chunk in response.content.iter_chunked(64 * 1024):
- chunks.extend(chunk)
- if len(chunks) > READER_ASSETS_MAX_COMPRESSED_BYTES:
- raise ValueError("reader assets response is too large")
- raw = bytes(chunks)
- reader_assets_cache = decode_reader_assets(raw)
- reader_assets_cache_at = now
- except Exception as exc:
- print(f"reader assets refresh failed: {exc}")
- reader_assets_cache_at = now
+ columns = {row[1] for row in source_connection.execute("PRAGMA table_info(postings)")}
+ has_metadata = source_connection.execute(
+ "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'metadata'"
+ ).fetchone()
+ version_row = source_connection.execute(
+ "SELECT value FROM metadata WHERE key = 'tokenizer_version'"
+ ).fetchone() if has_metadata else None
+ has_content_fts = bool(source_connection.execute(
+ "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'content_fts'"
+ ).fetchone())
+ anchor_columns = {
+ row[1] for row in source_connection.execute("PRAGMA table_info(snippet_anchors)")
+ }
+ has_snippet_anchors = {"token_hash", "doc_number", "char_offset"} <= anchor_columns
+ document_count = int(source_connection.execute("SELECT COUNT(*) FROM documents").fetchone()[0])
+ fts_count = int(source_connection.execute("SELECT COUNT(*) FROM content_fts").fetchone()[0]) if has_content_fts else 0
+ if self.in_memory:
+ connection = sqlite3.connect(":memory:", check_same_thread=False)
+ source_connection.backup(connection)
+ except Exception:
+ if connection is not source_connection:
+ connection.close()
+ source_connection.close()
+ raise
finally:
- if response is not None:
- response.release()
- return reader_assets_cache
-
-def decode_varint(data: bytes, offset: int) -> tuple[int, int]:
- value = 0
- shift = 0
- while offset < len(data):
- byte = data[offset]
- offset += 1
- value |= (byte & 0x7F) << shift
- if not byte & 0x80:
- return value, offset
- shift += 7
- raise ValueError("truncated varint")
-
-def records_digest() -> bytes:
- digest = hashlib.sha256()
- for record in records:
- for value in (
- record.get("_file_lower", ""),
- record.get("_repo_lower", ""),
- record.get("_folder_path_lower", ""),
- ):
- digest.update(value.encode("utf-8"))
- digest.update(b"\0")
- return digest.digest()
-
-
-def load_ngram_index(path: Path, width: int, record_count: int, expected_digest: bytes) -> tuple[dict, bytes]:
- try:
- data = gzip.decompress(path.read_bytes())
- if not data.startswith(NGRAM_MAGIC) or len(data) < 37 or data[4] != width:
- return {}, b""
- if data[5:37] != expected_digest:
- return {}, b""
- offset = 37
- stored_record_count, offset = decode_varint(data, offset)
- count, offset = decode_varint(data, offset)
- if stored_record_count != record_count:
- return {}, b""
- postings = {}
- for _ in range(count):
- key_length, offset = decode_varint(data, offset)
- key = data[offset:offset + key_length].decode("utf-8")
- offset += key_length
- posting_count, offset = decode_varint(data, offset)
- posting_start = offset
- posting_length, offset = decode_varint(data, offset)
- posting_start = offset
- offset += posting_length
- if offset > len(data):
- raise ValueError("truncated postings")
- postings[key] = (posting_start, offset, posting_count)
- if offset != len(data):
- raise ValueError("trailing postings")
- return postings, data
- except (OSError, ValueError, UnicodeDecodeError, EOFError):
- return {}, b""
-
-
-def load_word_index(path: Path, record_count: int, expected_digest: bytes) -> dict:
- try:
- data = gzip.decompress(path.read_bytes())
- if not data.startswith(WORD_INDEX_MAGIC):
- return {}
- offset = len(WORD_INDEX_MAGIC)
- tokenizer_length, offset = decode_varint(data, offset)
- tokenizer = data[offset:offset + tokenizer_length].decode("utf-8")
- offset += tokenizer_length
- if tokenizer != WORD_INDEX_TOKENIZER or data[offset:offset + 32] != expected_digest:
- return {}
- offset += 32
- stored_record_count, offset = decode_varint(data, offset)
- count, offset = decode_varint(data, offset)
- if stored_record_count != record_count:
- return {}
- postings = {}
- for _ in range(count):
- key_length, offset = decode_varint(data, offset)
- key = data[offset:offset + key_length].decode("utf-8")
- offset += key_length
- posting_count, offset = decode_varint(data, offset)
- posting_length, offset = decode_varint(data, offset)
- posting_start = offset
- offset += posting_length
- if offset > len(data):
- raise ValueError("truncated postings")
- decoded = decode_postings(memoryview(data)[posting_start:offset])
- if len(decoded) != posting_count:
- raise ValueError("posting count mismatch")
- postings[key] = decoded
- if offset != len(data):
- raise ValueError("trailing postings")
- return postings
- except (OSError, ValueError, UnicodeDecodeError, EOFError):
- return {}
-
-def decode_postings(data: bytes) -> set[int]:
- record_ids = set()
- offset = 0
- previous = 0
- while offset < len(data):
- delta, offset = decode_varint(data, offset)
- previous += delta
- record_ids.add(previous)
- return record_ids
-
-
-def resolve_postings(value, buffer: bytes) -> set[int]:
- if isinstance(value, tuple):
- start, end, _ = value
- return decode_postings(memoryview(buffer)[start:end])
- return set(value) if isinstance(value, (set, frozenset)) else decode_postings(value)
-
-
-def word_postings(token: str) -> set[int]:
- value = word_index.get(token)
- return set(value) if value is not None else set()
-
-
-def word_posting_count(token: str) -> int:
- value = word_index.get(token)
- return len(value) if value is not None else 0
-
-def literal_candidates(query: str, repos=None) -> set[int] | None:
- width = 3 if len(query) >= 3 and ngram_postings.get(3) else 2
- postings = ngram_postings.get(width)
- if not postings or len(query) < width:
- return None
- candidates = None
- for index in range(len(query) - width + 1):
- encoded = postings.get(query[index:index + width])
- if encoded is None:
+ if source_connection is not connection:
+ source_connection.close()
+ return {
+ "connection": connection,
+ "has_doc_counts": "doc_count" in columns,
+ "tokenizer_version": version_row[0] if version_row else "legacy",
+ "has_content_fts": has_content_fts,
+ "has_snippet_anchors": has_snippet_anchors,
+ "document_count": document_count,
+ "fts_count": fts_count,
+ }
+
+ def _install_candidate(self, source: str, signature: tuple[int, int, int], candidate: dict) -> None:
+ connection = candidate["connection"]
+ old_connection = self.connections.get(source)
+ self.connections[source] = connection
+ self.has_doc_counts[source] = candidate["has_doc_counts"]
+ self.tokenizer_versions[source] = candidate["tokenizer_version"]
+ self.has_content_fts[source] = candidate["has_content_fts"]
+ self.has_snippet_anchors[source] = candidate["has_snippet_anchors"]
+ self.document_counts[source] = candidate["document_count"]
+ self.fts_counts[source] = candidate["fts_count"]
+ self.file_signatures[source] = signature
+ self.generation += 1
+ if old_connection is not None:
+ if self.connection_readers.get(old_connection, 0):
+ self.retired_connections.add(old_connection)
+ else:
+ old_connection.close()
+
+ @contextmanager
+ def _lease_source(self, source: str):
+ with self.lock:
+ connection = self.connections.get(source)
+ if connection is None or self.closed:
+ lease = None
+ else:
+ self.connection_readers[connection] = self.connection_readers.get(connection, 0) + 1
+ lease = (connection, {
+ "has_doc_counts": self.has_doc_counts.get(source, False),
+ "tokenizer_version": self.tokenizer_versions.get(source),
+ "has_content_fts": self.has_content_fts.get(source, False),
+ "has_snippet_anchors": self.has_snippet_anchors.get(source, False),
+ })
+ try:
+ yield lease
+ finally:
+ if lease is None:
+ return
+ with self.lock:
+ remaining = self.connection_readers.get(connection, 1) - 1
+ if remaining:
+ self.connection_readers[connection] = remaining
+ else:
+ self.connection_readers.pop(connection, None)
+ if connection in self.retired_connections:
+ self.retired_connections.remove(connection)
+ connection.close()
+
+ def close(self) -> None:
+ with self.refresh_lock:
+ with self.lock:
+ self.closed = True
+ connections = set(self.connections.values()) | self.retired_connections
+ self.connections.clear()
+ self.retired_connections = {
+ connection for connection in connections
+ if self.connection_readers.get(connection, 0)
+ }
+ for connection in connections - self.retired_connections:
+ connection.close()
+
+ def search_source(self, source: str, query: str, exact: bool = False) -> set[int]:
+ with self.lock:
+ missing = source not in self.connections
+ if missing:
+ self.refresh()
+ with self._lease_source(source) as lease:
+ if lease is None:
+ return set()
+ connection, state = lease
+ return self._search_source_connection(connection, query, exact, state)
+
+ def _search_source_connection(self, connection: sqlite3.Connection, query: str, exact: bool, state: dict) -> set[int]:
+ version = state["tokenizer_version"]
+ has_content_fts = state["has_content_fts"]
+ has_doc_counts = state["has_doc_counts"]
+ if exact and has_content_fts:
+ normalized_query = normalize_text(query)
+ try:
+ if has_wildcard_query(normalized_query):
+ required_tokens = wildcard_required_tokens(normalized_query) if version == TOKENIZER_VERSION else []
+ candidates = self._search_postings(connection, has_doc_counts, required_tokens) if required_tokens else None
+ fixed_parts = [part for part in re.split(r"[*?]+", normalized_query) if len(part) >= 3]
+ if fixed_parts:
+ match_query = " AND ".join(f'"{part.replace(chr(34), chr(34) * 2)}"' for part in fixed_parts)
+ fts_candidates = {
+ int(row[0]) for row in connection.execute(
+ "SELECT rowid FROM content_fts WHERE content_fts MATCH ?", (match_query,)
+ )
+ }
+ candidates = fts_candidates if candidates is None else candidates & fts_candidates
+ elif candidates is None:
+ candidates = {
+ int(row[0]) for row in connection.execute("SELECT rowid FROM content_fts")
+ }
+ return self.verify_wildcard_content(connection, candidates, query, fts_filtered=True)
+ if len(normalized_query) >= 3:
+ phrase = normalized_query.replace('"', '""')
+ rows = connection.execute(
+ "SELECT rowid FROM content_fts WHERE content_fts MATCH ?", (f'"{phrase}"',)
+ )
+ return {int(row[0]) for row in rows}
+ tokens = query_tokens(query) if version == TOKENIZER_VERSION else query_terms(query)
+ if version == TOKENIZER_VERSION:
+ tokens = [*tokens, *literal_query_tokens(query)]
+ # Short non-CJK literals (Latin substrings or punctuation-separated
+ # characters) are not provable from whole-token/unigram postings, so
+ # verify them against the normalized FTS content directly.
+ if not tokens or not re.fullmatch(r"[\u4e00-\u9fff\u3400-\u4dbf]+", normalized_query):
+ rows = connection.execute(
+ "SELECT rowid FROM content_fts WHERE instr(content, ?) > 0", (normalized_query,)
+ )
+ return {int(row[0]) for row in rows}
+ candidates = self._search_postings(connection, has_doc_counts, tokens)
+ return candidates
+ except sqlite3.OperationalError:
+ pass
+ if exact and version == TOKENIZER_VERSION and "*" not in query and "?" not in query:
+ tokens = [*query_tokens(query), *literal_query_tokens(query)]
+ else:
+ tokens = query_tokens(query) if version == TOKENIZER_VERSION else query_terms(query)
+ if not tokens:
return set()
- matches = resolve_postings(encoded, ngram_posting_buffers.get(width, b""))
- candidates = matches if candidates is None else candidates & matches
+ matched = None
+ token_hashes = [hashlib.sha256(token.encode("utf-8")).digest()[:16] for token in tokens]
+ if has_doc_counts:
+ placeholders = ",".join("?" for _ in token_hashes)
+ rows = connection.execute(
+ f"SELECT token_hash, docs, doc_count FROM postings WHERE token_hash IN ({placeholders})",
+ token_hashes,
+ ).fetchall()
+ if len(rows) != len(token_hashes):
+ return set()
+ posting_rows = [(docs, doc_count) for _token_hash, docs, doc_count in rows]
+ posting_rows.sort(key=lambda item: item[1])
+ else:
+ posting_rows = []
+ for token_hash in token_hashes:
+ row = connection.execute("SELECT docs FROM postings WHERE token_hash = ?", (token_hash,)).fetchone()
+ if row is None:
+ return set()
+ posting_rows.append((row[0], None))
+ for docs, _doc_count in posting_rows:
+ token_matches = decode_doc_numbers(docs, matched)
+ matched = token_matches if matched is None else matched & token_matches
+ if not matched:
+ return set()
+ result = matched or set()
+ if exact and has_content_fts and has_wildcard_query(query):
+ result = self.verify_wildcard_content(connection, result, query)
+ return result
+
+ def _search_postings(self, connection: sqlite3.Connection, has_doc_counts: bool, tokens: list[str]) -> set[int]:
+ if not tokens:
+ return set()
+ matched = None
+ token_hashes = [hashlib.sha256(token.encode("utf-8")).digest()[:16] for token in tokens]
+ if has_doc_counts:
+ placeholders = ",".join("?" for _ in token_hashes)
+ rows = connection.execute(
+ f"SELECT token_hash, docs, doc_count FROM postings WHERE token_hash IN ({placeholders})",
+ token_hashes,
+ ).fetchall()
+ if len(rows) != len(token_hashes):
+ return set()
+ posting_rows = [(docs, doc_count) for _token_hash, docs, doc_count in rows]
+ posting_rows.sort(key=lambda item: item[1])
+ else:
+ posting_rows = []
+ for token_hash in token_hashes:
+ row = connection.execute("SELECT docs FROM postings WHERE token_hash = ?", (token_hash,)).fetchone()
+ if row is None:
+ return set()
+ posting_rows.append((row[0], None))
+ for docs, _doc_count in posting_rows:
+ token_matches = decode_doc_numbers(docs, matched)
+ matched = token_matches if matched is None else matched & token_matches
+ if not matched:
+ return set()
+ return matched or set()
+
+ def verify_wildcard_content(
+ self,
+ connection: sqlite3.Connection,
+ candidates: set[int],
+ query: str,
+ fts_filtered: bool = False,
+ ) -> set[int]:
if not candidates:
return set()
- if repos:
- allowed = set()
- for repo in repos:
- allowed.update(repo_records_map.get(repo, []))
- candidates &= allowed
- return candidates or set()
-INITIAL_PAGE_SIZE = 100
-SIDEBAR_DATA_PATH = Path("data/sidebar")
-RECORD_KEY_MAP = {
- "r": "Repo",
- "f": "File",
- "e": "Extension",
- "d": "Folder",
- "s": "Size",
- "t": "HasTxt",
-}
-TREE_KEY_MAP = {
- "n": "name",
- "c": "count",
- "df": "hasDirectFiles",
- "ch": "children",
-}
-BROWSER_ENTRY_KEY_MAP = {
- "d": "folders",
- "f": "files",
-}
-BROWSER_FOLDER_ITEM_KEY_MAP = {
- "n": "name",
- "c": "count",
-}
-BROWSER_FILE_ITEM_KEY_MAP = {
- "n": "name",
- "e": "ext",
- "t": "hasTxt",
- "s": "size",
-}
-
-def build_record_relative_path(rec: dict) -> str:
- file_name = rec.get("File", "")
- extension = rec.get("Extension", "")
- full_name = f"{file_name}.{extension}" if extension else file_name
- folders = rec.get("Folder", []) or []
- return posixpath.join(*folders, full_name) if folders else full_name
-
-
-def build_reader_asset_key(rec: dict) -> str:
- return f"{rec.get('Repo', '')}\0{build_record_relative_path(rec)}"
-
-def build_record_link(rec: dict) -> str:
- repo = rec.get("Repo", "")
- rel_path = build_record_relative_path(rec)
- return f"https://huggingface.co/datasets/{repo}/resolve/main/{quote(rel_path, safe='/')}"
-
-def build_record_path_url(rec: dict) -> str:
- repo = rec.get("Repo", "")
- rel_path = build_record_relative_path(rec)
- return f"https://huggingface.co/datasets/{repo}/blob/main/{quote(rel_path, safe='/')}"
-
-def build_txt_relative_path(rec: dict) -> str:
- rel_path = build_record_relative_path(rec)
- if "." in rel_path.rsplit("/", 1)[-1]:
- rel_path = rel_path.rsplit(".", 1)[0]
- return f"{rel_path}.txt"
-
-def is_allowed_download_host(hostname: str | None) -> bool:
- if not hostname:
- return False
- return hostname in ALLOWED_DOWNLOAD_HOSTS or hostname.endswith(ALLOWED_DOWNLOAD_HOST_SUFFIXES)
-
-def validate_download_url(url: str) -> str:
- parsed = urlparse(url)
- if (
- parsed.scheme != "https"
- or not is_allowed_download_host(parsed.hostname)
- or parsed.port not in (None, 443)
- or parsed.username is not None
- or parsed.password is not None
- ):
- raise ValueError(f"不允许的下载跳转地址: {parsed.hostname or 'unknown'}")
- return url
-
-def normalize_download_url(url: str) -> str:
- parsed = urlparse(url)
- if parsed.hostname == "hf-mirror.com":
- return parsed._replace(netloc="huggingface.co").geturl()
- return url
-
-def validate_voiceofml_source_url(url: str) -> str:
- url = normalize_download_url(url)
- validate_download_url(url)
- parsed = urlparse(url)
- if parsed.hostname != "huggingface.co" or not parsed.path.startswith("/datasets/VoiceOfML/"):
- raise ValueError("只允许 VoiceOfML 数据集文件")
- return url
-
-
-READER_ASSET_SOURCE_RE = re.compile(
- r"^/datasets/vomebook/Reader-Assets/resolve/main/objects/[0-9a-f]{2}/[0-9a-f]{64}/(?:[a-z0-9-]+/)?(?:document\.pdf|book\.epub|document\.docx|document\.html)$"
-)
-VOICEOFML_READER_SOURCE_RE = re.compile(r"^/datasets/VoiceOfML/[A-Za-z0-9._-]+/(?:resolve|raw)/main/.+$")
-
-
-def validate_reader_source_url(url: str) -> str:
- url = normalize_download_url(url)
- validate_download_url(url)
- parsed = urlparse(url)
- if parsed.hostname != "huggingface.co" or parsed.fragment or parsed.query:
- raise ValueError("不允许的阅读来源")
- decoded_path = unquote(parsed.path)
- if "\\" in decoded_path or "\x00" in decoded_path or any(part in (".", "..") for part in decoded_path.split("/")):
- raise ValueError("不允许的阅读来源路径")
- if VOICEOFML_READER_SOURCE_RE.fullmatch(decoded_path) or READER_ASSET_SOURCE_RE.fullmatch(decoded_path):
- return url
- raise ValueError("不允许的阅读来源")
-
-
-async def open_download_response(
- session,
- url: str,
- timeout: aiohttp.ClientTimeout,
- *,
- method: str = "GET",
- request_headers: Optional[dict[str, str]] = None,
-):
- method = method.upper()
- if method not in ("GET", "HEAD"):
- raise ValueError("不支持的上游请求方法")
- current_url = validate_download_url(url)
- headers = {"Accept-Encoding": "identity"}
- headers.update(request_headers or {})
- for _ in range(MAX_REDIRECTS + 1):
- request_method = getattr(session, "request", None)
- if request_method is not None:
- response = await request_method(
- method,
- current_url,
- allow_redirects=False,
- timeout=timeout,
- headers=headers,
+ normalized_query = normalize_text(query)
+ fixed_parts = [part for part in re.split(r"[*?]+", normalized_query) if len(part) >= 3]
+ if fixed_parts and not fts_filtered:
+ match_query = " AND ".join(f'"{part.replace(chr(34), chr(34) * 2)}"' for part in fixed_parts)
+ try:
+ fts_candidates = {
+ int(row[0])
+ for row in connection.execute(
+ "SELECT rowid FROM content_fts WHERE content_fts MATCH ?",
+ (match_query,),
+ )
+ }
+ candidates &= fts_candidates
+ except sqlite3.OperationalError:
+ pass
+ glob_pattern = "*" + normalized_query.replace("[", "[[]") + "*"
+ matched: set[int] = set()
+ ordered = sorted(candidates)
+ for offset in range(0, len(ordered), 500):
+ batch = ordered[offset:offset + 500]
+ placeholders = ",".join("?" for _ in batch)
+ rows = connection.execute(
+ f"SELECT rowid FROM content_fts WHERE rowid IN ({placeholders}) AND (content GLOB ?) = 1",
+ [*batch, glob_pattern],
)
+ matched.update(int(row[0]) for row in rows)
+ return matched
+
+ def summaries(self, doc_ids: list[str]) -> dict[str, str]:
+ grouped = {}
+ for doc_id in doc_ids:
+ grouped.setdefault(doc_id.split(":", 1)[0], []).append(doc_id)
+ with self.lock:
+ missing = any(source not in self.connections for source in grouped)
+ if missing:
+ self.refresh()
+ summaries = {}
+ for source, source_doc_ids in grouped.items():
+ with self._lease_source(source) as lease:
+ if lease is None:
+ continue
+ connection, _state = lease
+ for offset in range(0, len(source_doc_ids), 500):
+ batch = source_doc_ids[offset:offset + 500]
+ rows = connection.execute(
+ f"SELECT doc_id, summary FROM documents WHERE doc_id IN ({','.join('?' for _ in batch)})",
+ batch,
+ )
+ summaries.update(rows)
+ return summaries
+
+ def matched_snippets(self, doc_ids: list[str], query: str, exact: bool = False, timings=None) -> dict[str, dict]:
+ normalized_query = normalize_text(query.strip())
+ # A fixed wildcard fragment can occur long before the actual match.
+ # Let the original-text snippet endpoint locate the verified span.
+ if exact and has_wildcard_query(normalized_query):
+ return {}
+ if exact:
+ needle = normalized_query
else:
- response = await session.get(
- current_url,
- allow_redirects=False,
- timeout=timeout,
- headers=headers,
- )
- if response.status not in (301, 302, 303, 307, 308):
- return response
- location = response.headers.get("Location")
- response.release()
- if not location:
- raise ValueError("上游重定向缺少地址")
- current_url = validate_download_url(urljoin(current_url, location))
- raise ValueError("上游重定向次数过多")
-
-def sanitize_archive_name(value) -> str:
- name = str(value or "file").replace("\\", "/").rsplit("/", 1)[-1]
- name = "".join(ch for ch in name if ord(ch) >= 32 and ch != "\x7f").strip()
- if not name or name in (".", ".."):
- name = "file"
- return name[:MAX_ARCHIVE_NAME_LENGTH]
-
-def build_browser_file_name(file_item: dict) -> str:
- name = str(file_item.get("name", "") or "file")
- ext = str(file_item.get("ext", "") or "")
- if not ext:
- return name
- if name.lower().endswith(f".{ext.lower()}"):
- return name
- return f"{name}.{ext}"
-
-def enrich_browser_entry(repo: str, path: str, entry: dict) -> dict:
- decoded = dict(entry or {})
- decoded["current_path"] = decoded.get("current_path", path)
- folders = []
- for folder in decoded.get("folders", []) or []:
- item = dict(folder)
- if "path" not in item:
- name = item.get("name", "")
- item["path"] = f"{path}/{name}" if path else name
- folders.append(item)
- files = []
- for file_item in decoded.get("files", []) or []:
- item = dict(file_item)
- full_name = build_browser_file_name(item)
- rel_path = f"{path}/{full_name}" if path else full_name
- item["link"] = item.get("link") or f"https://huggingface.co/datasets/{repo}/resolve/main/{quote(rel_path, safe='/')}"
- item["path"] = item.get("path") or f"https://huggingface.co/datasets/{repo}/blob/main/{quote(rel_path, safe='/')}"
- files.append(item)
- decoded["folders"] = folders
- decoded["files"] = files
- return decoded
-
-def decode_search_payload(data) -> list[dict]:
- if not isinstance(data, dict):
- return []
- repos = data.get("rp", []) or []
- folders = data.get("fd", []) or []
- records = []
- for item in data.get("rc", []) or []:
- if not isinstance(item, list) or len(item) < 6:
- continue
- repo = repos[item[0]] if isinstance(item[0], int) and 0 <= item[0] < len(repos) else ""
- folder = folders[item[3]] if isinstance(item[3], int) and 0 <= item[3] < len(folders) else []
- records.append({
- "Repo": repo,
- "File": item[1],
- "Extension": item[2],
- "Folder": folder,
- "Size": item[4],
- "HasTxt": bool(item[5]),
- })
- return records
-
-def decode_tree_node(node: dict, parent_path: str = "", is_root: bool = False) -> dict:
- name = node.get("n", "")
- path = "" if is_root else (f"{parent_path}/{name}" if parent_path else name)
- children = [decode_tree_node(child, path, False) for child in node.get("ch", [])]
- decoded = {
- "name": name,
- "path": path,
- "count": node.get("c", 0),
- "hasDirectFiles": node.get("df", False),
- "hasChildren": len(children) > 0,
- "showSelfToggle": bool(path and node.get("df", False) and len(children) > 0),
- "children": children,
- }
- if is_root:
- decoded["isRoot"] = True
- return decoded
-
-def decode_folder_tree_data(data: dict) -> dict:
- decoded = {}
- for repo, nodes in (data or {}).items():
- decoded[repo] = [decode_tree_node(node, "", True) for node in nodes]
- return decoded
+ needle = next(iter(query_tokens(query)), "")
+ if not needle:
+ return {}
+ grouped = {}
+ for doc_id in doc_ids:
+ source, _, number = doc_id.partition(":")
+ if number.isdigit():
+ grouped.setdefault(source, []).append((doc_id, int(number)))
+ with self.lock:
+ missing = any(source not in self.connections for source in grouped)
+ if missing:
+ self.refresh()
+ snippets = {}
+ for source, source_docs in grouped.items():
+ with self._lease_source(source) as lease:
+ if lease is None:
+ continue
+ connection, state = lease
+ if not state["has_content_fts"]:
+ continue
+ self._matched_snippets_source(
+ snippets, connection, state, source_docs, needle, query, exact, timings,
+ )
+ return snippets
+
+ def _matched_snippets_source(self, snippets, connection, state, source_docs, needle, query, exact, timings) -> None:
+ doc_id_by_number = {number: doc_id for doc_id, number in source_docs}
+ numbers = list(doc_id_by_number)
+ anchored_numbers = set()
+ # Exact snippets retain the established literal path. Normal snippets can seek
+ # directly to a token anchor without scanning FTS content with instr().
+ if not exact and state["has_snippet_anchors"]:
+ token_hash = hashlib.sha256(needle.encode("utf-8")).digest()[:16]
+ for offset in range(0, len(numbers), 500):
+ batch = numbers[offset:offset + 500]
+ placeholders = ",".join("?" for _ in batch)
+ try:
+ anchor_started = time.perf_counter()
+ rows = connection.execute(
+ f"""
+ SELECT content_fts.rowid, snippet_anchors.char_offset, length(content),
+ substr(content, max(snippet_anchors.char_offset - 219, 1), 440 + length(?))
+ FROM content_fts JOIN snippet_anchors ON snippet_anchors.doc_number = content_fts.rowid
+ WHERE snippet_anchors.token_hash = ? AND content_fts.rowid IN ({placeholders})
+ """,
+ [needle, token_hash, *batch],
+ ).fetchall()
+ if timings is not None:
+ timings["snippet_anchor_sql"] = timings.get("snippet_anchor_sql", 0) + (time.perf_counter() - anchor_started) * 1000
+ for number, char_offset, content_length, text in rows:
+ payload_started = time.perf_counter()
+ payload = build_normalized_snippet_payload(text or "", query)
+ if timings is not None:
+ timings["snippet_payload"] = timings.get("snippet_payload", 0) + (time.perf_counter() - payload_started) * 1000
+ if not payload["highlights"]:
+ continue
+ extracted_start = max(int(char_offset) - 219, 1)
+ if extracted_start > 1 and not payload["snippet"].startswith("..."):
+ payload["snippet"] = "..." + payload["snippet"]
+ if extracted_start + len(text or "") <= int(content_length or 0) and not payload["snippet"].endswith("..."):
+ payload["snippet"] += "..."
+ snippets[doc_id_by_number[int(number)]] = payload
+ anchored_numbers.add(int(number))
+ except sqlite3.OperationalError:
+ break
+ numbers = [number for number in numbers if number not in anchored_numbers]
+ for offset in range(0, len(numbers), 500):
+ batch = numbers[offset:offset + 500]
+ placeholders = ",".join("?" for _ in batch)
+ try:
+ fallback_started = time.perf_counter()
+ rows = connection.execute(
+ f"""
+ SELECT rowid, instr(content, ?) AS match_position, length(content),
+ substr(content, max(instr(content, ?) - 220, 1), 440 + length(?))
+ FROM content_fts
+ WHERE rowid IN ({placeholders})
+ """,
+ [needle, needle, needle, *batch],
+ ).fetchall()
+ if timings is not None:
+ timings["snippet_fallback_sql"] = timings.get("snippet_fallback_sql", 0) + (time.perf_counter() - fallback_started) * 1000
+ for number, match_position, content_length, text in rows:
+ payload_started = time.perf_counter()
+ extracted_start = max(int(match_position or 0) - 220, 1)
+ if exact and not has_wildcard_query(query):
+ payload = build_normalized_exact_snippet_payload(text or "", query)
+ if not payload["highlights"]:
+ payload = build_snippet_payload(text or "", query, exact=True)
+ else:
+ payload = build_snippet_payload(text or "", query, exact=exact)
+ if not payload["highlights"]:
+ payload = build_snippet_payload(text or "", needle, exact=True)
+ if timings is not None:
+ timings["snippet_payload"] = timings.get("snippet_payload", 0) + (time.perf_counter() - payload_started) * 1000
+ if payload["snippet"] and extracted_start > 1 and not payload["snippet"].startswith("..."):
+ payload["snippet"] = "..." + payload["snippet"]
+ if payload["snippet"] and extracted_start + len(text or "") <= int(content_length or 0) and not payload["snippet"].endswith("..."):
+ payload["snippet"] += "..."
+ snippets[doc_id_by_number[int(number)]] = payload
+ except sqlite3.OperationalError:
+ continue
+fulltext_databases: Optional[FulltextDatabases] = None
-def decode_browser_entry(entry: dict, current_path: str = "") -> dict:
- return {
- "folders": [
- {
- "name": item.get("n", ""),
- "path": f"{current_path}/{item.get('n', '')}" if current_path else item.get("n", ""),
- "count": item.get("c", 0),
- }
- for item in entry.get("d", [])
- ],
- "files": [
- {
- "name": item.get("n", ""),
- "ext": item.get("e", ""),
- "hasTxt": item.get("t", False),
- "size": item.get("s", ""),
- }
- for item in entry.get("f", [])
- ],
- "current_path": current_path,
- }
+def tokenize(text: str) -> list[str]:
+ return query_terms(text)
-def decode_folder_browser_data(data: dict) -> dict:
- decoded = {}
- for repo, repo_browser in (data or {}).items():
- decoded[repo] = {}
- for path, entry in (repo_browser or {}).items():
- decoded[repo][path] = decode_browser_entry(entry, path)
- return decoded
+def compile_exact_query(query: str) -> tuple[str, re.Pattern | None]:
+ normalized_query = normalize_text(query)
+ if has_wildcard_query(normalized_query):
+ expression = re.escape(normalized_query).replace(r"\*", ".*?").replace(r"\?", ".")
+ return normalized_query, re.compile(expression, re.DOTALL)
+ return normalized_query, None
-def tokenize(text: str) -> list[str]:
- tokens = []
- text_lower = text.lower()
- alpha_tokens = re.findall(r"[a-z0-9]+", text_lower)
- tokens.extend(alpha_tokens)
- chinese_text = re.sub(r"[a-z0-9\s]+", " ", text_lower)
- chinese_text = re.sub(r"[^\u4e00-\u9fff\u3400-\u4dbf\s]+", " ", chinese_text)
- chinese_tokens = jieba.lcut(chinese_text)
- tokens.extend([t.strip() for t in chinese_tokens if t.strip()])
- chinese_token_set = set(chinese_tokens)
- for ch in text_lower:
- if "\u4e00" <= ch <= "\u9fff" or "\u3400" <= ch <= "\u4dbf":
- if ch not in chinese_token_set:
- tokens.append(ch)
- return list(set(tokens))
+def matches_normalized_exact_query(text: str, normalized_query: str, pattern: re.Pattern | None = None) -> bool:
+ return pattern.search(text) is not None if pattern is not None else normalized_query in text
def matches_exact_query(text: str, query: str) -> bool:
- if "*" in query or "?" in query:
- pattern = re.escape(query).replace(r"\*", ".*").replace(r"\?", ".")
- return re.search(pattern, text, re.IGNORECASE) is not None
- return query.lower() in text.lower()
-
-def edit_distance(s1: str, s2: str, max_dist: int = 2) -> int:
- if abs(len(s1) - len(s2)) > max_dist:
- return 999
- prev = list(range(len(s2) + 1))
- for i, c1 in enumerate(s1):
- curr = [i + 1]
- for j, c2 in enumerate(s2):
- cost = 0 if c1 == c2 else 1
- curr.append(min(prev[j + 1] + 1, curr[j] + 1, prev[j] + cost))
- if min(curr) > max_dist:
- return 999
- prev = curr
- return prev[-1]
-
-def could_be_fuzzy(tok: str, word: str, max_dist: int = 2) -> bool:
- if abs(len(tok) - len(word)) > max_dist:
- return False
- if len(tok) < 4:
- return True
- counts: dict[str, int] = {}
- for ch in tok:
- counts[ch] = counts.get(ch, 0) + 1
- diff = 0
- for ch in word:
- if counts.get(ch, 0) > 0:
- counts[ch] -= 1
+ normalized_query, pattern = compile_exact_query(query)
+ return matches_normalized_exact_query(normalize_text(text), normalized_query, pattern)
+
+def has_wildcard_query(query: str) -> bool:
+ return "*" in query or "?" in query
+
+def normalized_text_map(text: str) -> tuple[str, list[tuple[int, int]]]:
+ normalized = []
+ offsets = []
+ for index, character in enumerate(text):
+ value = normalize_text(character)
+ normalized.append(value)
+ offsets.extend((index, index + 1) for _ in value)
+ return "".join(normalized), offsets
+
+def snippet_match_spans(text: str, query: str, exact: bool = False) -> list[tuple[int, int]]:
+ normalized_text, offsets = normalized_text_map(text)
+ normalized_query = normalize_text(query.strip())
+ if not normalized_text or not normalized_query or not offsets:
+ return []
+ normalized_spans = []
+ if exact:
+ if has_wildcard_query(normalized_query):
+ pattern = re.escape(normalized_query).replace(r"\*", ".*?").replace(r"\?", ".")
+ match = re.search(pattern, normalized_text, re.DOTALL)
+ if match:
+ fixed_parts = [part for part in re.split(r"[*?]+", normalized_query) if part]
+ for part in fixed_parts:
+ index = normalized_text.find(part, match.start(), match.end())
+ if index != -1:
+ normalized_spans.append((index, index + len(part)))
+ if not normalized_spans and match.end() > match.start():
+ normalized_spans.append(match.span())
else:
- diff += 1
- if diff > 2 * max_dist:
- return False
- for remaining in counts.values():
- diff += remaining
- if diff > 2 * max_dist:
- return False
- return True
-
-def build_word_index():
- global word_index, vocab_by_len
- word_index = {}
- for idx, rec in enumerate(records):
- file_name = rec.get("File", "")
- folders = rec.get("Folder", [])
- for tok in tokenize(" ".join([file_name] + folders)):
- word_index.setdefault(tok, set()).add(idx)
- vocab_by_len = {}
- for word in word_index:
- vocab_by_len.setdefault(len(word), []).append(word)
-
-
-def build_vocab_by_len():
- global vocab_by_len
- vocab_by_len = {}
- for word in word_index:
- vocab_by_len.setdefault(len(word), []).append(word)
-
+ index = normalized_text.find(normalized_query)
+ if index != -1:
+ normalized_spans.append((index, index + len(normalized_query)))
+ else:
+ for token in query_tokens(query):
+ normalized_token = normalize_text(token)
+ index = normalized_text.find(normalized_token)
+ if index != -1:
+ normalized_spans.append((index, index + len(normalized_token)))
+ return [
+ (offsets[start][0], offsets[end - 1][1])
+ for start, end in normalized_spans
+ if start < end and end <= len(offsets)
+ ]
-def build_indexes(build_words=True):
- global _data_generation, word_index, folder_index, extension_counts, repo_counts, repo_list, extension_list, repo_extension_counts, repo_records_map, sorted_by_name, sorted_by_size, repo_sorted_by_name, repo_sorted_by_size, txt_record_indices, repo_txt_record_indices, reader_record_indices, repo_reader_record_indices, reader_record_key_indices
- _data_generation += 1
+def build_snippet_payload(text: str, query: str, window: int = 110, exact: bool = False) -> dict:
+ query = (query or "").strip()
+ if not text or not query:
+ return {"snippet": "", "highlights": []}
+ spans = snippet_match_spans(text, query, exact)
+ if not spans:
+ condensed = re.sub(r"\s+", " ", text).strip()
+ return {"snippet": condensed[: window * 2], "highlights": []}
+ match_start, match_end = min(spans, key=lambda span: span[0])
+ start = max(0, match_start - window)
+ end = min(len(text), match_end + window)
+ snippet = re.sub(r"\s+", " ", text[start:end]).strip()
+ if start > 0:
+ snippet = "..." + snippet
+ if end < len(text):
+ snippet = snippet + "..."
+ highlights = list(dict.fromkeys(
+ re.sub(r"\s+", " ", text[span_start:span_end]).strip()
+ for span_start, span_end in spans
+ if span_start < end and span_end > start
+ ))
+ return {"snippet": snippet, "highlights": [value for value in highlights if value]}
+
+def build_normalized_snippet_payload(text: str, query: str, window: int = 110) -> dict:
+ """Build a normal-search snippet when text already came from normalized content_fts."""
+ query = (query or "").strip()
+ if not text or not query:
+ return {"snippet": "", "highlights": []}
+ spans = []
+ for token in query_tokens(query):
+ token = normalize_text(token)
+ index = text.find(token)
+ if index != -1:
+ spans.append((index, index + len(token)))
+ if not spans:
+ return {"snippet": re.sub(r"\s+", " ", text).strip()[: window * 2], "highlights": []}
+ match_start, match_end = min(spans, key=lambda span: span[0])
+ start = max(0, match_start - window)
+ end = min(len(text), match_end + window)
+ snippet = re.sub(r"\s+", " ", text[start:end]).strip()
+ if start > 0:
+ snippet = "..." + snippet
+ if end < len(text):
+ snippet += "..."
+ highlights = list(dict.fromkeys(
+ re.sub(r"\s+", " ", text[span_start:span_end]).strip()
+ for span_start, span_end in spans
+ if span_start < end and span_end > start
+ ))
+ return {"snippet": snippet, "highlights": [value for value in highlights if value]}
+
+def build_normalized_exact_snippet_payload(text: str, query: str, window: int = 110) -> dict:
+ needle = normalize_text(query.strip())
+ index = text.find(needle)
+ if index == -1:
+ return {"snippet": re.sub(r"\s+", " ", text).strip()[: window * 2], "highlights": []}
+ start = max(0, index - window)
+ end = min(len(text), index + len(needle) + window)
+ snippet = re.sub(r"\s+", " ", text[start:end]).strip()
+ if start > 0:
+ snippet = "..." + snippet
+ if end < len(text):
+ snippet += "..."
+ return {"snippet": snippet, "highlights": [needle]}
+
+def build_first_match_snippet(text: str, query: str, window: int = 110, exact: bool = False) -> str:
+ return build_snippet_payload(text, query, window, exact)["snippet"]
+
+def load_json_gz(path: Path):
+ return json.loads(gzip.decompress(path.read_bytes()).decode("utf-8"))
+
+def build_indexes() -> None:
+ global word_index, latin_all_index, latin_file_index, latin_vocabulary_all, source_records_map, extension_counts, source_extension_counts, name_order, size_order, source_name_order, source_size_order
word_index = {}
- folder_index = {}
+ latin_all_index = {}
+ latin_file_index = {}
+ latin_vocabulary_all = set()
+ source_records_map = {}
extension_counts = {}
- repo_extension_counts = {}
- repo_counts = {}
- repo_records_map = {}
- repo_sorted_by_name = {}
- repo_sorted_by_size = {}
- txt_record_indices = []
- repo_txt_record_indices = {}
- reader_record_indices = []
- repo_reader_record_indices = {}
- reader_record_key_indices = {}
+ source_extension_counts = {}
for idx, rec in enumerate(records):
- repo = rec.get("Repo", "")
- repo_counts[repo] = repo_counts.get(repo, 0) + 1
- ext = (rec.get("Extension") or "").lower()
- if ext:
- extension_counts[ext] = extension_counts.get(ext, 0) + 1
- repo_ext = repo_extension_counts.setdefault(repo, {})
- repo_ext[ext] = repo_ext.get(ext, 0) + 1
- file_name = rec.get("File", "")
- folders = rec.get("Folder", [])
- if repo not in folder_index:
- folder_index[repo] = {}
- for depth in range(len(folders) + 1):
- folder_path = "/".join(folders[:depth]) if depth > 0 else ""
- folder_index[repo].setdefault(folder_path, []).append(idx)
- repo_records_map.setdefault(repo, []).append(idx)
- if rec.get("HasTxt"):
- txt_record_indices.append(idx)
- repo_txt_record_indices.setdefault(repo, []).append(idx)
- if ext in READER_EXTENSIONS:
- reader_record_indices.append(idx)
- repo_reader_record_indices.setdefault(repo, []).append(idx)
- reader_record_key_indices[build_reader_asset_key(rec)] = idx
- folders_lower = [f.lower() for f in (folders or [])]
- rec["_file_lower"] = file_name.lower()
- rec["_folders_lower"] = folders_lower
- rec["_folder_text"] = " ".join(folders_lower)
- rec["_folder_path_lower"] = "/".join(folders_lower)
- rec["_folder_path"] = "/".join(folders)
- rec["_ext_lower"] = ext
- rec["_repo_lower"] = repo.lower()
- repo_list = sorted(
- [{"name": r, "count": c} for r, c in repo_counts.items()],
- key=lambda x: x["name"],
+ source = rec["source"]
+ extension = rec["extension"]
+ source_records_map.setdefault(source, []).append(idx)
+ extension_counts[extension] = extension_counts.get(extension, 0) + 1
+ source_counts_for_ext = source_extension_counts.setdefault(source, {})
+ source_counts_for_ext[extension] = source_counts_for_ext.get(extension, 0) + 1
+ file_text = " ".join([rec["display_name"], rec["source_name"]])
+ path_text = rec["display_rel_path"]
+ tokens = index_tokens(f"{file_text} {path_text}")
+ rec["_file_search_text"] = normalize_text(file_text)
+ rec["_path_search_text"] = normalize_text(path_text)
+ rec["_search_text"] = normalize_text(f"{file_text} {path_text}")
+ rec["_rank_path_key"] = rec["display_rel_path"].lower()
+ for token in tokens:
+ word_index.setdefault(token, set()).add(idx)
+ if LATIN_RE.fullmatch(token):
+ latin_all_index.setdefault(token, set()).add(idx)
+ latin_vocabulary_all.add(token)
+ for token in TOKEN_RE.findall(rec["_file_search_text"]):
+ if LATIN_RE.fullmatch(token):
+ latin_file_index.setdefault(token, set()).add(idx)
+ name_order = sorted(
+ range(len(records)),
+ key=lambda idx: (records[idx]["display_rel_path"].lower(), records[idx]["doc_id"]),
)
- if build_words:
- build_word_index()
- else:
- word_index = {}
- global vocab_by_len
- vocab_by_len = {}
- extension_list = sorted(extension_counts.keys())
- all_indices = list(range(len(records)))
- sorted_by_name = sorted(all_indices, key=lambda i: (records[i].get("File") or "").lower())
- sorted_by_size = sorted(all_indices, key=lambda i: (
- -(records[i].get("Size") if isinstance(records[i].get("Size"), (int, float)) else 0),
- (records[i].get("File") or "").lower(),
- ))
- for repo, repo_indices in repo_records_map.items():
- repo_sorted_by_name[repo] = sorted(repo_indices, key=lambda i: (records[i].get("File") or "").lower())
- repo_sorted_by_size[repo] = sorted(repo_indices, key=lambda i: (
- -(records[i].get("Size") if isinstance(records[i].get("Size"), (int, float)) else 0),
- (records[i].get("File") or "").lower(),
- ))
-
-def load_data():
- global records, folder_tree_data, folder_browser_data, ngram_postings, ngram_posting_buffers, word_index
+ size_order = sorted(
+ range(len(records)),
+ key=lambda idx: (
+ -(records[idx].get("size") or 0),
+ records[idx]["display_rel_path"].lower(),
+ records[idx]["doc_id"],
+ ),
+ )
+ source_name_order = {source: [] for source in source_records_map}
+ source_size_order = {source: [] for source in source_records_map}
+ for idx in name_order:
+ source_name_order[records[idx]["source"]].append(idx)
+ for idx in size_order:
+ source_size_order[records[idx]["source"]].append(idx)
+
+def load_data() -> None:
+ global records, record_map, record_map_index, sources, source_counts, folder_tree_data, folder_browser_data
start = time.time()
- if DATA_PATH.exists():
- raw = gzip.decompress(DATA_PATH.read_bytes())
- records = decode_search_payload(json.loads(raw.decode("utf-8")))
- else:
- print("⚠ 数据文件不存在")
- records = []
- return
- if FOLDER_TREE_PATH.exists():
- folder_tree_data = decode_folder_tree_data(json.loads(gzip.decompress(FOLDER_TREE_PATH.read_bytes()).decode("utf-8")))
- else:
- folder_tree_data = {}
- if FOLDER_BROWSER_PATH.exists():
- folder_browser_data = decode_folder_browser_data(json.loads(gzip.decompress(FOLDER_BROWSER_PATH.read_bytes()).decode("utf-8")))
- else:
- folder_browser_data = {}
- print(f"📖 已加载 {len(records)} 条记录 ({time.time() - start:.2f}s)")
- build_indexes(build_words=False)
- digest = records_digest()
- word_index = load_word_index(WORD_INDEX_PATH, len(records), digest)
- if word_index:
- build_vocab_by_len()
- else:
- build_word_index()
- loaded_ngrams = {
- width: load_ngram_index(path, width, len(records), digest)
- for width, path in NGRAM_PATHS.items()
- }
- ngram_postings = {width: loaded[0] for width, loaded in loaded_ngrams.items()}
- ngram_posting_buffers = {width: loaded[1] for width, loaded in loaded_ngrams.items()}
- print(f"🧭 已加载词索引: {len(word_index)}")
- print(f"🔎 已加载 ngram 索引: 2={len(ngram_postings[2])}, 3={len(ngram_postings[3])}")
- build_initial_payloads()
- precompute_injected_html()
-
-def score_record(rec_idx, query_tokens, search_folders=True):
- rec = records[rec_idx]
+ payload = load_json_gz(DATA_PATH)
+ records = payload.get("records", [])
+ sources = payload.get("sources", [])
+ source_counts = {item["slug"]: item.get("count", 0) for item in sources}
+ record_map = {rec["doc_id"]: rec for rec in records}
+ record_map_index = {rec["doc_id"]: idx for idx, rec in enumerate(records)}
+ folder_tree_data = load_json_gz(FOLDER_TREE_PATH)
+ folder_browser_data = load_json_gz(FOLDER_BROWSER_PATH)
+ build_indexes()
+ print(f"loaded {len(records)} txt records in {time.time() - start:.2f}s")
+
+def score_record(idx: int, query_tokens: list[str], search_paths: bool = True) -> int:
score = 0
- file_name = rec.get("_file_lower", "")
- repo = rec.get("_repo_lower", "")
- folder_text = rec.get("_folder_text", "")
- for tok in query_tokens:
- if tok in file_name:
+ file_text = records[idx]["_file_search_text"]
+ path_text = records[idx]["_path_search_text"]
+ for token in query_tokens:
+ if token in file_text:
score += 3
- if search_folders and tok in folder_text:
+ if search_paths and token in path_text:
score += 2
- if tok in repo:
- score += 1
return score
-def apply_filters(indices, repos=None, extensions=None, folders=None, min_size=None, max_size=None, folder_match_mode="prefix"):
- extensions_lower = [e.lower() for e in extensions] if extensions else None
+def relevance_page(indices, query_tokens: list[str], search_paths: bool, start: int, page_size: int) -> list[int]:
+ limit = start + page_size
+ if limit <= 0:
+ return []
+ ranked = heapq.nsmallest(
+ limit,
+ indices,
+ key=lambda idx: (-score_record(idx, query_tokens, search_paths), records[idx]["_rank_path_key"], records[idx]["doc_id"]),
+ )
+ return ranked[start:limit]
+
+def ordered_page(indices, ordering: list[int], start: int, page_size: int) -> list[int]:
+ candidates = set(indices)
+ if not candidates or page_size <= 0:
+ return []
+ selected = []
+ matched = 0
+ for idx in ordering:
+ if idx not in candidates:
+ continue
+ if matched >= start:
+ selected.append(idx)
+ if len(selected) >= page_size:
+ break
+ matched += 1
+ return selected
+
+def apply_filters(indices: list[int], sources_filter=None, folders=None, min_size=None, max_size=None):
result = []
for idx in indices:
rec = records[idx]
- if repos and rec.get("Repo") not in repos:
+ if sources_filter is not None and rec["source"] not in sources_filter:
continue
- if extensions_lower:
- if rec.get("_ext_lower", "") not in extensions_lower:
- continue
if folders:
- rec_folders = rec.get("Folder", []) or []
- rec_path = rec.get("_folder_path", "/".join(rec_folders))
- matched = False
- for f in folders:
- f_clean = f.strip("/")
- if folder_match_mode == "exact":
- if rec_path == f_clean:
- matched = True
- break
- elif f_clean == "":
- matched = True
- break
- elif rec_path == f_clean or rec_path.startswith(f_clean + "/"):
- matched = True
- break
+ rel_dir = "/".join(rec.get("display_dirs", []))
+ matched = any(rel_dir == folder or rel_dir.startswith(folder + "/") for folder in folders)
if not matched:
continue
- size = rec.get("Size")
- if size and isinstance(size, (int, float)) and size > 0:
- if min_size is not None and size < min_size:
- continue
- if max_size is not None and size > max_size:
- continue
+ size = rec.get("size") or 0
+ if min_size is not None and size < min_size:
+ continue
+ if max_size is not None and size > max_size:
+ continue
result.append(idx)
return result
-def trim_results(raw):
- keys = ("Repo", "File", "Extension", "Link", "Path", "Folder", "Size", "HasTxt")
- trimmed = []
- for r in raw:
- item = {k: r.get(k, "" if k != "Folder" else []) for k in keys}
- item["Link"] = r.get("Link") or build_record_link(r)
- item["Path"] = r.get("Path") or build_record_path_url(r)
- trimmed.append(item)
- return trimmed
-
-def sort_key_name(idx: int):
- return (records[idx].get("File") or "").lower()
-
-def sort_key_size(idx: int):
- return (
- -(records[idx].get("Size") if isinstance(records[idx].get("Size"), (int, float)) else 0),
- (records[idx].get("File") or "").lower(),
- )
-
-def paginate_indices(indices: list[int], page: int, page_size: int) -> tuple[int, list[dict]]:
- total = len(indices)
- start = (page - 1) * page_size
- return total, [records[i] for i in indices[start:start + page_size]]
-
-def get_initial_indices(repos=None) -> list[int]:
- if repos:
- indices = []
- for repo in repos:
- indices.extend(repo_records_map.get(repo, []))
- return indices
- return list(range(len(records)))
-
-def build_response(results: list[dict], total: int, page: int, page_size: int) -> dict:
- return {
- "results": trim_results(results),
- "total": total,
- "page": page,
- "page_size": page_size,
- }
-
-def trim_initial_results(raw: list[dict]) -> list[dict]:
- keys = ("Repo", "File", "Extension", "Folder", "Size", "HasTxt")
- return [
- {key: record.get(key, [] if key == "Folder" else "") for key in keys}
- for record in raw
- ]
-
-def build_initial_payload(repo: str | None = None) -> dict:
- if repo:
- indices = repo_records_map.get(repo, [])
- result_records = [records[i] for i in indices[:INITIAL_PAGE_SIZE]]
- mode = "repo"
- total = len(indices)
- else:
- result_records = records[:INITIAL_PAGE_SIZE]
- mode = "global"
- total = len(records)
- return {
- "mode": mode,
- "repo": repo,
- "sort": "relevance",
- "page": 1,
- "page_size": INITIAL_PAGE_SIZE,
- "total": total,
- "results": trim_initial_results(result_records),
- }
+def has_filters(sources_filter=None, folders=None, min_size=None, max_size=None) -> bool:
+ return bool(sources_filter is not None or folders or min_size is not None or max_size is not None)
-def build_initial_payloads():
- global initial_payload_global, initial_payload_by_repo
- initial_payload_global = build_initial_payload(None)
- initial_payload_by_repo = {
- repo: build_initial_payload(repo)
- for repo in repo_records_map
- }
-def precompute_injected_html():
- global injected_html_cache
- index_path = Path("static/index.html")
- if not index_path.exists():
- injected_html_cache = {}
- return
- html = index_path.read_text(encoding="utf-8")
- injected_html_cache = {None: inject_initial_payload(html, None)}
- for repo in repo_records_map:
- injected_html_cache[repo] = inject_initial_payload(html, repo.split("/")[-1])
-
-def load_sidebar_payload(path: Path):
- if not path.exists() or not path.is_file():
+def empty_query_order(sources_filter=None, folders=None, min_size=None, max_size=None, sort="relevance") -> list[int] | range | None:
+ if folders or min_size is not None or max_size is not None:
return None
- try:
- return json.loads(path.read_text(encoding="utf-8"))
- except Exception:
+ if sources_filter is None:
+ if sort == "name":
+ return name_order
+ if sort == "size":
+ return size_order
+ if sort == "relevance":
+ return range(len(records))
return None
+ if len(sources_filter) != 1:
+ return None
+ source = sources_filter[0]
+ if sort == "name":
+ return source_name_order.get(source, [])
+ if sort == "size":
+ return source_size_order.get(source, [])
+ if sort == "relevance":
+ return source_records_map.get(source, [])
+ return None
-def build_global_bootstrap():
- sidebar = load_sidebar_payload(SIDEBAR_DATA_PATH / "global.json") or {}
+def trim_record(rec: dict) -> dict:
return {
- "generation": _data_generation,
- "repos": repo_list,
- "extensions": [{"name": name, "count": extension_counts[name]} for name in sorted(extension_counts)],
- "random_txt": {"available": bool(reader_record_indices), "count": len(reader_record_indices)},
- "sidebar": sidebar,
+ "doc_id": rec["doc_id"],
+ "Source": rec["source"],
+ "SourceName": rec["source_name"],
+ "File": rec["display_name"],
+ "Extension": rec["extension"],
+ "Folder": rec["display_dirs"],
+ "DisplayPath": rec["display_rel_path"],
+ "Size": rec["size"],
+ "HasTxt": True,
}
-def inject_initial_payload(html: str, repo_short: str | None = None) -> str:
- repo = f"VoiceOfML/{repo_short}" if repo_short else None
- payload = initial_payload_by_repo.get(repo) if repo else initial_payload_global
- if not payload:
- return html
- script = (
- '"
- )
- app_script = re.search(r'', html)
- if app_script:
- return html[:app_script.start()] + script + "\n" + app_script.group(0) + html[app_script.end():]
- return html.replace("", script + "\n")
-
-def apply_mixed_folder_filters(indices, self_folders=None, subtree_folders=None):
- self_set = {str(path).strip("/") for path in (self_folders or []) if str(path).strip("/")}
- subtree_set = {str(path).strip("/") for path in (subtree_folders or []) if str(path).strip("/")}
- if not self_set and not subtree_set:
- return indices
- result = []
- for idx in indices:
- rec = records[idx]
- rec_folders = rec.get("Folder", []) or []
- rec_path = "/".join(rec_folders)
- if rec_path in self_set:
- result.append(idx)
+def add_summaries(items: list[dict], query: str = "", matched_snippets: bool = False, content_indices: set[int] | None = None, exact: bool = False, timings=None) -> list[dict]:
+ if fulltext_databases is None or not items:
+ return items
+ summaries_started = time.perf_counter()
+ summaries = fulltext_databases.summaries([item["doc_id"] for item in items])
+ if timings is not None:
+ timings["summaries"] = (time.perf_counter() - summaries_started) * 1000
+ content_doc_ids = [
+ item["doc_id"]
+ for item in items
+ if content_indices is not None and record_map_index.get(item["doc_id"]) in content_indices
+ ]
+ snippets_started = time.perf_counter()
+ matched = fulltext_databases.matched_snippets(content_doc_ids, query, exact=exact, timings=timings) if matched_snippets and query else {}
+ if timings is not None and matched_snippets and query:
+ timings["snippets"] = (time.perf_counter() - snippets_started) * 1000
+ for item in items:
+ summary = summaries.get(item["doc_id"], "")
+ payload = matched.get(item["doc_id"])
+ item["snippet"] = payload["snippet"] if payload else summary
+ if payload:
+ item["snippet_highlights"] = payload["highlights"]
+ needs_matched_snippet = matched_snippets and item["doc_id"] in content_doc_ids
+ item["snippet_complete"] = bool(payload) or (not needs_matched_snippet and item["doc_id"] in summaries)
+ return items
+
+def latin_substring_candidates(term: str, search_paths: bool) -> set[int]:
+ index = latin_all_index if search_paths else latin_file_index
+ candidates: set[int] = set()
+ for token in latin_vocabulary_all:
+ if term in token:
+ candidates |= index.get(token, ())
+ return candidates
+
+def metadata_prefilter(terms: list[str], search_paths: bool) -> set[int] | None:
+ candidates: set[int] | None = None
+ for term in terms:
+ if LATIN_RE.fullmatch(term):
+ term_candidates = latin_substring_candidates(term, search_paths)
+ else:
+ term_candidates = None
+ for token in query_tokens(term):
+ indices = word_index.get(token)
+ if not indices:
+ term_candidates = set()
+ break
+ term_candidates = set(indices) if term_candidates is None else term_candidates & indices
+ if not term_candidates:
+ break
+ if term_candidates is None:
continue
- matched = False
- for depth in range(1, len(rec_folders) + 1):
- prefix = "/".join(rec_folders[:depth])
- if prefix in subtree_set:
- matched = True
- break
- if matched:
- result.append(idx)
- return result
-
-def search(q="", repos=None, extensions=None, folders=None, min_size=None, max_size=None, page=1, page_size=100, search_folders=True, sort="relevance", exact=False, folder_match_mode="prefix", folder_selfs=None, folder_subtrees=None):
- q = q.strip()
- key = (_data_generation, q,
- tuple(repos) if repos is not None else (),
- tuple(extensions) if extensions is not None else (),
- tuple(folders) if folders is not None else (),
- min_size, max_size, page, page_size, search_folders,
- sort, exact, folder_match_mode,
- tuple(folder_selfs) if folder_selfs is not None else (),
- tuple(folder_subtrees) if folder_subtrees is not None else ())
- now = time.monotonic()
- cached = _search_cache.get(key)
- if cached and now - cached[0] < SEARCH_CACHE_TTL_SECONDS:
- return cached[1]
- result = _search_uncached(q, repos, extensions, folders,
- min_size, max_size, page, page_size,
- search_folders, sort, exact,
- folder_match_mode, folder_selfs,
- folder_subtrees)
- _search_cache[key] = (now, result)
- if len(_search_cache) > 500:
- oldest_key = min(_search_cache, key=lambda k: _search_cache[k][0])
- del _search_cache[oldest_key]
- return result
+ candidates = term_candidates if candidates is None else candidates & term_candidates
+ if not candidates:
+ return set()
+ return candidates
+
+def metadata_matches(q: str, exact: bool, search_paths: bool) -> set[int]:
+ field = "_search_text" if search_paths else "_file_search_text"
+ if exact:
+ terms = query_terms(q)
+ normalized_query, pattern = compile_exact_query(q)
+ # CJK literals necessarily contain all of their indexed unigram/bigram
+ # tokens, so candidate intersection is lossless there. Latin substrings
+ # are resolved through the per-field Latin token index, and
+ # punctuation-only queries retain the full-range scan.
+ if has_wildcard_query(q):
+ candidate_set = None
+ for token in wildcard_required_tokens(normalized_query):
+ token_indices = word_index.get(token, set())
+ candidate_set = set(token_indices) if candidate_set is None else candidate_set & token_indices
+ if not candidate_set:
+ return set()
+ candidates = candidate_set if candidate_set is not None else range(len(records))
+ elif not terms:
+ candidates = range(len(records))
+ else:
+ candidates = metadata_prefilter(terms, search_paths) or set()
+ return {
+ idx for idx in candidates
+ if matches_normalized_exact_query(records[idx][field], normalized_query, pattern)
+ }
+ terms = query_terms(q)
+ if not terms:
+ return set()
+ candidates = metadata_prefilter(terms, search_paths)
+ if candidates is None:
+ candidates = range(len(records))
+ return {
+ idx for idx in candidates
+ if all(term in records[idx][field] for term in terms)
+ }
-def _search_uncached(q="", repos=None, extensions=None, folders=None, min_size=None, max_size=None, page=1, page_size=100, search_folders=True, sort="relevance", exact=False, folder_match_mode="prefix", folder_selfs=None, folder_subtrees=None):
+def search(q="", sources_filter=None, folders=None, min_size=None, max_size=None, page=1, page_size=100, sort="relevance", exact=False, search_paths=True, timings=None):
+ started = time.perf_counter()
q = q.strip()
-
- if exact and q:
- matched_indices = []
- if "*" in q or "?" in q:
- _exact_re = re.compile(re.escape(q).replace(r"\*", ".*").replace(r"\?", "."), re.IGNORECASE)
- _is_match = lambda t: _exact_re.search(t) is not None
- else:
- _q_lower = q.lower()
- _is_match = lambda t: _q_lower in t
- candidate_indices = get_initial_indices(repos)
- if "*" not in q and "?" not in q:
- candidates = literal_candidates(_q_lower, repos)
- if candidates is not None:
- candidate_indices = sorted(candidates)
- for idx in candidate_indices:
- rec = records[idx]
- file_name = rec.get("_file_lower", "")
- folder_path = rec.get("_folder_path_lower", "")
- repo_name = rec.get("_repo_lower", "")
- if _is_match(file_name) or _is_match(repo_name) or (search_folders and _is_match(folder_path)):
- matched_indices.append(idx)
- if folder_match_mode == "mixed":
- filtered = apply_mixed_folder_filters(matched_indices, folder_selfs, folder_subtrees)
- filtered = apply_filters(filtered, None, extensions, None, min_size, max_size)
- else:
- filtered = apply_filters(matched_indices, None, extensions, folders, min_size, max_size, folder_match_mode)
- if sort == "name":
- filtered.sort(key=sort_key_name)
- elif sort == "size":
- filtered.sort(key=sort_key_size)
- total, results = paginate_indices(filtered, page, page_size)
- return build_response(results, total, page, page_size)
+ start = (page - 1) * page_size
+ preordered = empty_query_order(sources_filter, folders, min_size, max_size, sort) if not q else None
+ if preordered is not None:
+ result_items = [trim_record(records[idx]) for idx in preordered[start:start + page_size]]
+ summaries_started = time.perf_counter()
+ result_items = add_summaries(result_items, timings=timings)
+ if timings is not None:
+ timings["decorate"] = (time.perf_counter() - summaries_started) * 1000
+ timings["engine"] = (time.perf_counter() - started) * 1000
+ return {"results": result_items, "total": len(preordered), "page": page, "page_size": page_size}
if not q:
- has_record_filters = extensions or folders or min_size is not None or max_size is not None or folder_match_mode == "mixed"
- can_use_precomputed = not has_record_filters and (not repos or len(repos) == 1)
- if can_use_precomputed:
- if repos and len(repos) == 1:
- repo = repos[0]
- if sort == "name":
- indices = repo_sorted_by_name.get(repo, [])
- elif sort == "size":
- indices = repo_sorted_by_size.get(repo, [])
- else:
- indices = repo_records_map.get(repo, [])
- else:
- if sort == "name":
- indices = sorted_by_name
- elif sort == "size":
- indices = sorted_by_size
- else:
- indices = list(range(len(records)))
- total, results = paginate_indices(indices, page, page_size)
- else:
- if repos:
- indices = []
- for r in repos:
- indices.extend(repo_records_map.get(r, []))
- else:
- indices = list(range(len(records)))
- if folder_match_mode == "mixed":
- filtered = apply_mixed_folder_filters(indices, folder_selfs, folder_subtrees)
- filtered = apply_filters(filtered, repos, extensions, None, min_size, max_size)
- else:
- filtered = apply_filters(indices, repos, extensions, folders, min_size, max_size, folder_match_mode)
- if sort == "name":
- filtered.sort(key=sort_key_name)
- elif sort == "size":
- filtered.sort(key=sort_key_size)
- total, results = paginate_indices(filtered, page, page_size)
- return build_response(results, total, page, page_size)
- query_tokens = tokenize(q)
- exact_matches = None
- exact_tokens = [(tok, word_posting_count(tok)) for tok in query_tokens if tok in word_index]
- if exact_tokens:
- exact_tokens.sort(key=lambda x: x[1])
- exact_matches = word_postings(exact_tokens[0][0])
- for tok, _ in exact_tokens[1:]:
- exact_matches.intersection_update(word_postings(tok))
- fuzzy_matches = set()
- for i, tok in enumerate(query_tokens):
- if tok in word_index:
- continue
- candidates = set()
- tok_len = len(tok)
- for delta in (-2, -1, 0, 1, 2):
- for vocab_word in vocab_by_len.get(tok_len + delta, []):
- if not could_be_fuzzy(tok, vocab_word, 2):
- continue
- dist = edit_distance(tok, vocab_word, max_dist=2)
- if dist <= 2:
- candidates.update(word_postings(vocab_word))
- if len(candidates) >= 200:
- break
- if len(candidates) >= 200:
- break
- if candidates:
- if i == 0 and not exact_matches:
- fuzzy_matches = candidates
- else:
- fuzzy_matches.update(candidates)
- if exact_matches:
- all_matched = list(exact_matches)
- if fuzzy_matches:
- all_matched.extend([i for i in fuzzy_matches if i not in exact_matches])
- elif fuzzy_matches:
- all_matched = list(fuzzy_matches)
+ indices = list(range(len(records)))
+ filter_started = time.perf_counter()
+ filtered = apply_filters(indices, sources_filter, folders, min_size, max_size)
+ if timings is not None:
+ timings["filter"] = (time.perf_counter() - filter_started) * 1000
else:
- all_matched = []
- literal_query = q.lower()
- if literal_query:
- literal_matches = []
- initial_indices = get_initial_indices(repos)
- candidates = literal_candidates(literal_query, repos)
- if candidates is not None:
- initial_indices = sorted(candidates)
- skip = set(all_matched)
- for idx in initial_indices:
- if skip and idx in skip:
- continue
- rec = records[idx]
- file_name = rec.get("_file_lower", "")
- folder_path = rec.get("_folder_path_lower", "")
- repo_name = rec.get("_repo_lower", "")
- if literal_query in file_name or literal_query in repo_name or (search_folders and literal_query in folder_path):
- literal_matches.append(idx)
- if literal_matches:
- all_matched.extend(idx for idx in literal_matches if idx not in skip)
- if not search_folders and query_tokens and all_matched:
- filtered_by_file = []
- for idx in all_matched:
- rec = records[idx]
- file_name = rec.get("_file_lower", "")
- repo = rec.get("_repo_lower", "")
- if all(tok in file_name or tok in repo for tok in query_tokens):
- filtered_by_file.append(idx)
- all_matched = filtered_by_file
- if folder_match_mode == "mixed":
- filtered = apply_mixed_folder_filters(all_matched, folder_selfs, folder_subtrees)
- filtered = apply_filters(filtered, repos, extensions, None, min_size, max_size)
+ tokens = tokenize(q)
+ metadata_started = time.perf_counter()
+ indices = list(metadata_matches(q, exact, search_paths))
+ if timings is not None:
+ timings["metadata"] = (time.perf_counter() - metadata_started) * 1000
+ filter_started = time.perf_counter()
+ filtered = apply_filters(indices, sources_filter, folders, min_size, max_size)
+ if timings is not None:
+ timings["filter"] = (time.perf_counter() - filter_started) * 1000
+ if sort == "relevance":
+ total = len(filtered)
+ rank_started = time.perf_counter()
+ paged_indices = relevance_page(filtered, tokens, search_paths, start, page_size)
+ if timings is not None:
+ timings["rank"] = (time.perf_counter() - rank_started) * 1000
+ total = len(filtered)
+ if q and sort == "relevance":
+ selected = paged_indices
+ elif sort == "name":
+ sort_started = time.perf_counter()
+ selected = ordered_page(filtered, name_order, start, page_size)
+ if timings is not None:
+ timings["sort"] = (time.perf_counter() - sort_started) * 1000
+ elif sort == "size":
+ sort_started = time.perf_counter()
+ selected = ordered_page(filtered, size_order, start, page_size)
+ if timings is not None:
+ timings["sort"] = (time.perf_counter() - sort_started) * 1000
else:
- filtered = apply_filters(all_matched, repos, extensions, folders, min_size, max_size, folder_match_mode)
- scored = [(idx, score_record(idx, query_tokens, search_folders)) for idx in filtered]
+ selected = filtered[start:start + page_size]
+ result_items = [trim_record(records[idx]) for idx in selected]
+ summaries_started = time.perf_counter()
+ result_items = add_summaries(result_items, timings=timings)
+ if timings is not None:
+ timings["decorate"] = (time.perf_counter() - summaries_started) * 1000
+ timings["engine"] = (time.perf_counter() - started) * 1000
+ return {"results": result_items, "total": total, "page": page, "page_size": page_size}
+
+def fulltext_search(q="", sources_filter=None, folders=None, min_size=None, max_size=None, page=1, page_size=100, sort="relevance", exact=False, search_paths=True, timings=None):
+ started = time.perf_counter()
+ q = q.strip()
+ if not q or fulltext_databases is None:
+ return search(q, sources_filter, folders, min_size, max_size, page, page_size, sort, exact, search_paths, timings)
+ source_slugs = sources_filter if sources_filter is not None else list(source_counts)
+ if not source_slugs:
+ return {"results": [], "total": 0, "page": max(1, page), "page_size": page_size}
+ matched_doc_ids = set()
+ sqlite_started = time.perf_counter()
+ for source_slug in source_slugs:
+ matched_doc_ids.update(f"{source_slug}:{doc_number}" for doc_number in fulltext_databases.search_source(source_slug, q, exact))
+ if timings is not None:
+ timings["sqlite"] = (time.perf_counter() - sqlite_started) * 1000
+ content_indices = {idx for doc_id in matched_doc_ids if (idx := record_map_index.get(doc_id)) is not None}
+ indices = set(content_indices)
+ metadata_started = time.perf_counter()
+ indices.update(metadata_matches(q, exact, search_paths))
+ if timings is not None:
+ timings["metadata"] = (time.perf_counter() - metadata_started) * 1000
+ filter_started = time.perf_counter()
+ filtered = apply_filters(indices, sources_filter, folders, min_size, max_size)
+ if timings is not None:
+ timings["filter"] = (time.perf_counter() - filter_started) * 1000
+ tokens = query_tokens(q)
if sort == "relevance":
- scored.sort(key=lambda x: (-x[1], x[0]))
+ total = len(filtered)
+ start = (page - 1) * page_size
+ rank_started = time.perf_counter()
+ selected = relevance_page(filtered, tokens, search_paths, start, page_size)
+ if timings is not None:
+ timings["rank"] = (time.perf_counter() - rank_started) * 1000
elif sort == "name":
- scored.sort(key=lambda x: sort_key_name(x[0]))
+ sort_started = time.perf_counter()
+ selected = ordered_page(filtered, name_order, (page - 1) * page_size, page_size)
+ if timings is not None:
+ timings["sort"] = (time.perf_counter() - sort_started) * 1000
elif sort == "size":
- scored.sort(key=lambda x: sort_key_size(x[0]))
- total = len(scored)
- start = (page - 1) * page_size
- paged = scored[start:start + page_size]
- results = [records[idx] for idx, _ in paged]
- return build_response(results, total, page, page_size)
-
-def build_filter_folder_tree(repo: str) -> list[dict]:
- if repo in folder_tree_data:
- return folder_tree_data[repo]
- if repo not in folder_index:
- return []
- repo_name_short = repo.split("/")[-1]
- tree_root = {
- "name": repo_name_short,
- "path": "",
- "children": [],
- "count": 0,
- "hasDirectFiles": False,
- "hasChildren": False,
- "showSelfToggle": False,
- "isRoot": True,
- }
- node_map = {"": tree_root}
- for path, idx_list in folder_index[repo].items():
- if path == "":
- tree_root["count"] = len(idx_list)
- continue
- parts = path.split("/")
- for depth in range(1, len(parts) + 1):
- key = "/".join(parts[:depth])
- if key not in node_map:
- parent_key = "/".join(parts[:depth - 1]) if depth > 1 else ""
- parent = node_map[parent_key]
- node = {
- "name": parts[depth - 1],
- "path": key,
- "children": [],
- "count": 0,
- "hasDirectFiles": False,
- "hasChildren": False,
- "showSelfToggle": False,
- }
- node_map[key] = node
- parent["children"].append(node)
- node_map[path]["count"] = len(idx_list)
- direct_file_dirs = set()
- for idx in repo_records_map.get(repo, []):
- rec = records[idx]
- folders = rec.get("Folder", []) or []
- direct_file_dirs.add("/".join(folders))
- for path_key, node in node_map.items():
- node["hasDirectFiles"] = path_key in direct_file_dirs
- node["children"].sort(key=lambda child: child["name"])
- node["hasChildren"] = len(node["children"]) > 0
- node["showSelfToggle"] = bool(path_key and node["hasDirectFiles"] and node["hasChildren"])
- return [tree_root]
-
-def get_folder_contents(repo: str, path: str) -> dict:
- if repo in folder_browser_data:
- repo_browser = folder_browser_data.get(repo, {}) or {}
- if path in repo_browser:
- return enrich_browser_entry(repo, path, repo_browser[path])
- if path == "" and "" in repo_browser:
- return enrich_browser_entry(repo, path, repo_browser[""])
- path_parts = [p for p in path.strip("/").split("/") if p]
- path_depth = len(path_parts)
- repo_indices = repo_records_map.get(repo, [])
- repo_records = [records[i] for i in repo_indices]
-
- def is_under(rec_folders, prefix):
- return rec_folders[:len(prefix)] == prefix
- matching = [rec for rec in repo_records if is_under(rec.get("Folder", []) or [], path_parts)]
- subfolder_names = set()
- subfolder_counts = {}
- for rec in matching:
- folders = rec.get("Folder", []) or []
- if len(folders) > path_depth:
- name = folders[path_depth]
- subfolder_names.add(name)
- subfolder_counts[name] = subfolder_counts.get(name, 0) + 1
- result_folders = []
- for fname in sorted(subfolder_names):
- full_path = "/".join(path_parts + [fname]) if path_parts else fname
- result_folders.append({"name": fname, "path": full_path, "count": subfolder_counts.get(fname, 0)})
- direct_files = []
- for rec in matching:
- folders = rec.get("Folder", []) or []
- if len(folders) == path_depth:
- direct_files.append({
- "name": rec.get("File", ""),
- "ext": rec.get("Extension", ""),
- "link": rec.get("Link") or build_record_link(rec),
- "path": rec.get("Path") or build_record_path_url(rec),
- "hasTxt": rec.get("HasTxt", False),
- "size": rec.get("Size", ""),
- })
- direct_files.sort(key=lambda f: f["name"])
- return {"folders": result_folders, "files": direct_files, "current_path": path}
-
-async def warm_txt_proxy_loop(app: FastAPI):
- while True:
- await asyncio.sleep(TXT_WARM_INTERVAL_SECONDS)
- if time.monotonic() - last_user_request_at < TXT_WARM_IDLE_SECONDS:
- continue
- pool = txt_record_indices
- if not pool:
- continue
- rec = records[random.choice(pool)]
- remote_path = build_txt_relative_path(rec)
- remote_url = TXT_SPACE_RAW_BASE + "/" + quote(remote_path, safe="/")
- try:
- async with app.state.http_session.get(
- remote_url,
- headers={"Range": "bytes=0-0", "Accept-Encoding": "identity"},
- timeout=aiohttp.ClientTimeout(total=15, connect=5, sock_read=5),
- ) as response:
- await response.content.read(1)
- except asyncio.CancelledError:
- raise
- except Exception as e:
- print(f"txt 预热异常: {e}")
+ sort_started = time.perf_counter()
+ selected = ordered_page(filtered, size_order, (page - 1) * page_size, page_size)
+ if timings is not None:
+ timings["sort"] = (time.perf_counter() - sort_started) * 1000
+ total = len(filtered)
+ summaries_started = time.perf_counter()
+ result_items = add_summaries(
+ [trim_record(records[idx]) for idx in selected],
+ q,
+ matched_snippets=True,
+ content_indices=content_indices,
+ exact=exact,
+ timings=timings,
+ )
+ if timings is not None:
+ timings["decorate"] = (time.perf_counter() - summaries_started) * 1000
+ timings["engine"] = (time.perf_counter() - started) * 1000
+ return {"results": result_items, "total": total, "page": page, "page_size": page_size}
+
+def get_doc_storage_path(doc: dict) -> Path:
+ return BASE_DIR / doc["storage_root"] / doc["storage_rel_path"]
+
+def read_preview_text(doc_id: str, file_path: Path) -> str:
+ with preview_text_cache_lock:
+ cached = preview_text_cache.get(doc_id)
+ if cached is not None:
+ return cached
+ try:
+ size = file_path.stat().st_size
+ except OSError:
+ size = PREVIEW_CACHE_MAX_ENTRY_BYTES + 1
+ text = file_path.read_text(encoding="utf-8", errors="ignore")
+ if size <= PREVIEW_CACHE_MAX_ENTRY_BYTES:
+ with preview_text_cache_lock:
+ preview_text_cache[doc_id] = text
+ while len(preview_text_cache) > PREVIEW_CACHE_MAX_ENTRIES:
+ preview_text_cache.pop(next(iter(preview_text_cache)), None)
+ return text
+
+def get_folder_contents(source_slug: str, path: str) -> dict:
+ source_browser = folder_browser_data.get(source_slug, {})
+ if path in source_browser:
+ entry = dict(source_browser[path])
+ entry["current_path"] = path
+ return entry
+ return {"folders": [], "files": [], "current_path": path}
+
+def _cached_payload(cache, lock, key: tuple, builder, ttl: int, max_entries: int, timings=None):
+ cache_started = time.perf_counter()
+ now = time.monotonic()
+ with lock:
+ cached = cache.get(key)
+ if cached and now - cached[0] < ttl:
+ if timings is not None:
+ timings["cache"] = "hit"
+ timings["cache_lookup"] = (time.perf_counter() - cache_started) * 1000
+ return cached[1]
+ if timings is not None:
+ timings["cache"] = "miss"
+ timings["cache_lookup"] = (time.perf_counter() - cache_started) * 1000
+ # Serialize identical cold keys so concurrent requests compute once instead
+ # of stampeding the builder. The per-key lock is removed once it is no longer
+ # contended so the map stays bounded to the number of in-flight keys.
+ with cache_key_locks_guard:
+ key_lock = cache_key_locks.setdefault(key, threading.Lock())
+ try:
+ with key_lock:
+ with lock:
+ cached = cache.get(key)
+ if cached and now - cached[0] < ttl:
+ return cached[1]
+ value = builder()
+ inserted_at = time.monotonic()
+ with lock:
+ cache[key] = (inserted_at, value)
+ while len(cache) > max_entries:
+ oldest_key = min(cache, key=lambda item: cache[item][0])
+ cache.pop(oldest_key, None)
+ return value
+ finally:
+ with cache_key_locks_guard:
+ if cache_key_locks.get(key) is key_lock:
+ cache_key_locks.pop(key, None)
+
+def cached_payload(key: tuple, builder, timings=None):
+ return _cached_payload(
+ api_response_cache, api_response_cache_lock, key, builder,
+ API_CACHE_TTL_SECONDS, API_CACHE_MAX_ENTRIES, timings,
+ )
+
+def cached_search_payload(key: tuple, builder, timings=None):
+ return _cached_payload(
+ search_response_cache, search_response_cache_lock, key, builder,
+ SEARCH_CACHE_TTL_SECONDS, SEARCH_CACHE_MAX_ENTRIES, timings,
+ )
+
+def ready_fulltext_sources() -> set[str]:
+ if fulltext_databases is None:
+ return set()
+ with fulltext_databases.lock:
+ return {
+ source for source in fulltext_databases.connections
+ if fulltext_databases.tokenizer_versions.get(source) == TOKENIZER_VERSION
+ and fulltext_databases.has_content_fts.get(source, False)
+ and fulltext_databases.has_snippet_anchors.get(source, False)
+ and fulltext_databases.document_counts.get(source) == source_counts.get(source)
+ and fulltext_databases.fts_counts.get(source) == source_counts.get(source)
+ }
@asynccontextmanager
async def lifespan(app: FastAPI):
+ global fulltext_databases
+ fulltext_refresh_stop.clear()
load_data()
- app.state.http_session = aiohttp.ClientSession(
- timeout=aiohttp.ClientTimeout(total=300, connect=15),
- headers={"User-Agent": "VoiceOfML-Search/1.0"},
- auto_decompress=False,
- )
- app.state.upstream_semaphore = asyncio.Semaphore(8)
- app.state.reader_assets_lock = asyncio.Lock()
- app.state.txt_warm_task = asyncio.create_task(warm_txt_proxy_loop(app))
+ # Mounted indexes can be large. Defer their first open/backup to the background
+ # refresher so FastAPI reaches readiness before full-text data is available.
+ fulltext_databases = FulltextDatabases(FULLTEXT_DIR, in_memory=FULLTEXT_IN_MEMORY, refresh=False)
+ ready = ready_fulltext_sources()
+ expected = set(source_counts)
+ if ready != expected:
+ threading.Thread(target=refresh_fulltext_loop, daemon=True).start()
try:
yield
finally:
- app.state.txt_warm_task.cancel()
- try:
- await app.state.txt_warm_task
- except asyncio.CancelledError:
- pass
- await app.state.http_session.close()
-app = FastAPI(title="VoiceOfML Search", version="2.0", lifespan=lifespan)
+ fulltext_refresh_stop.set()
+ fulltext_databases.close()
+app = FastAPI(title="VOMEBOOK Search", version="1.0", lifespan=lifespan)
app.add_middleware(GZipMiddleware, minimum_size=500)
-app.add_middleware(
- CORSMiddleware,
- allow_origins=["*"],
- allow_methods=["*"],
- allow_headers=["*"],
- expose_headers=["Accept-Ranges", "Content-Range", "Content-Length", "ETag", "Last-Modified"],
-)
-
-@app.middleware("http")
-async def track_user_activity(request: Request, call_next):
- global last_user_request_at
- if request.url.path not in ("/health", "/favicon.ico", "/api/ping"):
- last_user_request_at = time.monotonic()
- return await call_next(request)
+app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
class SearchRequest(BaseModel):
- q: str = ""
- repos: Optional[list[str]] = None
- extensions: Optional[list[str]] = None
+ q: str = Field(default="", max_length=256)
+ sources: Optional[list[str]] = None
folders: Optional[list[str]] = None
- folder_match_mode: str = "prefix"
- folder_selfs: Optional[list[str]] = None
- folder_subtrees: Optional[list[str]] = None
- min_size: Optional[int] = None
- max_size: Optional[int] = None
+ min_size: Optional[int] = Field(default=None, ge=0)
+ max_size: Optional[int] = Field(default=None, ge=0)
page: int = Field(default=1, ge=1)
- page_size: int = Field(default=100, ge=1, le=500)
- search_folders: bool = True
- sort: str = "relevance"
+ page_size: int = Field(default=100, ge=1, le=200)
+ sort: Literal["relevance", "name", "size"] = "relevance"
exact: bool = False
-@app.post("/api/search")
-
-def api_search(body: SearchRequest):
- repos = [r.strip() for r in body.repos] if body.repos else None
- extensions = [e.strip().lower() for e in body.extensions] if body.extensions else None
- folders = [f.strip() for f in body.folders] if body.folders else None
- folder_selfs = [f.strip() for f in body.folder_selfs] if body.folder_selfs else None
- folder_subtrees = [f.strip() for f in body.folder_subtrees] if body.folder_subtrees else None
- return JSONResponse(search(
- body.q, repos, extensions, folders, body.min_size, body.max_size,
- body.page, body.page_size, body.search_folders, body.sort, body.exact,
- body.folder_match_mode, folder_selfs, folder_subtrees,
- ))
-@app.post("/api/search/{repo_name}")
-
-def api_search_repo(repo_name: str, body: SearchRequest):
- full_repo = f"VoiceOfML/{repo_name}"
- if full_repo not in repo_counts:
- return JSONResponse({"error": f"仓库 {full_repo} 不存在", "results": [], "total": 0}, status_code=404)
- extensions = [e.strip().lower() for e in body.extensions] if body.extensions else None
- folders = [f.strip() for f in body.folders] if body.folders else None
- folder_selfs = [f.strip() for f in body.folder_selfs] if body.folder_selfs else None
- folder_subtrees = [f.strip() for f in body.folder_subtrees] if body.folder_subtrees else None
- return JSONResponse(search(
- body.q, [full_repo], extensions, folders, body.min_size, body.max_size,
- body.page, body.page_size, body.search_folders, body.sort, body.exact,
- body.folder_match_mode, folder_selfs, folder_subtrees,
- ))
-@app.get("/api/repos")
+ search_paths: bool = True
+ fulltext: bool = False
-def api_repos():
- return JSONResponse(cached_payload(("repos", _data_generation), lambda: repo_list), headers=METADATA_CACHE_HEADERS)
-@app.get("/api/reader-assets")
+class ZipRequest(BaseModel):
+ doc_ids: list[str] = Field(default_factory=list, max_length=500)
-async def api_reader_assets():
- return JSONResponse(await get_reader_assets(), headers={"Cache-Control": "public, max-age=300"})
-@app.get("/api/extensions")
+def create_zip_download_token(doc_ids: list[str]) -> str:
+ now = time.monotonic()
+ with zip_download_lock:
+ expired = [token for token, (created, _ids) in zip_download_tokens.items() if now - created > ZIP_TOKEN_TTL_SECONDS]
+ for token in expired:
+ zip_download_tokens.pop(token, None)
+ while len(zip_download_tokens) >= ZIP_TOKEN_MAX_ENTRIES:
+ oldest = min(zip_download_tokens, key=lambda token: zip_download_tokens[token][0])
+ zip_download_tokens.pop(oldest, None)
+ token = secrets.token_urlsafe(24)
+ zip_download_tokens[token] = (now, list(dict.fromkeys(doc_ids)))
+ return token
+
+def get_zip_download_token(token: str) -> list[str] | None:
+ with zip_download_lock:
+ item = zip_download_tokens.get(token)
+ if item is None or time.monotonic() - item[0] > ZIP_TOKEN_TTL_SECONDS:
+ with zip_download_lock:
+ zip_download_tokens.pop(token, None)
+ return None
+ return item[1]
-def api_extensions(repo: Optional[str] = Query(default=None)):
- def build():
- if repo:
- full_repo = f"VoiceOfML/{repo}"
- ext_count = repo_extension_counts.get(full_repo, {})
- else:
- ext_count = extension_counts
- sorted_exts = sorted(ext_count.keys())
- return [{"name": e, "count": ext_count[e]} for e in sorted_exts]
- return JSONResponse(cached_payload(("extensions", _data_generation, repo or ""), build), headers=METADATA_CACHE_HEADERS)
-@app.get("/api/folders/{repo_name}")
-
-def api_folders(repo_name: str):
- full_repo = f"VoiceOfML/{repo_name}"
- if full_repo not in repo_counts:
- return JSONResponse({"error": f"仓库 {full_repo} 不存在"}, status_code=404)
- return JSONResponse(cached_payload(("folders", full_repo), lambda: build_filter_folder_tree(full_repo)))
-@app.get("/api/folders/{repo_name}/contents")
-
-def api_folder_contents(repo_name: str, path: str = Query(default="")):
- full_repo = f"VoiceOfML/{repo_name}"
- if full_repo not in repo_counts:
- return JSONResponse({"error": f"仓库 {full_repo} 不存在"}, status_code=404)
- return JSONResponse(cached_payload(("folder_contents", full_repo, path), lambda: get_folder_contents(full_repo, path)))
-@app.get("/api/random")
+def zip_entries(doc_ids: list[str]) -> list[tuple[str, Path]]:
+ entries = []
+ for doc_id in dict.fromkeys(doc_ids):
+ rec = record_map.get(doc_id)
+ if not rec:
+ continue
+ file_path = get_doc_storage_path(rec)
+ if not file_path.exists():
+ continue
+ source = re.sub(r'[:*?"<>|/\\\x00-\x1f]', "_", str(rec["source"])).strip(" .") or "source"
+ title = re.sub(
+ r'[:*?"<>|/\\\x00-\x1f]',
+ lambda match: {":": "：", "*": "＊", "?": "？", '"': "＂", "<": "＜", ">": "＞", "|": "｜", "/": "／", "\\": "＼"}.get(match.group(0), " "),
+ str(rec.get("display_name") or Path(str(rec["display_rel_path"])).stem),
+ ).strip(" .") or "article"
+ title_bytes = title.encode("utf-8")
+ if len(title_bytes) > 180:
+ cutoff = 180
+ while cutoff > 0 and (title_bytes[cutoff] & 0xC0) == 0x80:
+ cutoff -= 1
+ title = title_bytes[:cutoff].decode("utf-8", errors="ignore").rstrip(" .")
+ safe_doc_id = re.sub(r"[^a-zA-Z0-9._-]+", "_", doc_id)
+ entries.append((f"{source}/{title} [{safe_doc_id}].txt", file_path))
+ return entries
+
+def zip_response(doc_ids: list[str]):
+ if len(doc_ids) > 500:
+ return JSONResponse({"error": "too many files", "max_files": 500}, status_code=400)
+ entries = zip_entries(doc_ids)
+ if not entries:
+ return JSONResponse({"error": "no files"}, status_code=400)
+ import zipfile
+ temporary = tempfile.NamedTemporaryFile(prefix="vomebook-", suffix=".zip", delete=False)
+ archive_path = Path(temporary.name)
+ temporary.close()
+ try:
+ with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as archive:
+ for member_name, file_path in entries:
+ archive.write(file_path, arcname=member_name)
+ except Exception:
+ archive_path.unlink(missing_ok=True)
+ raise
+ return FileResponse(
+ archive_path,
+ media_type="application/zip",
+ filename="vomebook_batch.zip",
+ headers={
+ "Cache-Control": "no-store",
+ "Content-Encoding": "identity",
+ },
+ background=BackgroundTask(archive_path.unlink, missing_ok=True),
+ )
-def api_random(repo: Optional[str] = Query(default=None)):
- if repo:
- full_repo = f"VoiceOfML/{repo}"
- repo_indices = repo_records_map.get(full_repo, [])
- if not repo_indices:
- return JSONResponse({"error": "无可用记录"}, status_code=404)
- rec = dict(records[random.choice(repo_indices)])
+def run_search(body: SearchRequest, sources_filter=None, timings=None):
+ selected_sources = sources_filter if sources_filter is not None else body.sources
+ if fulltext_databases is None:
+ fulltext_generation = 0
else:
- if not records:
- return JSONResponse({"error": "无可用记录"}, status_code=404)
- rec = dict(random.choice(records))
- rec["Link"] = rec.get("Link") or build_record_link(rec)
- rec["Path"] = rec.get("Path") or build_record_path_url(rec)
- return JSONResponse(rec)
-
-@app.get("/api/random-txt")
-def api_random_txt(repo: Optional[str] = Query(default=None)):
- if repo:
- full_repo = f"VoiceOfML/{repo}"
- pool = repo_txt_record_indices.get(full_repo, [])
- else:
- pool = txt_record_indices
- if not pool:
- return JSONResponse({"error": "无可用 txt"}, status_code=404)
- rec = dict(records[random.choice(pool)])
- rec["Link"] = rec.get("Link") or build_record_link(rec)
- rec["Path"] = rec.get("Path") or build_record_path_url(rec)
- return JSONResponse(rec)
-
-@app.get("/api/random-txt/status")
-def api_random_txt_status(repo: Optional[str] = Query(default=None)):
- if repo:
- full_repo = f"VoiceOfML/{repo}"
- count = len(repo_txt_record_indices.get(full_repo, []))
- else:
- count = len(txt_record_indices)
- return JSONResponse({"available": count > 0, "count": count}, headers=METADATA_CACHE_HEADERS)
-
-def converted_reader_indices(assets: dict, full_repo: str | None = None) -> dict[int, dict]:
- converted = {}
- for key, asset in assets.get("f", {}).items():
- if asset.get("s") != 2 or (full_repo and not key.startswith(f"{full_repo}\0")):
- continue
- index = reader_record_key_indices.get(key)
- if index is not None:
- converted[index] = asset
- return converted
-
-
-@app.get("/api/random-reader")
-async def api_random_reader(repo: Optional[str] = Query(default=None)):
- full_repo = f"VoiceOfML/{repo}" if repo else None
- original = repo_reader_record_indices.get(full_repo, []) if full_repo else reader_record_indices
- converted = converted_reader_indices(await get_reader_assets(), full_repo)
- pool = list(dict.fromkeys([*original, *converted]))
- if not pool:
- return JSONResponse({"error": "无可读文档"}, status_code=404)
- index = random.choice(pool)
- rec = dict(records[index])
- rec["Link"] = rec.get("Link") or build_record_link(rec)
- rec["Path"] = rec.get("Path") or build_record_path_url(rec)
- asset = converted.get(index)
- if asset:
- rec["ReaderLink"] = f"https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/{asset['p']}"
- rec["ReaderExtension"] = {"p": "pdf", "e": "epub", "d": "docx", "h": "html"}[asset["m"]]
- rec["DownloadLink"] = rec["Link"]
- return JSONResponse(rec)
-
-@app.get("/api/random-reader/status")
-async def api_random_reader_status(repo: Optional[str] = Query(default=None)):
- full_repo = f"VoiceOfML/{repo}" if repo else None
- original = repo_reader_record_indices.get(full_repo, []) if full_repo else reader_record_indices
- converted = converted_reader_indices(await get_reader_assets(), full_repo)
- count = len(set(original).union(converted))
- return JSONResponse({"available": count > 0, "count": count}, headers=METADATA_CACHE_HEADERS)
-
-@app.get("/api/bootstrap")
-def api_bootstrap():
- return JSONResponse(
- cached_payload(("bootstrap", _data_generation), build_global_bootstrap),
- headers=METADATA_CACHE_HEADERS,
+ with fulltext_databases.lock:
+ fulltext_generation = fulltext_databases.generation
+ key = (
+ "search",
+ fulltext_generation,
+ tuple(selected_sources or []),
+ tuple(body.folders or []),
+ body.q,
+ body.min_size,
+ body.max_size,
+ body.page,
+ body.page_size,
+ body.sort,
+ body.exact,
+ body.search_paths,
+ body.fulltext,
)
+ def build_payload():
+ result = (
+ fulltext_search(
+ body.q, selected_sources, body.folders, body.min_size, body.max_size,
+ body.page, body.page_size, body.sort, body.exact, body.search_paths,
+ timings,
+ )
+ if body.fulltext
+ else search(
+ body.q, selected_sources, body.folders, body.min_size, body.max_size,
+ body.page, body.page_size, body.sort, body.exact, body.search_paths,
+ timings,
+ )
+ )
+ payload = dict(result)
+ requested_sources = set(selected_sources or source_counts)
+ payload["indexing"] = bool(body.fulltext and not requested_sources.issubset(ready_fulltext_sources()))
+ payload["index_generation"] = fulltext_generation
+ return payload
-@app.get("/api/ping")
-def api_ping():
- return PlainTextResponse(status_code=204, headers={"Cache-Control": "no-store"})
+ return cached_search_payload(key, build_payload, timings)
+
+@app.post("/api/search")
+
+def api_search(body: SearchRequest):
+ return JSONResponse(run_search(body))
+@app.post("/api/search/{source_slug}")
+def api_search_source(source_slug: str, body: SearchRequest):
+ if source_slug not in source_counts:
+ return JSONResponse({"error": "source not found", "results": [], "total": 0}, status_code=404)
+ return JSONResponse(run_search(body, [source_slug]))
-READER_UPSTREAM_RESPONSE_HEADERS = (
- "Content-Type", "Content-Length", "Content-Range", "Accept-Ranges",
- "ETag", "Last-Modified", "Cache-Control",
-)
+@app.get("/api/ping")
+def api_ping():
+ return Response(status_code=204, headers={"Cache-Control": "no-store"})
+@app.get("/api/sources")
+def api_sources():
+ return JSONResponse(cached_payload(("sources",), lambda: sources))
+@app.get("/api/index-status")
-@app.api_route("/api/reader-content", methods=["GET", "HEAD"])
-async def api_reader_content(request: Request, url: str = Query(...)):
+def api_index_status():
+ if fulltext_databases is None:
+ loaded = []
+ versions = {}
+ else:
+ fulltext_databases.refresh()
+ loaded = sorted(ready_fulltext_sources())
+ versions = dict(fulltext_databases.tokenizer_versions)
+ expected = sorted(source_counts)
+ building = [source for source in expected if source not in loaded]
+ build_state = "ready" if not building else "building"
try:
- target_url = validate_reader_source_url(url)
- except ValueError as exc:
- return JSONResponse({"error": str(exc)}, status_code=403)
- forwarded_headers = {
- name: request.headers[name]
- for name in ("Range", "If-Range", "If-None-Match", "If-Modified-Since")
- if name in request.headers
- }
- semaphore = app.state.upstream_semaphore
- await semaphore.acquire()
- release_semaphore = True
+ status_payload = json.loads(INDEX_BUILD_STATUS_PATH.read_text(encoding="utf-8"))
+ if building and status_payload.get("state") == "failed":
+ build_state = "failed"
+ except Exception:
+ pass
try:
- upstream = await open_download_response(
- app.state.http_session,
- target_url,
- aiohttp.ClientTimeout(total=180, connect=15, sock_read=60),
- method=request.method,
- request_headers=forwarded_headers,
- )
- except asyncio.CancelledError:
- raise
- except Exception as exc:
- print(f"阅读代理异常: {exc}")
- return JSONResponse({"error": str(exc) or "上游阅读文件加载失败"}, status_code=502)
- finally:
- if release_semaphore and "upstream" not in locals():
- semaphore.release()
- headers = {name: upstream.headers[name] for name in READER_UPSTREAM_RESPONSE_HEADERS if name in upstream.headers}
- headers.setdefault("Cache-Control", "public, max-age=300")
- headers["Content-Encoding"] = "identity"
- status = upstream.status
- if request.method == "HEAD":
- upstream.release()
- semaphore.release()
- return Response(status_code=status, headers=headers)
- if status not in (200, 206, 304, 416):
- upstream.release()
- semaphore.release()
- return JSONResponse({"error": f"上游阅读文件加载失败: HTTP {status}"}, status_code=status)
-
- async def stream_reader_content():
- try:
- async with upstream:
- async for chunk in upstream.content.iter_chunked(65536):
- yield chunk
- except Exception as exc:
- print(f"阅读代理异常: {exc}")
- raise
- finally:
- semaphore.release()
+ bucket_status = json.loads(BUCKET_INDEX_STATUS_PATH.read_text(encoding="utf-8"))
+ except Exception:
+ bucket_status = None
+ return JSONResponse({
+ "ready": not building,
+ "state": build_state,
+ "loaded_sources": loaded,
+ "building_sources": building,
+ "versions": versions,
+ "bucket": bucket_status,
+ })
+@app.get("/api/extensions")
- release_semaphore = False
- return StreamingResponse(stream_reader_content(), status_code=status, headers=headers)
+def api_extensions(source: Optional[str] = Query(default=None)):
+ counts = cached_payload(("extensions", source or ""), lambda: source_extension_counts.get(source, {}) if source else extension_counts)
+ return JSONResponse([{"name": ext, "count": count} for ext, count in sorted(counts.items())])
+@app.get("/api/folders/{source_slug}")
+def api_folders(source_slug: str):
+ return JSONResponse(cached_payload(("folders", source_slug), lambda: folder_tree_data.get(source_slug, [])))
+@app.get("/api/folders/{source_slug}/contents")
-@app.get("/api/download")
+def api_folder_contents(source_slug: str, path: str = Query(default="")):
+ return JSONResponse(cached_payload(("folder_contents", source_slug, path), lambda: get_folder_contents(source_slug, path)))
+@app.get("/api/random")
-async def api_download(file: str = Query(...), link: str = Query(...)):
- target_url = link
- try:
- target_url = validate_voiceofml_source_url(target_url)
- except ValueError:
- return JSONResponse({"error": "只允许 HuggingFace 域名"}, status_code=403)
- session = app.state.http_session
- semaphore = app.state.upstream_semaphore
- await semaphore.acquire()
- release_semaphore = True
- try:
- upstream = await open_download_response(
- session,
- target_url,
- aiohttp.ClientTimeout(total=120, connect=15, sock_read=30),
- )
- except asyncio.CancelledError:
- raise
- except Exception as e:
- print(f"下载代理异常: {e}")
- return JSONResponse({"error": str(e) or "上游下载失败"}, status_code=502)
- finally:
- if release_semaphore and "upstream" not in locals():
- semaphore.release()
- if upstream.status != 200:
- status = upstream.status
- upstream.release()
- semaphore.release()
- return JSONResponse({"error": f"上游下载失败: HTTP {status}"}, status_code=502)
- encoded_filename = quote(sanitize_archive_name(file), safe="")
- content_disposition = f"attachment; filename*=UTF-8''{encoded_filename}"
-
- async def stream_response():
- try:
- async with upstream:
- async for chunk in upstream.content.iter_chunked(65536):
- yield chunk
- except Exception as e:
- print(f"下载代理异常: {e}")
- raise
- finally:
- semaphore.release()
- release_semaphore = False
- return StreamingResponse(
- stream_response(),
- media_type="application/octet-stream",
- headers={
- "Content-Disposition": content_disposition,
- "Cache-Control": "no-cache",
+def api_random(source: Optional[str] = Query(default=None)):
+ if source:
+ indices = source_records_map.get(source, [])
+ if not indices:
+ return JSONResponse({"error": "no record"}, status_code=404)
+ return JSONResponse(trim_record(records[random.choice(indices)]))
+ if not records:
+ return JSONResponse({"error": "no record"}, status_code=404)
+ return JSONResponse(trim_record(random.choice(records)))
+@app.get("/api/preview/{doc_id}")
+
+def api_preview(doc_id: str):
+ rec = record_map.get(doc_id)
+ if not rec:
+ return JSONResponse({"error": "not found"}, status_code=404)
+ file_path = get_doc_storage_path(rec)
+ if not file_path.exists():
+ return JSONResponse({"error": f"missing file: {file_path.name}"}, status_code=404)
+ text = read_preview_text(doc_id, file_path)
+ return JSONResponse({"doc_id": doc_id, "title": rec["display_name"], "path": rec["display_rel_path"], "source": rec["source_name"], "text": text})
+@app.get("/api/snippet/{doc_id}")
+
+def api_snippet(doc_id: str, q: str = Query(default="", max_length=256), exact: bool = Query(default=False)):
+ rec = record_map.get(doc_id)
+ if not rec:
+ return JSONResponse({"error": "not found"}, status_code=404)
+ file_path = get_doc_storage_path(rec)
+ if not file_path.exists():
+ return JSONResponse({"error": f"missing file: {file_path.name}"}, status_code=404)
+ payload = cached_payload(
+ ("snippet", doc_id, q, exact),
+ lambda: {
+ "doc_id": doc_id,
+ **build_snippet_payload(file_path.read_text(encoding="utf-8", errors="ignore"), q, exact=exact),
},
)
-@app.get("/api/download/check")
+ return JSONResponse(payload)
+@app.get("/api/download/{doc_id}")
+
+def api_download(doc_id: str):
+ rec = record_map.get(doc_id)
+ if not rec:
+ return JSONResponse({"error": "not found"}, status_code=404)
+ file_path = get_doc_storage_path(rec)
+ if not file_path.exists():
+ return JSONResponse({"error": f"missing file: {file_path.name}"}, status_code=404)
+ return FileResponse(
+ file_path,
+ media_type="text/plain; charset=utf-8",
+ filename=f"{rec['display_name']}.txt",
+ content_disposition_type="attachment",
+ headers={"Content-Encoding": "identity"},
+ )
+@app.post("/api/zip")
-async def api_download_check(link: str = Query(...)):
- target_url = link
- try:
- target_url = validate_voiceofml_source_url(target_url)
- except ValueError as e:
- return JSONResponse({"error": str(e)}, status_code=403)
- session = app.state.http_session
- try:
- upstream = await open_download_response(
- session,
- target_url,
- aiohttp.ClientTimeout(total=30, connect=10, sock_read=10),
- )
- except Exception as e:
- print(f"下载检查异常: {e}")
- return JSONResponse({"error": str(e) or "上游下载失败"}, status_code=502)
- try:
- if upstream.status != 200:
- return JSONResponse({"error": f"上游下载失败: HTTP {upstream.status}"}, status_code=502)
- return JSONResponse({"ok": True})
- finally:
- upstream.release()
-TXT_DIR.mkdir(parents=True, exist_ok=True)
-
-def resolve_safe_txt_path(raw_path: str) -> Path:
- decoded_path = unquote(raw_path).replace("\\", "/").strip("/")
- normalized = posixpath.normpath(decoded_path)
- if normalized in ("", ".") or normalized.startswith("../") or normalized == "..":
- raise ValueError("invalid txt path")
- target = (TXT_DIR / normalized).resolve()
- base = TXT_DIR.resolve()
- if target != base and base not in target.parents:
- raise ValueError("invalid txt path")
- return target
-
-@app.get("/txt/{txt_path:path}")
-async def serve_txt_proxy(txt_path: str):
+def api_zip(req: ZipRequest):
try:
- local_path = resolve_safe_txt_path(txt_path)
- except ValueError:
- return JSONResponse({"error": "invalid txt path"}, status_code=400)
- if local_path.exists() and local_path.is_file():
- return FileResponse(local_path, media_type="text/plain; charset=utf-8")
- remote_path = unquote(txt_path).replace("\\", "/").strip("/")
- remote_url = TXT_SPACE_RAW_BASE + "/" + quote(remote_path, safe="/")
- session = app.state.http_session
+ return zip_response(req.doc_ids)
+ except Exception as exc:
+ print(f"zip build failed: {exc}")
+ return JSONResponse({"error": "zip failed"}, status_code=500)
+
+@app.post("/api/zip-prepare")
+def api_zip_prepare(req: ZipRequest):
+ valid_ids = [
+ doc_id for doc_id in dict.fromkeys(req.doc_ids)
+ if (record := record_map.get(doc_id)) is not None and get_doc_storage_path(record).exists()
+ ]
+ if not valid_ids:
+ return JSONResponse({"error": "no files"}, status_code=400)
+ token = create_zip_download_token(valid_ids)
+ return JSONResponse({"download_url": f"/api/zip-download/{token}", "file_count": len(valid_ids)})
+
+@app.get("/api/zip-download/{token}")
+def api_zip_download_token(token: str):
+ doc_ids = get_zip_download_token(token)
+ if doc_ids is None:
+ return JSONResponse({"error": "download expired"}, status_code=404)
try:
- upstream = await open_download_response(
- session,
- remote_url,
- aiohttp.ClientTimeout(total=60, connect=10, sock_read=20),
- )
- except Exception as e:
- print(f"txt 代理异常: {e}")
- return JSONResponse({"error": str(e) or "txt 读取失败"}, status_code=502)
- if upstream.status != 200:
- status = upstream.status
- upstream.release()
- return JSONResponse({"error": f"txt 不存在: HTTP {status}"}, status_code=404 if status == 404 else 502)
-
- async def stream_txt():
- try:
- async with upstream:
- async for chunk in upstream.content.iter_chunked(65536):
- yield chunk
- except Exception as e:
- print(f"txt 代理异常: {e}")
- raise
-
- return StreamingResponse(stream_txt(), media_type="text/plain; charset=utf-8")
-
-HASHED_ASSET_RE = re.compile(r"\.[0-9a-f]{12}\.(?:js|css|mjs)$")
-
-
-class HashedStaticFiles(StaticFiles):
- async def get_response(self, path: str, scope):
- response = await super().get_response(path, scope)
- if response.status_code == 200 and HASHED_ASSET_RE.search(path):
- response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
- return response
-
+ return zip_response(doc_ids)
+ except Exception as exc:
+ print(f"zip build failed: {exc}")
+ return JSONResponse({"error": "zip failed"}, status_code=500)
+
+@app.post("/api/zip-download")
+async def api_zip_download(request: Request):
+ if request.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/x-www-form-urlencoded":
+ return JSONResponse({"error": "unsupported content type"}, status_code=415)
+ body_bytes = await request.body()
+ if len(body_bytes) > 65536:
+ return JSONResponse({"error": "request too large"}, status_code=413)
+ body = body_bytes.decode("utf-8", errors="ignore")
+ doc_ids = parse_qs(body, keep_blank_values=False).get("doc_id", [])
+ valid_ids = [
+ doc_id for doc_id in dict.fromkeys(doc_ids)
+ if (record := record_map.get(doc_id)) is not None and get_doc_storage_path(record).exists()
+ ]
+ if not valid_ids:
+ return JSONResponse({"error": "no files"}, status_code=400)
+ token = create_zip_download_token(valid_ids)
+ return RedirectResponse(f"/api/zip-download/{token}", status_code=303)
+
+@app.get("/api/zip-download")
+def api_zip_download_missing_token():
+ return JSONResponse({"error": "download token required"}, status_code=400)
+@app.get("/data/fulltext/{path:path}")
+def block_fulltext_artifacts(path: str):
+ return JSONResponse({"error": "not found"}, status_code=404)
+app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static"), html=True), name="static")
+app.mount("/icons", StaticFiles(directory=str(BASE_DIR / "static/icons")), name="icons")
+app.mount("/data/initial", StaticFiles(directory=str(BASE_DIR / "data/initial")), name="initial-data")
+app.mount("/data/sidebar", StaticFiles(directory=str(BASE_DIR / "data/sidebar")), name="sidebar-data")
+@lru_cache(maxsize=1)
+def _manifest_payload() -> dict:
+ return json.loads((BASE_DIR / "static/manifest.json").read_text(encoding="utf-8"))
-app.mount("/static", HashedStaticFiles(directory="static", html=True), name="static")
-app.mount("/icons", StaticFiles(directory="static/icons"), name="icons")
@app.get("/manifest.json")
def serve_manifest():
- p = Path("static/manifest.json")
- if p.exists():
- return JSONResponse(json.loads(p.read_text(encoding="utf-8")))
- return JSONResponse({"error": "manifest not found"}, status_code=404)
+ return JSONResponse(_manifest_payload())
+@lru_cache(maxsize=1)
+def _sw_source() -> str:
+ return (BASE_DIR / "static/sw.js").read_text(encoding="utf-8")
+
@app.get("/sw.js")
def serve_sw():
- p = Path("static/sw.js")
- if p.exists():
- return PlainTextResponse(p.read_text(encoding="utf-8"), media_type="application/javascript")
- return PlainTextResponse("", status_code=404)
-
-@app.get("/data/initial/global.json")
-def serve_initial_global():
- return JSONResponse(cached_payload(("initial_global",), lambda: initial_payload_global or build_initial_payload(None)))
-
-@app.get("/data/initial/repos/{repo_name}.json")
-def serve_initial_repo(repo_name: str):
- full_repo = f"VoiceOfML/{repo_name}"
- payload = initial_payload_by_repo.get(full_repo)
- if not payload:
- return JSONResponse({"error": f"仓库 {full_repo} 不存在"}, status_code=404)
- return JSONResponse(cached_payload(("initial_repo", full_repo), lambda: payload))
-
-@app.get("/data/initial/manifest.json")
-def serve_initial_manifest():
- def build():
- urls = ["/data/initial/global.json"]
- for repo in sorted(initial_payload_by_repo):
- urls.append(f"/data/initial/repos/{quote(repo.split('/')[-1], safe='')}.json")
- return {"urls": urls}
- return JSONResponse(cached_payload(("initial_manifest",), build))
-
-def read_sidebar_json(path: Path):
- data = load_sidebar_payload(path)
- if data is None:
- return JSONResponse({"error": "sidebar data not found"}, status_code=404)
- return JSONResponse(
- cached_payload(("sidebar_file", _data_generation, str(path)), lambda: data),
- headers=METADATA_CACHE_HEADERS,
- )
+ return PlainTextResponse(_sw_source(), media_type="application/javascript")
-@app.get("/data/sidebar/global.json")
-def serve_sidebar_global():
- return read_sidebar_json(SIDEBAR_DATA_PATH / "global.json")
-
-@app.get("/data/sidebar/repos/{repo_name}.json")
-def serve_sidebar_repo(repo_name: str):
- full_repo = f"VoiceOfML/{repo_name}"
- if full_repo not in repo_counts:
- return JSONResponse({"error": f"仓库 {full_repo} 不存在"}, status_code=404)
- return read_sidebar_json(SIDEBAR_DATA_PATH / "repos" / f"{quote(repo_name, safe='')}.json")
-
-@app.get("/data/sidebar/manifest.json")
-def serve_sidebar_manifest():
- return read_sidebar_json(SIDEBAR_DATA_PATH / "manifest.json")
-
-@app.get("/")
-async def serve_root():
- html = injected_html_cache.get(None)
- if html is not None:
- return HTMLResponse(html)
- index_path = Path("static/index.html")
- if index_path.exists():
- return HTMLResponse(inject_initial_payload(index_path.read_text(encoding="utf-8"), None))
- return PlainTextResponse("Frontend shell unavailable", status_code=503)
+def refresh_fulltext_loop() -> None:
+ while not fulltext_refresh_stop.wait(FULLTEXT_REFRESH_INTERVAL_SECONDS):
+ try:
+ if fulltext_databases is not None:
+ fulltext_databases.refresh()
+ if ready_fulltext_sources() == set(source_counts):
+ break
+ except Exception as exc:
+ print(f"fulltext_refresh_failed={exc}")
+
+def initial_payload_for_path(rest_of_path: str) -> str | None:
+ source_slug = unquote((rest_of_path or "").strip("/"))
+ if source_slug:
+ payload_path = BASE_DIR / "data" / "initial" / "sources" / f"{quote(source_slug, safe='')}.json"
+ else:
+ payload_path = BASE_DIR / "data" / "initial" / "global.json"
+ if not payload_path.exists():
+ return None
+ return payload_path.read_text(encoding="utf-8")
+def sidebar_payload_for_path(rest_of_path: str) -> str | None:
+ source_slug = unquote((rest_of_path or "").strip("/"))
+ if source_slug:
+ payload_path = BASE_DIR / "data" / "sidebar" / "sources" / f"{quote(source_slug, safe='')}.json"
+ else:
+ payload_path = BASE_DIR / "data" / "sidebar" / "global.json"
+ if not payload_path.exists():
+ return None
+ return payload_path.read_text(encoding="utf-8")
+
+@lru_cache(maxsize=64)
+def render_index_html(rest_of_path: str = "") -> str:
+ html = (BASE_DIR / "static/index.html").read_text(encoding="utf-8")
+ initial_payload = initial_payload_for_path(rest_of_path)
+ sidebar_payload = sidebar_payload_for_path(rest_of_path)
+ global_sidebar_payload = sidebar_payload_for_path("")
+ scripts = []
+ if initial_payload:
+ initial_payload = initial_payload.replace("{initial_payload}')
+ if sidebar_payload:
+ sidebar_payload = sidebar_payload.replace("{sidebar_payload}')
+ if global_sidebar_payload:
+ global_sidebar_payload = global_sidebar_payload.replace("{global_sidebar_payload}')
+ if not scripts:
+ return html
+ script = "\n".join(scripts) + "\n"
+ marker = '