69 lines
2.7 KiB
Python
69 lines
2.7 KiB
Python
"""Canonical page content and links shared by retrieval, scoring, and citations."""
|
|
from pathlib import Path
|
|
|
|
from .config import get_settings
|
|
from .models import KnowledgeDocument, KnowledgePage
|
|
|
|
|
|
def text_url(document_id: int, page: int) -> str:
|
|
return f"/api/knowledge/documents/{document_id}/pages/{page}"
|
|
|
|
|
|
def image_url(document_id: int, page: int) -> str:
|
|
return f"{text_url(document_id, page)}/image"
|
|
|
|
|
|
def page_text(source: dict) -> str:
|
|
blocks = [str(source.get("text") or "").strip()]
|
|
blocks.extend("表格: " + str(table.get("rows", table)) for table in source.get("tables", []))
|
|
return "\n".join(block for block in blocks if block)
|
|
|
|
|
|
def relative_image_path(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
root = get_settings().processed_path.resolve()
|
|
path = Path(value).resolve()
|
|
if not path.is_relative_to(root):
|
|
raise ValueError("页面图像路径不在处理目录内")
|
|
return str(path.relative_to(root))
|
|
|
|
|
|
def page_image_path(db, doc: KnowledgeDocument, page: int) -> Path | None:
|
|
root = get_settings().processed_path.resolve()
|
|
record = db.get(KnowledgePage, (doc.id, page))
|
|
path = (root / record.image_path).resolve() if record and record.image_path else None
|
|
return path if path and path.is_relative_to(root) and path.is_file() else None
|
|
|
|
|
|
def page_links(db, document_id: int | None, page: int | None) -> dict:
|
|
"""Stable citation links plus the page type, so callers can label the source."""
|
|
empty = {"url": None, "image_url": None, "page_type": None, "section": None}
|
|
if not document_id or not page:
|
|
return empty
|
|
doc = db.get(KnowledgeDocument, document_id)
|
|
if not doc:
|
|
return empty
|
|
record = db.get(KnowledgePage, (document_id, page))
|
|
if not record:
|
|
return empty
|
|
return {"url": text_url(document_id, page),
|
|
"image_url": image_url(document_id, page) if page_image_path(db, doc, page) else None,
|
|
"page_type": record.page_type or "unknown",
|
|
"section": record.section or ""}
|
|
|
|
|
|
def evidence_links(db, evidence) -> dict:
|
|
return page_links(db, evidence.document_id, evidence.page)
|
|
|
|
|
|
def page_types(db, document_ids) -> dict[int, str]:
|
|
"""Page type per page of the given documents, for labelling citations in bulk."""
|
|
from sqlalchemy import select as _select
|
|
ids = [item for item in dict.fromkeys(document_ids or []) if item]
|
|
if not ids:
|
|
return {}
|
|
rows = db.execute(_select(KnowledgePage.document_id, KnowledgePage.page, KnowledgePage.page_type)
|
|
.where(KnowledgePage.document_id.in_(ids))).all()
|
|
return {(document_id, page): (page_type or "unknown") for document_id, page, page_type in rows}
|