"-joined words, and # outline-number/punctuation differences), scanning bookmarks in document order so # repeated titles stay in sync. An item only anchors when its bookmark has a deeper # child bookmark — i.e. the item actually carries supporting documents in the packet # — which keeps procedural items (whose bookmark merely points at the agenda front # matter) from claiming a section. # --------------------------------------------------------------------------- # _OUTLINE_MATCH_MIN_LEN = 6 # squeezed item title shorter than this is too generic _OUTLINE_MATCH_PREFIX = 28 # compare this many squeezed leading chars def _squeeze_title(title: str) -> str: """Lowercased alphanumeric-only form of a title (tags + enumerator stripped). Collapsing to ``[a-z0-9]`` makes the item title and the packet bookmark title line up despite HTML markup, ``
``-joined words, non-breaking/zero-width spaces, and outline-number/punctuation differences.""" t = _TAG_RE.sub(" ", title or "") t = _ENUM_RE.sub("", t.strip()) return re.sub(r"[^a-z0-9]+", "", t.lower()) def _outline_anchors(outline: list[dict], titles: list[str]) -> list[int | None]: """For each agenda item, the 0-indexed packet page from its bookmark (or None). ``outline`` is :func:`pdf_utils.extract_pdf_outline`'s flat bookmark list. Anchors only content-bearing items (whose bookmark has a deeper child) and never the front-matter page; everything else is ``None`` so the caller can fall back to text anchoring. """ anchors: list[int | None] = [None] * len(titles) if not outline: return anchors # A bookmark is content-bearing iff the next bookmark nests under it (its own # attachment/report), so its page is where the item's documents actually start. has_child = [ i + 1 < len(outline) and outline[i + 1]["level"] > outline[i]["level"] for i in range(len(outline)) ] cand = [ (outline[i]["page"], _squeeze_title(outline[i]["title"]), has_child[i]) for i in range(len(outline)) if outline[i]["page"] is not None ] pos = 0 # bookmarks are consumed in document order to disambiguate repeats for i, title in enumerate(titles): key = _squeeze_title(title) if len(key) < _OUTLINE_MATCH_MIN_LEN: continue for j in range(pos, len(cand)): page, bkey, child = cand[j] if not bkey: continue if bkey.startswith(key[:_OUTLINE_MATCH_PREFIX]) or key.startswith( bkey[:_OUTLINE_MATCH_PREFIX] ): pos = j + 1 if child and page > 0: anchors[i] = page break return anchors def _slice_packet_for_item( pages: list[str], titles: list[str], index: int, outline: list[dict] | None = None, ) -> tuple[str, dict]: """Slice an agenda packet down to the selected item's section. The packet is one compiled PDF. We locate each item's section by **two anchoring strategies, outline first**: 1. *Outline* — match the item to its bookmark in the packet's PDF outline (the packet author's own item→page map). Deterministic and exact when present. 2. *Text* — fallback when an item has no bookmark: anchor the item's title to a packet *content* page by whitespace-normalized substring match, skipping outline/TOC pages (which list several item titles at once). The selected item's slice runs from its anchor page to the next later-anchored item's page (text anchors are capped; outline anchors aren't, since they're exact). When the item can't be confidently located by either strategy, we don't guess — the caller falls back to the whole packet. Returns ``(sliced_text, info)`` where ``info`` is ``{"sliced", "pages", "start", "end", "method"}`` (``start``/``end`` are 0-indexed, end-exclusive page bounds, or ``None`` when not sliced; ``method`` is ``"outline"``/``"text"``/``""``). ``sliced_text`` is empty when no confident anchor was found. """ norm_pages = [_norm_ws(p) for p in pages] keys = [_title_key(t) for t in titles] probes = [k[:_ITEM_ANCHOR_KEY_LEN] for k in keys] out_anchors = _outline_anchors(outline or [], titles) def distinctive(k: str) -> bool: return len(k) >= _ITEM_ANCHOR_MIN_LEN def text_anchor(idx: int) -> int | None: """First content page whose text contains item ``idx``'s title and at most one *other* item title (i.e. not an outline/TOC page).""" if not (0 <= idx < len(keys)) or not distinctive(keys[idx]): return None probe = probes[idx] for pi, npg in enumerate(norm_pages): if probe not in npg: continue others = sum(1 for j, p in enumerate(probes) if j != idx and distinctive(keys[j]) and p and p in npg) if others < _ITEM_OUTLINE_OTHER_TITLES: return pi return None def anchor(idx: int) -> int | None: """Item ``idx``'s packet start page — its bookmark if any, else its title.""" if 0 <= idx < len(out_anchors) and out_anchors[idx] is not None: return out_anchors[idx] return text_anchor(idx) not_sliced = {"sliced": False, "pages": "", "start": None, "end": None, "method": ""} start = anchor(index) if start is None: return "", dict(not_sliced) method = "outline" if (0 <= index < len(out_anchors) and out_anchors[index] is not None) else "text" later = [a for a in (anchor(j) for j in range(index + 1, len(keys))) if a is not None and a > start] if later: end = min(later) elif method == "outline": end = len(pages) # exact map — run to the packet end else: end = min(len(pages), start + _ITEM_SLICE_MAX_PAGES) text = "\n\n".join(p for p in pages[start:end] if p).strip() if not text: return "", dict(not_sliced) return text, {"sliced": True, "pages": f"{start + 1}-{end}", "start": start, "end": end, "method": method} def _outline_slice_for_item( outline: list[dict], titles: list[str], index: int, page_count: int, ) -> dict: """Outline-anchored slice info for one item — **without** per-page text. The page viewer renders by page *index*, so when the packet's PDF outline anchors the item we never need to extract the whole packet's text. We match the item to its bookmark exactly as :func:`_outline_anchors` does (squeezed-title prefix, in document order, only content-bearing bookmarks) so the viewer's start page agrees with the report slicer — then bound the section at the **next bookmark of the same or a higher level** (the item's next sibling or the next section), i.e. the bookmark's natural extent. That is the item's true page span; we fall back to the packet end only when no such bookmark follows. Returns the same ``info`` dict shape (``method`` always ``"outline"``), or ``{"sliced": False, ...}`` when the item has no content-bearing bookmark — the caller then falls back to the text-based slice. """ not_sliced = {"sliced": False, "pages": "", "start": None, "end": None, "method": ""} if page_count <= 0 or not outline or not (0 <= index < len(titles)): return dict(not_sliced) # Candidate bookmarks in document order, each carrying its level and whether it is # content-bearing (has a deeper child = the item's own attachments/reports). has_child = [ i + 1 < len(outline) and outline[i + 1]["level"] > outline[i]["level"] for i in range(len(outline)) ] cand = [ (outline[i]["page"], outline[i]["level"], _squeeze_title(outline[i]["title"]), has_child[i]) for i in range(len(outline)) if outline[i]["page"] is not None ] pos = 0 # bookmarks consumed in document order to disambiguate repeated titles for ti, title in enumerate(titles): key = _squeeze_title(title) if len(key) < _OUTLINE_MATCH_MIN_LEN: if ti == index: return dict(not_sliced) continue match_j = None for j in range(pos, len(cand)): page, level, bkey, child = cand[j] if bkey and (bkey.startswith(key[:_OUTLINE_MATCH_PREFIX]) or key.startswith(bkey[:_OUTLINE_MATCH_PREFIX])): match_j, pos = j, j + 1 break if ti != index: continue if match_j is None: return dict(not_sliced) page, level, _, child = cand[match_j] if not (child and page > 0): return dict(not_sliced) end = page_count for k in range(match_j + 1, len(cand)): npage, nlevel = cand[k][0], cand[k][1] if nlevel <= level and npage > page: end = npage break if end <= page: return dict(not_sliced) return {"sliced": True, "pages": f"{page + 1}-{end}", "start": page, "end": end, "method": "outline"} return dict(not_sliced) # --------------------------------------------------------------------------- # # Build the agenda item tree from a parsed packet — bookmarks first, text fallback. # --------------------------------------------------------------------------- # def _items_from_outline(outline: list[dict], page_count: int) -> list[AgendaItem]: """Build the agenda item tree from the packet's PDF bookmarks. Surfaces level-0 (top-level items + section headers) and level-1 (sub-items) bookmarks as agenda items; deeper bookmarks are an item's own attachments and are not surfaced. An item is content-bearing (``has_pages``) iff its bookmark has a deeper child AND points beyond the agenda front matter (page index > 0); its page span runs to the next bookmark of the same-or-higher level (its next sibling / section). 0-indexed ``start``/``end`` (end-exclusive) plus a 1-indexed ``pages`` string, matching the slicer. """ if not outline: return [] has_child = [ i + 1 < len(outline) and outline[i + 1]["level"] > outline[i]["level"] for i in range(len(outline)) ] items: list[AgendaItem] = [] n = 0 for i, bm in enumerate(outline): level = int(bm.get("level", 0) or 0) if level > 1: # attachments / reports — not surfaced as agenda items continue page = bm.get("page") number, name = _split_enumerator(bm.get("title", "")) if not name: continue content = bool(has_child[i] and page is not None and page > 0) start = end = 0 pages_str = "" if content: sec_end = page_count for k in range(i + 1, len(outline)): npage, nlevel = outline[k].get("page"), int(outline[k].get("level", 0) or 0) if npage is not None and nlevel <= level and npage > page: sec_end = npage break if sec_end > page: start, end, pages_str = page, sec_end, f"{page + 1}-{sec_end}" else: content = False n += 1 items.append(AgendaItem( id=f"outline:{n}", number=number, name=name, summary="", is_section=(level == 0 and not number), level=level, status="", attachments=0, has_pages=content, pages=pages_str, start=start, end=end, )) return items def _items_from_text(pages: list[str], agenda_pages: str) -> list[AgendaItem]: """Text fallback when the packet has no bookmarks. Parses the user-marked agenda page range into an item list, then text-anchors each item into the later packet pages to fill ``start``/``end``/``has_pages``/``pages``. """ if not pages: return [] a_start, a_end = _parse_page_range(agenda_pages, len(pages)) agenda_text = "\n\n".join(p for p in pages[a_start:a_end] if p).strip() items = _parse_agenda_outline(agenda_text) if not items: return [] titles = [f"{it['number']} {it['name']}".strip() for it in items] for idx, it in enumerate(items): _, info = _slice_packet_for_item(pages, titles, idx, outline=[]) if info.get("sliced"): it["has_pages"] = True it["pages"] = info["pages"] it["start"] = int(info["start"]) it["end"] = int(info["end"]) return items def _parse_page_range(agenda_pages: str, page_count: int) -> tuple[int, int]: """Parse a 1-indexed inclusive page range (``"1-3"`` / ``"5"``) to a 0-indexed, end-exclusive ``(start, end)`` slice, clamped to ``[0, page_count]``. Empty input defaults to the first few pages (a typical agenda front section).""" s = (agenda_pages or "").strip() if not s: return 0, min(page_count, 3) if page_count else 0 m = re.match(r"^\s*(\d+)\s*(?:[-–to]+\s*(\d+))?\s*$", s) if not m: return 0, min(page_count, 3) if page_count else 0 lo = max(1, int(m.group(1))) hi = int(m.group(2)) if m.group(2) else lo if hi < lo: lo, hi = hi, lo start = max(0, lo - 1) end = hi if page_count <= 0 else min(page_count, hi) if end <= start: end = min(page_count or hi, start + 1) return start, end def parse_agenda_outline_from_packet( upload_id: str, agenda_pages: str = "" ) -> AgendaOutline: """Parse an uploaded packet into agenda items — bookmarks first, text fallback. Returns the item tree (each item carrying ``has_pages`` + 0-indexed ``start``/``end`` + a 1-indexed ``pages`` string) plus ``source`` (``outline`` | ``text`` | ``none``), a ``confidence`` hint, a message, and a suggested agenda ``title``. Item names truncated by the PDF bookmarks are restored deterministically from the agenda (table-of-contents) page text — no LLM. """ def result(items, source, confidence, message, title="") -> AgendaOutline: return {"items": items, "source": source, "confidence": confidence, "message": message, "model": "", "title": title} if not upload_id: return result([], "none", "empty", "Upload an agenda packet first.") if not agenda_pages: with _PKT_LOCK: meta = _PKT_META.get(upload_id) agenda_pages = (meta or {}).get("agenda_pages", "") outline = cached_packet_outline(upload_id) page_count = cached_packet_page_count(upload_id) if page_count <= 0: return result([], "none", "empty", "The packet is no longer on the server — re-upload it.") if outline: items = _items_from_outline(outline, page_count) if items: # Restore truncated bookmark titles + a default agenda name from the TOC text. agenda_text = _agenda_text_for(upload_id, agenda_pages, page_count) if agenda_text: _untruncate_item_names(items, agenda_text) return result(items, "outline", "good", "Parsed from the packet's PDF bookmarks.", title=_agenda_title_from_text(agenda_text)) pages = cached_packet_pages(upload_id) items = _items_from_text(pages, agenda_pages) if not items: return result([], "none", "empty", "No agenda items could be parsed — set the agenda page range, " "or add items/ranges manually.") a_start, a_end = _parse_page_range(agenda_pages, len(pages)) agenda_text = "\n\n".join(p for p in pages[a_start:a_end] if p).strip() n_numbered = sum(1 for it in items if it["number"] and not it["is_section"]) confidence = "good" if n_numbered >= 3 else "poor" return result(items, "text", confidence, "Parsed from the agenda text (no PDF bookmarks) — verify the page " "ranges.", title=_agenda_title_from_text(agenda_text)) # --------------------------------------------------------------------------- # # Local packet cache (keyed by upload_id). # # The uploaded Agenda Packet is one multi-MB PDF used by every feature: the page # viewer renders pages from it, the item report reads its text, the agent searches # it. We save it once to a local file at upload time and reuse it everywhere. Per-page # text and the bookmark outline (the next-biggest costs) are extracted from the saved # PDF once and memoized alongside. The cache is keyed by the server-minted upload_id. # --------------------------------------------------------------------------- # _PKT_DIR = Path( os.getenv("PACKET_CACHE_DIR") or (Path(tempfile.gettempdir()) / "agenda_packets") ) _PKT_META: "OrderedDict[str, dict]" = OrderedDict() # uid -> {path, name, agenda_pages} _PKT_PAGES: "OrderedDict[str, list]" = OrderedDict() # uid -> [page text] _PKT_OUTLINE: "OrderedDict[str, list]" = OrderedDict() # uid -> [bookmark] _PKT_PAGECOUNT: "OrderedDict[str, int]" = OrderedDict() # uid -> page count _PKT_CACHE_MAX = 24 # Gradio runs API handlers in a threadpool, so the caches above and the in-flight # map below are shared across threads. One lock guards every mutation/read of them. # The expensive work (text extraction) runs OUTSIDE the lock via the single-flight # helper, so the lock is only ever held briefly. _PKT_LOCK = threading.Lock() # pypdfium2 (PDFium) is NOT thread-safe — it keeps global state, so two threads doing # any PDFium work at once (rendering pages, counting pages, extracting text/outline) # corrupt the native heap and abort the process ("corrupted double-linked list"). # Gradio runs API handlers in a threadpool, so EVERY PDFium call in this module must be # serialized behind this one global lock. _PDFIUM_LOCK = threading.Lock() class _Flight: """One in-flight expensive op for a cache key — followers share its result.""" __slots__ = ("event", "result", "error") def __init__(self) -> None: self.event = threading.Event() self.result = None self.error: BaseException | None = None _PKT_PAGES_FLIGHTS: "dict[str, _Flight]" = {} # text extractions def _compute_once(flights: "dict[str, _Flight]", key: str, producer): """Run ``producer()`` once across threads for ``key``; share its result/error. The first caller becomes the leader and runs ``producer()`` **without holding** ``_PKT_LOCK`` (it may parse for many seconds); concurrent callers become followers that wait on the leader's event and return the same result (or re-raise the same exception). This collapses a burst of rapid item clicks into a single extraction instead of N duplicate, mutually-starving ones. """ with _PKT_LOCK: flight = flights.get(key) leader = flight is None if leader: flight = _Flight() flights[key] = flight if not leader: flight.event.wait() if flight.error is not None: raise flight.error return flight.result try: flight.result = producer() except BaseException as e: # noqa: BLE001 - shared with followers, then re-raised flight.error = e finally: with _PKT_LOCK: flights.pop(key, None) flight.event.set() if flight.error is not None: raise flight.error return flight.result def _evict_packet(uid: str) -> None: """Drop a packet's file + every derived cache entry. Caller holds ``_PKT_LOCK``.""" meta = _PKT_META.pop(uid, None) _PKT_PAGES.pop(uid, None) _PKT_OUTLINE.pop(uid, None) _PKT_PAGECOUNT.pop(uid, None) if meta: try: Path(meta["path"]).unlink() except OSError: pass def upload_packet( file_bytes: bytes, filename: str, agenda_pages: str = "", upload_id: str = "" ) -> dict: """Persist an uploaded agenda packet and parse its agenda outline. Writes the PDF under ``_PKT_DIR`` keyed by a server-minted ``upload_id`` (or the one passed in, for a re-upload — which overwrites in place and refreshes the parse), registers the cache meta, evicting the least-recently-used packet past the cap. Returns ``{upload_id, name, size, page_count, agenda_pages, items, source, confidence, message, model}``. """ uid = upload_id or secrets.token_urlsafe(12) _PKT_DIR.mkdir(parents=True, exist_ok=True) safe = re.sub(r"[^A-Za-z0-9_-]", "_", uid) path = _PKT_DIR / f"{safe}.pdf" path.write_bytes(file_bytes) name = (filename or "agenda-packet.pdf").strip() if not name.lower().endswith(".pdf"): name += ".pdf" with _PKT_LOCK: # A re-upload reuses the uid — clear any stale derived caches for it. _PKT_PAGES.pop(uid, None) _PKT_OUTLINE.pop(uid, None) _PKT_PAGECOUNT.pop(uid, None) _PKT_META[uid] = {"path": str(path), "name": name, "agenda_pages": agenda_pages} _PKT_META.move_to_end(uid) while len(_PKT_META) > _PKT_CACHE_MAX: old_uid = next(iter(_PKT_META)) if old_uid == uid: break _evict_packet(old_uid) page_count = cached_packet_page_count(uid) outline_result = parse_agenda_outline_from_packet(uid, agenda_pages) return { "upload_id": uid, "name": name, "size": len(file_bytes), "page_count": page_count, "agenda_pages": agenda_pages, **outline_result, } def cached_packet(upload_id: str) -> tuple[Path, str] | None: """Local path + filename of an uploaded packet, or ``None`` if unknown/evicted. The packet is written at upload time, so this is a pure cache lookup. ``None`` means the upload is gone (tmp wiped on a Space restart, or LRU-evicted) — the caller surfaces a "re-upload" prompt. """ if not upload_id: return None with _PKT_LOCK: meta = _PKT_META.get(upload_id) if meta is not None and Path(meta["path"]).exists(): _PKT_META.move_to_end(upload_id) return Path(meta["path"]), meta["name"] # Bundled samples are always re-derivable from disk: on a cache miss (Space restart, # LRU eviction, or a saved sample from a prior session), rehydrate from samples/ # rather than 404. User uploads have no such source, so they still surface no_packet. if upload_id.startswith(_SAMPLE_UID_PREFIX): s = _SAMPLE_BY_ID.get(upload_id[len(_SAMPLE_UID_PREFIX):]) if s is not None and (_SAMPLES_DIR / s["file"]).exists(): return _rehydrate_sample(upload_id, s) return None def _rehydrate_sample(upload_id: str, s: dict) -> tuple[Path, str] | None: """Re-cache a bundled sample's PDF on disk + register its meta (no re-parse), so the /packet route can serve it after the in-memory cache lost it.""" _PKT_DIR.mkdir(parents=True, exist_ok=True) safe = re.sub(r"[^A-Za-z0-9_-]", "_", upload_id) path = _PKT_DIR / f"{safe}.pdf" try: if not path.exists(): path.write_bytes((_SAMPLES_DIR / s["file"]).read_bytes()) except OSError: return None with _PKT_LOCK: _PKT_META[upload_id] = {"path": str(path), "name": s["file"], "agenda_pages": s["agenda_pages"]} _PKT_META.move_to_end(upload_id) return path, s["file"] def _extract_pages_text(src: bytes | str) -> list[str]: """Per-page text via pypdfium2 (PDFium native). ~20-25x faster than pdfplumber's pdfminer backend on a 150-page packet (≈0.4s vs ≈9s), and the slicer's whitespace-normalized title anchoring lands on exactly the same pages — we only need the page *text*, not pdfplumber's layout analysis. Accepts PDF bytes or a path (renders from the locally-cached file). """ import pypdfium2 as pdfium out: list[str] = [] with _PDFIUM_LOCK: # PDFium is not thread-safe — serialize all access pdf = pdfium.PdfDocument(src) try: for i in range(len(pdf)): textpage = pdf[i].get_textpage() try: out.append(textpage.get_text_range()) finally: textpage.close() finally: pdf.close() return out def cached_packet_pages(upload_id: str) -> list[str]: """Per-page text of the cached packet PDF — extracted once and memoized. Drives the slicer's text fallback (and the item report). Reads the locally-saved PDF. Returns ``[]`` when the upload is unknown/evicted. The page viewer avoids this whole-packet text pass when the PDF outline anchors the item (see :func:`agenda_item_pages`). Concurrent callers share one extraction. """ if not upload_id: return [] with _PKT_LOCK: cached = _PKT_PAGES.get(upload_id) if cached is not None: _PKT_PAGES.move_to_end(upload_id) return cached cp = cached_packet(upload_id) if cp is None: return [] def _extract() -> list[str]: with _PKT_LOCK: cached = _PKT_PAGES.get(upload_id) if cached is not None: _PKT_PAGES.move_to_end(upload_id) return cached pages = _extract_pages_text(str(cp[0])) # CPU — no lock held with _PKT_LOCK: _PKT_PAGES[upload_id] = pages _PKT_PAGES.move_to_end(upload_id) while len(_PKT_PAGES) > _PKT_CACHE_MAX: _PKT_PAGES.popitem(last=False) return pages return _compute_once(_PKT_PAGES_FLIGHTS, upload_id, _extract) def cached_packet_page_count(upload_id: str) -> int: """Number of pages in the cached packet PDF — read once and memoized. The outline-only slice (page viewer) needs just the page count for its end bound, not the per-page text. Opens the locally-saved PDF (no text extraction). Returns 0 when the upload is unknown/evicted. """ if not upload_id: return 0 with _PKT_LOCK: cached = _PKT_PAGECOUNT.get(upload_id) if cached is not None: _PKT_PAGECOUNT.move_to_end(upload_id) return cached cp = cached_packet(upload_id) if cp is None: return 0 import pypdfium2 as pdfium with _PDFIUM_LOCK: # PDFium is not thread-safe — serialize all access pdf = pdfium.PdfDocument(str(cp[0])) try: count = len(pdf) finally: pdf.close() with _PKT_LOCK: _PKT_PAGECOUNT[upload_id] = count _PKT_PAGECOUNT.move_to_end(upload_id) while len(_PKT_PAGECOUNT) > _PKT_CACHE_MAX: _PKT_PAGECOUNT.popitem(last=False) return count def cached_packet_outline(upload_id: str) -> list[dict]: """The cached packet PDF's bookmark outline — read once and memoized. Feeds the bookmarks-first parse and the slicer's outline anchoring (see :func:`_outline_anchors`). Reads the locally-saved PDF; returns ``[]`` when the upload is unknown/evicted or the packet carries no outline. """ if not upload_id: return [] with _PKT_LOCK: cached = _PKT_OUTLINE.get(upload_id) if cached is not None: _PKT_OUTLINE.move_to_end(upload_id) return cached cp = cached_packet(upload_id) if cp is None: return [] try: with _PDFIUM_LOCK: # extract_pdf_outline uses PDFium — serialize all access outline = extract_pdf_outline(Path(cp[0]).read_bytes()) except Exception: # noqa: BLE001 - a packet with no/garbled outline just falls back outline = [] with _PKT_LOCK: _PKT_OUTLINE[upload_id] = outline _PKT_OUTLINE.move_to_end(upload_id) while len(_PKT_OUTLINE) > _PKT_CACHE_MAX: _PKT_OUTLINE.popitem(last=False) return outline def _agenda_pages_for(upload_id: str) -> str: with _PKT_LOCK: meta = _PKT_META.get(upload_id) return (meta or {}).get("agenda_pages", "") # --------------------------------------------------------------------------- # # Deterministic title untruncation — recover full item titles (and a default agenda # name) by searching the agenda (table-of-contents) page text. PDF bookmark titles are # commonly truncated (~50 chars) and may drop a department prefix; the agenda text holds # each item's full title, so we find the truncated substring there and lift the whole # enclosing title block. No LLM. # --------------------------------------------------------------------------- # _UNTRUNC_MIN_KEY = 24 # squeezed bookmark shorter than this isn't worth searching _UNTRUNC_MAX_LEN = 320 # never adopt a runaway "title" longer than this # A line that ends an item's title block (its metadata / a page footer / the AGENDA # banner). Continuation lines (wrapped title text) match none of these. _TITLE_META_RE = re.compile( r"(?i)^\s*(item category|presenter|action required|action|fiscal impact|department|" r"sponsor|recommended action|recommendation|background|summary|whereas|now therefore)\b" ) _PAGE_FOOTER_RE = re.compile(r"(?i)(page \d+ of \d+|agenda page \d+)") # Lenient enumerator at the START of a line ("11.", "a)", "iv.", or even "1 " / "2 "). _ENUM_LINE_RE = re.compile(r"^\s*(?:\d{1,3}|[a-zA-Z]|[ivxlcdmIVXLCDM]{1,6})[.):]?\s+\S") # Lenient enumerator to strip from a recovered full title (allows the bare "1 " form). _ENUM_LEAD_RE = re.compile(r"^\s*(?:\d{1,3}|[a-z]|[ivxlcdm]{1,6})[.):]?\s+", re.IGNORECASE) def _is_title_boundary(line: str) -> bool: s = line.strip() if not s: return True return bool(_TITLE_META_RE.match(s) or _PAGE_FOOTER_RE.search(s) or _OUTLINE_START_RE.match(s)) def _title_block(lines: list[str], hit: int) -> str: """The full title block of agenda ``lines`` that the matched line ``hit`` belongs to. Walks up to the block's enumerator line (or a boundary above) and down across wrapped continuation lines, stopping at the next enumerator / a metadata line / a page footer. """ top = hit while top > 0: if _ENUM_LINE_RE.match(lines[top]): break if _is_title_boundary(lines[top - 1]): break top -= 1 bot = hit while bot + 1 < len(lines): nxt = lines[bot + 1] if _is_title_boundary(nxt) or _ENUM_LINE_RE.match(nxt): break bot += 1 return " ".join(ln.strip() for ln in lines[top:bot + 1] if ln.strip()) def _untruncate_item_names(items: list[dict], agenda_text: str) -> int: """Replace truncated bookmark item names with their full title from the agenda text. For each item, find its (enumerator-stripped, alphanumeric-squeezed) bookmark title as a substring of the squeezed agenda text — in document order to disambiguate repeats — then lift the enclosing title block. Adopts it only when it's genuinely longer. Returns the number of names improved. """ lines = agenda_text.splitlines() squeezed_chars: list[str] = [] line_of: list[int] = [] # squeezed char index -> agenda line index for li, ln in enumerate(lines): for ch in ln: if ch.isalnum(): squeezed_chars.append(ch.lower()) line_of.append(li) squeezed = "".join(squeezed_chars) if not squeezed: return 0 cursor = 0 changed = 0 for it in items: name = it.get("name", "") bkey = _squeeze_title(name) if len(bkey) < _UNTRUNC_MIN_KEY: continue q = squeezed.find(bkey, cursor) if q == -1: # fall back to a global search (order drift) q = squeezed.find(bkey) if q == -1: continue cursor = q + len(bkey) full = _title_block(lines, line_of[q]) full = _ENUM_LEAD_RE.sub("", full).strip() if full and len(name) < len(full) <= _UNTRUNC_MAX_LEN: it["name"] = full changed += 1 return changed # Header lines (before the AGENDA banner) that name the meeting + its date. _AGENDA_DATE_RE = re.compile( r"(?i)(?:(?:mon|tue|wed|thu|fri|sat|sun)[a-z]*,?\s+)?" r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\.?\s+\d{1,2},?\s+\d{4}" ) _AGENDA_BODY_RE = re.compile( r"(?i)\b(board|council|commission|committee|authority|district|trustees|supervisors|" r"township|village|city|county|district|department|agency)\b" ) def _agenda_title_from_text(agenda_text: str) -> str: """A concise default agenda name (governing body + date) from the header lines. Deterministic, best-effort: scans the lines before the AGENDA banner for a governing- body line and a date. Returns ``""`` when nothing usable is found (caller falls back to the file name).""" lines = [re.sub(r"\s+", " ", ln).strip() for ln in agenda_text.splitlines()] lines = [ln for ln in lines if ln] end = len(lines) for i, ln in enumerate(lines): if _OUTLINE_START_RE.match(ln): end = i break header = lines[:end] or lines[:15] date = "" for ln in header: m = _AGENDA_DATE_RE.search(ln) if m: date = m.group(0).strip(" ,") break body = "" for ln in header: if _AGENDA_DATE_RE.search(ln) or any(c.isdigit() for c in ln): continue if 4 <= len(ln) <= 70 and _AGENDA_BODY_RE.search(ln): body = ln.title() if ln.isupper() else ln break return " — ".join(p for p in (body, date) if p)[:90] def _agenda_text_for(upload_id: str, agenda_pages: str, page_count: int) -> str: """Text of just the marked agenda (table-of-contents) pages — a targeted PDFium read so big packets don't pay a full text extraction at parse time.""" cp = cached_packet(upload_id) if cp is None or page_count <= 0: return "" a_start, a_end = _parse_page_range(agenda_pages, page_count) import pypdfium2 as pdfium out: list[str] = [] with _PDFIUM_LOCK: # PDFium is not thread-safe — serialize all access pdf = pdfium.PdfDocument(str(cp[0])) try: for i in range(max(0, a_start), min(len(pdf), a_end)): textpage = pdf[i].get_textpage() try: out.append(textpage.get_text_range()) finally: textpage.close() finally: pdf.close() return "\n\n".join(p for p in out if p).strip() def _packet_documents(upload_id: str) -> list[dict]: """Chroma documents for the whole uploaded packet (one doc; chunking splits it). Replaces the old per-event document loader. Used by the report, the agent's semantic search, and the summarize/report tools. """ cp = cached_packet(upload_id) if cp is None: return [] name = cp[1] pages = cached_packet_pages(upload_id) text = "\n\n".join(p for p in pages if p).strip() if not text: return [] return [{ "doc_id": f"{upload_id}:packet", "text": text, "metadata": {"source": "packet", "upload_id": upload_id, "file_name": name}, }] # --------------------------------------------------------------------------- # # Summarize (chunk the agenda portion -> Gemma-4) # --------------------------------------------------------------------------- # def summarize_agenda(upload_id: str, model: str = "") -> Iterator[dict]: """Stream a summary of the uploaded agenda's table-of-contents portion. Summarizes the text of the page range the user marked as the agenda (not the whole multi-hundred-page packet). Yields progress dicts ``{"stage", "message", "summary", "chunks", "model"}`` where ``stage`` is ``working`` | ``done`` | ``error``. Args: upload_id: the uploaded packet handle. model: which GGUF to run (local backend) — ``"e4b"`` (default) or ``"26b"``. """ def msg(stage: str, message: str = "", summary: str = "", chunks: int = 0, model: str = "") -> dict: return {"stage": stage, "message": message, "summary": summary, "chunks": chunks, "model": model} if not upload_id: yield msg("error", "Upload an agenda packet first.") return yield msg( "working", "Reading the agenda and summarizing. The model may cold-start on the first " "call — this can take a minute.", ) pages = cached_packet_pages(upload_id) if not pages: yield msg("error", "The packet is no longer on the server — re-upload it.") return a_start, a_end = _parse_page_range(_agenda_pages_for(upload_id), len(pages)) text = "\n\n".join(p for p in pages[a_start:a_end] if p).strip() if not text: yield msg( "error", "The agenda pages have no extractable text — they may be scanned / " "image-only. Try the Report tab, which reads the whole packet.", ) return n_chunks = len(chunk_text(text)) try: if LLM_BACKEND == "local": from webapp import local_llm summary = local_llm.gpu_summarize(text, model) else: summary = summarize_text(text, complete=_complete) except Exception as e: # noqa: BLE001 yield msg("error", f"Summarization failed: {type(e).__name__}: {e}") return yield msg( "done", f"Summarized the agenda pages {a_start + 1}-{a_end} ({n_chunks} block(s)).", summary=summary or "_(empty summary)_", chunks=n_chunks, model=_model_label(model), ) # --------------------------------------------------------------------------- # # Report streaming (shared by the whole-packet and per-item reports) # --------------------------------------------------------------------------- # def _report_msg(stage: str, message: str = "", report: str = "", frac: float = 0.0, docs: int = 0, model: str = "", code: str = "") -> dict: return {"stage": stage, "message": message, "report": report, "frac": frac, "docs": docs, "model": model, "code": code} def _stream_report(docs: list[dict], question: str, max_sections: int, built_from: str, model: str = "") -> Iterator[dict]: """Run a query-framed report over pre-fetched ``docs`` and stream progress. Bridges :func:`chroma.report.generate_report` (remote) or :func:`webapp.local_llm.gpu_report` (in-process ZeroGPU) to ``working`` / ``done`` / ``error`` frames. ``built_from`` is surfaced in the done frame; ``model`` selects the GGUF (local backend). """ import queue import threading def done_frame(report: str) -> dict: return _report_msg( "done", built_from, report=report or "_(empty report)_", frac=1.0, docs=len(docs), model=_model_label(model), ) # Local backend: stream straight from the in-process GPU generator. if LLM_BACKEND == "local": from webapp import local_llm try: for frame in local_llm.gpu_report(docs, question, max_sections, model): if "error" in frame: yield _report_msg("error", f"Report generation failed: {frame['error']}") return if frame.get("done"): yield done_frame(frame.get("report", "")) return yield _report_msg("working", frame.get("message", "Working…"), frac=float(frame.get("frac", 0.0))) except Exception as e: # noqa: BLE001 yield _report_msg("error", f"Report generation failed: {type(e).__name__}: {e}") return # Remote backend: bridge generate_report's progress callback to this generator. q: "queue.Queue[tuple]" = queue.Queue() def cb(frac: float, message: str) -> None: q.put(("progress", frac, message)) def worker() -> None: try: report = generate_report( docs, question, complete=_complete, max_chunks=int(max_sections), progress=cb, ) q.put(("done", report)) except Exception as e: # noqa: BLE001 q.put(("error", f"{type(e).__name__}: {e}")) threading.Thread(target=worker, daemon=True).start() while True: kind, *rest = q.get() if kind == "progress": frac, message = rest yield _report_msg("working", message, frac=float(frac)) elif kind == "done": yield done_frame(rest[0]) return else: # error yield _report_msg("error", f"Report generation failed: {rest[0]}") return def _stream_item_answer(docs: list[dict], question: str, max_sections: int, built_from: str, engine: str = "auto", model: str = "") -> Iterator[dict]: """Answer one agenda item in a single pass and stream progress. Bridges :func:`webapp.local_llm.gpu_answer` (in-process ZeroGPU, token-streamed) or :func:`chroma.answer.answer_question` (remote) to ``working`` / ``done`` / ``error`` frames. ``engine`` is the user's choice — ``"single"`` (semantic search / one-shot answer) or ``"mapreduce"`` (read every section) — defaulting to ``"auto"`` (size-based heuristic). ``model`` selects the GGUF (local backend only). ``working`` frames carry the partial ``report`` so the UI renders the answer as it streams. """ import queue import threading def done_frame(report: str) -> dict: return _report_msg( "done", built_from, report=report or "_(empty answer)_", frac=1.0, docs=len(docs), model=_model_label(model), ) # Local backend: stream tokens straight from the in-process GPU generator. if LLM_BACKEND == "local": from webapp import local_llm try: for frame in local_llm.gpu_answer(docs, question, max_sections, engine, model): if "error" in frame: yield _report_msg("error", f"Answer generation failed: {frame['error']}") return if frame.get("done"): yield done_frame(frame.get("report", "")) return yield _report_msg("working", frame.get("message", "Working…"), report=frame.get("report", ""), frac=float(frame.get("frac", 0.0))) except Exception as e: # noqa: BLE001 yield _report_msg("error", f"Answer generation failed: {type(e).__name__}: {e}") return # Remote backend: single-pass answer (or map-reduce fallback) in a worker thread. from chroma.answer import answer_question, budget_for, resolve_engine char_budget, max_tokens = budget_for(max_sections) q: "queue.Queue[tuple]" = queue.Queue() def cb(frac: float, message: str) -> None: q.put(("progress", frac, message)) def worker() -> None: try: if resolve_engine(docs, engine, char_budget=char_budget) == "mapreduce": report = generate_report(docs, question, complete=_complete, max_chunks=int(max_sections), progress=cb) else: report = answer_question(docs, question, complete=_complete, char_budget=char_budget, max_tokens=max_tokens, progress=cb) q.put(("done", report)) except Exception as e: # noqa: BLE001 q.put(("error", f"{type(e).__name__}: {e}")) threading.Thread(target=worker, daemon=True).start() while True: kind, *rest = q.get() if kind == "progress": frac, message = rest yield _report_msg("working", message, frac=float(frac)) elif kind == "done": yield done_frame(rest[0]) return else: # error yield _report_msg("error", f"Answer generation failed: {rest[0]}") return # --------------------------------------------------------------------------- # # Agenda Report (query-framed over the whole uploaded packet) # --------------------------------------------------------------------------- # def generate_agenda_report( upload_id: str, question: str, max_sections: int = 60, model: str = "", ) -> Iterator[dict]: """Stream a query-framed markdown report over the whole uploaded packet. Chunks the packet into an ephemeral in-memory ChromaDB collection, mines every chunk for facts relevant to ``question`` (map) and writes the findings up as markdown (reduce) — see :func:`chroma.report.generate_report`. Yields progress dicts ``{"stage", "frac", "message", "report", "docs", "model", "code"}`` where ``stage`` is ``working`` | ``done`` | ``error``. Args: upload_id: the uploaded packet handle. question: the free-form question framing the report. max_sections: cap on how many chunks are mined (higher = more thorough, slower). model: which GGUF to run (local backend) — ``"e4b"`` (default) or ``"26b"``. """ question = (question or "").strip() if not upload_id: yield _report_msg("error", "Upload an agenda packet, then ask a question.") return if not question: yield _report_msg("error", "Enter a question to frame the report.") return yield _report_msg("working", "Reading the agenda packet…", frac=0.02) docs = _packet_documents(upload_id) if not docs: yield _report_msg( "error", "The packet is no longer on the server — re-upload it.", code="no_packet", ) return built_from = (f"Built from the agenda packet, chunked into an ephemeral ChromaDB " "collection.") yield from _stream_report(docs, question, max_sections, built_from, model) # --------------------------------------------------------------------------- # # Agenda Item Report (slice the packet to one item's section, report over it) # --------------------------------------------------------------------------- # def _item_documents( upload_id: str, item_titles: list[str], item_index: int, start: int = 0, end: int = 0, ) -> tuple[list[dict], dict]: """Build report documents scoped to one agenda item. Reads the packet from the local cache and scopes it to the item's pages. When an explicit page range is given (``end > start``, the user-adjusted range from the client), that range is used directly (clamped); otherwise the item is located by anchoring (outline first, text fallback — see :func:`_slice_packet_for_item`). The item's own agenda line is always included as a focusing document. Returns ``(documents, info)`` with ``info`` = ``{"sliced", "pages", "note"}``. """ cp = cached_packet(upload_id) packet_name = cp[1] if cp is not None else "Agenda Packet" pages = cached_packet_pages(upload_id) if cp is not None else [] outline = cached_packet_outline(upload_id) if cp is not None else [] title = item_titles[item_index] if 0 <= item_index < len(item_titles) else "" base_meta = {"source": "packet", "upload_id": str(upload_id)} docs: list[dict] = [] if title: docs.append({ "doc_id": f"{upload_id}:item", "text": f"Selected agenda item: {title}", "metadata": {**base_meta, "file_name": "Selected agenda item"}, }) if not pages: return docs, {"sliced": False, "pages": "", "note": "", "code": "no_packet"} # Explicit (possibly user-adjusted) range wins — clamp to the packet. s = max(0, int(start or 0)) e = min(len(pages), int(end or 0)) if e > s: sliced_text = "\n\n".join(p for p in pages[s:e] if p).strip() if sliced_text: info = {"sliced": True, "pages": f"{s + 1}-{e}", "start": s, "end": e, "method": "explicit"} docs.append({ "doc_id": f"{upload_id}:packet-slice", "text": sliced_text, "metadata": {**base_meta, "file_name": packet_name, "pages": info["pages"]}, }) info["note"] = (f"Reporting on pages {info['pages']} of {len(pages)} " "(the item's backup pages).") return docs, info sliced_text, info = _slice_packet_for_item( pages, item_titles, item_index, outline=outline) if info["sliced"]: docs.append({ "doc_id": f"{upload_id}:packet-slice", "text": sliced_text, "metadata": {**base_meta, "file_name": packet_name, "pages": info["pages"]}, }) how = ("from the packet's bookmarks" if info.get("method") == "outline" else "by matching the item's title in the packet") info["note"] = (f"Sliced the agenda packet to this item's section " f"(pages {info['pages']} of {len(pages)}, {how}).") else: docs.append({ "doc_id": f"{upload_id}:packet", "text": "\n\n".join(p for p in pages if p), "metadata": {**base_meta, "file_name": packet_name}, }) info["note"] = ("Couldn't locate this item's pages in the packet, so this " "report covers the full packet, framed by the item. Set the " "item's page range to scope it.") return docs, info def generate_agenda_item_report( upload_id: str, item_titles: list[str], item_index: int, question: str, start: int = 0, end: int = 0, max_sections: int = 60, engine: str = "auto", model: str = "", ) -> Iterator[dict]: """Stream a query-framed report scoped to a single agenda item. Scopes the packet to the item's backup pages — using the explicit (user-adjusted) ``start``/``end`` range when given, else anchoring by bookmark/title (see :func:`_item_documents`) — and answers the question over that slice in a single streamed pass. When the pages can't be located the answer falls back to the full packet, with a note saying so. Args: upload_id: the uploaded packet handle. item_titles: every agenda item's ``"