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

fixed code implicit special and ran black for CI

Browse Source
This commit is contained in:
ppanchigar
2026-02-04 10:57:40 -06:00
parent 6751a5ab7a
commit 5b619b2f5f
12 changed files with 81 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/codes/code_funcs.py
+7 -2
View File
@@ -229,8 +229,13 @@ def code_implicit_special(service, filename):
"ST": ["ST Codes TBD"],
}
code_answer_dict = {}
if llm_answer_final[0] in special_case_mapping:
code_answer_dict["PROCEDURE_CD"] = special_case_mapping[llm_answer_final]
# Check if llm_answer_final is a list and has at least one element, and if the first element is in the special_case_mapping
if (
isinstance(llm_answer_final, list)
and len(llm_answer_final) > 0
and llm_answer_final[0] in special_case_mapping
):
code_answer_dict["PROCEDURE_CD"] = special_case_mapping[llm_answer_final[0]]
code_answer_dict["PROCEDURE_CD_DESC"] = llm_answer_final
return code_answer_dict
src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py
+3 -1
View File
@@ -34,9 +34,11 @@ def prompt_exhibit_level(
# Normalize CLAIM_TYPE_CD if present: convert single-element lists to strings
# This field should be a single value even though it's a 1:N field
if "CLAIM_TYPE_CD" in llm_answer_final:
llm_answer_final["CLAIM_TYPE_CD"] = string_utils.normalize_one_to_one_field_value(
llm_answer_final["CLAIM_TYPE_CD"] = (
string_utils.normalize_one_to_one_field_value(
"CLAIM_TYPE_CD", llm_answer_final["CLAIM_TYPE_CD"]
)
)
return llm_answer_final
src/pipelines/clients/clover/prompts/prompt_calls.py
+3 -1
View File
@@ -40,9 +40,11 @@ def prompt_exhibit_level(
# Normalize CLAIM_TYPE_CD if present: convert single-element lists to strings
# This field should be a single value even though it's a 1:N field
if "CLAIM_TYPE_CD" in llm_answer_final:
llm_answer_final["CLAIM_TYPE_CD"] = string_utils.normalize_one_to_one_field_value(
llm_answer_final["CLAIM_TYPE_CD"] = (
string_utils.normalize_one_to_one_field_value(
"CLAIM_TYPE_CD", llm_answer_final["CLAIM_TYPE_CD"]
)
)
return llm_answer_final
src/pipelines/runner.py
+6 -2
View File
@@ -108,11 +108,15 @@ def main(client: str = "saas", testing=False, test_params={}):
"""
# Check if batch_id is specified
if not config.BATCH_ID:
raise ValueError("batch_id is required. Please specify batch_id in your command")
raise ValueError(
"batch_id is required. Please specify batch_id in your command"
)
# Check if max_workers is not negative
if config.MAX_WORKERS < 0:
raise ValueError(f"Invalid configuration: MAX_WORKERS must be non-negative (got {config.MAX_WORKERS})")
raise ValueError(
f"Invalid configuration: MAX_WORKERS must be non-negative (got {config.MAX_WORKERS})"
)
# Validate client
registry = get_registry()
src/pipelines/saas/file_processing.py
-1
View File
@@ -228,7 +228,6 @@ def run_one_to_one_prompts(
# Add HSC's N/A if field doesn't exist yet
one_to_one_results[key] = value
one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one(
one_to_one_results, filename
)
src/pipelines/saas/main.py
+6 -2
View File
@@ -86,11 +86,15 @@ def safe_process_file(item, constants, run_timestamp):
def main(testing=False, test_params={}):
# Check if batch_id is specified
if not config.BATCH_ID:
raise ValueError("batch_id is required. Please specify batch_id in your command")
raise ValueError(
"batch_id is required. Please specify batch_id in your command"
)
# Check if max_workers is not negative
if config.MAX_WORKERS < 0:
raise ValueError(f"Invalid configuration: MAX_WORKERS must be non-negative (got {config.MAX_WORKERS})")
raise ValueError(
f"Invalid configuration: MAX_WORKERS must be non-negative (got {config.MAX_WORKERS})"
)
# Set up per-file logging (creates separate log files for each document)
logging_utils.setup_per_file_logging()
src/pipelines/shared/extraction/tin_npi_funcs.py
+30 -14
View File
@@ -30,7 +30,9 @@ def run_provider_info_fields(
all_tins, all_npis, tin_matches, npi_matches = get_all_tins_and_npis(contract_text)
exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis = get_tin_extraction_quality(tin_matches, npi_matches, filename)
exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis = (
get_tin_extraction_quality(tin_matches, npi_matches, filename)
)
relevant_pages = get_relevant_pages(all_tins, all_npis, text_dict, filename)
@@ -224,7 +226,10 @@ def normalize_to_lists(provider: dict):
return tins, npis, names
def deunknown_and_deduplicate(group_tins, group_npis, group_names, other_tins, other_npis, other_names):
def deunknown_and_deduplicate(
group_tins, group_npis, group_names, other_tins, other_npis, other_names
):
# De-Unknown GROUP fields
group_tins = [v for v in group_tins if not string_utils.is_empty(v)]
@@ -234,20 +239,25 @@ def deunknown_and_deduplicate(group_tins, group_npis, group_names, other_tins, o
# Deduplicate while preserving order and remove any "UNKNOWN" entries and overlaps between group and other
group_tins = list(dict.fromkeys(group_tins))
other_tins = list(dict.fromkeys(other_tins))
other_tins = [v for v in other_tins if v not in group_tins and not string_utils.is_empty(v)]
other_tins = [
v for v in other_tins if v not in group_tins and not string_utils.is_empty(v)
]
group_npis = list(dict.fromkeys(group_npis))
other_npis = list(dict.fromkeys(other_npis))
other_npis = [v for v in other_npis if v not in group_npis and not string_utils.is_empty(v)]
other_npis = [
v for v in other_npis if v not in group_npis and not string_utils.is_empty(v)
]
group_names = list(dict.fromkeys(group_names))
other_names = list(dict.fromkeys(other_names))
other_names = [v for v in other_names if v not in group_names and not string_utils.is_empty(v)]
other_names = [
v for v in other_names if v not in group_names and not string_utils.is_empty(v)
]
return group_tins, group_npis, group_names, other_tins, other_npis, other_names
def merge_provider_info_with_one_to_one(one_to_one_results: dict, filename: str):
"""
Merges provider_info into one_to_one_results.
@@ -272,11 +282,11 @@ def merge_provider_info_with_one_to_one(one_to_one_results: dict, filename: str)
elif not isinstance(provider_name, list):
provider_name = []
prov_info_json = one_to_one_results[
"PROV_INFO_JSON"
] # Returns list
prov_info_json = one_to_one_results["PROV_INFO_JSON"] # Returns list
prov_info_json_list = (
json.loads(prov_info_json) if isinstance(prov_info_json, str) else prov_info_json
json.loads(prov_info_json)
if isinstance(prov_info_json, str)
else prov_info_json
)
# Track if we found any name matches
found_match = False
@@ -316,9 +326,11 @@ def merge_provider_info_with_one_to_one(one_to_one_results: dict, filename: str)
for name in provider_name:
group_names.add(name)
group_tins, group_npis, group_names, other_tins, other_npis, other_names = deunknown_and_deduplicate(
group_tins, group_npis, group_names, other_tins, other_npis, other_names = (
deunknown_and_deduplicate(
group_tins, group_npis, group_names, other_tins, other_npis, other_names
)
)
# Filter out records with NO_IDENTIFIERS_FOUND
prov_info_json_list = [
@@ -370,7 +382,11 @@ def deduplicate_providers(
name = prov_info_dict.get("NAME")
# Skip providers where all critical identification fields are unknown
if (string_utils.is_empty(tin) and string_utils.is_empty(npi) and string_utils.is_empty(name)):
if (
string_utils.is_empty(tin)
and string_utils.is_empty(npi)
and string_utils.is_empty(name)
):
continue
key = (tin, npi, name)
@@ -380,6 +396,7 @@ def deduplicate_providers(
return valid_providers
def flag_missed_prov_info(page_text, providers, page_num, filename):
# Get all TINs and NPIs found via regex on this page
regex_tins = [
@@ -423,7 +440,6 @@ def flag_missed_prov_info(page_text, providers, page_num, filename):
)
def get_relevant_pages(all_tins, all_npis, text_dict, filename):
# If there are no identifiers, return early with default values
if not all_tins and not all_npis:
@@ -451,6 +467,7 @@ def get_relevant_pages(all_tins, all_npis, text_dict, filename):
relevant_pages = [relevant_pages[0]]
return relevant_pages
def get_prov_info_json(relevant_pages, text_dict, filename, payer_name):
if not relevant_pages:
prov_info_json = [
@@ -475,7 +492,6 @@ def get_prov_info_json(relevant_pages, text_dict, filename, payer_name):
return prov_info_json
def get_tin_extraction_quality(tin_matches, npi_matches, filename):
# Log OCR corrections for monitoring
exact_tins = [tin for tin, quality in tin_matches if quality == "EXACT"]
src/pipelines/shared/postprocessing/aarete_derived.py
+4 -1
View File
@@ -234,7 +234,10 @@ def get_crosswalk_fields(answer_dicts: list, constants: Constants):
)
# Assign the result: convert to string if it's a single-value field with one element
if to_field_name in SINGLE_VALUE_FIELDS and len(to_field_answer_list) == 1:
if (
to_field_name in SINGLE_VALUE_FIELDS
and len(to_field_answer_list) == 1
):
answer_dict[to_field_name] = to_field_answer_list[0]
else:
answer_dict[to_field_name] = to_field_answer_list
src/tests/test_tin_npi_funcs.py
-1
View File
@@ -30,7 +30,6 @@ class TestTinNpiFuncs:
# Test no matches
assert tin_npi_funcs.get_all_matches("No TINs here", pattern) == []
@patch("src.pipelines.saas.prompts.prompt_calls.provider_name_match_check")
def test_merge_provider_info_with_hybrid_smart_chunking(
self, mock_name_match_check
src/utils/string_utils.py
+4 -6
View File
@@ -867,7 +867,9 @@ def normalize_one_to_one_field_value(field_name: str, value) -> str | list:
return str(value) if value is not None else "N/A"
def normalize_one_to_one_answers_dict(answers_dict: dict[str, str | list]) -> dict[str, str | list]:
def normalize_one_to_one_answers_dict(
answers_dict: dict[str, str | list],
) -> dict[str, str | list]:
"""
Normalize all values in a 1:1 answers dictionary.
@@ -893,11 +895,7 @@ def normalize_one_to_one_answers_dict(answers_dict: dict[str, str | list]) -> di
"PROVIDER_NAME": ["Provider A", "Provider B"]
}
"""
DO_NOT_NORMALIZE = [
"PROV_INFO_JSON",
"PAYER_STATE",
"PROVIDER_STATE"
]
DO_NOT_NORMALIZE = ["PROV_INFO_JSON", "PAYER_STATE", "PROVIDER_STATE"]
normalized = {}
for field_name, value in answers_dict.items():
if field_name not in DO_NOT_NORMALIZE: