feat: IPTC domain taxonomy, TOC keywords, and image region metadata

AGENTS.md

@@ -129,8 +129,10 @@ This is the typed `ParseOutput` entry for all file types. The parser flow:
 
 1. **Profiles** the document via `profiling.doc_profiler.profile_document()`. PDF
    profiling uses `document_agent` as the single PyMuPDF feature source, then runs
-   VLM coarse classification with two fields: open semantic `category` (for example,
-   `Financial Prospectus`) and routing-only `routing_category`
+   VLM coarse classification with three fields: `category` (IPTC Media Topics
+   top-level codes from the numbered list in `profiling/domain_taxonomy.py`,
+   multiple allowed), `category_label` (open short phrase, for example
+   `Financial Prospectus`), and routing-only `routing_category`
    (`atlas/scanned/slides/generic`). Oversized non-atlas PDFs additionally run the
    structural anatomy stage once at the entry point and pass the resulting shard
    plan to PDF parsing.

apps/worker/app/services/document_agent/coarse_profile/classifier.py

@@ -19,6 +19,10 @@
 from app.services.document_agent.coarse_profile.prompts import COARSE_PROFILE_INSTRUCTIONS
 from app.services.document_agent.manifest import DocumentProfile, ToolContext, ToolResult
 from app.services.document_agent.visual import render_pages
+from app.services.document_parser.profiling.domain_taxonomy import (
+    OTHER_CODE,
+    domain_codes_from_ids,
+)
 from app.services.document_parser.profiling.taxonomy import PdfRoutingCategory
 
 PAGE_KIND_DEFINITIONS = {
@@ -137,10 +141,15 @@ def _parse_profile(raw: str) -> DocumentProfile:
     data = json.loads(raw)
     if not isinstance(data, dict):
         raise ValueError("coarse profile output must be a JSON object")
-    category = " ".join(str(data.get("category") or "unknown document").split()[:5])
-    routing_category = PdfRoutingCategory.normalize(
-        data.get("routing_category") or data.get("category")
-    ).value
+    try:
+        category = domain_codes_from_ids(data.get("domain_ids"))
+    except ValueError as exc:
+        logger.warning("[document_agent] invalid domain_ids, using OTHER: {}", exc)
+        category = [OTHER_CODE]
+    category_label = " ".join(
+        str(data.get("category_label") or "unknown document").split()[:5]
+    )
+    routing_category = PdfRoutingCategory.normalize(data.get("routing_category")).value
     raw_is_scanned = data.get("is_scanned")
     if isinstance(raw_is_scanned, bool):
         is_scanned = raw_is_scanned
@@ -155,7 +164,8 @@ def _parse_profile(raw: str) -> DocumentProfile:
         footer_y = None
     return DocumentProfile(
         is_scanned=is_scanned,
-        category=category or "unknown document",
+        category=category,
+        category_label=category_label or "unknown document",
         routing_category=routing_category,
         language=str(data.get("language") or "unknown"),
         header_y=header_y,
@@ -197,7 +207,8 @@ def classify(self) -> tuple[DocumentProfile, ToolResult]:
         if not model:
             profile = DocumentProfile(
                 is_scanned=False,
-                category="unknown document",
+                category=[OTHER_CODE],
+                category_label="unknown document",
                 routing_category=PdfRoutingCategory.GENERIC.value,
             )
             return profile, ToolResult(

apps/worker/app/services/document_agent/coarse_profile/prompts.py

@@ -1,11 +1,17 @@
 """Prompts for one-shot coarse document profiling."""
 
+from app.services.document_parser.profiling.domain_taxonomy import DOMAIN_MENU
+
 COARSE_PROFILE_INSTRUCTIONS = (
     "You are a document profile classifier. Use page-feature statistics "
     "and the provided page screenshots to classify the PDF. "
-    "Return strict JSON only with keys: is_scanned, category, routing_category, "
-    "language, header_y, footer_y. "
-    "category is a concise semantic document type in at most 5 English words. "
+    "Return strict JSON only with keys: is_scanned, domain_ids, category_label, "
+    "routing_category, language, header_y, footer_y. "
+    "domain_ids is a non-empty JSON array of integers taken from the numbered "
+    "IPTC Media Topics list below. Pick every numbered topic that describes what "
+    "the document is about; section titles without a number are not selectable. "
+    "Pick the Other entry alone, and only when no listed topic fits. "
+    "category_label is a concise semantic document type in at most 5 English words. "
     "routing_category must be one of atlas, scanned, slides, generic. "
     "Set routing_category=atlas only when pages are primarily drawing/detail "
     "sheets rather than prose. "
@@ -15,7 +21,8 @@
     "you observe (largest y) when any header is present, otherwise null; "
     "footer_y is the highest footer line you observe (smallest y) when any "
     "footer is present, otherwise null. When both are set, require "
-    "header_y < footer_y."
+    "header_y < footer_y.\n"
+    "Domain list:\n" + DOMAIN_MENU
 )
 
 __all__ = ["COARSE_PROFILE_INSTRUCTIONS"]

apps/worker/app/services/document_agent/manifest.py

@@ -49,7 +49,9 @@ def to_dict(self) -> dict[str, Any]:
 @dataclass
 class DocumentProfile:
     is_scanned: bool
-    category: str
+    category: list[str]
+    """IPTC Media Topic codes picked from the domain taxonomy (e.g. ``["04000000"]``)."""
+    category_label: str
     routing_category: str = "generic"
     language: str = "unknown"
     # Content-band margins as fractions of page height (top origin, y down).

apps/worker/app/services/document_agent/tools/find_toc_anchor_pages.py

@@ -16,6 +16,9 @@
     page_bands_map,
 )
 from app.services.document_agent.registry import has_page_full_text, has_page_labels, register_tool
+from app.services.document_parser.profiling.domain_taxonomy import (
+    ECONOMY_BUSINESS_FINANCE_CODE,
+)
 from app.services.document_parser.structure.body_boundary import normalize_match_text
 from app.services.document_parser.formats.pdf.pymupdf_subprocess import (
     run_in_child_process,
@@ -25,34 +28,65 @@
 
 # CJK and English TOC keywords used for first-pass anchor detection.
 TOC_KEYWORDS = frozenset({"目录", "目次", "contents", "table of contents"})
+# Strict extras for every domain: match line-start only.
+# Exhibit lists appear in filings, contracts, and legal documents alike.
+GLOBAL_PREFIX_TOC_KEYWORDS = frozenset({"exhibit index"})
+# Domain extras: only when that IPTC category was picked; match line-start only.
+DOMAIN_PREFIX_TOC_KEYWORDS: dict[str, frozenset[str]] = {
+    # SEC filings: "index to/of …" covers financial statements and exhibits.
+    ECONOMY_BUSINESS_FINANCE_CODE: frozenset({"index to", "index of"}),
+}
 
 MAX_ANCHOR_CANDIDATES = 20
+HEAD_WINDOW_MAX_PAGES = 10
 # English keywords must sit at a line edge once candidates overflow; CJK 目录/目次 are exempt.
 EDGE_ANCHORED_KEYWORDS = frozenset({"contents", "table of contents"})
 MIN_CONSECUTIVE_DUPLICATE_RUN = 2
 
-# Upper bound on consecutive lines joined when repairing a keyword split by
-# newlines (e.g. 目\\n录). Derived from the longest keyword character length.
-_MAX_KEYWORD_SPLIT_LINES = max(len(keyword) for keyword in TOC_KEYWORDS)
+
+def _prefix_toc_keywords_for_categories(categories: object) -> frozenset[str]:
+    keywords: set[str] = set(GLOBAL_PREFIX_TOC_KEYWORDS)
+    if not isinstance(categories, list):
+        return frozenset(keywords)
+    codes = {str(code) for code in categories}
+    for code, extras in DOMAIN_PREFIX_TOC_KEYWORDS.items():
+        if code in codes:
+            keywords.update(extras)
+    return frozenset(keywords)
+
+
+def _all_scan_keywords(
+    *,
+    prefix_keywords: frozenset[str] = frozenset(),
+) -> frozenset[str]:
+    return TOC_KEYWORDS | prefix_keywords
+
+
+def _max_keyword_split_lines(keywords: frozenset[str]) -> int:
+    return max((len(keyword) for keyword in keywords), default=1)
 
 
-def _match_toc_keyword_parts(parts: list[str]) -> str | None:
+def _match_toc_keyword_parts(
+    parts: list[str],
+    keywords: frozenset[str],
+) -> str | None:
     candidates = {normalize_match_text(parts[0])} if parts else set()
     for part in parts[1:]:
         next_candidates: set[str] = set()
         for prefix in candidates:
             for separator in ("", " "):
                 candidate = normalize_match_text(f"{prefix}{separator}{part}")
-                if any(keyword.startswith(candidate) for keyword in TOC_KEYWORDS):
+                if any(keyword.startswith(candidate) for keyword in keywords):
                     next_candidates.add(candidate)
         candidates = next_candidates
         if not candidates:
             return None
-    return next((candidate for candidate in candidates if candidate in TOC_KEYWORDS), None)
+    return next((candidate for candidate in candidates if candidate in keywords), None)
 
 
 def _merge_keyword_split_lines(
     lines: list[str],
+    keywords: frozenset[str],
 ) -> list[tuple[str, int, int]]:
     """Merge consecutive lines that together exactly equal one TOC keyword.
 
@@ -61,12 +95,13 @@ def _merge_keyword_split_lines(
     """
     merged: list[tuple[str, int, int]] = []
     index = 0
+    max_parts = _max_keyword_split_lines(keywords)
     while index < len(lines):
         joined: tuple[str, int, int] | None = None
-        upper = min(_MAX_KEYWORD_SPLIT_LINES, len(lines) - index)
+        upper = min(max_parts, len(lines) - index)
         for part_count in range(upper, 1, -1):
             parts = [lines[index + offset].strip() for offset in range(part_count)]
-            keyword = _match_toc_keyword_parts(parts)
+            keyword = _match_toc_keyword_parts(parts, keywords)
             if keyword is None:
                 continue
             joined = (keyword, index, index + part_count - 1)
@@ -80,29 +115,59 @@ def _merge_keyword_split_lines(
     return merged
 
 
-def _match_toc_keyword_in_line(normalized_line: str) -> str | None:
-    """Return the longest TOC keyword contained in a normalized line."""
+def _match_toc_keyword_in_line(
+    normalized_line: str,
+    *,
+    keywords: frozenset[str],
+    prefix_keywords: frozenset[str],
+) -> str | None:
+    """Return the longest matching TOC keyword for a normalized line.
+
+    Base keywords use containment; domain extras require a line-start prefix.
+    """
     if not normalized_line:
         return None
-    return next(
+    contained = next(
         (
             keyword
-            for keyword in sorted(TOC_KEYWORDS, key=len, reverse=True)
+            for keyword in sorted(keywords - prefix_keywords, key=len, reverse=True)
             if keyword in normalized_line
         ),
         None,
     )
+    prefixed = next(
+        (
+            keyword
+            for keyword in sorted(prefix_keywords, key=len, reverse=True)
+            if normalized_line.startswith(keyword)
+        ),
+        None,
+    )
+    if contained is None:
+        return prefixed
+    if prefixed is None:
+        return contained
+    return contained if len(contained) >= len(prefixed) else prefixed
 
 
-def _find_toc_text_matches(lines: list[str]) -> list[dict[str, Any]]:
-    """Match TOC keywords as line-level containment after keyword-split repair.
+def _find_toc_text_matches(
+    lines: list[str],
+    *,
+    keywords: frozenset[str],
+    prefix_keywords: frozenset[str] = frozenset(),
+) -> list[dict[str, Any]]:
+    """Match TOC keywords after keyword-split repair.
 
-    Cross-line handling stays keyword-internal only (e.g. 目+录). Hit rule is
-    ``keyword in normalized_line`` (longest match wins), not whole-line equality.
+    Cross-line handling stays keyword-internal only (e.g. 目+录). Base keywords
+    use ``keyword in normalized_line``; domain extras require line-start only.
     """
     matches: list[dict[str, Any]] = []
-    for raw_line, start_idx, end_idx in _merge_keyword_split_lines(lines):
-        keyword = _match_toc_keyword_in_line(normalize_match_text(raw_line))
+    for raw_line, start_idx, end_idx in _merge_keyword_split_lines(lines, keywords):
+        keyword = _match_toc_keyword_in_line(
+            normalize_match_text(raw_line),
+            keywords=keywords,
+            prefix_keywords=prefix_keywords,
+        )
         if keyword is None:
             continue
         matches.append(
@@ -120,14 +185,20 @@ def _scan_toc_matches(
     bands_by_page: dict[int, PageTextBands],
     *,
     page_count: int,
+    prefix_keywords: frozenset[str] = frozenset(),
 ) -> list[dict[str, Any]]:
+    keywords = _all_scan_keywords(prefix_keywords=prefix_keywords)
     matches: list[dict[str, Any]] = []
     for page in range(1, page_count + 1):
         bands = bands_by_page.get(page)
         if bands is None:
             continue
         texts = [" ".join(line.text.split()) for line in bands.lines]
-        for match in _find_toc_text_matches(texts):
+        for match in _find_toc_text_matches(
+            texts,
+            keywords=keywords,
+            prefix_keywords=prefix_keywords,
+        ):
             covered = bands.lines[match["line_index"] : match["line_end_index"] + 1]
             matches.append(
                 {
@@ -228,6 +299,13 @@ def _select_anchor_pages(
     return _match_pages(matches), narrowing
 
 
+def _head_window_pages(page_count: int) -> set[int]:
+    """First ``min(page_count, 10)`` pages when keyword scan found nothing."""
+    if page_count <= 0:
+        return set()
+    return set(range(1, min(page_count, HEAD_WINDOW_MAX_PAGES) + 1))
+
+
 @worker
 def _render_pages_worker(
     queue, pdf_path: str, pages: list[int], output_dir: str, dpi: int
@@ -269,11 +347,28 @@ def find_toc_anchor_pages(ctx: ToolContext, _args: dict[str, Any]) -> ToolResult
     start = time.monotonic()
     total_pages = ctx.blackboard.page_count
 
+    profile = ctx.blackboard.document_profile
+    prefix_keywords = _prefix_toc_keywords_for_categories(
+        getattr(profile, "category", None)
+    )
+    logger.info(
+        "[find.toc_anchor_pages] line-start TOC keywords enabled: {}",
+        sorted(prefix_keywords),
+    )
     keyword_matches = _scan_toc_matches(
         page_bands_map(ctx.blackboard.page_full_text_cache),
         page_count=total_pages,
+        prefix_keywords=prefix_keywords,
     )
     anchor_pages, narrowing = _select_anchor_pages(keyword_matches)
+    if not anchor_pages and narrowing.get("raw") == 0:
+        anchor_pages = _head_window_pages(total_pages)
+        narrowing["head_window"] = len(anchor_pages)
+        logger.info(
+            "[find.toc_anchor_pages] no keyword candidates; "
+            "using first {} pages as head-window fallback",
+            len(anchor_pages),
+        )
     logger.info("[find.toc_anchor_pages] candidate narrowing: {}", narrowing)
     if narrowing.get("gave_up"):
         logger.warning(

apps/worker/app/services/document_parser/assets/inline_asset.py

@@ -13,6 +13,7 @@ def build_image_asset_row(
     keywords: str = "",
     entities: str = "",
     asset_title: str = "",
+    extra_metadata: str = "",
 ) -> ParsedRow:
     return ParsedRow(
         content=content,
@@ -26,6 +27,7 @@ def build_image_asset_row(
         addtime=addtime,
         entities=entities,
         asset_title=asset_title,
+        extra_metadata=extra_metadata,
     )
 
 

apps/worker/app/services/document_parser/formats/image/parser.py

@@ -11,7 +11,7 @@
 from app.services.document_parser.tables.dataframe_helpers import process_dup_paths_df
 from app.services.document_parser.support.identifiers import gen_str_codes, get_str_time
 from app.services.document_parser.support.parser_rows import (
-    PARSER_ROW_COLUMNS,
+    PARSER_DF_COLUMNS,
     ParsedRow,
     ParsedRowsBuilder,
 )
@@ -236,7 +236,7 @@ def parse_image(
         from shared.core.constants import ProcessingConstants
 
         if discard_undersized_image_file(img_path, label=f"image {filename}"):
-            return pd.DataFrame(columns=list(PARSER_ROW_COLUMNS))
+            return pd.DataFrame(columns=list(PARSER_DF_COLUMNS))
 
         # Extract image content
         client = _get_vision_client()