Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import html | |
| from pathlib import Path | |
| from typing import Any | |
| import gradio as gr | |
| from src.export import safe_markdown_code | |
| from src.filters import MetadataFilter, filters_to_dict | |
| from src.indexing import save_and_ingest_pdf | |
| from src.store import list_documents | |
| from src.ui.content import EMPTY_LIBRARY_HTML | |
| from src.ui.helpers import int_value, status_html | |
| def read_uploaded_pdf(file_obj: object) -> tuple[bytes, str]: | |
| if isinstance(file_obj, str): | |
| path = Path(file_obj) | |
| return path.read_bytes(), path.name | |
| raw_path = getattr(file_obj, "path", None) | |
| raw_name = getattr(file_obj, "orig_name", None) | |
| if isinstance(raw_path, str) and raw_path: | |
| path = Path(raw_path) | |
| name = raw_name.strip() if isinstance(raw_name, str) and raw_name.strip() else path.name | |
| return path.read_bytes(), name | |
| if isinstance(file_obj, dict): | |
| raw_path = file_obj.get("path") | |
| raw_name = file_obj.get("orig_name") or file_obj.get("name") | |
| if isinstance(raw_path, str) and raw_path: | |
| path = Path(raw_path) | |
| name = raw_name.strip() if isinstance(raw_name, str) and raw_name.strip() else path.name | |
| return path.read_bytes(), name | |
| raise TypeError(f"Unsupported uploaded file type: {type(file_obj).__name__}") | |
| def read_uploaded_pdfs(file_obj: object) -> list[tuple[bytes, str]]: | |
| if file_obj is None: | |
| return [] | |
| if isinstance(file_obj, (list, tuple)): | |
| return [read_uploaded_pdf(item) for item in file_obj] | |
| return [read_uploaded_pdf(file_obj)] | |
| def build_filters(filenames: list[str] | None, page: int | None) -> dict[str, object] | None: | |
| payload: dict[str, object] = {} | |
| if filenames: | |
| payload["filenames"] = filenames | |
| if page is not None: | |
| payload["page"] = page | |
| return filters_to_dict(MetadataFilter.model_validate(payload)) if payload else None | |
| def refresh_docs() -> tuple[ | |
| dict[str, Any], | |
| dict[str, dict[str, object]], | |
| dict[str, Any], | |
| str, | |
| str, | |
| ]: | |
| docs = list_documents() | |
| choices = [str(doc["filename"]) for doc in docs] | |
| doc_map = {str(doc["filename"]): doc for doc in docs} | |
| if docs: | |
| document_label = "File" if len(docs) == 1 else "Files" | |
| chunk_count = sum(int_value(doc["chunk_count"]) for doc in docs) | |
| chunk_label = "Chunk" if chunk_count == 1 else "Chunks" | |
| summary = ( | |
| '<div class="library-stats" aria-label="Indexed library summary">' | |
| '<div class="library-stat">' | |
| f"<strong>{len(docs)}</strong><span>{document_label}</span>" | |
| "</div>" | |
| '<div class="library-stat">' | |
| f"<strong>{chunk_count}</strong><span>{chunk_label}</span>" | |
| "</div>" | |
| "</div>" | |
| ) | |
| else: | |
| summary = EMPTY_LIBRARY_HTML | |
| filenames_text = ( | |
| "\n".join(f"- {safe_markdown_code(name)}" for name in choices) if choices else "" | |
| ) | |
| return ( | |
| gr.update(choices=choices, value=[]), | |
| doc_map, | |
| gr.update( | |
| choices=["All pages"], | |
| value="All pages", | |
| interactive=False, | |
| ), | |
| summary, | |
| filenames_text, | |
| ) | |
| def pages_for_selection(doc_map: dict[str, Any], selected: list[str]) -> dict[str, Any]: | |
| if len(selected) != 1: | |
| return gr.update( | |
| choices=["All pages"], | |
| value="All pages", | |
| interactive=False, | |
| ) | |
| doc = doc_map.get(selected[0]) or {} | |
| raw_pages = doc.get("pages") | |
| pages = raw_pages if isinstance(raw_pages, list) else [] | |
| page_choices = ["All pages", *[str(page) for page in pages]] | |
| return gr.update( | |
| choices=page_choices, | |
| value="All pages", | |
| interactive=True, | |
| ) | |
| def scope_summary_html(selected: list[str] | None, page: str | None) -> str: | |
| names = [str(name) for name in (selected or []) if str(name).strip()] | |
| if not names: | |
| return ( | |
| '<div class="scope-summary scope-summary-empty">' | |
| '<span class="scope-summary-status">Waiting for scope</span>' | |
| '<span>Select an indexed document to start learning.</span>' | |
| "</div>" | |
| ) | |
| visible_names = ", ".join(html.escape(name, quote=False) for name in names[:2]) | |
| if len(names) > 2: | |
| visible_names += f" and {len(names) - 2} more" | |
| page_label = html.escape(str(page or "All pages"), quote=False) | |
| document_label = "document" if len(names) == 1 else "documents" | |
| return ( | |
| '<div class="scope-summary" aria-live="polite">' | |
| '<span class="scope-summary-status scope-summary-ready">Indexed & ready</span>' | |
| f'<strong>{len(names)} {document_label}</strong>' | |
| f'<span class="scope-summary-files">{visible_names}</span>' | |
| f'<span class="scope-summary-page">Page: {page_label}</span>' | |
| "</div>" | |
| ) | |
| def upload_pdf( | |
| file: object | None, | |
| ) -> tuple[str, dict[str, Any], dict[str, dict[str, object]], dict[str, Any], str, str]: | |
| payloads = read_uploaded_pdfs(file) | |
| if not payloads: | |
| choices, doc_map, page_dropdown, summary, filenames_text = refresh_docs() | |
| return ( | |
| status_html("Please select at least one PDF file."), | |
| choices, | |
| doc_map, | |
| page_dropdown, | |
| summary, | |
| filenames_text, | |
| ) | |
| successes: list[str] = [] | |
| failures: list[str] = [] | |
| chunks_total = 0 | |
| for file_bytes, filename in payloads: | |
| try: | |
| info = save_and_ingest_pdf(file_bytes, filename) | |
| except (OSError, RuntimeError, TypeError, ValueError) as error: | |
| failures.append(f"{filename}: {error}") | |
| continue | |
| successes.append(str(info["filename"])) | |
| chunks_total += int_value(info.get("chunks_indexed")) | |
| parts: list[str] = [] | |
| if successes: | |
| parts.append(f"Loaded {len(successes)} file · {chunks_total} chunks") | |
| if failures: | |
| parts.append(f"Could not load {len(failures)} file") | |
| details = "" | |
| if failures: | |
| items = "".join( | |
| f"<li><code>{html.escape(item, quote=False)}</code></li>" for item in failures | |
| ) | |
| details = f"<details><summary>View errors</summary><ul>{items}</ul></details>" | |
| body = (" · ".join(parts) if parts else "No valid files.") + details | |
| choices, doc_map, page_dropdown, summary, filenames_text = refresh_docs() | |
| return ( | |
| status_html(body, trusted=True), | |
| choices, | |
| doc_map, | |
| page_dropdown, | |
| summary, | |
| filenames_text, | |
| ) | |