From 38034f168bb055b46cf770c609e7fa19509892e6 Mon Sep 17 00:00:00 2001 From: Rahul Ailaboina Date: Thu, 5 Mar 2026 16:14:45 +0000 Subject: [PATCH] Merged in feature/TIN_stats_update (pull request #898) 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 --- src/pipelines/saas/main.py | 14 ++ src/qc_qa/__main__.py | 19 ++- src/qc_qa/pipeline.py | 269 ++++++++++++++++++++++++++++++- src/tests/test_qc_qa_pipeline.py | 135 +++++++++++++--- src/utils/io_utils.py | 17 ++ src/utils/string_utils.py | 49 ++++++ 6 files changed, 474 insertions(+), 29 deletions(-) diff --git a/src/pipelines/saas/main.py b/src/pipelines/saas/main.py index d6c4831..9eb9ffb 100644 --- a/src/pipelines/saas/main.py +++ b/src/pipelines/saas/main.py @@ -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()}") diff --git a/src/qc_qa/__main__.py b/src/qc_qa/__main__.py index 22860d3..39bb585 100644 --- a/src/qc_qa/__main__.py +++ b/src/qc_qa/__main__.py @@ -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/" diff --git a/src/qc_qa/pipeline.py b/src/qc_qa/pipeline.py index ce70fb9..98e17d3 100644 --- a/src/qc_qa/pipeline.py +++ b/src/qc_qa/pipeline.py @@ -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() diff --git a/src/tests/test_qc_qa_pipeline.py b/src/tests/test_qc_qa_pipeline.py index 5acd397..4200e0f 100644 --- a/src/tests/test_qc_qa_pipeline.py +++ b/src/tests/test_qc_qa_pipeline.py @@ -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 diff --git a/src/utils/io_utils.py b/src/utils/io_utils.py index 00eff0b..d47bb55 100644 --- a/src/utils/io_utils.py +++ b/src/utils/io_utils.py @@ -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": diff --git a/src/utils/string_utils.py b/src/utils/string_utils.py index 1d32615..1e422c9 100644 --- a/src/utils/string_utils.py +++ b/src/utils/string_utils.py @@ -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.