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 = ( '
' '
' f"{len(docs)}{document_label}" "
" '
' f"{chunk_count}{chunk_label}" "
" "
" ) 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 ( '
' 'Waiting for scope' 'Select an indexed document to start learning.' "
" ) 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 ( '
' 'Indexed & ready' f'{len(names)} {document_label}' f'{visible_names}' f'Page: {page_label}' "
" ) 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"
  • {html.escape(item, quote=False)}
  • " for item in failures ) details = f"
    View errors
    " 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, )