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

Merged in feature/provider-state (pull request #602)

Browse Source 
Feature/provider state

* Add prompt for extracting provider state information in investment prompts

* Add state normalization function and integrate into context field extraction

* black format string_utils.py; add tests for normalize_state_to_abbreviation function

* black format string_utils_test.py

* Add PROVIDER_STATE to investment column order

* Merged main into feature/provider-state


Approved-by: Katon Minhas
This commit is contained in:
Alex Galarce
2025-07-07 21:42:49 +00:00
parent 5a8adf838d
commit 479675b05c
5 changed files with 587 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/utils/string_utils.py
+171 -39
View File
@@ -10,8 +10,11 @@ import src.utils.llm_utils as llm_utils
from src import config
from src.enums.delimiters import Delimiter
from src.prompts import preprocessing_prompts
from src.regex.regex_patterns import (BACKTICK_PATTERN, PIPE_PATTERN,
TRIPLE_BACKTICK_PATTERN)
from src.regex.regex_patterns import (
BACKTICK_PATTERN,
PIPE_PATTERN,
TRIPLE_BACKTICK_PATTERN,
)
def extract_text_from_delimiters(
@@ -71,7 +74,8 @@ def extract_text_from_delimiters(
return matches[match_index]
def page_key_sort(page_key: str) -> tuple[int, float|str]:
def page_key_sort(page_key: str) -> tuple[int, float | str]:
"""Sorts page keys, prioritizing numeric keys first.
Args:
@@ -85,14 +89,14 @@ def page_key_sort(page_key: str) -> tuple[int, float|str]:
(0, 1.0)
>>> page_key_sort("A")
(1, "A")
Example Usage with mixed types:
>>> L = ["10", "2", "A", "1"]
>>> sort(L, key=page_key_sort)
>>> print(L)
['1', '2', '10', 'A']
"""
try:
return (0, float(page_key)) # Numeric pages first, in numerical order
@@ -129,18 +133,18 @@ def json_parsing_search(response_text: str, field_list: list[str]) -> dict[str,
response_text = response_text.rsplit("}", 1)[0]
else:
response_text = response_text.rstrip(",") + "}"
field_l = []
answer_l = []
position_dict = {}
for f in field_list:
location = response_text.find('"' + f + '"')
if location != -1:
position_dict[location] = f
field_list = list(dict(sorted(position_dict.items())).values())
for f in field_list:
if f in response_text:
field_l.append(f)
@@ -253,6 +257,7 @@ def primary_string_to_dict(string_dict, filename):
data.append(dict_)
return data
def universal_json_load(string_dict: str):
"""Matches json dicts and list of dicts - NOT simple lists
Args:
@@ -262,10 +267,10 @@ def universal_json_load(string_dict: str):
Raises:
ValueError: If no valid JSON object is found in the input string. In addition,
the traceback as well as the offending string are printed.
"""
# Try to match dictionary or list of dictionaries first
dict_match = re.search(r'\{.*\}|\[\s*?\{.*\}\s*?\]', string_dict, re.DOTALL)
dict_match = re.search(r"\{.*\}|\[\s*?\{.*\}\s*?\]", string_dict, re.DOTALL)
if dict_match:
matched_str = dict_match.group()
try:
@@ -275,32 +280,36 @@ def universal_json_load(string_dict: str):
print(f"Original string: {string_dict}")
print(f"Traceback: {traceback.format_exc()}")
raise ValueError(f"JSON parsing failed: {e.msg} at position {e.pos}") from e
# If no dict pattern found, try to match list of strings
list_match = re.search(r'\[(.*?)\]', string_dict, re.DOTALL)
list_match = re.search(r"\[(.*?)\]", string_dict, re.DOTALL)
if list_match:
matched_str = list_match.group()
try:
# Clean up the string list before parsing
cleaned_str = re.sub(r'(?<!\\)"', '"', matched_str) # Handle escaped quotes
cleaned_str = re.sub(r"'", '"', cleaned_str) # Replace single quotes with double quotes
cleaned_str = re.sub(
r"'", '"', cleaned_str
) # Replace single quotes with double quotes
return json.loads(cleaned_str)
except json.JSONDecodeError as e:
print(f"Failed to parse list: {e}")
print(f"Original string: {string_dict}")
print(f"Traceback: {traceback.format_exc()}")
raise ValueError(f"JSON parsing failed: {e.msg} at position {e.pos}") from e
return {}
reimbursement_strings = [ # These are for `method='keyword'`
reimbursement_strings = [ # These are for `method='keyword'`
"%",
"$",
"percent",
"billed charges"
"billed charges",
]
reimb_regex = r"(?<![$%])(?:\$\d+|\d+[$%])(?![$%])"
# TODO: check should this always be checking just page 1? Bc in file_processing and table_utils it is but in preprocessing_funcs it's checking specific pages
def contains_reimbursement(text, page="1", method="keyword"):
"""
@@ -329,7 +338,8 @@ def contains_reimbursement(text, page="1", method="keyword"):
else:
raise ValueError("Invalid method. Choose 'keyword' or 'regex'.")
def count_reimbursements_in_exhibit(exhibit_text: str) -> int: #JUST by regex
def count_reimbursements_in_exhibit(exhibit_text: str) -> int: # JUST by regex
"""Counts reimbursements in an exhibit text. Reimbursements are detected by a regex
search as defined by `reimb_regex`.
@@ -341,15 +351,16 @@ def count_reimbursements_in_exhibit(exhibit_text: str) -> int: #JUST by regex
"""
return len(re.findall(reimb_regex, exhibit_text))
def is_empty(value: str | list | pd.Series, pd_mask: bool=True) -> bool | pd.Series:
def is_empty(value: str | list | pd.Series, pd_mask: bool = True) -> bool | pd.Series:
"""
Checks if a value is considered empty or invalid.
Args:
value: The value to check.
Can be a string, list, or pandas Series.
pd_mask (bool): Only relevant for Series inputs.
If True, returns a mask for empty values in a pandas Series.
pd_mask (bool): Only relevant for Series inputs.
If True, returns a mask for empty values in a pandas Series.
Otherwise, the function returns True iff the entire Series is empty.
Returns:
@@ -361,13 +372,15 @@ def is_empty(value: str | list | pd.Series, pd_mask: bool=True) -> bool | pd.Ser
- If `pd_mask` is False, returns True iff all elements are empty/invalid, or if the entire Series is empty.
"""
empty_values = [None, "", "N/A", "NA", "null", "none", "NaN", np.nan, "nan", "None"]
if isinstance(value, list): # Handle list inputs
return not value or all(is_empty(v) for v in value)
# if it's a pd.Series, return the mask (True for empty values)
if isinstance(value, pd.Series):
return value.isna() | (value.isin(empty_values)) if pd_mask else value.isna().all()
return (
value.isna() | (value.isin(empty_values)) if pd_mask else value.isna().all()
)
if pd.isna(value):
return True
@@ -378,9 +391,10 @@ def is_empty(value: str | list | pd.Series, pd_mask: bool=True) -> bool | pd.Ser
def datetime_str():
return datetime.now().strftime("[%Y-%m-%d %H:%M:%S]")
def extract_signature_page(text_dict: dict, filename: str) -> dict:
"""Extracts the signature page(s) from a contract text dictionary and returns it as a dictionary.
First tries to match the textract marker "this page has N signature(s)", then falls back to looking
First tries to match the textract marker "this page has N signature(s)", then falls back to looking
for the word "signature" if no matches are found. "this page has N signature(s)" is Textract's marker for signatures
Args:
@@ -394,19 +408,25 @@ def extract_signature_page(text_dict: dict, filename: str) -> dict:
signature_pages = {}
# This regex pattern matches the Textract marker for signatures
# It will look for both numeric and a few textual representations of numbers (one-ten)
textract_pattern = re.compile(r"this page has (?:\d+|one|two|three|four|five|six|seven|eight|nine|ten) signature", re.IGNORECASE)
textract_pattern = re.compile(
r"this page has (?:\d+|one|two|three|four|five|six|seven|eight|nine|ten) signature",
re.IGNORECASE,
)
# First pass: Try to match the full textract pattern
for page_num, page_text in text_dict.items():
if textract_pattern.search(page_text):
signature_pages[page_num] = page_text
# If no pages found with textract pattern, try with keyword
if not signature_pages:
page_items = list(text_dict.items())
for i, (page_num, page_text) in enumerate(page_items):
lower_text = page_text.lower()
if "signature page follow" in lower_text or "signature authorization" in lower_text:
if (
"signature page follow" in lower_text
or "signature authorization" in lower_text
):
# curr_page = int(float(page_num))
if page_num not in signature_pages:
signature_pages[page_num] = page_text # current page
@@ -414,8 +434,7 @@ def extract_signature_page(text_dict: dict, filename: str) -> dict:
next_page_num, next_page_text = page_items[i + 1]
# next_page = int(float(next_page_num))
if next_page_num not in signature_pages:
signature_pages[next_page_num] = next_page_text # next page
signature_pages[next_page_num] = next_page_text # next page
return signature_pages
@@ -424,9 +443,9 @@ def extract_effective_date_pages(text_dict: dict, filename: str) -> dict:
"""
Extracts the effective date page(s) from a contract text dictionary and returns it as a dictionary.
This function identifies pages that either contain the Textract marker
"this page has N signature(s)" or the keyword 'effective date:'.
Additionally, it ensures that the first page (page 1) is always included
This function identifies pages that either contain the Textract marker
"this page has N signature(s)" or the keyword 'effective date:'.
Additionally, it ensures that the first page (page 1) is always included
in the result, as it often contains effective date.
Args:
@@ -434,7 +453,7 @@ def extract_effective_date_pages(text_dict: dict, filename: str) -> dict:
filename (str): The name of the file being processed.
Returns:
dict: A dictionary containing the extracted effective date page(s),
dict: A dictionary containing the extracted effective date page(s),
where keys are page numbers and values are the corresponding page text.
"""
@@ -442,16 +461,129 @@ def extract_effective_date_pages(text_dict: dict, filename: str) -> dict:
# Regex pattern to match the Textract marker for signature pages which often has effective dates
# This pattern accounts for both numeric and textual representations of numbers (one-ten).
textract_pattern = re.compile(r"this page has (?:\d+|one|two|three|four|five|six|seven|eight|nine|ten) signature", re.IGNORECASE)
textract_pattern = re.compile(
r"this page has (?:\d+|one|two|three|four|five|six|seven|eight|nine|ten) signature",
re.IGNORECASE,
)
# Iterate through each page in the text dictionary.
for page_num, page_text in text_dict.items():
# page_num = int(float(page_num))
# Check if the page contains the Textract marker or the keyword 'effective date:'.
if textract_pattern.search(page_text) or 'effective date:' in page_text.lower():
if textract_pattern.search(page_text) or "effective date:" in page_text.lower():
effective_date_pages[page_num] = page_text
# Always include the first page (page 1) as it often contains effective date.
elif page_num == 1 or page_num == "1" or page_num == "1.0":
effective_date_pages[page_num] = page_text
return effective_date_pages
def normalize_state_to_abbreviation(state_value: str) -> str:
"""Convert state names or abbreviations to standardized two-letter abbreviations.
Handles various formats of state names and abbreviations, including full names,
common abbreviations, and variations. Returns the standardized two-letter abbreviation
or the original value if no match is found.
Args:
state_value (str): State name or abbreviation from extraction
Returns:
str: Two-letter state abbreviation or original value if no match found
"""
if not state_value or state_value in ["N/A", "UNKNOWN"]:
return state_value
# Dictionary mapping full state names to abbreviations
state_mapping = {
"alabama": "AL",
"alaska": "AK",
"arizona": "AZ",
"arkansas": "AR",
"california": "CA",
"colorado": "CO",
"connecticut": "CT",
"delaware": "DE",
"florida": "FL",
"georgia": "GA",
"hawaii": "HI",
"idaho": "ID",
"illinois": "IL",
"indiana": "IN",
"iowa": "IA",
"kansas": "KS",
"kentucky": "KY",
"louisiana": "LA",
"maine": "ME",
"maryland": "MD",
"massachusetts": "MA",
"michigan": "MI",
"minnesota": "MN",
"mississippi": "MS",
"missouri": "MO",
"montana": "MT",
"nebraska": "NE",
"nevada": "NV",
"new hampshire": "NH",
"new jersey": "NJ",
"new mexico": "NM",
"new york": "NY",
"north carolina": "NC",
"north dakota": "ND",
"ohio": "OH",
"oklahoma": "OK",
"oregon": "OR",
"pennsylvania": "PA",
"rhode island": "RI",
"south carolina": "SC",
"south dakota": "SD",
"tennessee": "TN",
"texas": "TX",
"utah": "UT",
"vermont": "VT",
"virginia": "VA",
"washington": "WA",
"west virginia": "WV",
"wisconsin": "WI",
"wyoming": "WY",
"district of columbia": "DC",
}
# Clean and normalize the input
cleaned_state = state_value.strip().lower()
# Check if it's already a valid 2-letter abbreviation
if len(cleaned_state) == 2 and cleaned_state.upper() in state_mapping.values():
return cleaned_state.upper()
# Check if it matches a full state name
if cleaned_state in state_mapping:
return state_mapping[cleaned_state]
# Handle common variations
variations = {
"calif": "CA",
"cal": "CA",
"fla": "FL",
"florida": "FL",
"ny": "NY",
"n.y.": "NY",
"tx": "TX",
"tex": "TX",
"penn": "PA",
"pa": "PA",
"mass": "MA",
"massachusetts": "MA",
"wash": "WA",
"washington state": "WA",
"d.c.": "DC",
"dc": "DC",
"washington dc": "DC",
}
if cleaned_state in variations:
return variations[cleaned_state]
# If no match found, return original value
return state_value