jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in bugfix/DAIP2-2138-reimbursement-primary-testing (pull request #910)

Browse Source 
Bugfix/DAIP2-2138 reimbursement primary testing

* updated table instructions for reimb term extraction

* updated split tables to include header rows

* issue fix

* Merged DEV into bugfix/DAIP2-2138-reimbursement-primary-testing

* black format fix


Approved-by: Katon Minhas
This commit is contained in:
Venkatakrishna Reddy Avula
2026-03-13 18:50:58 +00:00
committed by Katon Minhas
parent ec4617eeee
commit ddd1703e2f
2 changed files with 93 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pipelines/shared/extraction/page_funcs.py
+92 -3
View File
@@ -102,8 +102,23 @@ class Page:
# Table is small enough, don't split
return
# Identify header and title rows for proper splitting
header_row_index, title_row_indices = identify_header_and_title_rows(table_rows)
if title_row_indices:
logging.info(
f"Page {self.page_num}: Found {len(title_row_indices)} title row(s) "
f"at indices {title_row_indices}"
)
if header_row_index is not None:
logging.info(
f"Page {self.page_num}: Identified header row at index {header_row_index}"
)
# Split table into chunks based on cell count
table_chunks = split_table_by_cell_count(table_rows, num_columns, max_cells)
table_chunks = split_table_by_cell_count(
table_rows, num_columns, max_cells, header_row_index=header_row_index
)
# Create sub-pages
self.sub_pages = {}
@@ -329,19 +344,71 @@ def parse_table_rows(table_text: str, page_num: str = "") -> tuple[list[str], in
return rows, num_columns
def identify_header_and_title_rows(table_rows: list) -> tuple[Optional[int], list[int]]:
"""
Identify the header row and any title rows in a table.
A title row is identified when all non-empty cells contain the same value.
The header row is the first non-title row, typically containing column names.
Args:
table_rows: List of table rows (each row is a list)
Returns:
tuple[Optional[int], list[int]]: (header_row_index, list_of_title_row_indices)
"""
if not table_rows:
return None, []
title_row_indices = []
header_row_index = None
for i, row in enumerate(table_rows):
if not isinstance(row, list) or len(row) == 0:
continue
# Check if all non-empty cells in the row have the same value (title row pattern)
non_empty_values = [str(cell).strip() for cell in row if str(cell).strip()]
if len(non_empty_values) > 0 and len(set(non_empty_values)) == 1:
# All non-empty cells have the same value - likely a title row
title_row_indices.append(i)
logging.debug(
f"Detected title row at index {i}: '{non_empty_values[0]}' repeated across all cells"
)
elif header_row_index is None:
# First non-title row is likely the header
header_row_index = i
logging.debug(
f"Detected header row at index {i}: {row[:3]}{'...' if len(row) > 3 else ''}"
)
# If no header found (all rows are title rows or empty), use first row as header
if header_row_index is None and len(table_rows) > 0:
header_row_index = 0
logging.debug(f"No distinct header found, using first row (index 0) as header")
return header_row_index, title_row_indices
def split_table_by_cell_count(
table_rows: list[str], num_columns: int, max_cells: int = 40
table_rows: list[str],
num_columns: int,
max_cells: int = 40,
header_row_index: Optional[int] = None,
) -> list[str]:
"""
Split table rows into chunks based on cell count (rows × columns).
The product of rows × columns in each chunk will be <= max_cells.
This means tables with more columns will have fewer rows per chunk.
Headers are preserved at the beginning of each chunk for continuity.
Args:
table_rows: List of table row strings
table_rows: List of table row strings (each row is a list)
num_columns: Number of columns in the table
max_cells: Maximum number of cells (rows × columns) per chunk (default: 40)
header_row_index: Index of the header row to preserve in each chunk
Returns:
list[str]: List of table chunks, each as a string
@@ -354,9 +421,31 @@ def split_table_by_cell_count(
# Minimum of 1 row per chunk
max_rows = max(1, max_cells // num_columns)
# Get header row if specified
header_row = None
if header_row_index is not None and 0 <= header_row_index < len(table_rows):
header_row = table_rows[header_row_index]
logging.debug(
f"Header row will be preserved in sub-tables: {header_row[:3] if isinstance(header_row, list) else header_row}..."
)
chunks = []
for i in range(0, len(table_rows), max_rows):
chunk = table_rows[i : i + max_rows]
# For chunks after the first, prepend the header if available and not already included
if (
i > 0
and header_row is not None
and header_row_index is not None
and header_row_index < i
):
# Header was in a previous chunk, add it to this chunk
chunk = [header_row] + chunk
logging.debug(
f"Added header row to chunk starting at row {i} (chunk {len(chunks) + 1})"
)
chunk_strs = []
for row in chunk:
if isinstance(row, list):
src/prompts/prompt_templates.py
+1 -2
View File
@@ -601,8 +601,7 @@ REIMB_TERM: Describes the method by which the price of the Service is calculated
- When a unit of measure (e.g., "PMPM", "per member per month", "per visit", "per diem", "per case") appears in a table header, column label, or surrounding context, always append it to the REIMB_TERM for each rate in that table. For example, if the header says "Payment PMPM" and a row shows "$31.51", the REIMB_TERM should be "$31.51 PMPM", not just "$31.51".
- Include all SERVICE_TERM and REIMB_TERM pairs mentioned, even if they are redundant or overlapping. There should be at least one entry for each SERVICE_TERM and REIMB_TERM pair mentioned.
- If different reimbursement percentages or dollar values are mentioned for different effective dates, create separate entries for each effective date.
- For the tables: Extract EVERY ROW as a separate entry, including all line items, subtotals, totals, adjustments, charges, and calculated ratios. When rows involve different providers/entities, create separate entries for EACH provider/entity and include the provider/entity name in the SERVICE_TERM (e.g., "Provider Name – Service Category").
- Extract ONE entry per RATE STATEMENT in the contract. A rate statement includes percentages, dollar amounts, weight-based calculations (e.g., DRG weight), or other reimbursement methodologies.
- For the tables: Extract EVERY ROW as a separate entry, including all line items, subtotals, totals, adjustments, charges, calculated ratios, and all relevant information from every column in that row. ALWAYS include column headers and subheaders as part of the SERVICE_TERM to provide full context for the service, forming a complete sentence if necessary. If a row contains multiple reimbursement rate types (e.g., different rate columns such as facility vs non-facility, professional vs technical, inpatient vs outpatient, etc.), create separate entries for EACH rate type using the corresponding rate column header in the SERVICE_TERM. When rows involve different providers/entities, create separate entries for EACH provider/entity and include the provider/entity name in the SERVICE_TERM (e.g., "Provider Name – Service Category").
- GROUPING RULE: If multiple services are explicitly listed together (e.g., "Reference Lab and DME") AND share the same rate statement, extract as ONE entry preserving the exact combined SERVICE_TERM. Do not split them.
- SEPARATION RULE: If services appear in separate sentences or have different reimbursement methods, extract as separate entries.
- ENUMERATED SUB-CASES: When a paragraph introduces a service category and then lists sub-cases (i, ii, iii or a, b, c or 1, 2, 3) with different rates, extract each sub-case as a separate entry. For SERVICE_TERM, use the overall service description with only a brief sub-case label (e.g., "primary procedure", "second and subsequent procedures") - do NOT include billing logic, conditional clauses, or payment-determination language in the SERVICE_TERM. For REIMB_TERM, extract only the payment method (rate, percentage, dollar amount) - do NOT include the conditional clause that precedes it.