# -*- coding: utf-8 -*- """ Data Clarity CentralReach Data Warehouse Audit Runner ===================================================== Runs locally against the CentralReach Microsoft SQL Server database and audits both the current ``dw2`` schema and legacy ``insights`` schema. It exports a PHI-minimized JSON package that can be uploaded to the Data Clarity Audit Portal. Required packages: pip install pandas pymssql Connection settings are read only from environment variables. Never hard-code or upload live credentials. Required environment variables: CR_DW_SERVER CR_DW_USERNAME CR_DW_PASSWORD Optional environment variables: CR_DW_PORT=1433 CR_DW_DATABASE=insights CR_AUDIT_WINDOW_DAYS=730 CR_AUDIT_MAX_EXCEPTIONS=100 Example: python centralreach_audit_runner_sanitized.py --client-code AFL \ --client-name "AFL" --output AFL_audit.json """ from datetime import date, datetime, timezone import argparse import json import decimal import math import os import traceback import pandas as pd PYMSSQL_IMPORT_ERROR = None try: import pymssql except Exception as _exc: pymssql = None PYMSSQL_IMPORT_ERROR = "%s: %s" % (type(_exc).__name__, str(_exc)) # ============================================================================ # CONNECTION SETTINGS # ============================================================================ SERVER = os.getenv("CR_DW_SERVER", "").strip() PORT = int(os.getenv("CR_DW_PORT", "1433")) DATABASE = os.getenv("CR_DW_DATABASE", "insights").strip() or "insights" USERNAME = os.getenv("CR_DW_USERNAME", "").strip() PASSWORD = os.getenv("CR_DW_PASSWORD", "") SCHEMAS = ["dw2", "insights"] CONNECTION_TIMEOUT_SECONDS = int(os.getenv("CR_DW_CONNECTION_TIMEOUT", "30")) QUERY_TIMEOUT_SECONDS = int(os.getenv("CR_DW_QUERY_TIMEOUT", "240")) TRANSACTION_LOOKBACK_DAYS = int(os.getenv("CR_AUDIT_WINDOW_DAYS", "730")) MAX_EXCEPTION_RECORDS_PER_OBJECT = int(os.getenv("CR_AUDIT_MAX_EXCEPTIONS", "100")) TIER_WEIGHTS = {"T1": 0.40, "T2": 0.30, "T3": 0.20, "T4": 0.10} TIER_ORDER = ["T1", "T2", "T3", "T4"] TIER_RANK = {"T1": 4, "T2": 3, "T3": 2, "T4": 1} # ============================================================================ # AUDIT MAP # ============================================================================ MODULES = [ { "key": "authorizations", "label": "Authorizations", "functional_area": "Caseload Mapping and Authorization Readiness", "dashboard_impact": "Authorization ownership, active authorization monitoring, service-code coverage, pace, renewals, and supervision planning.", }, { "key": "clinical", "label": "Clinical Productivity", "functional_area": "Service Delivery", "dashboard_impact": "Provided hours, service-code productivity, utilization, supervision compliance, and clinician performance.", }, { "key": "capacity", "label": "Capacity and HR", "functional_area": "Capacity Management", "dashboard_impact": "Client/provider rosters, active status, staffing, caseload capacity, and clinic reporting.", }, { "key": "financial", "label": "Financial", "functional_area": "Financial Health", "dashboard_impact": "Payments, payor reporting, collections, deposits, and revenue-cycle analysis.", }, ] def fld(name, tier, candidates, zero_is_missing=False, action=None): return { "name": name, "tier": tier, "candidates": candidates, "zero_is_missing": bool(zero_is_missing), "action": action or "Populate or correct this field in CentralReach and confirm it reaches the next Data Warehouse load.", } # Each logical audit source can have one DW2 object and/or one legacy Insights # object. Every source object that exists is audited independently. AUDIT_SOURCES = [ { "key": "AuthorizationLine", "module": "authorizations", "sources": [("dw2", ["Authorization"]), ("insights", ["PayorAuthorization"])], "row_id": ["AuthorizationId", "Id", "AuthId"], "date_candidates": ["EndDate", "AuthEndDate"], "filter_type": "authorization", "fields": [ fld("Client ID", "T1", ["ClientContactId", "ClientId", "AuthClientId"]), fld("Authorization Group ID", "T1", ["AuthorizationGroupId", "AuthGroupId"]), fld("Service Code ID", "T1", ["ServiceCodeId", "AuthServiceCodeId"]), fld("Authorization Start Date", "T1", ["StartDate", "AuthStartDate"]), fld("Authorization End Date", "T1", ["EndDate", "AuthEndDate"]), fld("Authorization Number", "T2", ["AuthorizationNumber", "AuthNumber"]), fld("Authorized Hours", "T2", ["TotalHours", "AuthorizedHours", "AuthTotalHours", "AuthHours"], True), fld("Payor / Insurance ID", "T2", ["InsuranceId", "ClientPayorId", "AuthGroupClientPayorId", "AuthPayorPlanId"]), ], }, { "key": "AuthorizationGroup", "module": "authorizations", "sources": [("dw2", ["AuthorizationGroup"]), ("insights", ["PayorAuthorization"])], "row_id": ["AuthorizationGroupId", "AuthGroupId", "Id", "AuthId"], "date_candidates": ["EndDate", "AuthEndDate"], "filter_type": "authorization", "fields": [ fld("Client ID", "T1", ["ClientContactId", "ClientId", "AuthClientId"]), fld("Authorization Manager ID", "T1", ["ManagerId", "AuthManagerId", "AuthManagerEmployeeId"]), fld("Payor / Insurance ID", "T1", ["InsuranceId", "ClientPayorId", "AuthGroupClientPayorId", "AuthPayorPlanId"]), fld("Authorization Start Date", "T1", ["StartDate", "AuthStartDate"]), fld("Authorization End Date", "T1", ["EndDate", "AuthEndDate"]), fld("Authorization Frequency", "T2", ["Frequency", "AuthFrequency"]), fld("Total Authorized Hours", "T2", ["TotalHours", "AuthGroupTotalHours", "AuthorizedHours"], True), fld("Authorization Number", "T3", ["AuthorizationNumber", "AuthNumber"]), ], }, { "key": "Resources", "module": "authorizations", "sources": [("dw2", ["Resources"]), ("insights", ["Resource"])], "row_id": ["ResourceId", "Id"], "filter_type": "soft_delete", "fields": [ fld("Resource ID", "T1", ["ResourceId", "Id"]), fld("Client / Contact ID", "T1", ["ContactId", "ResourceContactId", "ClientId"]), fld("Resource Name", "T2", ["Name", "ResourceName", "FileName"]), fld("Creation Date", "T3", ["CreationDate", "ResourceCreationDate"]), fld("Organization ID", "T3", ["OrganizationId"]), ], }, { "key": "BillingEntries", "module": "clinical", "sources": [("dw2", ["BillingEntriesCurrent"]), ("insights", ["TimeBilling"])], "row_id": ["BillingEntryId", "TimeBillingId", "Id"], "date_candidates": ["ServiceStartTime", "ServiceDate", "TimeBillingServiceDate", "TimeBillingDateOfService"], "filter_type": "billing", "fields": [ fld("Billing Entry ID", "T1", ["BillingEntryId", "TimeBillingId", "Id"]), fld("Client ID", "T1", ["ClientContactId", "TimeBillingClientId", "ClientId"]), fld("Provider ID", "T1", ["ProviderContactId", "TimeBillingProviderId", "ProviderId"]), fld("Service Code ID", "T1", ["ServiceCodeId", "TimeBillingServiceCodeId"]), fld("Service Start", "T2", ["ServiceStartTime", "TimeWorkedFrom", "TimeBillingStartTime", "ServiceDate", "TimeBillingServiceDate"]), fld("Service End", "T2", ["ServiceEndTime", "TimeWorkedTo", "TimeBillingEndTime"]), fld("Service Units", "T2", ["UnitsOfService", "ServiceUnits", "TimeBillingUnits"], True), fld("Insurance ID", "T3", ["InsuranceId", "TimeBillingInsuranceId", "ClientPayorId"]), fld("Organization ID", "T3", ["OrganizationId"]), ], }, { "key": "ServiceCodes", "module": "clinical", "sources": [("dw2", ["ServiceCodes", "ServiceCode"]), ("insights", ["ServiceCode", "ServiceCodes"])], "row_id": ["ServiceCodeId", "Id"], "filter_type": "soft_delete", "fields": [ fld("Service Code ID", "T1", ["ServiceCodeId", "Id"]), fld("Service Code", "T1", ["Code", "ServiceCode"]), fld("Billable Status", "T2", ["IsBillable", "BillableStatus", "ServiceCodeIsBillable"]), fld("Calculation Type", "T2", ["CalcType", "CalculationType", "ServiceCodeCalculationType"]), fld("Service Category", "T3", ["ServiceCategory", "ServiceCodeCategory", "Category"]), fld("Minutes Per Unit", "T3", ["MinutesPerUnit"], True), ], }, { "key": "Contacts", "module": "capacity", "sources": [("dw2", ["Contacts"]), ("insights", ["Client", "Contacts"])], "row_id": ["ContactId", "ClientId", "Id"], "filter_type": "contacts", "fields": [ fld("Contact / Client ID", "T1", ["ContactId", "ClientId", "Id"]), fld("First Name", "T1", ["FirstName", "ClientFirstName"]), fld("Last Name", "T1", ["LastName", "ClientLastName"]), fld("Active Status", "T2", ["IsActive", "ActiveStatus", "ClientActiveStatus"]), fld("Primary Office", "T3", ["PrimaryOfficeLocationId", "OfficeLocationId", "ClientOfficeLocationName"]), fld("Organization ID", "T3", ["OrganizationId", "ClientOrganizationId"]), ], }, { "key": "Employees", "module": "capacity", "sources": [("dw2", ["Employee"]), ("insights", ["Employee"])], "row_id": ["EmployeeId", "ContactId", "Id"], "filter_type": "none", "fields": [ fld("Employee ID", "T1", ["EmployeeId", "ContactId", "Id"]), fld("Employment Type", "T2", ["EmployeeType", "EmploymentType"]), fld("Employment Position", "T2", ["EmploymentPosition", "Position"]), fld("Hire Date", "T3", ["HireDate", "EmployeeHireDate"]), fld("Department", "T3", ["Department", "EmployeeDepartment"]), fld("Organization ID", "T3", ["OrganizationId", "EmployeeOrganizationId"]), ], }, { "key": "Payments", "module": "financial", "sources": [("dw2", ["Payments"]), ("insights", ["Payment"])], "row_id": ["Id", "PaymentId", "PaymentID"], "date_candidates": ["RecordedDate", "PaymentRecordDate", "DepositDate", "PaymentDepositDate"], "filter_type": "payments", "fields": [ fld("Payment ID", "T1", ["Id", "PaymentId", "PaymentID"]), fld("Billing Entry ID", "T1", ["BillingEntryId", "PaymentBillingEntryId"]), fld("Payor ID", "T1", ["PayorId", "PaymentPayorId", "ClientPayorId"]), fld("Payment Amount", "T1", ["Amount", "PaymentAmount"], True), fld("Recorded Date", "T2", ["RecordedDate", "PaymentRecordDate"]), fld("Deposit Date", "T3", ["DepositDate", "PaymentDepositDate"]), fld("Payment Type", "T3", ["PaymentType", "PaymentTypeName"]), fld("Organization ID", "T3", ["OrganizationId"]), ], }, { "key": "ClientPayor", "module": "financial", "sources": [ ("dw2", ["ContactInsuranceCompanies"]), ("insights", ["ClientPayor", "PayorPlan"]), ], "row_id": ["InsuranceId", "ClientPayorId", "PayorPlanId", "Id"], "filter_type": "soft_delete", "fields": [ fld("Payor Record ID", "T1", ["InsuranceId", "ClientPayorId", "PayorPlanId", "Id"]), fld("Client / Contact ID", "T1", ["ContactId", "ClientId", "ClientPayorClientId"]), fld("Plan ID", "T2", ["PlanId", "PayorPlanId"]), fld("Payor / Company ID", "T2", ["CompanyId", "PayorId", "InsuranceCompanyId"]), fld("Member ID", "T3", ["MemberId", "SubscriberId", "ClientPayorMemberId"]), fld("Organization ID", "T3", ["OrganizationId"]), ], }, ] # ============================================================================ # HELPERS # ============================================================================ def qi(name): return "[" + str(name).replace("]", "]]" ) + "]" def connect_dw(): if pymssql is None: raise RuntimeError( "Python could not import pymssql. Install it with: pip install pymssql. " "Import error: %s" % (PYMSSQL_IMPORT_ERROR or "unknown") ) missing = [ name for name, value in { "CR_DW_SERVER": SERVER, "CR_DW_USERNAME": USERNAME, "CR_DW_PASSWORD": PASSWORD, }.items() if not value ] if missing: raise RuntimeError( "Missing required environment variables: %s" % ", ".join(missing) ) connect_kwargs = dict( server=SERVER, port=str(PORT), user=USERNAME, password=PASSWORD, database=DATABASE, login_timeout=CONNECTION_TIMEOUT_SECONDS, timeout=QUERY_TIMEOUT_SECONDS, charset="UTF-8", as_dict=False, autocommit=True, tds_version="7.4", appname="Data Clarity CentralReach DW Audit", ) # pymssql 2.2.8+ supports explicit encryption and read-only intent. # The fallback allows an older pymssql build to connect using its default # TLS negotiation, while still avoiding an external ODBC driver. try: connection = pymssql.connect( encryption="require", read_only=True, **connect_kwargs ) except TypeError: connection = pymssql.connect(**connect_kwargs) connector = "pymssql %s / %s" % ( getattr(pymssql, "__version__", "unknown"), getattr(pymssql, "get_dbversion", lambda: "FreeTDS")(), ) return connection, connector def resolve_name(candidates, available_lower): for candidate in candidates: found = available_lower.get(str(candidate).lower()) if found: return found return None def is_present_python(value, zero_is_missing=False): if value is None: return False if isinstance(value, str): text = value.strip() if text == "": return False if zero_is_missing: try: return float(text) != 0.0 except Exception: return True return True if isinstance(value, (int, float, decimal.Decimal)): try: number = float(value) if math.isnan(number): return False if zero_is_missing and number == 0.0: return False except Exception: pass return True if isinstance(value, (bytes, bytearray)): return len(value) > 0 return True def readiness_status(score): if score is None: return "Not Scored" if score >= 95: return "Ready" if score >= 85: return "Mostly Ready" if score >= 70: return "At Risk" return "Not Ready" def present_sql(column_name, zero_is_missing=False): text_present = "NULLIF(LTRIM(RTRIM(CONVERT(nvarchar(4000), %s))), '') IS NOT NULL" % qi(column_name) if zero_is_missing: return "(%s AND (TRY_CONVERT(decimal(38,10), %s) IS NULL OR TRY_CONVERT(decimal(38,10), %s) <> 0))" % ( text_present, qi(column_name), qi(column_name), ) return "(" + text_present + ")" def load_inventory(cursor): placeholders = ",".join("%s" for _ in SCHEMAS) sql = ( "SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, IS_NULLABLE, ORDINAL_POSITION " "FROM INFORMATION_SCHEMA.COLUMNS " "WHERE TABLE_SCHEMA IN (%s) " "ORDER BY TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION" % placeholders ) cursor.execute(sql, tuple(SCHEMAS)) inventory_rows = [] columns_by_object = {} for schema_name, table_name, column_name, data_type, is_nullable, ordinal_position in cursor.fetchall(): key = (schema_name, table_name) columns_by_object.setdefault(key, []).append(column_name) inventory_rows.append( { "Schema": schema_name, "Table": table_name, "FullName": "%s.%s" % (schema_name, table_name), "Column": column_name, "DataType": data_type, "IsNullable": str(is_nullable).upper() == "YES", "OrdinalPosition": int(ordinal_position), } ) return inventory_rows, columns_by_object def source_objects_for_spec(spec, columns_by_object): found = [] for schema_name, table_candidates in spec["sources"]: available = { table_name.lower(): table_name for (schema, table_name) in columns_by_object if schema.lower() == schema_name.lower() } actual = resolve_name(table_candidates, available) if actual: found.append((schema_name, actual)) return found def make_filter_sql(spec, columns_lower): conditions = [] def col(*names): return resolve_name(names, columns_lower) deleted_flag = col("IsDeleted", "Deleted") deleted_date = col("DeletedDate", "DeletedOn", "AuthDeletedDate", "ClientDeletedOn") void_flag = col("IsVoid", "IsVoided", "TimeBillingVoided") void_date = col("VoidedDate", "VoidDate", "PaymentVoidedDate") if spec.get("filter_type") in {"authorization", "soft_delete", "billing", "contacts"}: if deleted_flag: conditions.append("COALESCE(%s, 0) = 0" % qi(deleted_flag)) elif deleted_date: conditions.append("%s IS NULL" % qi(deleted_date)) if spec.get("filter_type") == "billing": if void_flag: conditions.append("COALESCE(%s, 0) = 0" % qi(void_flag)) elif void_date: conditions.append("%s IS NULL" % qi(void_date)) if spec.get("filter_type") == "payments" and void_date: conditions.append("%s IS NULL" % qi(void_date)) date_col = resolve_name(spec.get("date_candidates", []), columns_lower) if TRANSACTION_LOOKBACK_DAYS > 0 and date_col: if spec.get("filter_type") == "authorization": conditions.append( "(%s IS NULL OR %s >= DATEADD(day, -%d, SYSUTCDATETIME()))" % (qi(date_col), qi(date_col), TRANSACTION_LOOKBACK_DAYS) ) elif spec.get("filter_type") in {"billing", "payments"}: conditions.append( "%s >= DATEADD(day, -%d, SYSUTCDATETIME())" % (qi(date_col), TRANSACTION_LOOKBACK_DAYS) ) return " AND ".join(conditions) if conditions else "1 = 1" def tier_stats(field_rows, module_key=None): rows = field_rows if module_key is None else [r for r in field_rows if r["Module"] == module_key] output = [] for tier in TIER_ORDER: tier_rows = [r for r in rows if r["Tier"] == tier] present = sum(int(r["Present"]) for r in tier_rows) total = sum(int(r["Total"]) for r in tier_rows) pct_raw = present / total if total else 0.0 output.append( { "Tier": tier, "TierWeight": TIER_WEIGHTS[tier], "Present": present, "Total": total, "Missing": total - present, "PctFilledRaw": pct_raw, "PctFilled": round(pct_raw * 100.0, 2), "WeightedContributionRaw": pct_raw * TIER_WEIGHTS[tier], } ) return output def weighted_score(stats): active = [r for r in stats if r["Total"] > 0] if not active: return None total_weight = sum(r["TierWeight"] for r in active) if total_weight <= 0: return None return sum(r["WeightedContributionRaw"] for r in active) / total_weight * 100.0 def worst_tier(stats): active = [r for r in stats if r["Total"] > 0] if not active: return None return sorted(active, key=lambda r: (r["PctFilledRaw"], -TIER_RANK[r["Tier"]]))[0]["Tier"] def build_definitions(): module_lookup = {m["key"]: m for m in MODULES} rows = [] for spec in AUDIT_SOURCES: module = module_lookup[spec["module"]] source_names = [] for schema_name, candidates in spec["sources"]: source_names.extend("%s.%s" % (schema_name, candidate) for candidate in candidates) for field_spec in spec["fields"]: rows.append( { "Module": spec["module"], "ModuleName": module["label"], "LogicalTable": spec["key"], "CandidateSourceObjects": ", ".join(source_names), "ExpectedField": field_spec["name"], "Tier": field_spec["tier"], "TierWeight": TIER_WEIGHTS[field_spec["tier"]], "CandidateColumns": ", ".join(field_spec["candidates"]), "ZeroIsMissing": field_spec["zero_is_missing"], "RecommendedAction": field_spec["action"], } ) return rows # ============================================================================ # AUDIT EXECUTION # ============================================================================ def run_audit(): generated_at = datetime.now(timezone.utc) module_lookup = {m["key"]: m for m in MODULES} definition_rows = build_definitions() errors = [] field_rows = [] table_rows = [] exception_rows = [] missing_column_rows = [] inventory_rows = [] driver = None database_name = DATABASE login_name = None try: connection, driver = connect_dw() except Exception as exc: errors.append( { "Scope": "Connection", "LogicalTable": None, "SourceObject": None, "Error": str(exc), "Detail": traceback.format_exc(), } ) return build_outputs( generated_at, driver, database_name, login_name, inventory_rows, definition_rows, field_rows, table_rows, exception_rows, missing_column_rows, errors, connection_status="Failed", ) try: cursor = connection.cursor() try: cursor.execute("SELECT DB_NAME(), SUSER_SNAME()") database_name, login_name = cursor.fetchone() except Exception as exc: errors.append( { "Scope": "Connection Test", "LogicalTable": None, "SourceObject": None, "Error": str(exc), "Detail": "SELECT DB_NAME(), SUSER_SNAME()", } ) inventory_rows, columns_by_object = load_inventory(cursor) if not inventory_rows: errors.append( { "Scope": "Schema Discovery", "LogicalTable": None, "SourceObject": None, "Error": "Connected, but no columns were visible in the dw2 or insights schemas.", "Detail": "Confirm this login has SELECT permission and metadata visibility for the CentralReach warehouse views.", } ) for spec in AUDIT_SOURCES: module = module_lookup[spec["module"]] source_objects = source_objects_for_spec(spec, columns_by_object) if not source_objects: expected_sources = [] for schema_name, candidates in spec["sources"]: expected_sources.extend("%s.%s" % (schema_name, t) for t in candidates) errors.append( { "Scope": "Source Object", "LogicalTable": spec["key"], "SourceObject": None, "Error": "No candidate source object was found.", "Detail": "Candidates: " + ", ".join(expected_sources), } ) table_rows.append( { "Module": spec["module"], "ModuleName": module["label"], "LogicalTable": spec["key"], "SourceSchema": None, "SourceTable": None, "SourceObject": None, "SourceVersion": None, "TableExists": False, "RowCount": 0, "RowsWithGaps": 0, "GapRatePct": 0.0, "Score": None, "Status": "Not Scored", "WorstTier": None, "FieldsExpected": len(spec["fields"]), "FieldsFound": 0, "FilterApplied": None, } ) continue for schema_name, table_name in source_objects: full_name = "%s.%s" % (schema_name, table_name) columns = columns_by_object[(schema_name, table_name)] columns_lower = {c.lower(): c for c in columns} resolved_fields = [] for field_spec in spec["fields"]: actual_column = resolve_name(field_spec["candidates"], columns_lower) resolved_fields.append((field_spec, actual_column)) if not actual_column: missing_column_rows.append( { "Module": spec["module"], "ModuleName": module["label"], "LogicalTable": spec["key"], "SourceSchema": schema_name, "SourceTable": table_name, "SourceObject": full_name, "ExpectedField": field_spec["name"], "CandidateColumns": ", ".join(field_spec["candidates"]), "Tier": field_spec["tier"], "RecommendedAction": field_spec["action"], } ) row_id_column = resolve_name(spec["row_id"], columns_lower) where_sql = make_filter_sql(spec, columns_lower) select_parts = ["COUNT_BIG(*) AS [TotalRows]"] missing_conditions = [] for index, (field_spec, actual_column) in enumerate(resolved_fields): if actual_column: expression = present_sql(actual_column, field_spec["zero_is_missing"]) select_parts.append( "SUM(CASE WHEN %s THEN CAST(1 AS bigint) ELSE CAST(0 AS bigint) END) AS %s" % (expression, qi("Present_%d" % index)) ) missing_conditions.append("NOT " + expression) else: missing_conditions.append("1 = 1") any_missing_sql = "(" + " OR ".join(missing_conditions) + ")" if missing_conditions else "1 = 0" select_parts.append( "SUM(CASE WHEN %s THEN CAST(1 AS bigint) ELSE CAST(0 AS bigint) END) AS [GapRows]" % any_missing_sql ) aggregate_sql = "SELECT %s FROM %s.%s WHERE %s" % ( ", ".join(select_parts), qi(schema_name), qi(table_name), where_sql, ) values = {} total_rows = 0 gap_rows = 0 query_ok = True try: cursor.execute(aggregate_sql) record = cursor.fetchone() names = [d[0] for d in cursor.description] values = dict(zip(names, record)) total_rows = int(values.get("TotalRows") or 0) gap_rows = int(values.get("GapRows") or 0) except Exception as exc: query_ok = False errors.append( { "Scope": "Aggregate Query", "LogicalTable": spec["key"], "SourceObject": full_name, "Error": str(exc), "Detail": aggregate_sql, } ) current_fields = [] for index, (field_spec, actual_column) in enumerate(resolved_fields): if query_ok and actual_column: present = int(values.get("Present_%d" % index) or 0) total = total_rows elif query_ok and not actual_column: present = 0 total = total_rows if total_rows > 0 else 1 else: present = 0 total = 0 missing = max(total - present, 0) pct = present / total * 100.0 if total else 0.0 row = { "Module": spec["module"], "ModuleName": module["label"], "LogicalTable": spec["key"], "SourceSchema": schema_name, "SourceTable": table_name, "SourceObject": full_name, "SourceVersion": "DW2" if schema_name.lower() == "dw2" else "Insights", "Field": field_spec["name"], "ActualColumn": actual_column, "Tier": field_spec["tier"], "TierWeight": TIER_WEIGHTS[field_spec["tier"]], "Present": present, "Total": total, "Missing": missing, "PctFilled": round(pct, 2), "Status": readiness_status(pct if total else None), "ColumnExists": bool(actual_column), "ZeroIsMissing": field_spec["zero_is_missing"], "RecommendedAction": field_spec["action"], } field_rows.append(row) current_fields.append(row) object_stats = tier_stats(current_fields) object_score = weighted_score(object_stats) table_rows.append( { "Module": spec["module"], "ModuleName": module["label"], "LogicalTable": spec["key"], "SourceSchema": schema_name, "SourceTable": table_name, "SourceObject": full_name, "SourceVersion": "DW2" if schema_name.lower() == "dw2" else "Insights", "TableExists": True, "RowCount": total_rows, "RowsWithGaps": gap_rows, "GapRatePct": round(gap_rows / total_rows * 100.0 if total_rows else 0.0, 2), "Score": round(object_score, 1) if object_score is not None else None, "Status": readiness_status(object_score), "WorstTier": worst_tier(object_stats), "FieldsExpected": len(spec["fields"]), "FieldsFound": sum(1 for _, c in resolved_fields if c), "FilterApplied": where_sql, } ) # Limited row-level exception sample. No names or free-text fields # are selected unless they are themselves part of the audit map. if query_ok and gap_rows > 0 and MAX_EXCEPTION_RECORDS_PER_OBJECT > 0: sample_columns = [] if row_id_column: sample_columns.append("%s AS [__RowId]" % qi(row_id_column)) else: sample_columns.append("CAST(NULL AS nvarchar(100)) AS [__RowId]") for index, (_, actual_column) in enumerate(resolved_fields): if actual_column: sample_columns.append("%s AS %s" % (qi(actual_column), qi("F%d" % index))) sample_sql = "SELECT TOP (%d) %s FROM %s.%s WHERE (%s) AND %s" % ( MAX_EXCEPTION_RECORDS_PER_OBJECT, ", ".join(sample_columns), qi(schema_name), qi(table_name), where_sql, any_missing_sql, ) try: cursor.execute(sample_sql) sample_names = [d[0] for d in cursor.description] for sample_index, record in enumerate(cursor.fetchall()): item = dict(zip(sample_names, record)) raw_id = item.get("__RowId") row_id = str(raw_id) if raw_id is not None else "sample_%d" % (sample_index + 1) missing_items = [] for index, (field_spec, actual_column) in enumerate(resolved_fields): value = item.get("F%d" % index) if actual_column else None if not actual_column or not is_present_python(value, field_spec["zero_is_missing"]): missing_items.append((field_spec, actual_column, value)) if not missing_items: continue top = sorted([x[0]["tier"] for x in missing_items], key=lambda t: -TIER_RANK[t])[0] for field_spec, actual_column, value in missing_items: exception_rows.append( { "Module": spec["module"], "ModuleName": module["label"], "LogicalTable": spec["key"], "SourceSchema": schema_name, "SourceTable": table_name, "SourceObject": full_name, "RowKey": "%s|%s" % (full_name, row_id), "RowId": row_id, "TopTier": top, "MissingField": field_spec["name"], "MissingTier": field_spec["tier"], "CurrentValue": None if value is None else str(value), "Issue": "Expected source column was not found." if not actual_column else "Required value is blank, null, or zero.", "RecommendedAction": field_spec["action"], } ) except Exception as exc: errors.append( { "Scope": "Exception Sample Query", "LogicalTable": spec["key"], "SourceObject": full_name, "Error": str(exc), "Detail": sample_sql, } ) return build_outputs( generated_at, driver, database_name, login_name, inventory_rows, definition_rows, field_rows, table_rows, exception_rows, missing_column_rows, errors, connection_status="Connected", ) except Exception as exc: errors.append( { "Scope": "Script", "LogicalTable": None, "SourceObject": None, "Error": str(exc), "Detail": traceback.format_exc(), } ) return build_outputs( generated_at, driver, database_name, login_name, inventory_rows, definition_rows, field_rows, table_rows, exception_rows, missing_column_rows, errors, connection_status="Failed", ) finally: connection.close() def build_outputs( generated_at, driver, database_name, login_name, inventory_rows, definition_rows, field_rows, table_rows, exception_rows, missing_column_rows, errors, connection_status, ): module_score_rows = [] tier_output_rows = [] for module in MODULES: module_key = module["key"] module_fields = [r for r in field_rows if r["Module"] == module_key] module_tables = [r for r in table_rows if r["Module"] == module_key] stats = tier_stats(field_rows, module_key) score = weighted_score(stats) for stat in stats: tier_output_rows.append( { "Scope": "Module", "Module": module_key, "ModuleName": module["label"], "Tier": stat["Tier"], "TierWeight": stat["TierWeight"], "Present": stat["Present"], "Total": stat["Total"], "Missing": stat["Missing"], "PctFilled": stat["PctFilled"], "WeightedContributionPct": round(stat["WeightedContributionRaw"] * 100.0, 2), } ) total_rows = sum(int(r["RowCount"]) for r in module_tables) gap_rows = sum(int(r["RowsWithGaps"]) for r in module_tables) module_score_rows.append( { "Module": module_key, "ModuleName": module["label"], "FunctionalArea": module["functional_area"], "DashboardImpact": module["dashboard_impact"], "Score": round(score, 1) if score is not None else None, "Status": readiness_status(score), "WorstTier": worst_tier(stats), "RowCount": total_rows, "RowsWithGaps": gap_rows, "GapRatePct": round(gap_rows / total_rows * 100.0 if total_rows else 0.0, 2), "FieldChecks": sum(int(r["Total"]) for r in module_fields), "SourceObjectsExpected": sum(len(spec["sources"]) for spec in AUDIT_SOURCES if spec["module"] == module_key), "SourceObjectsFound": sum(1 for r in module_tables if bool(r["TableExists"])), } ) overall_stats = tier_stats(field_rows) overall_tier_rows = [] for stat in overall_stats: overall_tier_rows.append( { "Scope": "Overall", "Module": None, "ModuleName": "Overall", "Tier": stat["Tier"], "TierWeight": stat["TierWeight"], "Present": stat["Present"], "Total": stat["Total"], "Missing": stat["Missing"], "PctFilled": stat["PctFilled"], "WeightedContributionPct": round(stat["WeightedContributionRaw"] * 100.0, 2), } ) tier_output_rows = overall_tier_rows + tier_output_rows scored_modules = [r for r in module_score_rows if r["Score"] is not None] overall_score = ( sum(float(r["Score"]) for r in scored_modules) / len(scored_modules) if scored_modules else None ) total_rows = sum(int(r["RowCount"]) for r in table_rows) gap_rows = sum(int(r["RowsWithGaps"]) for r in table_rows) schemas_found = sorted({r["Schema"] for r in inventory_rows}) source_objects_found = sorted({r["SourceObject"] for r in table_rows if r.get("SourceObject")}) summary_rows = [ { "ConnectionStatus": connection_status, "ReadinessScore": round(overall_score, 1) if overall_score is not None else None, "ReadinessStatus": readiness_status(overall_score), "TotalRowsAudited": total_rows, "RowsWithGaps": gap_rows, "CleanRows": max(total_rows - gap_rows, 0), "GapRatePct": round(gap_rows / total_rows * 100.0 if total_rows else 0.0, 2), "ModulesScored": len(scored_modules), "SourceObjectsFound": len(source_objects_found), "DW2ObjectsFound": sum(1 for name in source_objects_found if name.lower().startswith("dw2.")), "InsightsObjectsFound": sum(1 for name in source_objects_found if name.lower().startswith("insights.")), "SchemasVisible": ", ".join(schemas_found), "VisibleSchemaColumns": len(inventory_rows), "MissingColumnCount": len(missing_column_rows), "ExceptionRows": len(exception_rows), "AuditErrorCount": len(errors), "AuditWindowDays": TRANSACTION_LOOKBACK_DAYS, "GeneratedAtUTC": generated_at, "Server": SERVER, "Database": database_name, "Login": login_name, "Connector": driver, } ] # Explicit columns keep Power BI Navigator stable even when a table has no rows. return { "AuditSummary": pd.DataFrame(summary_rows), "ModuleScores": pd.DataFrame(module_score_rows), "TierStats": pd.DataFrame( tier_output_rows, columns=["Scope", "Module", "ModuleName", "Tier", "TierWeight", "Present", "Total", "Missing", "PctFilled", "WeightedContributionPct"], ), "FieldStats": pd.DataFrame( field_rows, columns=["Module", "ModuleName", "LogicalTable", "SourceSchema", "SourceTable", "SourceObject", "SourceVersion", "Field", "ActualColumn", "Tier", "TierWeight", "Present", "Total", "Missing", "PctFilled", "Status", "ColumnExists", "ZeroIsMissing", "RecommendedAction"], ), "TableStats": pd.DataFrame( table_rows, columns=["Module", "ModuleName", "LogicalTable", "SourceSchema", "SourceTable", "SourceObject", "SourceVersion", "TableExists", "RowCount", "RowsWithGaps", "GapRatePct", "Score", "Status", "WorstTier", "FieldsExpected", "FieldsFound", "FilterApplied"], ), "Exceptions": pd.DataFrame( exception_rows, columns=["Module", "ModuleName", "LogicalTable", "SourceSchema", "SourceTable", "SourceObject", "RowKey", "RowId", "TopTier", "MissingField", "MissingTier", "CurrentValue", "Issue", "RecommendedAction"], ), "MissingColumns": pd.DataFrame( missing_column_rows, columns=["Module", "ModuleName", "LogicalTable", "SourceSchema", "SourceTable", "SourceObject", "ExpectedField", "CandidateColumns", "Tier", "RecommendedAction"], ), "SchemaInventory": pd.DataFrame( inventory_rows, columns=["Schema", "Table", "FullName", "Column", "DataType", "IsNullable", "OrdinalPosition"], ), "AuditDefinitions": pd.DataFrame(definition_rows), "AuditErrors": pd.DataFrame( errors, columns=["Scope", "LogicalTable", "SourceObject", "Error", "Detail"], ), } # ============================================================================ # JSON EXPORT / COMMAND-LINE ENTRY POINT # ============================================================================ EXPORT_TABLES = [ "AuditSummary", "ModuleScores", "TierStats", "FieldStats", "TableStats", "Exceptions", "MissingColumns", "SchemaInventory", "AuditDefinitions", "AuditErrors", ] def json_value(value): if value is None: return None if isinstance(value, (datetime, date)): return value.isoformat() if isinstance(value, decimal.Decimal): return float(value) if isinstance(value, float) and math.isnan(value): return None try: if pd.isna(value): return None except Exception: pass if hasattr(value, "item"): try: return value.item() except Exception: pass return value def frame_records(frame): records = [] for row in frame.to_dict(orient="records"): records.append({key: json_value(value) for key, value in row.items()}) return records def build_export_package(outputs, client_code, client_name): payload = { "reportFormat": "soza-centralreach-audit", "schemaVersion": "2.1.0", "generatedAtUTC": datetime.now(timezone.utc).isoformat(), "client": { "clientCode": client_code, "clientName": client_name, "auditName": "CentralReach Reporting Readiness Audit", }, "source": { "database": DATABASE, "schemas": SCHEMAS, "auditWindowDays": TRANSACTION_LOOKBACK_DAYS, "connector": "pymssql", }, } for name in EXPORT_TABLES: payload[name] = frame_records(outputs[name]) # Remove sensitive connection metadata from the client-facing package. for row in payload.get("AuditSummary", []): row.pop("Server", None) row.pop("Login", None) row.pop("Connector", None) return payload def parse_args(): parser = argparse.ArgumentParser( description="Run the CentralReach DW2/Insights readiness audit locally and export JSON." ) parser.add_argument("--client-code", required=True, help="Short client code, for example AFL.") parser.add_argument("--client-name", required=True, help="Client display name.") parser.add_argument("--output", default="centralreach-audit.json", help="Output JSON path.") return parser.parse_args() def main(): args = parse_args() outputs = run_audit() package = build_export_package(outputs, args.client_code, args.client_name) with open(args.output, "w", encoding="utf-8") as handle: json.dump(package, handle, indent=2, ensure_ascii=False) summary = package.get("AuditSummary", [{}])[0] print("Audit package written:", os.path.abspath(args.output)) print("Connection status:", summary.get("ConnectionStatus")) print("Readiness score:", summary.get("ReadinessScore")) print("Rows audited:", summary.get("TotalRowsAudited")) print("Audit errors:", summary.get("AuditErrorCount")) if __name__ == "__main__": main()