Spaces:
Running
Running
File size: 5,035 Bytes
aa0a3e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | from pathlib import Path
from typing import Any
from urllib.parse import quote, urlparse
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, Response
from . import runtime, source_files
from .doc_store import get_doc, random_doc, variants_for_title
from .document_identity import content_hash
from .storage_lock import serving_lock
router = APIRouter()
@router.get("/api/preview/{doc_id}")
async def preview(doc_id: str, request: Request):
if runtime.index_status() == "restoring":
return JSONResponse({"error": "index restore in progress"}, status_code=503)
with serving_lock():
stored = get_doc(doc_id)
if stored:
payload = await preview_payload(stored)
content_hash = str(payload["content_hash"])
etag = f'"{content_hash}"'
if request.headers.get("if-none-match") == etag:
return Response(status_code=304, headers={"ETag": etag, "Cache-Control": "private, max-age=0, must-revalidate"})
first_page = next((file.get("initial_page") or file.get("pages", [None])[0] for file in payload.get("source_files", []) if file.get("initial_page") or file.get("pages")), None)
headers = {"ETag": etag, "Cache-Control": "private, max-age=0, must-revalidate"}
if first_page:
headers["Link"] = f"<{first_page}>; rel=preload; as=image"
return JSONResponse(payload, headers=headers)
return JSONResponse({"error": "not found"}, status_code=404)
@router.get("/api/random")
async def random_preview():
if runtime.index_status() == "restoring":
return JSONResponse({"error": "index restore in progress"}, status_code=503)
with serving_lock():
stored = random_doc()
if stored:
return await preview_payload(stored)
return JSONResponse({"error": "not found"}, status_code=404)
async def preview_payload(source: dict[str, Any]) -> dict[str, Any]:
doc_id = str(source.get("doc_id") or "")
title = source.get("title")
variants: list[dict[str, Any]] = []
if title:
with serving_lock():
variants = variants_for_title(str(title), doc_id, source.get("publication_name"))
source_items = []
for index, url in enumerate(source.get("source_files") or []):
source_url = str(url)
hostname = (urlparse(source_url).hostname or "").lower()
entry = await source_files.source_mirror_entry(source_url) if hostname in {"raw.githubusercontent.com", "github.com"} else None
page_paths = entry.get("page_previews", {}).get("paths", []) if entry else []
pages = [
f"{source_files.SOURCE_MIRROR_BASE_URL}/{quote(path, safe='/')}"
for path in page_paths
if isinstance(path, str) and path.startswith("archives") and ".." not in Path(path).parts
]
item = {"index": index, "name": Path(source_url.split("?", 1)[0]).name or f"文件 {index + 1}", "url": source_url}
mirror_path = entry.get("path") if entry else None
if (not source_url.lower().split("?", 1)[0].endswith(".pdf")
and isinstance(mirror_path, str) and mirror_path.startswith("archives")
and ".." not in Path(mirror_path).parts):
item["preview_url"] = f"{source_files.SOURCE_MIRROR_BASE_URL}/{quote(mirror_path, safe='/')}"
if pages:
item["page_count"] = len(page_paths)
article_page = int((source.get("article") or {}).get("page_start") or 1)
requested_numbers = [number for number in (article_page, article_page + 1) if 1 <= number <= len(page_paths)]
item["pages"] = [
f"{source_files.SOURCE_MIRROR_BASE_URL}/{quote(page_paths[number - 1], safe='/')}"
for number in requested_numbers
if isinstance(page_paths[number - 1], str) and page_paths[number - 1].startswith("archives") and ".." not in Path(page_paths[number - 1]).parts
]
item["page_numbers"] = requested_numbers[:len(item["pages"])]
if item["pages"]:
item["initial_page"] = item["pages"][0]
item["initial_page_number"] = item["page_numbers"][0]
source_items.append(item)
return {
"doc_id": doc_id,
"content_hash": content_hash(source),
"title": title,
"publication_name": source.get("publication_name"),
"publication_type": source.get("publication_type"),
"source_file_count": int(source.get("source_file_count") or len(source.get("source_files") or [])),
"date_display": source.get("date_display") or [],
"tag_names": source.get("tag_names") or [],
"authors": source.get("authors") or [],
"content": source.get("content") or "",
"article_id": source.get("article_id"),
"publication_id": source.get("publication_id"),
"archive_id": source.get("archive_id"),
"article": source.get("article") or {},
"source_files": source_items,
"variants": variants,
}
|