Files
doczyai-pipelines/archive/scripts/cnc_hotfixes/cnc_hotfix_effective_date_utils.py
T
Katon Minhas afb6d5185d Merged in feature/lesser-table-caching-refactor-hybrid (pull request #847)
Feature/lesser table caching refactor hybrid

* chore: Remove unused duplicate main.py from shared pipeline

* fix: Correct crosswalk paths in aarete_derived.py

* chore: Remove unused documentation files from fieldExtraction

* docs: Add documentation files to documentation folder

* docs: Update README with uv setup, expanded project structure, and branching conventions

* docs: Add uv installation steps with Ubuntu/WSL emphasis

* Enable prompt caching for all remaining LLM calls

- Add _INSTRUCTION() functions for: EXHIBIT_HEADER, EXHIBIT_LINKAGE,
  EXHIBIT_TITLE_MATCH, DATE_FIX, DERIVED_TERM_DATE, CHECK_PROVIDER_NAME_MATCH,
  SPECIAL_CASE_ASSIGNMENT
- Update all invoke_claude() calls in saas and clover pipelines to use
  cache=True with corresponding _INSTRUCTION() functions
- Add new instructions to get_cacheable_instructions() for cache warming
- Update tests for new instruction functions

Functions now using caching:
- prompt_exhibit_level
- prompt_exhibit_lesser (EXHIBIT_LEVEL_LESSER_OF)
- prompt_fee_schedule_breakout
- prompt_grouper_breakout
- prompt_special_case_assignment
- prompt_exhibit_linkage
- prompt_exhibit_header
- prompt_smart_chunked (ONE_TO_ONE templates)
- prompt_date_fix
- prompt_derived_term_date
- prompt_exhibit_title_match
- provider_name_match_check

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Reorder

* feat: Add bcbs_promise client pipeline with OFFSET_TERM extraction

- Add new bcbs_promise client with HSC-based OFFSET_TERM field extraction
- Extract full paragraph text of offset/recoupment provisions from contracts
- Derive OFFSET_INDICATOR (Y/N) from OFFSET_TERM presence
- Fix reorder_columns to preserve extra columns not in COLUMN_ORDER
- Update QC/QA output path to outputs/qc_qa/

* fix: Update dev deps and test assertions for QC/QA output path

- Add pytest/pytest-mock to dev dependencies for mypy type checking
- Update test assertions to expect outputs/qc_qa instead of qa_qc_output

* style: Apply black formatting to prompt_templates.py

* Merge main, move scripts

* Archive some scripts

* update py version

* remove .py version file

* Remove ASCII characters

* Restore testbed code

* restore tracking

* Update testbed metrics

* Enable prompt caching for CODE_LAST_CHECK, FILL_BILL_TYPE, DUAL_LOB_CHECK, and GROUPER_BREAKOUT

- Add CODE_LAST_CHECK_INSTRUCTION() for service specificity classification
- Add FILL_BILL_TYPE_INSTRUCTION() for bill type code determination
- Add DUAL_LOB_CHECK_INSTRUCTION() for Medicare/Medicaid classification
- Update code_funcs.py to use caching for CODE_LAST_CHECK, FILL_BILL_TYPE, GROUPER_BREAKOUT
- Update postprocessing_funcs.py to use caching for DUAL_LOB_CHECK
- Add new instructions to get_cacheable_instructions() for cache warming
- Add unit tests for new instruction functions

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Fix postprocessing_funcs to remove invalid columns

* Merge branch 'main' into feature/lesser-table-caching-refactor-hybrid

* Revert prompt caching changes from aed1b73c

* update formatting

* Update imports


Approved-by: Sha Brown
Approved-by: Praneel Panchigar
2026-01-26 16:52:55 +00:00

139 lines
5.3 KiB
Python

import re
import hybrid_smart_chunking_funcs
regex_backticks = r"\|([^|]+)\|"
# TODO: leave this one in here; don't move it to prompts until we get confirmation that we want to handle meridian differently.
def get_effective_date_meridian_prompt(context):
prompt_template = f""""
Please analyze the following contract and extract ONLY the Signature Date. Follow these precise rules in order:
1. Look for Signature Date
2. If multiple matches are found:
- Prioritize the latest full date
3. If only partial dates are found, return N/A.
4. Return the date in YYYY-MM-DD format or N/A if date not found. Enclose just your final answer in |pipes|.
Here is the contract to analyze:
## START CONTRACT TEXT ##
{context}
## END CONTRACT TEXT ##
Before returning the output, VERIFY the date format is YYYY-MM-DD. If the date is not found, return N/A.
Finally state "Answer: |YYYY-MM-DD| or |N/A|"
"""
return prompt_template
def chunk_with_include_exclude_keywords(
text_dict: dict[str, str],
include_keywords: list[str],
exclude_keywords: list[str],
case_sensitive: bool = False,
apply_date_filtering: bool = True,
):
"""Look for pages that include any of the include_keywords and exclude any pages that contain any of the exclude_keywords.
Args:
text_dict (dict[str, str]): A dictionary keyed by string page number and valued
by string page contents
include_keywords (list[str]): A list of string keywords to include
exclude_keywords (list[str]): A list of string keywords to exclude
case_sensitive (bool): A flag for converting all keywords and text to lower case,
in order to ignore case altogether.
apply_date_filtering (bool): A flag to filter out the pages not containing a date.
Returns:
list[str]: The sorted list of string page numbers that include all include_keywords and exclude any exclude_keywords
"""
if not case_sensitive:
include_keywords = [keyword.lower() for keyword in include_keywords]
exclude_keywords = [keyword.lower() for keyword in exclude_keywords]
text_dict = {key: value.lower() for key, value in text_dict.items()}
if apply_date_filtering:
pages_containing_date = set()
for page_num, page_content in text_dict.items():
dates_list = re.findall(hybrid_smart_chunking_funcs.DATE_PATTERN, page_content)
if len(dates_list) > 0:
pages_containing_date.add(page_num)
pages_containing_date = sorted(pages_containing_date, key=int)
# debug lines for inclusion and exclusion keywords found on each page
# print(f"Pages containing date: {pages_containing_date}")
d = list() # to record details
for page_num, page_content in text_dict.items():
if page_num in pages_containing_date:
inclusion_keywords = set()
exclusion_keywords = set()
contains_inclusion_keyword = False
for keyword in include_keywords:
contains_inclusion_keyword = True
if keyword in page_content:
inclusion_keywords.add(keyword)
for keyword in exclude_keywords:
if keyword in page_content:
exclusion_keywords.add(keyword)
# print(f"Page #: {page_num} contains inclusion keywords: {inclusion_keywords} and contain exclusion keywords: {exclusion_keywords} | Inclusion Flag: {contains_inclusion_keyword}")
if contains_inclusion_keyword:
d.append(
{
"Page Number": page_num,
"Inclusion Keywords": inclusion_keywords,
"Exclusion Keywords": exclusion_keywords,
}
)
# print(f"Page #: {page_num} contains inclusion keywords: {inclusion_keywords} and contain exclusion keywords: {exclusion_keywords}")
page_list = []
for page in text_dict.keys():
if not apply_date_filtering or (
apply_date_filtering and page in pages_containing_date
):
if any(
keyword in text_dict[page] for keyword in include_keywords
) and not any(keyword in text_dict[page] for keyword in exclude_keywords):
# if str(int(page) - 1) in text_dict.keys():
# page_list.append(str(int(page) - 1))
# print(f"Taking page #: {str(int(page) - 1)} (as buffer for following page)")
page_list.append(page)
# print(f"Taking page #: {page}")
# if str(int(page) + 1) in text_dict.keys():
# page_list.append(str(int(page) + 1))
# # print(f"Taking page #: {str(int(page) + 1)} (as buffer for previous page)")
page_list_sorted = sorted([int(page_str) for page_str in set(page_list)])
page_list_sorted_str = list(map(str, page_list_sorted))
# print(", ".join([page for page in page_list_sorted_str]))
# print(f"Obtained total pages: {len(page_list_sorted_str)}")
return page_list_sorted_str, d