ok

app/services/reconciliation_service.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import io
 import os
+import re
 import time
 import unicodedata
 from collections import Counter
@@ -18,8 +19,6 @@ from app.core.logger import get_logger
 _logger = get_logger(__name__)
 _settings = get_settings()
 
-MAX_CONCURRENT_DOWNLOADS = 10
-MAX_CONCURRENT_JOBS = 5
 DOWNLOAD_TIMEOUT = 60
 DOWNLOAD_MAX_RETRIES = 3
 DOWNLOAD_BACKOFF_FACTOR = 0.5
@@ -45,13 +44,17 @@ def extract_extension(url: str) -> str:
     return os.path.splitext(path)[1].lstrip(".").lower()
 
 
+def _normalize_str(x: str) -> str:
+    if not pd.notna(x) or x == 'nan':
+        return x
+    return re.sub(r'\s+', ' ', unicodedata.normalize('NFKC', x))
+
+
 def normalize_dataframe(df: pd.DataFrame) -> pd.DataFrame:
     for col in df.columns:
         if pd.api.types.is_string_dtype(df[col]):
             s = df[col].astype(str).str.strip()
-            s = s.map(lambda x: unicodedata.normalize('NFKC', x) if x != 'nan' else x)
-            s = s.str.replace(r'\s+', ' ', regex=True)
-            df[col] = s.replace({'nan': pd.NA, 'None': pd.NA, '': pd.NA, 'null': pd.NA})
+            df[col] = s.apply(_normalize_str).replace({'nan': pd.NA, 'None': pd.NA, '': pd.NA, 'null': pd.NA})
         elif pd.api.types.is_numeric_dtype(df[col]):
             df[col] = df[col].replace({pd.NA: None})
     return df
@@ -65,10 +68,7 @@ async def download_file_with_retry(session: aiohttp.ClientSession, url: str, cor
                 return await response.read()
         except aiohttp.ClientError as e:
             wait_time = DOWNLOAD_BACKOFF_FACTOR * (2 ** attempt)
-            _logger.warning(
-                f"Download attempt {attempt+1} failed for {url}. Retrying in {wait_time}s. Error: {e}",
-                extra={"correlation_id": correlation_id}
-            )
+            _logger.warning(f"Download attempt {attempt+1} failed for {url}. Retrying in {wait_time}s. Error: {e}", extra={"correlation_id": correlation_id})
             await asyncio.sleep(wait_time)
     raise DownloadError(f"Failed to download {url} after {DOWNLOAD_MAX_RETRIES} retries.")
 
@@ -85,7 +85,8 @@ def read_to_dataframe(data: bytes, ext: str, correlation_id: str) -> pd.DataFram
             return pd.read_excel(io.BytesIO(data), engine="xlrd")
         elif ext == "parquet":
             return pd.read_parquet(io.BytesIO(data))
-        raise UnsupportedFormatError(f"File format '{ext}' is not supported.")
+        else:
+            raise UnsupportedFormatError(f"File format '{ext}' is not supported.")
     except Exception as e:
         _logger.error(f"Failed to parse {ext} file: {str(e)}", extra={"correlation_id": correlation_id})
         raise ReconciliationError(f"Corrupted or unparseable file: {str(e)}")
@@ -97,7 +98,6 @@ def compare_schemas(df_src: pd.DataFrame, df_dst: pd.DataFrame) -> Dict[str, Any
     missing_in_src = list(dst_cols - src_cols)
     missing_in_dst = list(src_cols - dst_cols)
     type_mismatches = []
-    
     common_cols = src_cols & dst_cols
     for col in common_cols:
         if df_src[col].dtype != df_dst[col].dtype:
@@ -106,7 +106,6 @@ def compare_schemas(df_src: pd.DataFrame, df_dst: pd.DataFrame) -> Dict[str, Any
                 "source_type": str(df_src[col].dtype),
                 "destination_type": str(df_dst[col].dtype)
             })
-            
     fully_match = not missing_in_src and not missing_in_dst and not type_mismatches
     return {
         "fully_match": fully_match,
@@ -121,22 +120,19 @@ def compare_schemas(df_src: pd.DataFrame, df_dst: pd.DataFrame) -> Dict[str, Any
 def compare_rows(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols: List[str]) -> Tuple[int, int, int, int]:
     src_subset = df_src[common_cols].astype(str).agg('|'.join, axis=1)
     dst_subset = df_dst[common_cols].astype(str).agg('|'.join, axis=1)
-    
     src_counts = Counter(src_subset)
     dst_counts = Counter(dst_subset)
-    
     identical = sum(min(count, dst_counts.get(row, 0)) for row, count in src_counts.items())
     missing = len(df_src) - identical
     extra = len(df_dst) - identical
-    
     return identical, 0, missing, extra
 
 
 def reconcile_columns(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols: List[str]) -> List[Dict[str, Any]]:
     results = []
     for col in common_cols:
-        src_nn = int(df_src[col].count())
-        dst_nn = int(df_dst[col].count())
+        src_nn = int(df_src[col].notna().sum())
+        dst_nn = int(df_dst[col].notna().sum())
         status = "fully_match" if src_nn == dst_nn else "partial_match"
         results.append({
             "column_name": col,
@@ -151,51 +147,51 @@ def reconcile_columns(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols: L
 
 
 def reconcile_numeric_columns(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols: List[str], numeric_keywords: Optional[Set[str]] = None) -> List[Dict[str, Any]]:
-    numeric_keywords = numeric_keywords or DEFAULT_NUMERIC_KEYWORDS
+    if numeric_keywords is None:
+        numeric_keywords = DEFAULT_NUMERIC_KEYWORDS
     results = []
-    
     for col in common_cols:
-        if any(kw in col.lower() for kw in numeric_keywords):
-            if pd.api.types.is_numeric_dtype(df_src[col]) and pd.api.types.is_numeric_dtype(df_dst[col]):
-                src_sum = df_src[col].sum()
-                dst_sum = df_dst[col].sum()
-                diff = src_sum - dst_sum
-                abs_diff = abs(diff)
-                pct_diff = (abs_diff / src_sum * 100) if src_sum != 0 else 0.0
-                
-                src_nn = int(df_src[col].count())
-                dst_nn = int(df_dst[col].count())
-                status = "fully_match" if abs_diff < 1e-9 else "partial_match"
-                
-                results.append({
-                    "column_name": col,
-                    "status": status,
-                    "source_non_null": src_nn,
-                    "destination_non_null": dst_nn,
-                    "matching_values": src_nn,
-                    "mismatching_values": abs(src_nn - dst_nn),
-                    "difference_count": abs(src_nn - dst_nn),
-                    "source_total": round(float(src_sum), 2),
-                    "destination_total": round(float(dst_sum), 2),
-                    "total_difference": round(float(diff), 2),
-                    "absolute_difference": round(float(abs_diff), 2),
-                    "percentage_difference": round(float(pct_diff), 5),
-                    "source_min": round(float(df_src[col].min()), 2),
-                    "source_max": round(float(df_src[col].max()), 2),
-                    "source_avg": round(float(df_src[col].mean()), 2),
-                    "dest_min": round(float(df_dst[col].min()), 2),
-                    "dest_max": round(float(df_dst[col].max()), 2),
-                    "dest_avg": round(float(df_dst[col].mean()), 2)
-                })
+        is_numeric_name = any(kw in col.lower() for kw in numeric_keywords)
+        is_numeric_dtype = pd.api.types.is_numeric_dtype(df_src[col]) and pd.api.types.is_numeric_dtype(df_dst[col])
+        if is_numeric_name and is_numeric_dtype:
+            src_sum = df_src[col].sum()
+            dst_sum = df_dst[col].sum()
+            diff = src_sum - dst_sum
+            abs_diff = abs(diff)
+            pct_diff = (abs_diff / src_sum * 100) if src_sum != 0 else 0.0
+            src_nn = int(df_src[col].notna().sum())
+            dst_nn = int(df_dst[col].notna().sum())
+            status = "fully_match" if abs_diff < 1e-9 else "partial_match"
+            results.append({
+                "column_name": col,
+                "status": status,
+                "source_non_null": src_nn,
+                "destination_non_null": dst_nn,
+                "matching_values": src_nn,
+                "mismatching_values": abs(src_nn - dst_nn),
+                "difference_count": abs(src_nn - dst_nn),
+                "source_total": round(float(src_sum), 2),
+                "destination_total": round(float(dst_sum), 2),
+                "total_difference": round(float(diff), 2),
+                "absolute_difference": round(float(abs_diff), 2),
+                "percentage_difference": round(float(pct_diff), 5),
+                "source_min": round(float(df_src[col].min()), 2),
+                "source_max": round(float(df_src[col].max()), 2),
+                "source_avg": round(float(df_src[col].mean()), 2),
+                "dest_min": round(float(df_dst[col].min()), 2),
+                "dest_max": round(float(df_dst[col].max()), 2),
+                "dest_avg": round(float(df_dst[col].mean()), 2)
+            })
     return results
 
 
 def reconcile_date_columns(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols: List[str]) -> List[Dict[str, Any]]:
     results = []
     for col in common_cols:
-        if pd.api.types.is_datetime64_any_dtype(df_src[col]) and pd.api.types.is_datetime64_any_dtype(df_dst[col]):
-            src_nn = int(df_src[col].count())
-            dst_nn = int(df_dst[col].count())
+        is_date = pd.api.types.is_datetime64_any_dtype(df_src[col]) and pd.api.types.is_datetime64_any_dtype(df_dst[col])
+        if is_date:
+            src_nn = int(df_src[col].notna().sum())
+            dst_nn = int(df_dst[col].notna().sum())
             results.append({
                 "column_name": col,
                 "status": "fully_match" if df_src[col].equals(df_dst[col]) else "partial_match",
@@ -238,7 +234,7 @@ def analyze_missing_data(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols
     return results
 
 
-def _build_failed_pair(job_id: str, src_url: str, dst_url: str, started_at: str, status: str, errors: List[str]) -> Dict[str, Any]:
+def _build_failed_pair_result(job_id: str, src_url: str, dst_url: str, started_at: str, status: str, errors: List[str], schema: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
     return {
         "job_id": job_id,
         "source_file": src_url,
@@ -248,13 +244,11 @@ def _build_failed_pair(job_id: str, src_url: str, dst_url: str, started_at: str,
         "processing_time_ms": 0,
         "status": status,
         "summary": {"overall_status": status, "overall_match_percentage": 0.0},
-        "schema": {"fully_match": False, "source_columns": 0, "destination_columns": 0},
-        "columns": [], 
-        "numeric_columns": [], 
-        "date_columns": [],
-        "duplicates": {}, 
-        "missing_values": [],
-        "errors": errors
+        "schema": schema or {"fully_match": False, "source_columns": 0, "destination_columns": 0},
+        "columns": [], "numeric_columns": [], "date_columns": [],
+        "duplicates": {}, "missing_values": [],
+        "errors": errors,
+        "warnings": []
     }
 
 
@@ -274,15 +268,15 @@ async def process_pair(src_url: str, dst_url: str, src_data: bytes, dst_data: by
         
         df_src = normalize_dataframe(df_src.copy())
         df_dst = normalize_dataframe(df_dst.copy())
-        df_src = df_src.infer_objects()
-        df_dst = df_dst.infer_objects()
+        df_src.infer_objects()
+        df_dst.infer_objects()
         
         schema_report = compare_schemas(df_src, df_dst)
         if not schema_report["fully_match"]:
             status = "SCHEMA_MISMATCH"
             warnings.append("Schema mismatch detected.")
         
-        common_cols = list(df_src.columns.intersection(df_dst.columns))
+        common_cols = df_src.columns.intersection(df_dst.columns).tolist()
         
         identical_rows, _, missing_rows, extra_rows = compare_rows(df_src, df_dst, common_cols)
         
@@ -326,13 +320,16 @@ async def process_pair(src_url: str, dst_url: str, src_data: bytes, dst_data: by
         summary = {"overall_status": "FAILED", "overall_match_percentage": 0.0}
         schema_report, col_reports, num_reports, date_reports, dup_report, missing_report = {}, [], [], [], {}, []
     
+    completed_at = datetime.utcnow().isoformat()
+    processing_time_ms = round((time.time() - start_time) * 1000, 2)
+    
     return {
         "job_id": job_id,
         "source_file": src_url,
         "destination_file": dst_url,
         "started_at": started_at,
-        "completed_at": datetime.utcnow().isoformat(),
-        "processing_time_ms": round((time.time() - start_time) * 1000, 2),
+        "completed_at": completed_at,
+        "processing_time_ms": processing_time_ms,
         "status": status,
         "summary": summary,
         "schema": schema_report,
@@ -349,16 +346,23 @@ async def process_pair(src_url: str, dst_url: str, src_data: bytes, dst_data: by
 async def reconcile_pair(pair: Dict[str, str], job_id: str, numeric_keywords: Optional[Set[str]] = None) -> Dict[str, Any]:
     src_url = pair["source"]
     dst_url = pair["destination"]
+    
     started_at = datetime.utcnow().isoformat()
     
     src_ext = extract_extension(src_url)
     dst_ext = extract_extension(dst_url)
     
     if src_ext not in SUPPORTED_EXTENSIONS or dst_ext not in SUPPORTED_EXTENSIONS:
-        return _build_failed_pair(job_id, src_url, dst_url, started_at, "FAILED", [f"Unsupported format. Source: {src_ext}, Dest: {dst_ext}"])
+        return _build_failed_pair_result(
+            job_id, src_url, dst_url, started_at, "FAILED",
+            [f"Unsupported format. Source: {src_ext}, Dest: {dst_ext}"]
+        )
     
     if src_ext != dst_ext:
-        return _build_failed_pair(job_id, src_url, dst_url, started_at, "FILE_TYPE_MISMATCH", [f"File type mismatch. Source: {src_ext}, Dest: {dst_ext}"])
+        return _build_failed_pair_result(
+            job_id, src_url, dst_url, started_at, "FILE_TYPE_MISMATCH",
+            [f"File type mismatch. Source: {src_ext}, Dest: {dst_ext}"]
+        )
     
     async with aiohttp.ClientSession() as session:
         try:
@@ -368,9 +372,13 @@ async def reconcile_pair(pair: Dict[str, str], job_id: str, numeric_keywords: Op
             )
         except DownloadError as e:
             _logger.error(f"Download failed for job {job_id}: {str(e)}", extra={"correlation_id": job_id})
-            return _build_failed_pair(job_id, src_url, dst_url, started_at, "FAILED", [str(e)])
+            return _build_failed_pair_result(
+                job_id, src_url, dst_url, started_at, "FAILED", [str(e)]
+            )
     
     if len(src_data) > _MAX_FILE_SIZE or len(dst_data) > _MAX_FILE_SIZE:
-        return _build_failed_pair(job_id, src_url, dst_url, started_at, "FAILED", ["File size exceeds maximum allowed limit"])
+        return _build_failed_pair_result(
+            job_id, src_url, dst_url, started_at, "FAILED", ["File size exceeds maximum allowed limit"]
+        )
     
     return await process_pair(src_url, dst_url, src_data, dst_data, src_ext, job_id, numeric_keywords)