"""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}