diff --git a/apps/worker/app/services/connect_builder/summary_builder.py b/apps/worker/app/services/connect_builder/summary_builder.py index f55dec47..c02a57c4 100644 --- a/apps/worker/app/services/connect_builder/summary_builder.py +++ b/apps/worker/app/services/connect_builder/summary_builder.py @@ -3,6 +3,8 @@ Reads doc_nav.json and in-memory chunks, generates ``summary`` at every non-leaf via deterministic covers assembly or LLM aggregation (with self_only). +Persists document-level ``top_summary`` on doc_nav (default LLM); section-level +LLM remains opt-in via ``use_llm``. Usage (standalone): from app.services.connect_builder.summary_builder import enrich_doc_nav_summaries @@ -381,30 +383,70 @@ def _build_nav_top_summary( self_only_lookup: Optional[Dict[str, str]] = None, source_file_name: str = "", ) -> str: - """Build navigation-facing top summary from enriched doc_nav.json.""" - sections = doc_nav.get("sections", []) + """Build document-level top summary from already-enriched section nodes. + + Children are never re-summarized with LLM here — section LLM is controlled + only by ``enrich_doc_nav_summaries(use_llm=...)``. This path may optionally + LLM-aggregate the document overview from child contributions. + """ + sections = list(doc_nav.get("sections") or []) if not sections: return "" file_name = source_file_name or str(doc_nav.get("file_name") or "") - virtual_doc_node = { - "title": "Document Overview", - "children": sections, - } - return _recursive_summarize_nav( - virtual_doc_node, - use_llm=use_llm, + # Fill any missing child summaries deterministically without enabling LLM. + for section in sections: + if isinstance(section, dict): + _recursive_summarize_nav( + section, + use_llm=False, + self_only_lookup=self_only_lookup, + source_file_name=file_name, + ) + + child_titles = _child_title_list(sections, is_top_level=True) + child_rows = _child_contribution_rows(sections, is_top_level=True) + contrib_len = sum(len(contrib) for _, contrib in child_rows) + deterministic = _deterministic_section_summary( is_top_level=True, - self_only_lookup=self_only_lookup, - source_file_name=file_name, + self_only="", + child_titles=child_titles, + ) + if not use_llm or contrib_len <= SUMMARY_MAX_LEN: + return deterministic + + llm_result = _llm_summarize( + node_name="Document Overview", + self_only="", + child_rows=child_rows, + max_tokens=NAVIGATION_TOP_SUMMARY_MAX_TOKENS, ) + return llm_result or deterministic + + +def _persist_doc_nav_top_summary( + *, + file_dir: str, + doc_nav: Dict[str, Any], + top_summary: str, +) -> str: + """Write document-level top_summary once onto doc_nav and save.""" + cleaned = str(top_summary or "").strip() + if cleaned: + doc_nav["top_summary"] = cleaned + elif "top_summary" in doc_nav: + doc_nav.pop("top_summary", None) + _save_doc_nav(file_dir, doc_nav) + return cleaned def enrich_doc_nav_summaries( document_workspace_dir: str, source_file: Optional[str] = None, force: bool = False, - use_llm: bool = True, + use_llm: bool = False, + *, + top_summary_use_llm: bool = True, chunks: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, str]: """Enrich doc_nav.json with bottom-up recursive summaries. @@ -413,14 +455,16 @@ def enrich_doc_nav_summaries( document_workspace_dir: Absolute path to the temporary document workspace. source_file: If given, only process this file. Otherwise process all. force: If True, regenerate even if summaries already exist. - use_llm: If True, use LLM when contribution length sum exceeds threshold. + use_llm: Section-level LLM summaries (default off). + top_summary_use_llm: Document-level top summary LLM (default on). chunks: In-memory parse chunks for exact-path self_only extraction. Returns: Dict mapping file_name → top-level summary string. """ results: Dict[str, str] = {} - mode_label = "LLM" if use_llm else "title-concat" + section_mode = "LLM" if use_llm else "title-concat" + top_mode = "LLM" if top_summary_use_llm else "title-concat" if source_file: targets = [source_file] @@ -445,21 +489,39 @@ def enrich_doc_nav_summaries( source_file_name=source_file_name, ) - if not force and _doc_nav_has_enriched_summaries(doc_nav): + existing_top = str(doc_nav.get("top_summary") or "").strip() + if ( + not force + and _doc_nav_has_enriched_summaries(doc_nav) + and existing_top + ): logger.debug( f"Summaries already exist in {DOC_NAV_FILENAME} for {file_name}, skipping" ) - results[file_name] = _build_nav_top_summary( + results[file_name] = existing_top + continue + + if not force and _doc_nav_has_enriched_summaries(doc_nav): + logger.info( + f"📝 Building missing {DOC_NAV_FILENAME} top_summary for {file_name} " + f"(top_mode={top_mode})" + ) + top_summary = _build_nav_top_summary( doc_nav, - use_llm=use_llm, + use_llm=top_summary_use_llm, self_only_lookup=self_only_lookup, source_file_name=source_file_name, ) + results[file_name] = _persist_doc_nav_top_summary( + file_dir=file_dir, + doc_nav=doc_nav, + top_summary=top_summary, + ) continue logger.info( f"📝 Enriching {DOC_NAV_FILENAME} summaries for {file_name} " - f"(mode={mode_label})" + f"(section_mode={section_mode}, top_mode={top_mode})" ) for section in doc_nav.get("sections", []): @@ -470,31 +532,36 @@ def enrich_doc_nav_summaries( source_file_name=source_file_name, ) - _save_doc_nav(file_dir, doc_nav) - logger.info(f"✅ doc_nav summaries saved for {file_name}") - top_summary = _build_nav_top_summary( doc_nav, - use_llm=use_llm, + use_llm=top_summary_use_llm, self_only_lookup=self_only_lookup, source_file_name=source_file_name, ) - results[file_name] = top_summary + results[file_name] = _persist_doc_nav_top_summary( + file_dir=file_dir, + doc_nav=doc_nav, + top_summary=top_summary, + ) + logger.info(f"✅ doc_nav summaries saved for {file_name}") return results def load_nav_top_summary(file_dir: str, file_name: str = "") -> str: - """Load doc_nav.json and extract the navigation top summary.""" + """Load persisted doc_nav top_summary, with deterministic fallback.""" doc_nav = _load_doc_nav(file_dir) - if doc_nav is not None: - source_file_name = file_name or str(doc_nav.get("file_name") or "") - return _build_nav_top_summary( - doc_nav, - use_llm=False, - source_file_name=source_file_name, - ) - return "" + if doc_nav is None: + return "" + existing = str(doc_nav.get("top_summary") or "").strip() + if existing: + return existing + source_file_name = file_name or str(doc_nav.get("file_name") or "") + return _build_nav_top_summary( + doc_nav, + use_llm=False, + source_file_name=source_file_name, + ) def build_section_summary_lookup(file_dir: str) -> Dict[str, str]: @@ -530,11 +597,7 @@ def _walk(node: Dict[str, Any]) -> None: _walk(section) if "Root" not in lookup: - top_summary = _build_nav_top_summary( - doc_nav, - use_llm=False, - source_file_name=source_file_name, - ) + top_summary = load_nav_top_summary(file_dir, source_file_name) if top_summary: lookup["Root"] = top_summary diff --git a/apps/worker/app/services/document_ingestion/success_finalization.py b/apps/worker/app/services/document_ingestion/success_finalization.py index 29ecd4be..9acc0e3a 100644 --- a/apps/worker/app/services/document_ingestion/success_finalization.py +++ b/apps/worker/app/services/document_ingestion/success_finalization.py @@ -48,7 +48,6 @@ def finalize_parse_success( job_context=job_context, source_file_name=source_file_name, ) - _attach_document_top_summary(result_package.chunks, document_top_summary) _refresh_processing_stages(job_context) lifecycle_service.update_progress( @@ -92,6 +91,7 @@ def finalize_parse_success( stored_count=stored_count, delivery_mode="url", section_summaries=section_summaries, + document_top_summary=document_top_summary, ) if finalization_response.get("status") != "success": logger.error( @@ -137,6 +137,7 @@ def _enrich_document_navigation( ) -> tuple[str, dict[str, str]]: document_top_summary = "" section_summaries: dict[str, str] = {} + enrich_results: dict[str, str] = {} add_dir = artifact.add_dir parsed_contents_df = artifact.dataframe if add_dir and source_file_name: @@ -153,16 +154,27 @@ def _enrich_document_navigation( "summary_use_llm", False, ) - enrich_doc_nav_summaries( + top_summary_use_llm = JobMetadataHelper.get_parsing_param( + job_context.job_metadata, + "top_summary_use_llm", + True, + ) + enrich_results = enrich_doc_nav_summaries( document_root_for_enrich, source_file=source_file_name, use_llm=summary_use_llm, + top_summary_use_llm=top_summary_use_llm, chunks=chunks, ) section_summaries = build_section_summary_lookup(str(add_dir)) except Exception as exc: logger.warning(f"doc_nav enrichment failed (non-fatal): {exc}") - document_top_summary = load_nav_top_summary(str(add_dir), source_file_name) + enrich_results = {} + document_top_summary = str( + enrich_results.get(source_file_name) or "" + ).strip() + if not document_top_summary: + document_top_summary = load_nav_top_summary(str(add_dir), source_file_name) return document_top_summary, section_summaries @@ -181,21 +193,6 @@ def _refresh_processing_stages(job_context: ParseJobContext) -> None: job_context.job_metadata["stages"] = stages -def _attach_document_top_summary( - chunks: list[dict[str, Any]], - document_top_summary: str, -) -> None: - if not document_top_summary: - return - - for chunk in chunks: - metadata = chunk.get("metadata") - if not isinstance(metadata, dict): - metadata = {} - chunk["metadata"] = metadata - metadata["document_top_summary"] = document_top_summary - - def _record_processing_completion( *, job_id: str, diff --git a/apps/worker/app/services/document_parser/assets/image_size_filter.py b/apps/worker/app/services/document_parser/assets/image_size_filter.py new file mode 100644 index 00000000..3c2c59dc --- /dev/null +++ b/apps/worker/app/services/document_parser/assets/image_size_filter.py @@ -0,0 +1,46 @@ +"""Shared minimum image size gate used across parsers. + +Discard before rename / VLM summary so undersized icons never enter the +asset pipeline. +""" + +from __future__ import annotations + +from pathlib import Path + +from loguru import logger + +from shared.core.constants.processing import ProcessingConstants + + +def is_below_img_min_size(byte_size: int) -> bool: + """Return True when ``byte_size`` is under ``IMG_MIN_SIZE`` (10KB).""" + return byte_size < ProcessingConstants.IMG_MIN_SIZE + + +def discard_undersized_image_file( + path: Path | str, + *, + label: str = "image", +) -> bool: + """Delete ``path`` when it exists and is undersized. + + Returns True when the file was discarded (caller should skip rename/LLM). + Returns False when the file is large enough to keep, or does not exist. + """ + image_path = Path(path) + if not image_path.exists(): + return False + + file_size = image_path.stat().st_size + if not is_below_img_min_size(file_size): + return False + + logger.debug( + f"Skipping {label} (too small: {file_size / 1024:.1f} KB): {image_path}" + ) + try: + image_path.unlink() + except OSError as exc: + logger.debug(f"Failed to remove undersized {label} {image_path}: {exc}") + return True diff --git a/apps/worker/app/services/document_parser/assets/inline_asset.py b/apps/worker/app/services/document_parser/assets/inline_asset.py index c39917d6..27348a19 100644 --- a/apps/worker/app/services/document_parser/assets/inline_asset.py +++ b/apps/worker/app/services/document_parser/assets/inline_asset.py @@ -38,12 +38,18 @@ def build_table_asset_row( addtime: str, entities: str = "", asset_title: str = "", + image_refs: list[str] | None = None, ) -> ParsedRow: row_content = relative_path + # Multiline type channel carries table→image embeds + # (same pattern as PTXT\n[tables/...] for text rows). + type_value = "table" + if image_refs: + type_value = "\n".join(["table", *image_refs]) return ParsedRow( content=row_content, path=relative_path, - type="table", + type=type_value, keywords=keywords, summary=summary, know_id=know_id, diff --git a/apps/worker/app/services/document_parser/formats/docx/block_stream.py b/apps/worker/app/services/document_parser/formats/docx/block_stream.py index dafe28ab..3d8ddacf 100644 --- a/apps/worker/app/services/document_parser/formats/docx/block_stream.py +++ b/apps/worker/app/services/document_parser/formats/docx/block_stream.py @@ -5,6 +5,7 @@ import zipfile from app.services.document_parser.formats.docx.toc import detect_doc_tocs, detect_sdt_toc +from app.services.document_parser.assets.image_size_filter import is_below_img_min_size from docx import Document from docx.oxml.table import CT_Tbl from docx.oxml.text.paragraph import CT_P @@ -134,6 +135,8 @@ def iter_block_items(doc_data): continue seen_rids.add(rid) data = docx.read("word/" + target) + if is_below_img_min_size(len(data)): + continue yield ( ele_num, None, @@ -234,9 +237,7 @@ def iter_block_items(doc_data): continue cell_seen_rids.add(rid) data = docx.read("word/" + target) - if ( - len(data) < 10 * 1024 - ): # Skip small images (<10KB, likely icons) + if is_below_img_min_size(len(data)): continue imgs_in_cell.append( { @@ -271,7 +272,7 @@ def iter_block_items(doc_data): f"Failed to convert VML cell image to PNG: {e}" ) continue - if len(png_data) < 10 * 1024: + if is_below_img_min_size(len(png_data)): continue orig_name = target.split("/")[-1] png_name = os.path.splitext(orig_name)[0] + ".png" diff --git a/apps/worker/app/services/document_parser/formats/docx/parser.py b/apps/worker/app/services/document_parser/formats/docx/parser.py index e195319f..f978ab77 100755 --- a/apps/worker/app/services/document_parser/formats/docx/parser.py +++ b/apps/worker/app/services/document_parser/formats/docx/parser.py @@ -584,9 +584,6 @@ def parse_docx( headings_stack[-1]["content"].append(text) elif label == "IMAGE": - if meta and meta.get("size", 0) < 10 * 1024: - continue - headings_stack = asset_accumulator.append_image( meta, headings_stack, diff --git a/apps/worker/app/services/document_parser/formats/image/parser.py b/apps/worker/app/services/document_parser/formats/image/parser.py index dd7fdd09..52ff4df6 100755 --- a/apps/worker/app/services/document_parser/formats/image/parser.py +++ b/apps/worker/app/services/document_parser/formats/image/parser.py @@ -87,7 +87,7 @@ def local_image_to_data_url(path, cut=True, min_size=None, max_size=None): if cut: file_size = path.stat().st_size # Bytes. - if file_size < min_size: # Smaller than 10 KB. + if file_size < min_size: logger.debug(f"Skipping {path} (too small: {file_size / 1024:.1f} KB)") return None if file_size >= max_size: # Larger than 5 MB. @@ -232,15 +232,13 @@ def parse_image( img_obj = Image.open(io.BytesIO(img_bytes)) img_obj.save(img_path) - # Early exit: skip images smaller than 10KB + # Early exit: skip images smaller than IMG_MIN_SIZE before VLM work. + from app.services.document_parser.assets.image_size_filter import ( + discard_undersized_image_file, + ) from shared.core.constants import ProcessingConstants - saved_size = os.path.getsize(img_path) - if saved_size < ProcessingConstants.IMG_MIN_SIZE: - logger.debug( - f"Skipping image {filename} (too small: {saved_size / 1024:.1f} KB)" - ) - os.remove(img_path) + if discard_undersized_image_file(img_path, label=f"image {filename}"): return pd.DataFrame(columns=list(PARSER_ROW_COLUMNS)) # Extract image content diff --git a/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py b/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py index d946db4f..06ff1ad0 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py +++ b/apps/worker/app/services/document_parser/formats/markdown/deferred_summary.py @@ -83,20 +83,46 @@ def apply_markdown_deferred_summaries( def replace_chunk_ref_in_rows( rows: list[list[str | int]], old_path: str, new_path: str ) -> None: + """Rewrite asset paths after deferred rename. + + Text/image rows store bracketed refs (``[tables/...]`` / ``[images/...]``). + Table rows store the bare relative path as ``content``. Both forms must move + with the renamed file; historically only the bracketed form was updated, + leaving table ``content`` stuck on the pre-rename path. + """ + if not old_path or old_path == new_path: + return + old_ref = build_chunk_ref(old_path) new_ref = build_chunk_ref(new_path) - if not old_ref or old_ref == new_ref: - return for row in rows: if len(row) > 0 and isinstance(row[0], str): - row[0] = row[0].replace(old_ref, new_ref) + updated = row[0] + if old_ref and new_ref and old_ref != new_ref: + updated = updated.replace(old_ref, new_ref) + # Table asset rows use the bare path as content (no brackets). + if updated == old_path: + updated = new_path + elif old_path in updated: + updated = updated.replace(old_path, new_path) + row[0] = updated if len(row) > 1 and row[1] == old_path: row[1] = new_path if len(row) > 2 and isinstance(row[2], str): - row[2] = row[2].replace(old_ref, new_ref) + updated_type = row[2] + if old_ref and new_ref and old_ref != new_ref: + updated_type = updated_type.replace(old_ref, new_ref) + if old_path in updated_type: + updated_type = updated_type.replace(old_path, new_path) + row[2] = updated_type if len(row) > 8 and isinstance(row[8], str): - row[8] = row[8].replace(old_ref, new_ref) + updated_connect = row[8] + if old_ref and new_ref and old_ref != new_ref: + updated_connect = updated_connect.replace(old_ref, new_ref) + if old_path in updated_connect: + updated_connect = updated_connect.replace(old_path, new_path) + row[8] = updated_connect def _run_deferred_summary_tasks( @@ -242,6 +268,11 @@ def _apply_image_summary_result( row = rows[row_index] _apply_asset_result_preserving_index(row, result) + # Table-embedded images keep a stable filename so in + # tables/*.html stays valid after summary generation. + if not original_task.rename_file: + return + img_title = result.title if not img_title: return @@ -285,10 +316,18 @@ def _apply_table_summary_result( table_dir = original_task.table_dir old_table_name = original_task.table_name - table_count = original_task.table_count + # Keep the original table-N index (same pattern as image rename). + table_num_match = re.match(r"table-(\d+)", str(old_table_name)) + table_num = ( + table_num_match.group(1) + if table_num_match + else str(old_table_name).split("-")[1] + if "-" in str(old_table_name) + else "0" + ) safe_title = sanitize_table_name_from_header(str(title)) new_table_name = path_handle( - f"table-{table_count} {safe_title}", mode="clean_single" + f"table-{table_num} {safe_title}", mode="clean_single" ) old_path = os.path.join(table_dir, f"{old_table_name}.html") new_path = os.path.join(table_dir, f"{new_table_name}.html") @@ -296,6 +335,8 @@ def _apply_table_summary_result( return os.rename(old_path, new_path) + old_relative_path = str(row[1]) if len(row) > 1 else f"tables/{old_table_name}.html" new_relative_path = f"tables/{new_table_name}.html" - replace_chunk_ref_in_rows(rows, str(row[1]), new_relative_path) + replace_chunk_ref_in_rows(rows, old_relative_path, new_relative_path) + row[0] = new_relative_path row[1] = new_relative_path diff --git a/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py b/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py index f2cb8240..c0467504 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py +++ b/apps/worker/app/services/document_parser/formats/markdown/deferred_task.py @@ -11,6 +11,7 @@ class ImageDeferredSummaryTask: image_dir: str image_name: str image_suffix: str + rename_file: bool = True @dataclass(frozen=True) @@ -19,7 +20,6 @@ class TableDeferredSummaryTask: table_html: str table_dir: str table_name: str - table_count: int @dataclass(frozen=True) diff --git a/apps/worker/app/services/document_parser/formats/markdown/image_asset.py b/apps/worker/app/services/document_parser/formats/markdown/image_asset.py index ed5f51ab..52305689 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/image_asset.py +++ b/apps/worker/app/services/document_parser/formats/markdown/image_asset.py @@ -7,6 +7,9 @@ from app.services.document_parser.support.identifiers import gen_str_codes from app.services.document_parser.formats.image.parser import perceptual_hash +from app.services.document_parser.assets.image_size_filter import ( + discard_undersized_image_file, +) from app.services.document_parser.assets.inline_asset import build_image_asset_row from app.services.document_parser.formats.markdown.deferred_task import ( ImageDeferredSummaryTask, @@ -27,6 +30,8 @@ class MarkdownImageAsset: cache_entry: dict[str, str] | None deferred_task: MarkdownDeferredSummaryTask | None should_advance_image_count: bool + relative_path: str | None + discarded_undersized: bool = False @dataclass(frozen=True) @@ -42,6 +47,7 @@ class MarkdownImageAssetRequest: seen_images: dict[str, dict[str, str]] summary_image: bool row_index: int + rename_on_summary: bool = True def build_markdown_image_asset( @@ -56,6 +62,13 @@ def build_markdown_image_asset( logger.warning(f"Image file not found, skipping rename: {request.image_path}") return _empty_asset(should_advance_image_count=True) + # Gate before rename / perceptual hash / deferred VLM summary. + if discard_undersized_image_file(source_path, label="markdown image"): + return _empty_asset( + should_advance_image_count=False, + discarded_undersized=True, + ) + with open(source_path, "rb") as image_file: image_binary_hash = perceptual_hash(image_file.read()) @@ -105,6 +118,7 @@ def build_markdown_image_asset( image_dir=request.image_dir, image_name=request.image_name, image_suffix=image_suffix, + rename_file=request.rename_on_summary, ) return MarkdownImageAsset( @@ -114,12 +128,15 @@ def build_markdown_image_asset( cache_entry=cache_entry, deferred_task=deferred_task, should_advance_image_count=True, + relative_path=relative_image_path, ) def build_markdown_image_name(*, image_count: int, last_context: str) -> str: - image_name_context = path_handle(last_context[:10], mode="clean_single") - return f"image-{str(image_count)}-{image_name_context}" + image_name_context = path_handle(last_context.strip(), mode="clean_single") + if image_name_context: + return f"image-{image_count}-{image_name_context}" + return f"image-{image_count}" def resolve_workspace_image_path( @@ -164,9 +181,10 @@ def _build_duplicate_image_asset( cache_entry: dict[str, str], timestamp: str, ) -> MarkdownImageAsset: + relative_path = cache_entry["relative_img_path"] row_values = _build_image_row_values( content=cache_entry["img_content"], - relative_path=cache_entry["relative_img_path"], + relative_path=relative_path, summary=cache_entry["img_summary_field"], know_id=cache_entry["temp_uid"], timestamp=timestamp, @@ -183,6 +201,7 @@ def _build_duplicate_image_asset( cache_entry=None, deferred_task=None, should_advance_image_count=False, + relative_path=relative_path, ) @@ -211,7 +230,11 @@ def _build_image_row_values( return cast(ParserRowValues, image_row.to_list()) -def _empty_asset(*, should_advance_image_count: bool) -> MarkdownImageAsset: +def _empty_asset( + *, + should_advance_image_count: bool, + discarded_undersized: bool = False, +) -> MarkdownImageAsset: return MarkdownImageAsset( content_item=None, row_values=None, @@ -219,4 +242,6 @@ def _empty_asset(*, should_advance_image_count: bool) -> MarkdownImageAsset: cache_entry=None, deferred_task=None, should_advance_image_count=should_advance_image_count, + relative_path=None, + discarded_undersized=discarded_undersized, ) diff --git a/apps/worker/app/services/document_parser/formats/markdown/parser.py b/apps/worker/app/services/document_parser/formats/markdown/parser.py index 4f552cbb..32b5094b 100755 --- a/apps/worker/app/services/document_parser/formats/markdown/parser.py +++ b/apps/worker/app/services/document_parser/formats/markdown/parser.py @@ -19,6 +19,9 @@ MarkdownTableAssetRequest, build_markdown_table_asset, ) +from app.services.document_parser.formats.markdown.table_embedded_images import ( + extract_table_embedded_images, +) from app.services.document_parser.support.parser_rows import ParsedRow from app.services.document_parser.support.path_helpers import find_matches_parsing from app.services.document_parser.formats.html.parser import ( @@ -417,8 +420,6 @@ def parse_md( if image_asset.should_advance_image_count: parser_state.image_count += 1 - # TODO for large and dense tables, such as "Epstein flight logs", - # integrate tabula-py as an independent extraction path to solve VLM hallucinations and misplacement # b. handle lines containing tables tb_bool, form, _ = identify_tables(line) if tb_bool: @@ -447,14 +448,30 @@ def parse_md( else: continue # Unknown form, skip + embedded = extract_table_embedded_images( + table_html=tb_str, + parser_state=parser_state, + output_dir=output_dir, + image_dir=img_dir, + summary_image=bool(base_llm_paras["summary_image"]), + ) + for image_asset in embedded.image_assets: + if image_asset.row_values is not None: + parser_state.append_row(image_asset.row_values) + if image_asset.deferred_task is not None: + parser_state.schedule_deferred_task(image_asset.deferred_task) + for image_ref in embedded.image_refs: + parser_state.append_content_item(f"\n{image_ref}\n") + table_asset = build_markdown_table_asset( MarkdownTableAssetRequest( - table_html=tb_str, + table_html=embedded.rewritten_html, table_dir=tb_dir, table_count=parser_state.table_count, timestamp=parser_state.timestamp, summary_table=bool(base_llm_paras["summary_table"]), row_index=len(parser_state.rows), + image_refs=embedded.image_refs, ) ) parser_state.append_content_item(table_asset.content_item) diff --git a/apps/worker/app/services/document_parser/formats/markdown/table_asset.py b/apps/worker/app/services/document_parser/formats/markdown/table_asset.py index 89558cb1..355dc7ec 100644 --- a/apps/worker/app/services/document_parser/formats/markdown/table_asset.py +++ b/apps/worker/app/services/document_parser/formats/markdown/table_asset.py @@ -34,6 +34,7 @@ class MarkdownTableAssetRequest: timestamp: str summary_table: bool row_index: int + image_refs: list[str] | None = None def build_markdown_table_asset( @@ -60,6 +61,7 @@ def build_markdown_table_asset( keywords="", know_id=gen_str_codes((request.table_html + str(request.table_count))), addtime=request.timestamp, + image_refs=request.image_refs or [], ) deferred_task = None @@ -69,7 +71,6 @@ def build_markdown_table_asset( table_html=request.table_html, table_dir=request.table_dir, table_name=table_name, - table_count=request.table_count - 1, ) return MarkdownTableAsset( diff --git a/apps/worker/app/services/document_parser/formats/markdown/table_embedded_images.py b/apps/worker/app/services/document_parser/formats/markdown/table_embedded_images.py new file mode 100644 index 00000000..ab82543d --- /dev/null +++ b/apps/worker/app/services/document_parser/formats/markdown/table_embedded_images.py @@ -0,0 +1,173 @@ +"""Extract and rewrite tags embedded inside MinerU HTML tables.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from bs4 import BeautifulSoup, Tag +from loguru import logger + +from app.services.document_parser.formats.markdown.image_asset import ( + MarkdownImageAsset, + MarkdownImageAssetRequest, + build_markdown_image_asset, + build_markdown_image_name, +) +from app.services.document_parser.formats.markdown.parse_state import MarkdownParseState +from shared.utils.chunk_refs import build_chunk_ref + +_IMG_SRC_RE = re.compile( + r"""]*\bsrc\s*=\s*(?P["'])(?P[^"']+)(?P=quote)""", + re.IGNORECASE, +) +_IMG_TAG_WITH_SRC_RE = re.compile( + r"""]*\bsrc\s*=\s*(?P["'])(?P[^"']+)(?P=quote)[^>]*/?\s*>""", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class TableEmbeddedImagesResult: + rewritten_html: str + """HTML with rewritten to stable images/image-N-* paths.""" + + image_assets: list[MarkdownImageAsset] + """Newly created image assets that should be registered as rows.""" + + image_refs: list[str] + """Chunk refs ([images/...]) for text content_items and table type channel.""" + + +def extract_table_embedded_images( + *, + table_html: str, + parser_state: MarkdownParseState, + output_dir: str, + image_dir: str, + summary_image: bool, +) -> TableEmbeddedImagesResult: + """Pull assets out of a table, rename them, and rewrite HTML srcs. + + Duplicate perceptual hashes and repeated srcs reuse the first stable path + without creating an extra image row. Missing files leave the original src. + """ + srcs = _unique_img_srcs(table_html) + if not srcs: + return TableEmbeddedImagesResult( + rewritten_html=table_html, + image_assets=[], + image_refs=[], + ) + + naming_context = _first_cell_text(table_html) + rewritten_html = table_html + image_assets: list[MarkdownImageAsset] = [] + image_refs: list[str] = [] + src_to_relative: dict[str, str] = {} + + for src in srcs: + if src in src_to_relative: + continue + + image_name = build_markdown_image_name( + image_count=parser_state.image_count, + last_context=naming_context, + ) + image_asset = build_markdown_image_asset( + MarkdownImageAssetRequest( + output_dir=output_dir, + image_dir=image_dir, + image_path=src, + image_name=image_name, + image_count=parser_state.image_count, + last_context=naming_context, + image_summary=naming_context or None, + timestamp=parser_state.timestamp, + seen_images=parser_state.seen_images, + summary_image=summary_image, + row_index=len(parser_state.rows) + len(image_assets), + rename_on_summary=False, + ) + ) + if image_asset.discarded_undersized: + rewritten_html = _strip_img_tags(rewritten_html, src) + continue + if image_asset.relative_path is None: + logger.warning( + f"Table-embedded image not found, leaving original src: {src}" + ) + continue + + src_to_relative[src] = image_asset.relative_path + image_refs.append(build_chunk_ref(image_asset.relative_path)) + + if image_asset.should_advance_image_count: + image_assets.append(image_asset) + # Advance immediately so the next distinct src gets a new image-N. + parser_state.image_count += 1 + if ( + image_asset.cache_key is not None + and image_asset.cache_entry is not None + ): + parser_state.seen_images[image_asset.cache_key] = ( + image_asset.cache_entry + ) + + for old_src, new_relative in src_to_relative.items(): + rewritten_html = _rewrite_img_src(rewritten_html, old_src, new_relative) + + return TableEmbeddedImagesResult( + rewritten_html=rewritten_html, + image_assets=image_assets, + image_refs=image_refs, + ) + + +def _unique_img_srcs(table_html: str) -> list[str]: + seen: set[str] = set() + ordered: list[str] = [] + for match in _IMG_SRC_RE.finditer(table_html): + src = match.group("src").strip() + if not src or src in seen: + continue + seen.add(src) + ordered.append(src) + return ordered + + +def _first_cell_text(table_html: str) -> str: + soup = BeautifulSoup(table_html, "html.parser") + table = soup.find("table") + if not isinstance(table, Tag): + return "" + for cell in table.find_all(["td", "th"]): + if not isinstance(cell, Tag): + continue + text = cell.get_text(strip=True) + if text: + return text + return "" + + +def _rewrite_img_src(html: str, old_src: str, new_src: str) -> str: + pattern = re.compile( + r"""(]*\bsrc\s*=\s*)(["'])""" + + re.escape(old_src) + + r"""\2""", + re.IGNORECASE, + ) + + def _replace(match: re.Match[str]) -> str: + return f"{match.group(1)}{match.group(2)}{new_src}{match.group(2)}" + + return pattern.sub(_replace, html) + + +def _strip_img_tags(html: str, src: str) -> str: + def _drop(match: re.Match[str]) -> str: + if match.group("src") == src: + return "" + return match.group(0) + + return _IMG_TAG_WITH_SRC_RE.sub(_drop, html) diff --git a/apps/worker/app/services/document_parser/orchestration/postprocess.py b/apps/worker/app/services/document_parser/orchestration/postprocess.py index 0fbb9f5c..28f30a96 100644 --- a/apps/worker/app/services/document_parser/orchestration/postprocess.py +++ b/apps/worker/app/services/document_parser/orchestration/postprocess.py @@ -11,6 +11,15 @@ from app.services.document_parser.support.stage_profiler import stage_timer from loguru import logger +_IMG_SRC_BASENAME_RE = re.compile( + r"""]*\bsrc\s*=\s*["']([^"']+)["']""", + re.IGNORECASE, +) +_HASH_IMAGE_RE = re.compile( + r"^[a-f0-9]{64}\.(?:jpg|jpeg|png|gif|webp)$", + re.IGNORECASE, +) + def apply_parse_postprocess( output_dir: str, @@ -39,19 +48,23 @@ def apply_parse_postprocess( def cleanup_unreferenced_images(output_dir: str) -> int: - """Remove UUID-named images that are not referenced by final parsed output.""" + """Remove hash-named MinerU images that are not referenced by table HTML. + + Images extracted as ``image-N-*`` are never matched by the hash pattern and + are always kept. Hash-named files still referenced from ``tables/*.html`` + (e.g. failed extraction) are also preserved. + """ image_dir = os.path.join(output_dir, "images") if not os.path.isdir(image_dir): return 0 - uuid_pattern = re.compile( - r"^[a-f0-9]{64}\.(?:jpg|jpeg|png|gif|webp)$", - re.IGNORECASE, - ) + protected_basenames = _collect_table_img_basenames(output_dir) removed_count = 0 for filename in os.listdir(image_dir): - if not uuid_pattern.match(filename): + if not _HASH_IMAGE_RE.match(filename): + continue + if filename in protected_basenames: continue file_path = os.path.join(image_dir, filename) @@ -68,3 +81,26 @@ def cleanup_unreferenced_images(output_dir: str) -> int: ) return removed_count + + +def _collect_table_img_basenames(output_dir: str) -> set[str]: + tables_dir = os.path.join(output_dir, "tables") + if not os.path.isdir(tables_dir): + return set() + + basenames: set[str] = set() + for filename in os.listdir(tables_dir): + if not filename.endswith(".html"): + continue + table_path = os.path.join(tables_dir, filename) + try: + with open(table_path, encoding="utf-8") as table_file: + html = table_file.read() + except OSError as exc: + logger.warning(f"Failed to read table HTML {table_path}: {exc}") + continue + for match in _IMG_SRC_BASENAME_RE.finditer(html): + src = match.group(1).strip() + if src: + basenames.add(os.path.basename(src)) + return basenames diff --git a/apps/worker/tests/contract/test_image_size_filter_contract.py b/apps/worker/tests/contract/test_image_size_filter_contract.py new file mode 100644 index 00000000..1b04bdb2 --- /dev/null +++ b/apps/worker/tests/contract/test_image_size_filter_contract.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import os +from pathlib import Path + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +from app.services.document_parser.assets.image_size_filter import ( # noqa: E402 + discard_undersized_image_file, + is_below_img_min_size, +) +from shared.core.constants.processing import ProcessingConstants # noqa: E402 + + +def test_is_below_img_min_size_boundary() -> None: + assert is_below_img_min_size(ProcessingConstants.IMG_MIN_SIZE - 1) + assert not is_below_img_min_size(ProcessingConstants.IMG_MIN_SIZE) + assert not is_below_img_min_size(ProcessingConstants.IMG_MIN_SIZE + 1) + + +def test_discard_undersized_image_file(tmp_path: Path) -> None: + small = tmp_path / "small.jpg" + small.write_bytes(b"x" * (ProcessingConstants.IMG_MIN_SIZE - 1)) + assert discard_undersized_image_file(small, label="unit") is True + assert not small.exists() + + large = tmp_path / "large.jpg" + large.write_bytes(b"y" * ProcessingConstants.IMG_MIN_SIZE) + assert discard_undersized_image_file(large, label="unit") is False + assert large.exists() diff --git a/apps/worker/tests/contract/test_markdown_table_rename_contract.py b/apps/worker/tests/contract/test_markdown_table_rename_contract.py new file mode 100644 index 00000000..6360181b --- /dev/null +++ b/apps/worker/tests/contract/test_markdown_table_rename_contract.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import os +from pathlib import Path + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +from app.services.document_parser.formats.markdown.deferred_summary import ( # noqa: E402 + replace_chunk_ref_in_rows, + _apply_table_summary_result, +) +from app.services.document_parser.formats.markdown.deferred_task import ( # noqa: E402 + TableDeferredSummaryTask, +) +from app.services.document_parser.formats.markdown.table_asset import ( # noqa: E402 + MarkdownTableAssetRequest, + build_markdown_table_asset, +) +from shared.services.ai.summary.model import AssetSummary, Entity # noqa: E402 +from shared.utils.chunk_refs import build_chunk_ref # noqa: E402 + + +def test_replace_chunk_ref_updates_bare_table_content_and_bracketed_text() -> None: + old_path = "tables/table-12 流程名称 招标文件.html" + new_path = "tables/table-12 招标文件及议标项目评审流程图.html" + rows: list[list[str | int]] = [ + [old_path, old_path, "table", 10, "", "table-12", "id", "", ""], + [ + f"see {build_chunk_ref(old_path)}", + "doc/section", + "ptxt", + 20, + "", + "", + "text-id", + "", + "", + ], + ] + + replace_chunk_ref_in_rows(rows, old_path, new_path) + + assert rows[0][0] == new_path + assert rows[0][1] == new_path + assert rows[1][0] == f"see {build_chunk_ref(new_path)}" + + +def test_table_deferred_task_has_no_legacy_count_field(tmp_path: Path) -> None: + asset = build_markdown_table_asset( + MarkdownTableAssetRequest( + table_html="
流程名称招标文件
", + table_dir=str(tmp_path), + table_count=12, + timestamp="2026-07-21 00:00:00", + summary_table=True, + row_index=0, + ) + ) + + assert asset.deferred_task is not None + assert isinstance(asset.deferred_task, TableDeferredSummaryTask) + assert not hasattr(asset.deferred_task, "table_count") + assert asset.deferred_task.table_name.startswith("table-12 ") + + +def test_apply_table_summary_rename_keeps_index_and_syncs_content( + tmp_path: Path, +) -> None: + old_stem = "table-12 流程名称 招标文件及议标项目评审流程图 流程编号" + old_relative = f"tables/{old_stem}.html" + old_file = tmp_path / f"{old_stem}.html" + old_file.write_text("
x
", encoding="utf-8") + + text_content = f"\n{build_chunk_ref(old_relative)}\n" + rows: list[list[str | int]] = [ + [old_relative, old_relative, "table", len(old_relative), "", "table-12", "t", "", ""], + [text_content, "doc/3、招标文件评审流程运行图", "ptxt", len(text_content), "", "", "x", "", ""], + ] + task = TableDeferredSummaryTask( + row_index=0, + table_html="
x
", + table_dir=str(tmp_path), + table_name=old_stem, + ) + + _apply_table_summary_result( + rows, + task, + 0, + AssetSummary( + title="招标文件及议标项目评审流程图", + summary="三类风险招标文件评审流程", + entities=[Entity(text="市场开发部", type="org")], + kind="table", + ), + ) + + new_relative = str(rows[0][1]) + assert new_relative.startswith("tables/table-12 ") + assert "招标文件及议标项目评审流程图" in new_relative + assert rows[0][0] == new_relative + assert build_chunk_ref(new_relative) in str(rows[1][0]) + assert not old_file.exists() + assert (tmp_path / Path(new_relative).name).exists() diff --git a/apps/worker/tests/contract/test_page_memory_navigation_contract.py b/apps/worker/tests/contract/test_page_memory_navigation_contract.py index 23e60c28..47913989 100644 --- a/apps/worker/tests/contract/test_page_memory_navigation_contract.py +++ b/apps/worker/tests/contract/test_page_memory_navigation_contract.py @@ -44,6 +44,7 @@ def test_page_doc_nav_uses_path_based_leaf_summaries_and_page_counts() -> None: def test_zip_result_service_builds_navigation_from_chunk_paths() -> None: doc_nav, hierarchy = ZipResultService()._build_navigation_outputs( # noqa: SLF001 + add_dir="", formatted_chunks=_page_chunks(), source_file_name="demo.pdf", ) @@ -57,6 +58,41 @@ def test_zip_result_service_builds_navigation_from_chunk_paths() -> None: } +def test_zip_result_service_prefers_enriched_on_disk_doc_nav(tmp_path) -> None: + enriched = { + "version": "1.0", + "file_name": "demo.pdf", + "top_summary": "Document overview from enrich", + "stats": {}, + "sections": [ + { + "title": "Kept From Disk", + "path": "demo.pdf/Kept From Disk", + "summary": "enriched section", + "chunk_count": 1, + "children": [], + } + ], + "resources": {"images": [], "tables": []}, + } + (tmp_path / "doc_nav.json").write_text( + __import__("json").dumps(enriched, ensure_ascii=False), + encoding="utf-8", + ) + + doc_nav, hierarchy = ZipResultService()._build_navigation_outputs( # noqa: SLF001 + add_dir=str(tmp_path), + formatted_chunks=_page_chunks(), + source_file_name="demo.pdf", + ) + + assert doc_nav is not None + assert doc_nav["top_summary"] == "Document overview from enrich" + assert doc_nav["sections"][0]["title"] == "Kept From Disk" + assert hierarchy == {"Kept From Disk": {}} + + + def _page_chunks() -> list[dict[str, object]]: return [ { diff --git a/apps/worker/tests/contract/test_table_embedded_images_contract.py b/apps/worker/tests/contract/test_table_embedded_images_contract.py new file mode 100644 index 00000000..0da70fad --- /dev/null +++ b/apps/worker/tests/contract/test_table_embedded_images_contract.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import os +from pathlib import Path + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") +os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test") +os.environ.setdefault("S3_BUCKET_NAME", "test-uploads") +os.environ.setdefault("S3_ACCESS_KEY_ID", "test") +os.environ.setdefault("S3_SECRET_ACCESS_KEY", "test") +os.environ.setdefault("S3_TEMP_PATH", "/tmp") + +from PIL import Image # noqa: E402 + +from app.services.document_parser.formats.markdown.parse_state import ( # noqa: E402 + MarkdownParseState, +) +from app.services.document_parser.formats.markdown.parser import ( # noqa: E402 + update_df_list, +) +from app.services.document_parser.formats.markdown.table_asset import ( # noqa: E402 + MarkdownTableAssetRequest, + build_markdown_table_asset, +) +from app.services.document_parser.formats.markdown.table_embedded_images import ( # noqa: E402 + extract_table_embedded_images, +) +from app.services.document_parser.orchestration.postprocess import ( # noqa: E402 + cleanup_unreferenced_images, +) +from shared.core.constants.processing import ProcessingConstants # noqa: E402 +from shared.services.chunks.dataframe_chunk_converter import ( # noqa: E402 + dataframe_to_chunks, +) +from shared.services.retrieval.agentic.evidence.renderer import ( # noqa: E402 + render_table_chunk_lines, +) + + +def _write_jpeg( + path: Path, + *, + size: tuple[int, int] = (640, 640), + min_bytes: int | None = None, +) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + # Solid colors compress too well; patterned noise keeps file size realistic. + width, height = size + pixels = bytearray(width * height * 3) + for index in range(len(pixels)): + pixels[index] = (index * 37) % 256 + Image.frombytes("RGB", size, bytes(pixels)).save(path, format="JPEG", quality=95) + if min_bytes is not None: + current_size = path.stat().st_size + if current_size < min_bytes: + with path.open("ab") as handle: + handle.write(b"\x00" * (min_bytes - current_size)) + + +def _make_parser_state() -> MarkdownParseState: + return MarkdownParseState( + relative_root="doc.pdf", + split_char="/", + llm_parameters={ + "summary_image": False, + "summary_table": False, + "summary_txt": False, + "stopwords": set(), + }, + timestamp="2026-01-01 00:00:00", + row_updater=update_df_list, + ) + + +def test_table_embedded_images_are_extracted_rewritten_and_linked( + tmp_path: Path, +) -> None: + output_dir = tmp_path / "doc" + images_dir = output_dir / "images" + tables_dir = output_dir / "tables" + images_dir.mkdir(parents=True) + tables_dir.mkdir(parents=True) + + hash_name = "a" * 64 + ".jpg" + image_path = images_dir / hash_name + _write_jpeg(image_path, min_bytes=ProcessingConstants.IMG_MIN_SIZE) + assert image_path.stat().st_size >= ProcessingConstants.IMG_MIN_SIZE + + table_html = ( + "" + f'' + "
流程名称评审流程图
说明审批节点
" + ) + + parser_state = _make_parser_state() + + embedded = extract_table_embedded_images( + table_html=table_html, + parser_state=parser_state, + output_dir=str(output_dir), + image_dir=str(images_dir), + summary_image=False, + ) + + assert len(embedded.image_assets) == 1 + assert len(embedded.image_refs) == 1 + assert hash_name not in embedded.rewritten_html + assert " None: + output_dir = tmp_path / "doc" + images_dir = output_dir / "images" + images_dir.mkdir(parents=True) + + hash_name = "c" * 64 + ".jpg" + image_path = images_dir / hash_name + _write_jpeg(image_path, size=(16, 16)) + assert image_path.stat().st_size < ProcessingConstants.IMG_MIN_SIZE + + table_html = ( + "" + f'
Logo
' + ) + + embedded = extract_table_embedded_images( + table_html=table_html, + parser_state=_make_parser_state(), + output_dir=str(output_dir), + image_dir=str(images_dir), + summary_image=False, + ) + + assert embedded.image_assets == [] + assert embedded.image_refs == [] + assert " None: captured: Dict[str, Any] = {} - def _fake_client() -> Any: + def _fake_client(**_kwargs: Any) -> Any: class _C: def chat_completion(self, **kwargs: Any) -> str: captured["messages"] = kwargs.get("messages") @@ -231,3 +231,76 @@ def chat_completion(self, **kwargs: Any) -> str: assert "[ChildB] ChildB" in user # legacy flat blob prompt removed assert "You will receive summaries of sub-sections" not in user + + +class TestDocNavTopSummaryPersistence: + def test_enrich_persists_top_summary_and_defaults_top_llm( + self, tmp_path, monkeypatch: pytest.MonkeyPatch + ) -> None: + import json + + from app.services.connect_builder.summary_builder import ( + enrich_doc_nav_summaries, + load_nav_top_summary, + ) + + captured: Dict[str, Any] = {} + + def _fake_llm(**kwargs: Any) -> str: + captured["is_top"] = kwargs.get("node_name") == "Document Overview" + captured["max_tokens"] = kwargs.get("max_tokens") + captured["calls"] = int(captured.get("calls") or 0) + 1 + return "LLM document overview" + + monkeypatch.setattr( + "app.services.connect_builder.summary_builder._llm_summarize", + _fake_llm, + ) + + file_dir = tmp_path / "report.pdf" + file_dir.mkdir() + long_leaf = "L" * (SUMMARY_MAX_LEN + 5) + doc_nav = { + "version": "1.0", + "file_name": "report.pdf", + "stats": {}, + "sections": [ + { + "title": "Chapter 1", + "path": "report.pdf/Chapter 1", + "summary": long_leaf, + "chunk_count": 1, + "children": [], + }, + { + "title": "Chapter 2", + "path": "report.pdf/Chapter 2", + "summary": long_leaf, + "chunk_count": 1, + "children": [], + }, + ], + "resources": {"images": [], "tables": []}, + } + (file_dir / "doc_nav.json").write_text( + json.dumps(doc_nav, ensure_ascii=False), + encoding="utf-8", + ) + + results = enrich_doc_nav_summaries( + str(tmp_path), + source_file="report.pdf", + use_llm=False, + top_summary_use_llm=True, + ) + assert results["report.pdf"] == "LLM document overview" + assert captured["calls"] == 1 + assert captured["is_top"] is True + + saved = json.loads((file_dir / "doc_nav.json").read_text(encoding="utf-8")) + assert saved["top_summary"] == "LLM document overview" + # Section leaves keep original summaries; top LLM must not rewrite them. + assert saved["sections"][0]["summary"] == long_leaf + assert load_nav_top_summary(str(file_dir), "report.pdf") == ( + "LLM document overview" + ) diff --git a/packages/shared-python/shared/models/schemas/job.py b/packages/shared-python/shared/models/schemas/job.py index 2a948f79..1ea775fd 100644 --- a/packages/shared-python/shared/models/schemas/job.py +++ b/packages/shared-python/shared/models/schemas/job.py @@ -38,6 +38,14 @@ class ParsingParams(BaseModel): "Increases parse time and API token cost." ), ) + top_summary_use_llm: bool = Field( + True, + description=( + "Use LLM for the document-level top_summary written to doc_nav. " + "Defaults to True because agentic document selection consumes this " + "field. Section-level summaries remain controlled by summary_use_llm." + ), + ) class JobCreateBase(BaseModel): diff --git a/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py b/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py index 513f8b78..e5ad3931 100644 --- a/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py +++ b/packages/shared-python/shared/services/chunks/dataframe_chunk_converter.py @@ -372,7 +372,7 @@ def dataframe_to_chunks(df: _ParserDataFrame | None) -> list[Dict[str, JsonValue for chunk in chunks: metadata = chunk["metadata"] relationship_refs = metadata.pop("_relationship_refs", []) - if chunk["type"] not in {"text", "page"}: + if chunk["type"] not in {"text", "page", "table"}: continue embed_connections = convert_refs_to_embed_connections( relationship_refs, resource_target_map diff --git a/packages/shared-python/shared/services/jobs/lifecycle/publication.py b/packages/shared-python/shared/services/jobs/lifecycle/publication.py index f29ed6ac..a8a3e747 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/publication.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/publication.py @@ -51,6 +51,7 @@ def publish_result( job_result_id: str, chunks: list[dict[str, Any]], section_summaries: dict[str, str] | None, + document_top_summary: str | None = None, ) -> JobPublicationOutcome: previous_document_scope = self._retrieval_publication.get_existing_document_scope( db, @@ -69,6 +70,7 @@ def publish_result( db, job_id=job_id, job_result_id=job_result_id, + top_summary=document_top_summary, ) cache_invalidation = self._build_cache_invalidation( diff --git a/packages/shared-python/shared/services/jobs/lifecycle/service.py b/packages/shared-python/shared/services/jobs/lifecycle/service.py index 3683fca3..c7953a00 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/service.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/service.py @@ -52,6 +52,7 @@ def finalize_job_success( stored_count: int = 0, delivery_mode: str = "url", section_summaries: Optional[Dict[str, str]] = None, + document_top_summary: Optional[str] = None, ) -> Dict[str, Any]: """Finalize a successful job in a single atomic transaction. @@ -78,6 +79,7 @@ def finalize_job_success( stored_count=stored_count, delivery_mode=delivery_mode, section_summaries=section_summaries, + document_top_summary=document_top_summary, ), should_commit=lambda finalization: finalization.response.should_commit(), build_response=lambda finalization: finalization.response.to_dict(), diff --git a/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py b/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py index d05b4143..2edfc2b6 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py @@ -79,6 +79,7 @@ def finalize( stored_count: int, delivery_mode: str, section_summaries: dict[str, str] | None, + document_top_summary: str | None = None, ) -> JobSuccessFinalization: job_result = self._result_writer.upsert_job_result( db, @@ -95,6 +96,7 @@ def finalize( job_result_id=job_result.id, chunks=chunks, section_summaries=section_summaries, + document_top_summary=document_top_summary, ) transition_outcome = self._state_machine.mark_completed_outcome( diff --git a/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py b/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py index db933802..3b8bd9c7 100644 --- a/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py +++ b/packages/shared-python/shared/services/retrieval/agentic/evidence/renderer.py @@ -183,6 +183,9 @@ def render_leaf_chunks( table_lines = render_table_chunk_lines( target, display_ref=display_ref, + chunk_by_id=chunk_by_id, + asset_lookup=asset_lookup, + rendered_ids=rendered_ids, ) content = content.replace(ref_str, "\n" + "\n".join(table_lines) + "\n") elif target_type == "image": @@ -227,6 +230,9 @@ def render_leaf_chunks( for line in render_table_chunk_lines( chunk, display_ref=display_ref, + chunk_by_id=chunk_by_id, + asset_lookup=asset_lookup, + rendered_ids=rendered_ids, ): if line.strip(): parts.append(f"{indent}┈ {line}") @@ -262,6 +268,9 @@ def render_table_chunk_lines( chunk: dict[str, Any], *, display_ref: str, + chunk_by_id: dict[str, dict[str, Any]] | None = None, + asset_lookup: dict[str, AssetLookupValue] | None = None, + rendered_ids: set[str] | None = None, ) -> list[str]: header = f"[Table: {display_ref}]" if display_ref else "[Table]" lines = [header] @@ -292,6 +301,65 @@ def render_table_chunk_lines( if caption: lines.append("Caption:") lines.append(str(caption).strip()) + + lines.extend( + _render_table_embedded_image_lines( + chunk, + chunk_by_id=chunk_by_id or {}, + asset_lookup=asset_lookup, + rendered_ids=rendered_ids, + ) + ) + return lines + + +def _render_table_embedded_image_lines( + table_chunk: dict[str, Any], + *, + chunk_by_id: dict[str, dict[str, Any]], + asset_lookup: dict[str, AssetLookupValue] | None, + rendered_ids: set[str] | None, +) -> list[str]: + metadata = table_chunk.get("chunk_metadata") or table_chunk.get("metadata") or {} + if not isinstance(metadata, dict): + return [] + + lines: list[str] = [] + for connection in metadata.get("connect_to") or []: + if not isinstance(connection, dict): + continue + if str(connection.get("relation") or "").strip() != "embeds": + continue + target_id = str(connection.get("target") or "").strip() + if not target_id: + continue + target = chunk_by_id.get(target_id) + if not target: + continue + target_type = ( + target.get("chunk_type") or target.get("type") or "" + ).strip().lower() + if target_type != "image": + continue + if rendered_ids is not None: + rendered_ids.add(target_id) + + file_path = target.get("file_path") or "" + asset_url = _lookup_asset_url(asset_lookup, target_id) + display_ref = asset_url or file_path + image_description = str(target.get("content") or "").strip() + ref_str = str(connection.get("ref") or "").strip() + if ref_str and ref_str in image_description: + image_description = image_description.replace(ref_str, "").strip() + + if display_ref: + lines.append(f"[Image: {display_ref}]") + elif image_description: + lines.append("[Image description]") + if image_description: + lines.extend( + line for line in image_description.split("\n") if line.strip() + ) return lines diff --git a/packages/shared-python/shared/services/retrieval/graph/service.py b/packages/shared-python/shared/services/retrieval/graph/service.py index 35d05fec..f5b9c51b 100644 --- a/packages/shared-python/shared/services/retrieval/graph/service.py +++ b/packages/shared-python/shared/services/retrieval/graph/service.py @@ -75,6 +75,7 @@ def publish_document_graph( namespace: str, document_id: str, job_result_id: str, + top_summary: str | None = None, ) -> None: document = db.execute( select(Document).where(Document.document_id == document_id) @@ -103,7 +104,11 @@ def publish_document_graph( types_breakdown[chunk_type or 'text'] += 1 chunks_count = len(chunk_meta_rows) - top_summary = extract_document_top_summary(chunk_metadata_list) + resolved_top_summary = str(top_summary or "").strip() + if not resolved_top_summary: + # Backward compatible fallback for older chunks that still carry + # per-chunk document_top_summary copies. + resolved_top_summary = extract_document_top_summary(chunk_metadata_list) # ── Clean up old graph data for this document ── self.remove_document_graph( @@ -137,7 +142,7 @@ def publish_document_graph( 'top_entities': top_entities, 'chunks_count': chunks_count, 'types': dict(types_breakdown), - 'top_summary': top_summary, + 'top_summary': resolved_top_summary, }, ) ) diff --git a/packages/shared-python/shared/services/retrieval/hydration/connected.py b/packages/shared-python/shared/services/retrieval/hydration/connected.py index 2267b91c..fcbc647b 100644 --- a/packages/shared-python/shared/services/retrieval/hydration/connected.py +++ b/packages/shared-python/shared/services/retrieval/hydration/connected.py @@ -31,7 +31,7 @@ async def hydrate_connected_target_rows( } target_ids_by_revision: dict[tuple[str, str], set[str]] = {} for row in rows: - if normalize_chunk_type(row.get('chunk_type')) not in ('text', 'page'): + if normalize_chunk_type(row.get('chunk_type')) not in ('text', 'page', 'table'): continue document_id = str(row.get('document_id') or '').strip() job_result_id = str(row.get('job_result_id') or '').strip() diff --git a/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py b/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py index 8c9a1b93..96534bce 100644 --- a/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py +++ b/packages/shared-python/shared/services/retrieval/hydration/result_assembly.py @@ -60,28 +60,14 @@ async def assemble_retrieval_results( assembled_row['content'] = _page_summary(row) assembled_row['content_source'] = 'summary' elif chunk_type == 'table': - assembled_row['content'] = _table_summary_content(row) + assembled_row['content'] = _compose_table_content(row, rows_by_chunk_id) assembled_row['content_source'] = 'summary' elif chunk_type == 'text': - connected_targets: list[tuple[int, str]] = [] - for target_id in iter_connected_target_ids(row): - target_row = rows_by_chunk_id.get(target_id) - if not target_row: - continue - if normalize_chunk_type(target_row.get('chunk_type')) != 'table': - continue - target_content = _table_summary_content(target_row) - if target_content: - sort_key = int(target_row.get('sort_order', 0) or 0) - connected_targets.append((sort_key, target_content)) - connected_targets.sort(key=lambda item: item[0]) - related_parts = [content for _, content in connected_targets] - - # TODO: Dedicated Large Table Agent - # For the Notebook/Agent environment, consider introducing a dedicated - # "Large Table Agent" that can fetch and query oversized tables via URL. + related_parts = _connected_media_parts(row, rows_by_chunk_id) if base_content and related_parts: assembled_row['content'] = '\n\n'.join([base_content, *related_parts]) + elif related_parts: + assembled_row['content'] = '\n\n'.join(related_parts) else: assembled_row['content'] = base_content assembled_row['content_source'] = 'content' @@ -100,6 +86,71 @@ def _page_summary(row: dict[str, Any]) -> str: return str(metadata.get('summary') or '').strip() +def _compose_table_content( + row: dict[str, Any], + rows_by_chunk_id: dict[str, dict[str, Any]], +) -> str: + parts = [_table_summary_content(row)] + parts.extend(_connected_image_parts(row, rows_by_chunk_id)) + return '\n\n'.join(part for part in parts if part) + + +def _connected_media_parts( + row: dict[str, Any], + rows_by_chunk_id: dict[str, dict[str, Any]], +) -> list[str]: + connected_targets: list[tuple[int, str]] = [] + for target_id in iter_connected_target_ids(row): + target_row = rows_by_chunk_id.get(target_id) + if not target_row: + continue + target_type = normalize_chunk_type(target_row.get('chunk_type')) + if target_type == 'table': + target_content = _compose_table_content(target_row, rows_by_chunk_id) + elif target_type == 'image': + target_content = _image_display_content(target_row) + else: + continue + if target_content: + sort_key = int(target_row.get('sort_order', 0) or 0) + connected_targets.append((sort_key, target_content)) + connected_targets.sort(key=lambda item: item[0]) + return [content for _, content in connected_targets] + + +def _connected_image_parts( + row: dict[str, Any], + rows_by_chunk_id: dict[str, dict[str, Any]], +) -> list[str]: + parts: list[str] = [] + for target_id in iter_connected_target_ids(row): + target_row = rows_by_chunk_id.get(target_id) + if not target_row: + continue + if normalize_chunk_type(target_row.get('chunk_type')) != 'image': + continue + content = _image_display_content(target_row) + if content: + parts.append(content) + return parts + + +def _image_display_content(row: dict[str, Any]) -> str: + display_ref = ( + str(row.get('asset_url') or '').strip() + or str(row.get('file_path') or '').strip() + ) + description = str(row.get('content') or '').strip() + lines: list[str] = [] + if display_ref: + lines.append(f'[Image: {display_ref}]') + elif description: + lines.append('[Image description]') + if description: + lines.extend(line for line in description.split('\n') if line.strip()) + return '\n'.join(lines) + + def _table_summary_content(row: dict[str, Any]) -> str: metadata = row.get('chunk_metadata') or row.get('metadata') or {} if not isinstance(metadata, dict): diff --git a/packages/shared-python/shared/services/retrieval/publication_service.py b/packages/shared-python/shared/services/retrieval/publication_service.py index 6e70aa71..8911c3b7 100644 --- a/packages/shared-python/shared/services/retrieval/publication_service.py +++ b/packages/shared-python/shared/services/retrieval/publication_service.py @@ -237,12 +237,18 @@ def publish_document_graph( *, job_id: str, job_result_id: str, + top_summary: str | None = None, ) -> None: job = db.execute(select(Job).where(Job.job_id == job_id)).scalar_one_or_none() if not job: raise RuntimeError(f"Job not found for graph publication: {job_id}") - self._publish_document_graph_for_job(db, job=job, job_result_id=job_result_id) + self._publish_document_graph_for_job( + db, + job=job, + job_result_id=job_result_id, + top_summary=top_summary, + ) def _publish_document_graph_for_job( self, @@ -250,6 +256,7 @@ def _publish_document_graph_for_job( *, job: Job, job_result_id: str, + top_summary: str | None = None, ) -> None: metadata = job.job_metadata or {} @@ -271,6 +278,7 @@ def _publish_document_graph_for_job( namespace=namespace, document_id=document_id, job_result_id=job_result_id, + top_summary=top_summary, ) def remove_document_graph( diff --git a/packages/shared-python/shared/services/storage/zip_chunk_schema.py b/packages/shared-python/shared/services/storage/zip_chunk_schema.py index 2cb36b43..7dc0be09 100644 --- a/packages/shared-python/shared/services/storage/zip_chunk_schema.py +++ b/packages/shared-python/shared/services/storage/zip_chunk_schema.py @@ -78,13 +78,19 @@ def format_chunks( if chunk_type == "text": metadata.update( - _format_text_metadata( - chunk=chunk, - chunk_type_str=chunk_type_str, - content=str(content), - existing_metadata=existing_metadata, - resource_target_map=resource_target_map, - ) + { + "tokens": existing_metadata.get("tokens") + or chunk.get("tokens", 0), + "keywords": existing_metadata.get("keywords") + or chunk.get("keywords", []), + "connect_to": _build_embed_connect_to( + chunk=chunk, + chunk_type_str=chunk_type_str, + content=str(content), + existing_metadata=existing_metadata, + resource_target_map=resource_target_map, + ), + } ) elif chunk_type == "image": if image_info: @@ -105,6 +111,13 @@ def format_chunks( "keywords", [] ) metadata["tokens"] = [] + metadata["connect_to"] = _build_embed_connect_to( + chunk=chunk, + chunk_type_str=chunk_type_str, + content=str(content), + existing_metadata=existing_metadata, + resource_target_map=resource_target_map, + ) elif chunk_type == "page": metadata["keywords"] = existing_metadata.get("keywords") or [] metadata["connect_to"] = existing_metadata.get("connect_to") or [] @@ -140,20 +153,17 @@ def _base_chunk_metadata( "summary": existing_metadata.get("summary") or chunk.get("summary", ""), "page_nums": existing_metadata.get("page_nums", []), } - document_top_summary = str(existing_metadata.get("document_top_summary") or "").strip() - if document_top_summary: - metadata["document_top_summary"] = document_top_summary return metadata -def _format_text_metadata( +def _build_embed_connect_to( *, chunk: dict[str, Any], chunk_type_str: Any, content: str, existing_metadata: dict[str, Any], resource_target_map: dict[str, str], -) -> dict[str, Any]: +) -> list[Any]: relationship_refs = parse_relationship_refs( chunk.get("type_raw") or chunk_type_str, content, @@ -168,11 +178,7 @@ def _format_text_metadata( or chunk.get("connectto"), resource_target_map, ) - return { - "tokens": existing_metadata.get("tokens") or chunk.get("tokens", 0), - "keywords": existing_metadata.get("keywords") or chunk.get("keywords", []), - "connect_to": merge_connections(embed_connections, related_connections), - } + return merge_connections(embed_connections, related_connections) def _resolve_table_file_path( diff --git a/packages/shared-python/shared/services/storage/zip_result_service.py b/packages/shared-python/shared/services/storage/zip_result_service.py index fe1f0c6d..8d6b229a 100644 --- a/packages/shared-python/shared/services/storage/zip_result_service.py +++ b/packages/shared-python/shared/services/storage/zip_result_service.py @@ -6,6 +6,8 @@ from __future__ import annotations +import json +import os from typing import Any from loguru import logger @@ -69,6 +71,7 @@ def generate_zip_package( statistics = self._schema.calculate_statistics(formatted_chunks) doc_nav, hierarchy = self._build_navigation_outputs( + add_dir=add_dir, formatted_chunks=formatted_chunks, source_file_name=source_file_name, ) @@ -120,13 +123,42 @@ def generate_zip_package( def _build_navigation_outputs( self, *, + add_dir: str, formatted_chunks: list[dict[str, Any]], source_file_name: str, ) -> tuple[dict[str, Any] | None, dict[str, Any]]: + """Prefer the already-enriched on-disk doc_nav; rebuild only as fallback.""" try: + existing = self._load_existing_doc_nav(add_dir) + if existing is not None: + hierarchy = self._schema.build_hierarchy_dict( + existing.get("sections", []) + ) + logger.info("Using enriched on-disk doc_nav.json for ZIP package") + return existing, hierarchy + doc_nav = self._schema.build_doc_nav(formatted_chunks, source_file_name) hierarchy = self._schema.build_hierarchy_dict(doc_nav.get("sections", [])) return doc_nav, hierarchy except Exception as exc: logger.warning(f"generate doc_nav.json fail {exc}") return None, {} + + @staticmethod + def _load_existing_doc_nav(add_dir: str) -> dict[str, Any] | None: + if not add_dir: + return None + path = os.path.join(add_dir, "doc_nav.json") + if not os.path.isfile(path): + return None + try: + with open(path, encoding="utf-8") as handle: + payload = json.load(handle) + except Exception as exc: + logger.warning(f"Failed to read existing doc_nav.json: {exc}") + return None + if not isinstance(payload, dict): + return None + if not isinstance(payload.get("sections"), list): + return None + return payload