molla

app/api/v1/reconcile.py

@@ -8,7 +8,7 @@ import uuid
 from typing import Any, Dict, List, Optional
 
 import aiohttp
-from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
 from pydantic import BaseModel
 
 from app.api.deps import require_auth
@@ -18,14 +18,15 @@ from app.services.reconciliation_service import (
     SUPPORTED_EXTENSIONS,
     analyze_duplicates,
     analyze_missing_data,
+    apply_column_mapping,
     compare_rows,
     compare_schemas,
     download_file_with_retry,
     normalize_dataframe,
+    read_to_dataframe,
     reconcile_columns,
     reconcile_date_columns,
     reconcile_numeric_columns,
-    read_to_dataframe,
 )
 
 router = APIRouter()
@@ -36,14 +37,19 @@ _MAX_PAIRS = 10
 semaphore_jobs = asyncio.Semaphore(5)
 
 
+class ColumnMapping(BaseModel):
+    source: str
+    destination: str
+
+
 class ReconciliationPair(BaseModel):
     source: str
     destination: str
+    column_mapping: Optional[List[ColumnMapping]] = None
 
 
 class ReconciliationUrlRequest(BaseModel):
     pairs: List[ReconciliationPair]
-    numeric_columns: Optional[List[str]] = None
 
 
 class ReconciliationPairResult(BaseModel):
@@ -85,12 +91,6 @@ def _failed_result(pair_index: int, error: str) -> Dict[str, Any]:
     }
 
 
-def _parse_numeric_keywords(numeric_columns: Optional[List[str]]) -> Optional[set]:
-    if not numeric_columns:
-        return None
-    return set(c.lower() for c in numeric_columns)
-
-
 def _validate_extensions(src_ext: str, dst_ext: str, idx: int) -> None:
     if src_ext not in SUPPORTED_EXTENSIONS:
         raise HTTPException(status_code=400, detail={"success": False, "message": f"Pair {idx}: Unsupported source format '{src_ext}'."})
@@ -112,12 +112,30 @@ def _build_response(results: List[Dict[str, Any]], start_time: float) -> Reconci
     )
 
 
+def _parse_column_mapping_str(value: Optional[str]) -> Optional[Dict[str, str]]:
+    if not value:
+        return None
+    try:
+        parsed = json.loads(value)
+        if isinstance(parsed, list):
+            return {item["destination"]: item["source"] for item in parsed if "source" in item and "destination" in item}
+    except (json.JSONDecodeError, KeyError, TypeError):
+        pass
+    return None
+
+
+def _parse_column_mapping_model(mappings: Optional[List[ColumnMapping]]) -> Optional[Dict[str, str]]:
+    if not mappings:
+        return None
+    return {m.destination: m.source for m in mappings}
+
+
 async def _process_file_pair(
     source_file: bytes,
     dest_file: bytes,
     source_ext: str,
     dest_ext: str,
-    numeric_keywords: Optional[set] = None,
+    column_mapping: Optional[Dict[str, str]] = None,
     pair_index: int = 0,
 ) -> Dict[str, Any]:
     if source_ext != dest_ext:
@@ -132,8 +150,11 @@ async def _process_file_pair(
 
         df_src = normalize_dataframe(df_src.copy())
         df_dst = normalize_dataframe(df_dst.copy())
-        df_src.infer_objects()
-        df_dst.infer_objects()
+        df_src = df_src.infer_objects()
+        df_dst = df_dst.infer_objects()
+
+        if column_mapping:
+            df_dst = apply_column_mapping(df_dst, column_mapping)
 
         schema_report = compare_schemas(df_src, df_dst)
         status = "MATCH" if schema_report["fully_match"] else "SCHEMA_MISMATCH"
@@ -145,7 +166,7 @@ async def _process_file_pair(
             status = "PARTIAL_MATCH"
 
         col_reports = reconcile_columns(df_src, df_dst, common_cols)
-        num_reports = reconcile_numeric_columns(df_src, df_dst, common_cols, numeric_keywords)
+        num_reports = reconcile_numeric_columns(df_src, df_dst, common_cols)
         date_reports = reconcile_date_columns(df_src, df_dst, common_cols)
         dup_report = analyze_duplicates(df_src, df_dst)
         missing_report = analyze_missing_data(df_src, df_dst, common_cols)
@@ -172,47 +193,55 @@ async def _process_file_pair(
     summary="Reconcile uploaded file pairs (up to 10 pairs)",
 )
 async def reconcile_files(
-    source_1: UploadFile = File(None, description="Source file for pair 1"),
-    destination_1: UploadFile = File(None, description="Destination file for pair 1"),
-    source_2: UploadFile = File(None, description="Source file for pair 2"),
-    destination_2: UploadFile = File(None, description="Destination file for pair 2"),
-    source_3: UploadFile = File(None, description="Source file for pair 3"),
-    destination_3: UploadFile = File(None, description="Destination file for pair 3"),
-    source_4: UploadFile = File(None, description="Source file for pair 4"),
-    destination_4: UploadFile = File(None, description="Destination file for pair 4"),
-    source_5: UploadFile = File(None, description="Source file for pair 5"),
-    destination_5: UploadFile = File(None, description="Destination file for pair 5"),
-    source_6: UploadFile = File(None, description="Source file for pair 6"),
-    destination_6: UploadFile = File(None, description="Destination file for pair 6"),
-    source_7: UploadFile = File(None, description="Source file for pair 7"),
-    destination_7: UploadFile = File(None, description="Destination file for pair 7"),
-    source_8: UploadFile = File(None, description="Source file for pair 8"),
-    destination_8: UploadFile = File(None, description="Destination file for pair 8"),
-    source_9: UploadFile = File(None, description="Source file for pair 9"),
-    destination_9: UploadFile = File(None, description="Destination file for pair 9"),
-    source_10: UploadFile = File(None, description="Source file for pair 10"),
-    destination_10: UploadFile = File(None, description="Destination file for pair 10"),
-    numeric_columns: Optional[str] = None,
+    source_1: UploadFile = File(None),
+    destination_1: UploadFile = File(None),
+    column_mapping_1: Optional[str] = Form(None),
+    source_2: UploadFile = File(None),
+    destination_2: UploadFile = File(None),
+    column_mapping_2: Optional[str] = Form(None),
+    source_3: UploadFile = File(None),
+    destination_3: UploadFile = File(None),
+    column_mapping_3: Optional[str] = Form(None),
+    source_4: UploadFile = File(None),
+    destination_4: UploadFile = File(None),
+    column_mapping_4: Optional[str] = Form(None),
+    source_5: UploadFile = File(None),
+    destination_5: UploadFile = File(None),
+    column_mapping_5: Optional[str] = Form(None),
+    source_6: UploadFile = File(None),
+    destination_6: UploadFile = File(None),
+    column_mapping_6: Optional[str] = Form(None),
+    source_7: UploadFile = File(None),
+    destination_7: UploadFile = File(None),
+    column_mapping_7: Optional[str] = Form(None),
+    source_8: UploadFile = File(None),
+    destination_8: UploadFile = File(None),
+    column_mapping_8: Optional[str] = Form(None),
+    source_9: UploadFile = File(None),
+    destination_9: UploadFile = File(None),
+    column_mapping_9: Optional[str] = Form(None),
+    source_10: UploadFile = File(None),
+    destination_10: UploadFile = File(None),
+    column_mapping_10: Optional[str] = Form(None),
     token: str = Depends(require_auth),
 ) -> ReconciliationResponse:
     start_time = time.time()
 
-    numeric_keywords = None
-    if numeric_columns:
-        try:
-            numeric_keywords = _parse_numeric_keywords(json.loads(numeric_columns))
-        except json.JSONDecodeError:
-            pass
-
     files = [
-        (source_1, destination_1), (source_2, destination_2), (source_3, destination_3),
-        (source_4, destination_4), (source_5, destination_5), (source_6, destination_6),
-        (source_7, destination_7), (source_8, destination_8), (source_9, destination_9),
-        (source_10, destination_10),
+        (source_1, destination_1, column_mapping_1),
+        (source_2, destination_2, column_mapping_2),
+        (source_3, destination_3, column_mapping_3),
+        (source_4, destination_4, column_mapping_4),
+        (source_5, destination_5, column_mapping_5),
+        (source_6, destination_6, column_mapping_6),
+        (source_7, destination_7, column_mapping_7),
+        (source_8, destination_8, column_mapping_8),
+        (source_9, destination_9, column_mapping_9),
+        (source_10, destination_10, column_mapping_10),
     ]
 
     pairs: List[tuple] = []
-    for idx, (src, dst) in enumerate(files, 1):
+    for idx, (src, dst, cm) in enumerate(files, 1):
         if src is None and dst is None:
             continue
         if src is None or dst is None:
@@ -220,18 +249,19 @@ async def reconcile_files(
         src_ext = _extract_extension(src.filename or "")
         dst_ext = _extract_extension(dst.filename or "")
         _validate_extensions(src_ext, dst_ext, idx)
-        pairs.append((src, dst, idx, src_ext, dst_ext))
+        pairs.append((src, dst, cm, idx, src_ext, dst_ext))
 
     if not pairs:
         raise HTTPException(status_code=400, detail={"success": False, "message": "At least one file pair is required."})
 
-    async def process_file_pair(src: UploadFile, dst: UploadFile, idx: int, src_ext: str, dst_ext: str) -> Dict[str, Any]:
+    async def process_file_pair(src: UploadFile, dst: UploadFile, cm_str: Optional[str], idx: int, src_ext: str, dst_ext: str) -> Dict[str, Any]:
         async with semaphore_jobs:
             src_data = await src.read()
             dst_data = await dst.read()
             if len(src_data) > _MAX_UPLOAD_BYTES or len(dst_data) > _MAX_UPLOAD_BYTES:
                 return _failed_result(idx, "File size exceeds maximum allowed limit")
-            return await _process_file_pair(src_data, dst_data, src_ext, dst_ext, numeric_keywords, idx)
+            mapping = _parse_column_mapping_str(cm_str)
+            return await _process_file_pair(src_data, dst_data, src_ext, dst_ext, mapping, idx)
 
     results = await asyncio.gather(*[process_file_pair(*p) for p in pairs])
     return _build_response(results, start_time)
@@ -256,27 +286,26 @@ async def reconcile_urls(
     for idx, pair in enumerate(body.pairs, 1):
         _validate_extensions(_extract_extension(pair.source), _extract_extension(pair.destination), idx)
 
-    numeric_keywords = _parse_numeric_keywords(body.numeric_columns)
+    async with aiohttp.ClientSession() as session:
+        async def process_url_pair(pair: ReconciliationPair, idx: int) -> Dict[str, Any]:
+            async with semaphore_jobs:
+                src_ext = _extract_extension(pair.source)
+                dst_ext = _extract_extension(pair.destination)
 
-    async def process_url_pair(pair: ReconciliationPair, idx: int) -> Dict[str, Any]:
-        async with semaphore_jobs:
-            job_id = str(uuid.uuid4())
-            src_ext = _extract_extension(pair.source)
-            dst_ext = _extract_extension(pair.destination)
-
-            async with aiohttp.ClientSession() as session:
                 try:
                     src_data, dst_data = await asyncio.gather(
-                        download_file_with_retry(session, pair.source, job_id),
-                        download_file_with_retry(session, pair.destination, job_id),
+                        download_file_with_retry(session, pair.source, str(uuid.uuid4())),
+                        download_file_with_retry(session, pair.destination, str(uuid.uuid4())),
                     )
                 except Exception as e:
                     return _failed_result(idx, str(e))
 
-            if len(src_data) > _MAX_UPLOAD_BYTES or len(dst_data) > _MAX_UPLOAD_BYTES:
-                return _failed_result(idx, "File size exceeds maximum allowed limit")
+                if len(src_data) > _MAX_UPLOAD_BYTES or len(dst_data) > _MAX_UPLOAD_BYTES:
+                    return _failed_result(idx, "File size exceeds maximum allowed limit")
+
+                mapping = _parse_column_mapping_model(pair.column_mapping)
+                return await _process_file_pair(src_data, dst_data, src_ext, dst_ext, mapping, idx)
 
-            return await _process_file_pair(src_data, dst_data, src_ext, dst_ext, numeric_keywords, idx)
+        results = await asyncio.gather(*[process_url_pair(pair, idx) for idx, pair in enumerate(body.pairs, 1)])
 
-    results = await asyncio.gather(*[process_url_pair(pair, idx) for idx, pair in enumerate(body.pairs, 1)])
     return _build_response(results, start_time)

app/services/reconciliation_service.py

@@ -60,6 +60,13 @@ def normalize_dataframe(df: pd.DataFrame) -> pd.DataFrame:
     return df
 
 
+def apply_column_mapping(df: pd.DataFrame, mapping: Dict[str, str]) -> pd.DataFrame:
+    valid_mapping = {old: new for old, new in mapping.items() if old in df.columns}
+    if valid_mapping:
+        df = df.rename(columns=valid_mapping)
+    return df
+
+
 async def download_file_with_retry(session: aiohttp.ClientSession, url: str, correlation_id: str) -> bytes:
     for attempt in range(DOWNLOAD_MAX_RETRIES):
         try:
@@ -68,7 +75,10 @@ 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.")
 
@@ -252,53 +262,65 @@ def _build_failed_pair_result(job_id: str, src_url: str, dst_url: str, started_a
     }
 
 
-async def process_pair(src_url: str, dst_url: str, src_data: bytes, dst_data: bytes, ext: str, job_id: str, numeric_keywords: Optional[Set[str]] = None) -> Dict[str, Any]:
+async def process_pair(
+    src_url: str,
+    dst_url: str,
+    src_data: bytes,
+    dst_data: bytes,
+    ext: str,
+    job_id: str,
+    column_mapping: Optional[Dict[str, str]] = None,
+    numeric_keywords: Optional[Set[str]] = None,
+) -> Dict[str, Any]:
     start_time = time.time()
     started_at = datetime.utcnow().isoformat()
     errors = []
     warnings = []
     status = "MATCH"
-    
+
     try:
         df_src = read_to_dataframe(src_data, ext, job_id)
         df_dst = read_to_dataframe(dst_data, ext, job_id)
-        
+
         if df_src.empty or df_dst.empty:
             raise EmptyDatasetError("One or both datasets are empty.")
-        
+
         df_src = normalize_dataframe(df_src.copy())
         df_dst = normalize_dataframe(df_dst.copy())
         df_src.infer_objects()
         df_dst.infer_objects()
-        
+
+        if column_mapping:
+            df_dst = apply_column_mapping(df_dst, column_mapping)
+
         schema_report = compare_schemas(df_src, df_dst)
         if not schema_report["fully_match"]:
             status = "SCHEMA_MISMATCH"
             warnings.append("Schema mismatch detected.")
-        
+
         common_cols = df_src.columns.intersection(df_dst.columns).tolist()
-        
+
         identical_rows, _, missing_rows, extra_rows = compare_rows(df_src, df_dst, common_cols)
-        
+
         if missing_rows > 0 or extra_rows > 0:
             status = "PARTIAL_MATCH" if status == "MATCH" else status
-        
+
         col_reports = reconcile_columns(df_src, df_dst, common_cols)
         num_reports = reconcile_numeric_columns(df_src, df_dst, common_cols, numeric_keywords)
         date_reports = reconcile_date_columns(df_src, df_dst, common_cols)
-        
+
         dup_report = analyze_duplicates(df_src, df_dst)
         missing_report = analyze_missing_data(df_src, df_dst, common_cols)
-        
+
         full_cols = sum(1 for c in col_reports if c["status"] == "fully_match")
         part_cols = sum(1 for c in col_reports if c["status"] == "partial_match")
-        
+
         if part_cols > 0 and status == "MATCH":
             status = "PARTIAL_MATCH"
-        
+
         total_max_rows = max(len(df_src), len(df_dst))
         match_pct = (identical_rows / total_max_rows * 100) if total_max_rows > 0 else 100.0
-        
+
         summary = {
             "total_columns": len(common_cols),
             "fully_matched_columns": full_cols,
@@ -319,17 +341,14 @@ async def process_pair(src_url: str, dst_url: str, src_data: bytes, dst_data: by
         errors.append(str(e))
         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": completed_at,
-        "processing_time_ms": processing_time_ms,
+        "completed_at": datetime.utcnow().isoformat(),
+        "processing_time_ms": round((time.time() - start_time) * 1000, 2),
         "status": status,
         "summary": summary,
         "schema": schema_report,
@@ -343,27 +362,32 @@ 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]:
+async def reconcile_pair(
+    pair: Dict[str, Any],
+    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_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_result(
             job_id, src_url, dst_url, started_at, "FILE_TYPE_MISMATCH",
             [f"File type mismatch. Source: {src_ext}, Dest: {dst_ext}"]
         )
-    
+
+    column_mapping: Optional[Dict[str, str]] = pair.get("column_mapping")
+
     async with aiohttp.ClientSession() as session:
         try:
             src_data, dst_data = await asyncio.gather(
@@ -375,10 +399,11 @@ async def reconcile_pair(pair: Dict[str, str], job_id: str, numeric_keywords: Op
             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_result(
-            job_id, src_url, dst_url, started_at, "FAILED", ["File size exceeds maximum allowed limit"]
+            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)
+
+    return await process_pair(src_url, dst_url, src_data, dst_data, src_ext, job_id, column_mapping, numeric_keywords)