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

Merged in feature/TIN_stats_update (pull request #898)

Browse Source 
Feature/TIN stats update

* Tin_stats_report

* black formatting

* minor changes

* black format applied

* Addressed the katons comments

* Merged DEV into feature/TIN_stats_update

* black reformat


Approved-by: Katon Minhas
This commit is contained in:
Rahul Ailaboina
2026-03-05 16:14:45 +00:00
committed by Katon Minhas
parent e99e554d23
commit 38034f168b
6 changed files with 474 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pipelines/saas/main.py
+14
View File
@@ -26,6 +26,8 @@ from src.parent_child import main as parent_child_main
from src.document_classification import main as dtc_main
from src.qc_qa.pipeline import (
generate_statistics,
generate_tin_contract_summary,
generate_tin_statistics,
run_qc_qa_pipeline,
save_qc_qa_outputs,
)
@@ -281,7 +283,19 @@ def main(testing=False, test_params={}):
# Generate validation statistics from CC version
logging.info("Generating QC/QA statistics...")
qc_qa_stats_df = generate_statistics(validated_cc_df)
tin_stats_df = generate_tin_statistics(validated_cc_df)
tin_contract_summary_df = generate_tin_contract_summary(validated_cc_df)
logging.info("QC/QA statistics generated successfully")
# Save QC/QA outputs (local + S3) - separate from main results
save_qc_qa_outputs(
validated_df=validated_cc_df,
stats_df=qc_qa_stats_df,
run_timestamp=run_timestamp,
write_to_s3=config.WRITE_TO_S3,
tin_stats_df=tin_stats_df,
tin_contract_summary_df=tin_contract_summary_df,
)
except Exception as e:
logging.error(f"QC/QA validation on CC results failed: {str(e)}")
logging.error(f"Full traceback:\n{traceback.format_exc()}")
src/qc_qa/__main__.py
+16 -3
View File
@@ -24,6 +24,8 @@ from src.config import get_arg_value
from src.qc_qa.pipeline import (
convert_excel_to_csv,
generate_statistics,
generate_tin_contract_summary,
generate_tin_statistics,
read_file_by_extension,
run_qc_qa_pipeline,
save_qc_qa_outputs,
@@ -33,6 +35,8 @@ from src.qc_qa.pipeline import (
def main():
"""Main entry point for standalone QC/QA validation."""
logging.basicConfig(level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s")
# Get arguments
input_file = get_arg_value("input_file", None)
output_file = get_arg_value("output_file", None)
@@ -84,7 +88,8 @@ def main():
# Generate statistics
logging.info("Generating statistics...")
output_stats_df = generate_statistics(validated_df)
tin_stats_df = generate_tin_statistics(validated_df)
tin_contract_summary_df = generate_tin_contract_summary(validated_df)
# Save to qa_qc_output folder (local) and S3 using centralized function
logging.info("Saving outputs to qa_qc_output and S3...")
@@ -93,6 +98,8 @@ def main():
stats_df=output_stats_df,
run_timestamp=run_timestamp,
write_to_s3=config.WRITE_TO_S3,
tin_stats_df=tin_stats_df,
tin_contract_summary_df=tin_contract_summary_df,
)
# Legacy: Save to user-specified output file if provided (supports CSV and Excel)
@@ -121,8 +128,14 @@ def main():
logging.warning(f"Warning: Could not remove temporary file {csv_file}: {e}")
logging.info("QC/QA Validation Complete!")
logging.info(f"\nOutputs saved to:")
logging.info(f" - Local: {config.QC_QA_OUTPUT_DIRECTORY}/{run_timestamp}/")
logging.info(
f"\nOutputs saved to: {config.QC_QA_OUTPUT_DIRECTORY}/{run_timestamp}/"
)
logging.info(f" - {config.BATCH_ID}-QC-QA-VALIDATED.csv")
logging.info(
f" - {config.BATCH_ID}-QC-QA-STATS.xlsx "
f"(sheets: QC-QA-STATS, QC-QA-TIN-STATS, QC-QA-TIN-CONTRACT-SUMMARY)"
)
if config.WRITE_TO_S3:
logging.info(
f" - S3: s3://{config.S3_OUTPUT_BUCKET}/{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/"
src/qc_qa/pipeline.py
+261 -8
View File
@@ -21,6 +21,7 @@ import pandas as pd
from src import config
from src.utils import qa_qc_utils as qc
from src.utils.string_utils import is_filled, extract_nested_values
import logging
# Suppress pandas FutureWarnings
@@ -599,6 +600,8 @@ def save_qc_qa_outputs(
stats_df: pd.DataFrame,
run_timestamp: str,
write_to_s3: bool = True,
tin_stats_df: pd.DataFrame = None,
tin_contract_summary_df: pd.DataFrame = None,
) -> None:
"""
Save QC/QA outputs to local qa_qc_output folder and optionally to S3.
@@ -618,28 +621,271 @@ def save_qc_qa_outputs(
from src.utils import io_utils
import logging
output_dir = os.path.join(config.QC_QA_OUTPUT_DIRECTORY, run_timestamp)
os.makedirs(output_dir, exist_ok=True)
# Save validated data locally
io_utils.write_local(validated_df, "", run_timestamp, "qc_qa_validated")
logging.info(
f"Saved QC/QA validated data to: {config.QC_QA_OUTPUT_DIRECTORY}/{run_timestamp}/{config.BATCH_ID}-QC-QA-VALIDATED.csv"
f"Saved QC/QA validated data to: {output_dir}/{config.BATCH_ID}-QC-QA-VALIDATED.csv"
)
# Save statistics locally
io_utils.write_local(stats_df, "", run_timestamp, "qc_qa_stats")
logging.info(
f"Saved QC/QA statistics to: {config.QC_QA_OUTPUT_DIRECTORY}/{run_timestamp}/{config.BATCH_ID}-QC-QA-STATS.csv"
)
# Save combined stats Excel with both stats as separate sheets
stats_excel_path = os.path.join(output_dir, f"{config.BATCH_ID}-QC-QA-STATS.xlsx")
with pd.ExcelWriter(stats_excel_path, engine="openpyxl") as writer:
stats_df.to_excel(writer, sheet_name="QC-QA-STATS", index=False)
if tin_stats_df is not None:
tin_stats_df.to_excel(writer, sheet_name="QC-QA-TIN-STATS", index=False)
if tin_contract_summary_df is not None:
tin_contract_summary_df.to_excel(
writer, sheet_name="QC-QA-TIN-CONTRACT-SUMMARY", index=False
)
logging.info(f"Saved combined stats Excel to: {stats_excel_path}")
# Upload to S3 if enabled
if write_to_s3:
logging.info("Uploading QC/QA outputs to S3...")
io_utils.write_s3(validated_df, "", run_timestamp, "qc_qa_validated")
io_utils.write_s3(stats_df, "", run_timestamp, "qc_qa_stats")
s3_key = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-QC-QA-STATS.xlsx"
try:
with open(stats_excel_path, "rb") as f:
config.S3_CLIENT.put_object(
Bucket=config.S3_OUTPUT_BUCKET, Key=s3_key, Body=f.read()
)
logging.info(f"Uploaded combined stats Excel to S3: {s3_key}")
except Exception as e:
logging.warning(f"Failed to upload combined Excel to S3: {e}")
logging.info(
f"Uploaded to S3: {config.S3_OUTPUT_BUCKET}/{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/"
)
def generate_tin_statistics(df: pd.DataFrame) -> pd.DataFrame:
"""
Generate summary statistics for the validated DataFrame.
Args:
df: Validated DataFrame
Returns:
DataFrame containing statistics
"""
import ast
cols = [
"FILENAME_TIN",
"PROV_GROUP_TIN",
"PROV_GROUP_NPI",
"PROV_GROUP_NAME_FULL",
"PROV_OTHER_TIN",
"PROV_OTHER_NPI",
"PROV_OTHER_NAME_FULL",
"PROV_INFO_JSON",
]
prov_cols_to_validate = [
"PROV_GROUP_TIN",
"PROV_GROUP_NPI",
"PROV_GROUP_NAME_FULL",
"PROV_OTHER_TIN",
"PROV_OTHER_NPI",
"PROV_OTHER_NAME_FULL",
]
lob_product_program_cols = [
"LOB",
"PRODUCT",
"PROGRAM",
"AARETE_DERIVED_LOB",
"AARETE_DERIVED_PROGRAM",
"NETWORK",
"AARETE_DERIVED_NETWORK",
]
effective_date_cols = [
"AARETE_DERIVED_EFFECTIVE_DT",
"REIMB_EFFECTIVE_DT",
"EFFECTIVE_DT",
]
rows = []
def append_metric(metric: str, filled: Optional[int], total: Optional[int]) -> None:
blank = None if filled is None or total is None else int(total - filled)
rows.append(
{
"Metric": metric,
"Filled": filled,
"Blank": blank,
"Total": total,
}
)
# Rows filled count per column
for col in cols:
if col not in df.columns:
append_metric(col, None, len(df))
continue
filled = df[col].apply(is_filled).sum()
append_metric(col, int(filled), len(df))
# PROV_INFO_JSON validation: check if each prov col's values appear in JSON
if "PROV_INFO_JSON" in df.columns:
col_filled_mask = {
col: df[col].apply(is_filled)
for col in prov_cols_to_validate
if col in df.columns
}
json_blank_mask = ~df["PROV_INFO_JSON"].apply(is_filled)
json_blank_total = int(json_blank_mask.sum())
for check_col in prov_cols_to_validate:
if check_col not in df.columns:
continue
col_filled = col_filled_mask[check_col]
checkable = int(
col_filled.sum()
) # denominator = all filled rows for this column
def value_in_json(row, _col=check_col):
if not is_filled(row["PROV_INFO_JSON"]):
return False # JSON empty — not a match
values = extract_nested_values(row[_col])
if not values:
return False # no extractable values — not a match
json_str = str(row["PROV_INFO_JSON"])
return all(v in json_str for v in values if v)
matched = df[col_filled].apply(value_in_json, axis=1).sum()
append_metric(
f"{check_col} value found in PROV_INFO_JSON (of filled {check_col} rows)",
int(matched),
checkable,
)
append_metric(
f"{check_col} filled when PROV_INFO_JSON is blank",
int((col_filled & json_blank_mask).sum()),
json_blank_total,
)
# LOB / Product / Program filled counts
for col in lob_product_program_cols:
if col not in df.columns:
append_metric(col, None, len(df))
continue
filled = df[col].apply(is_filled).sum()
append_metric(col, int(filled), len(df))
# Effective date filled counts
for col in effective_date_cols:
if col not in df.columns:
append_metric(col, None, len(df))
continue
filled = df[col].apply(is_filled).sum()
append_metric(col, int(filled), len(df))
result = pd.DataFrame(rows)
for col in ["Filled", "Blank", "Total"]:
result[col] = result[col].astype("Int64")
return result
def generate_tin_contract_summary(df: pd.DataFrame) -> pd.DataFrame:
"""
Generate one summary row per contract for provider fields filled when JSON is blank.
Args:
df: Validated DataFrame
Returns:
DataFrame containing per-contract summary statistics
"""
contract_col = qc.QC_CONTRACT_NAME
agreement_col = qc.QC_AGREEMENT_NAME
provider_cols = [
"PROV_GROUP_TIN",
"PROV_GROUP_NPI",
"PROV_GROUP_NAME_FULL",
"PROV_OTHER_TIN",
"PROV_OTHER_NPI",
"PROV_OTHER_NAME_FULL",
]
if contract_col not in df.columns:
return pd.DataFrame()
summary_rows = []
for contract_name, contract_df in df.groupby(contract_col, dropna=False):
row = {
"FILE_NAME": contract_name,
"Row Count": len(contract_df),
}
if agreement_col in contract_df.columns:
agreement_values = (
contract_df[agreement_col].dropna().astype(str).map(str.strip)
)
agreement_values = agreement_values[agreement_values != ""]
row[agreement_col] = (
agreement_values.iloc[0] if not agreement_values.empty else ""
)
json_filled_mask = (
contract_df["PROV_INFO_JSON"].apply(is_filled)
if "PROV_INFO_JSON" in contract_df.columns
else pd.Series(False, index=contract_df.index)
)
json_blank_mask = ~json_filled_mask
row["PROV_INFO_JSON Filled Rows"] = int(json_filled_mask.sum())
row["PROV_INFO_JSON Blank Rows"] = int(json_blank_mask.sum())
for col in provider_cols:
if col not in contract_df.columns:
row[f"{col} Filled Rows"] = None
row[f"{col} Filled When JSON Blank"] = None
continue
col_filled_mask = contract_df[col].apply(is_filled)
row[f"{col} Filled Rows"] = int(col_filled_mask.sum())
row[f"{col} Filled When JSON Blank"] = int(
(col_filled_mask & json_blank_mask).sum()
)
summary_rows.append(row)
result = pd.DataFrame(summary_rows)
preferred_order = [
"FILE_NAME",
agreement_col,
"Row Count",
"PROV_GROUP_TIN Filled Rows",
"PROV_GROUP_NPI Filled Rows",
"PROV_GROUP_NAME_FULL Filled Rows",
"PROV_OTHER_TIN Filled Rows",
"PROV_OTHER_NPI Filled Rows",
"PROV_OTHER_NAME_FULL Filled Rows",
"PROV_INFO_JSON Filled Rows",
"PROV_INFO_JSON Blank Rows",
"PROV_GROUP_TIN Filled When JSON Blank",
"PROV_GROUP_NPI Filled When JSON Blank",
"PROV_GROUP_NAME_FULL Filled When JSON Blank",
"PROV_OTHER_TIN Filled When JSON Blank",
"PROV_OTHER_NPI Filled When JSON Blank",
"PROV_OTHER_NAME_FULL Filled When JSON Blank",
]
result = result[[col for col in preferred_order if col in result.columns]]
int_cols = [
col for col in result.columns if col.endswith("Rows") or col == "Row Count"
]
for col in int_cols:
if col in result.columns:
result[col] = result[col].astype("Int64")
return result
def generate_statistics(df: pd.DataFrame) -> pd.DataFrame:
"""
Generate summary statistics for the validated DataFrame.
@@ -652,7 +898,14 @@ def generate_statistics(df: pd.DataFrame) -> pd.DataFrame:
"""
length_stats = {}
stats = {}
stats["percentage_filled"] = (df.notna().sum() / len(df) * 100).round(2).to_dict()
filled_mask = df.apply(
lambda col: col.map(
lambda x: x is not None
and not (isinstance(x, float) and pd.isna(x))
and str(x).strip() != ""
)
)
stats["percentage_filled"] = (filled_mask.sum() / len(df) * 100).round(2).to_dict()
if qc.QC_CONTRACT_NAME in df.columns:
stats["distinct_contract_count"] = df[qc.QC_CONTRACT_NAME].nunique()
src/tests/test_qc_qa_pipeline.py
+117 -18
View File
@@ -574,52 +574,151 @@ class TestPipelineFunctions:
assert "Manual Review Flag (Blank)" in result.columns
# Tests for save_qc_qa_outputs
def _patch_save_qc_qa_common(self, mocker):
"""Shared patches for save_qc_qa_outputs tests.
Returns (mock_excel_writer, mock_stats_df) where mock_stats_df is a
MagicMock whose .to_excel() is a no-op (avoids real file I/O when
pandas calls to_excel on the writer context).
"""
mocker.patch("os.makedirs")
mocker.patch("src.config.BATCH_ID", "test_batch")
mocker.patch("src.config.QC_QA_OUTPUT_DIRECTORY", "/tmp/qc_qa")
mock_writer_ctx = MagicMock()
mock_excel_writer = mocker.patch("src.qc_qa.pipeline.pd.ExcelWriter")
mock_excel_writer.return_value.__enter__ = MagicMock(
return_value=mock_writer_ctx
)
mock_excel_writer.return_value.__exit__ = MagicMock(return_value=False)
# Use a MagicMock DataFrame so .to_excel() is a no-op
mock_stats_df = MagicMock()
return mock_excel_writer, mock_stats_df
def test_save_qc_qa_outputs_local_only(self, sample_dataframe, mocker):
"""Test saving outputs locally without S3."""
mock_write_local = mocker.patch("src.utils.io_utils.write_local")
mock_write_s3 = mocker.patch("src.utils.io_utils.write_s3")
mock_excel_writer, mock_stats_df = self._patch_save_qc_qa_common(mocker)
save_qc_qa_outputs(
validated_df=sample_dataframe,
stats_df=sample_dataframe,
stats_df=mock_stats_df,
run_timestamp="run_20241114_10-00_test",
write_to_s3=False,
)
# Verify local writes were called
assert mock_write_local.call_count == 2
mock_write_local.assert_any_call(
# write_local called once for validated_df only (stats go to Excel)
mock_write_local.assert_called_once_with(
sample_dataframe, "", "run_20241114_10-00_test", "qc_qa_validated"
)
mock_write_local.assert_any_call(
sample_dataframe, "", "run_20241114_10-00_test", "qc_qa_stats"
)
# Verify S3 was NOT called
# Excel writer was invoked for stats
mock_excel_writer.assert_called_once()
# S3 was NOT called
mock_write_s3.assert_not_called()
def test_save_qc_qa_outputs_with_s3(self, sample_dataframe, mocker):
"""Test saving outputs locally and to S3."""
"""Test saving outputs locally and to S3 (validated_df + Excel upload)."""
mock_write_local = mocker.patch("src.utils.io_utils.write_local")
mock_write_s3 = mocker.patch("src.utils.io_utils.write_s3")
mock_excel_writer, mock_stats_df = self._patch_save_qc_qa_common(mocker)
mocker.patch("src.config.S3_OUTPUT_BUCKET", "test-bucket")
mock_s3_client = MagicMock()
mocker.patch("src.config.S3_CLIENT", mock_s3_client)
mocker.patch("builtins.open", mocker.mock_open(read_data=b""))
save_qc_qa_outputs(
validated_df=sample_dataframe,
stats_df=sample_dataframe,
stats_df=mock_stats_df,
run_timestamp="run_20241114_10-00_test",
write_to_s3=True,
)
# Verify local writes
assert mock_write_local.call_count == 2
# Verify S3 writes were called
assert mock_write_s3.call_count == 2
mock_write_s3.assert_any_call(
# write_local called once for validated_df
mock_write_local.assert_called_once_with(
sample_dataframe, "", "run_20241114_10-00_test", "qc_qa_validated"
)
mock_write_s3.assert_any_call(
sample_dataframe, "", "run_20241114_10-00_test", "qc_qa_stats"
# write_s3 called once for validated_df
mock_write_s3.assert_called_once_with(
sample_dataframe, "", "run_20241114_10-00_test", "qc_qa_validated"
)
# S3 put_object called once for the combined Excel
mock_s3_client.put_object.assert_called_once()
call_kwargs = mock_s3_client.put_object.call_args[1]
assert call_kwargs["Bucket"] == "test-bucket"
assert "QC-QA-STATS.xlsx" in call_kwargs["Key"]
def test_save_qc_qa_outputs_with_tin_stats(self, sample_dataframe, mocker):
"""Test that tin_stats_df and tin_contract_summary_df are written as extra sheets."""
mocker.patch("src.utils.io_utils.write_local")
mock_excel_writer, mock_stats_df = self._patch_save_qc_qa_common(mocker)
# Use MagicMocks so to_excel() calls are no-ops
tin_stats = MagicMock()
tin_summary = MagicMock()
save_qc_qa_outputs(
validated_df=sample_dataframe,
stats_df=mock_stats_df,
run_timestamp="run_20241114_10-00_test",
write_to_s3=False,
tin_stats_df=tin_stats,
tin_contract_summary_df=tin_summary,
)
# ExcelWriter created with the expected path
mock_excel_writer.assert_called_once()
excel_path_arg = mock_excel_writer.call_args[0][0]
assert "test_batch-QC-QA-STATS.xlsx" in excel_path_arg
# to_excel called on all three DataFrames (stats, tin_stats, tin_summary)
mock_stats_df.to_excel.assert_called_once()
tin_stats.to_excel.assert_called_once()
tin_summary.to_excel.assert_called_once()
def test_save_qc_qa_outputs_without_optional_tin_dfs(
self, sample_dataframe, mocker
):
"""Test that omitting tin_stats_df / tin_contract_summary_df writes only QC-QA-STATS sheet."""
mocker.patch("src.utils.io_utils.write_local")
mock_excel_writer, mock_stats_df = self._patch_save_qc_qa_common(mocker)
save_qc_qa_outputs(
validated_df=sample_dataframe,
stats_df=mock_stats_df,
run_timestamp="run_20241114_10-00_test",
write_to_s3=False,
)
# ExcelWriter created with the expected path
excel_path_arg = mock_excel_writer.call_args[0][0]
assert "test_batch-QC-QA-STATS.xlsx" in excel_path_arg
# Only stats_df.to_excel was called (no TIN sheets)
mock_stats_df.to_excel.assert_called_once()
def test_save_qc_qa_outputs_s3_failure_is_warned(self, sample_dataframe, mocker):
"""Test that S3 upload failure logs a warning rather than raising."""
mocker.patch("src.utils.io_utils.write_local")
mocker.patch("src.utils.io_utils.write_s3")
_, mock_stats_df = self._patch_save_qc_qa_common(mocker)
mocker.patch("src.config.S3_OUTPUT_BUCKET", "test-bucket")
mock_s3_client = MagicMock()
mock_s3_client.put_object.side_effect = Exception("S3 unavailable")
mocker.patch("src.config.S3_CLIENT", mock_s3_client)
mocker.patch("builtins.open", mocker.mock_open(read_data=b""))
# Should not raise — failure is caught and logged as warning
save_qc_qa_outputs(
validated_df=sample_dataframe,
stats_df=mock_stats_df,
run_timestamp="run_20241114_10-00_test",
write_to_s3=True,
)
# Tests for generate_statistics
src/utils/io_utils.py
+17
View File
@@ -688,6 +688,15 @@ def write_local(
quoting=1,
)
return None
elif output_type == "qc_qa_tin_stats":
output_dir = os.path.join(config.QC_QA_OUTPUT_DIRECTORY, run_timestamp)
os.makedirs(output_dir, exist_ok=True)
df.to_csv(
os.path.join(output_dir, f"{config.BATCH_ID}-QC-QA-TIN-STATS.csv"),
index=False,
quoting=1,
)
return None
elif output_type == "individual_cc":
# Individual CC results
base_filename = os.path.splitext(filename)[0].strip()
@@ -833,6 +842,12 @@ def write_s3(
config.S3_CLIENT.put_object(
Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer
)
elif output_type == "qc_qa_tin_stats":
output_path = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-QC-QA-TIN-STATS.csv"
csv_buffer = df.to_csv(index=False).encode()
config.S3_CLIENT.put_object(
Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer
)
elif output_type == "individual_cc":
output_path = (
f"{config.BATCH_ID}/{run_timestamp}/individual/{filename}-RESULTS-CC.csv"
@@ -898,6 +913,8 @@ def write_s3(
logging.info(f"Saved QC/QA validated data file: {output_path}")
elif output_type == "qc_qa_stats":
logging.info(f"Saved QC/QA statistics file: {output_path}")
elif output_type == "qc_qa_tin_stats":
logging.info(f"Saved QC/QA TIN statistics file: {output_path}")
elif output_type == "qc_qa_error":
logging.info(f"Saved QC/QA error file: {output_path}")
elif output_type == "individual_cc":
src/utils/string_utils.py
+49
View File
@@ -1024,6 +1024,55 @@ def detect_state(text: str) -> str | None:
return None
def is_filled(x) -> bool:
"""
Returns True if x contains a meaningful (non-empty) value.
Treats None, NaN, and strings like "", "[]", "nan" as empty.
Inverse of is_empty for scalar values, with additional handling for "[]".
Args:
x: The value to check.
Returns:
bool: True if x is considered filled, False otherwise.
"""
return (
x is not None
and not (isinstance(x, float) and pd.isna(x))
and str(x).strip() not in ("", "[]", "nan")
)
def extract_nested_values(val) -> list[str]:
"""
Recursively extract leaf values from arbitrarily nested string representations.
Handles strings that represent nested lists (e.g. "['a', ['b', 'c']]") by
parsing with ast.literal_eval and recursing into any list items.
Args:
val: The value to extract from; will be converted to string.
Returns:
list[str]: Flat list of non-empty leaf string values.
"""
s = str(val).strip()
try:
parsed = ast.literal_eval(s)
except Exception:
return [s] if s else []
if isinstance(parsed, list):
results = []
for item in parsed:
results.extend(extract_nested_values(str(item)))
return [r for r in results if r.strip()]
elif isinstance(parsed, str):
return extract_nested_values(parsed) if parsed != s else ([s] if s else [])
else:
return [str(parsed).strip()]
def token_signature(text: str) -> tuple:
"""
Generate a normalized token signature from a text string.