feat: add csv reconciliation

app/api/v1/reconcile.py

@@ -0,0 +1,282 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import time
+import uuid
+from typing import Any, Dict, List, Optional
+
+import aiohttp
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+from pydantic import BaseModel
+
+from app.api.deps import require_auth
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.services.reconciliation_service import (
+    SUPPORTED_EXTENSIONS,
+    analyze_duplicates,
+    analyze_missing_data,
+    compare_rows,
+    compare_schemas,
+    download_file_with_retry,
+    normalize_dataframe,
+    reconcile_columns,
+    reconcile_date_columns,
+    reconcile_numeric_columns,
+    read_to_dataframe,
+)
+
+router = APIRouter()
+_logger = get_logger(__name__)
+_settings = get_settings()
+_MAX_UPLOAD_BYTES = _settings.max_upload_bytes
+_MAX_PAIRS = 10
+semaphore_jobs = asyncio.Semaphore(5)
+
+
+class ReconciliationPair(BaseModel):
+    source: str
+    destination: str
+
+
+class ReconciliationUrlRequest(BaseModel):
+    pairs: List[ReconciliationPair]
+    numeric_columns: Optional[List[str]] = None
+
+
+class ReconciliationPairResult(BaseModel):
+    pair_index: int
+    status: str
+    error: Optional[str] = None
+    schema_report: Optional[Dict[str, Any]] = None
+    columns: Optional[List[Dict[str, Any]]] = None
+    numeric_columns: Optional[List[Dict[str, Any]]] = None
+    date_columns: Optional[List[Dict[str, Any]]] = None
+    duplicates: Optional[Dict[str, int]] = None
+    missing_values: Optional[List[Dict[str, Any]]] = None
+
+
+class ReconciliationResponse(BaseModel):
+    success: bool
+    total_pairs: int
+    completed_pairs: int
+    failed_pairs: int
+    processing_time_ms: float
+    results: List[ReconciliationPairResult]
+
+
+def _extract_extension(filename: str) -> str:
+    return os.path.splitext(filename)[1].lstrip(".").lower()
+
+
+def _failed_result(pair_index: int, error: str) -> Dict[str, Any]:
+    return {
+        "pair_index": pair_index,
+        "status": "FAILED",
+        "error": error,
+        "schema_report": None,
+        "columns": None,
+        "numeric_columns": None,
+        "date_columns": None,
+        "duplicates": None,
+        "missing_values": None,
+    }
+
+
+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}'."})
+    if dst_ext not in SUPPORTED_EXTENSIONS:
+        raise HTTPException(status_code=400, detail={"success": False, "message": f"Pair {idx}: Unsupported destination format '{dst_ext}'."})
+    if src_ext != dst_ext:
+        raise HTTPException(status_code=400, detail={"success": False, "message": f"Pair {idx}: File type mismatch - source '{src_ext}' != destination '{dst_ext}'."})
+
+
+def _build_response(results: List[Dict[str, Any]], start_time: float) -> ReconciliationResponse:
+    failed = sum(1 for r in results if r["status"] == "FAILED")
+    return ReconciliationResponse(
+        success=failed == 0,
+        total_pairs=len(results),
+        completed_pairs=len(results) - failed,
+        failed_pairs=failed,
+        processing_time_ms=round((time.time() - start_time) * 1000, 2),
+        results=results,
+    )
+
+
+async def _process_file_pair(
+    source_file: bytes,
+    dest_file: bytes,
+    source_ext: str,
+    dest_ext: str,
+    numeric_keywords: Optional[set] = None,
+    pair_index: int = 0,
+) -> Dict[str, Any]:
+    if source_ext != dest_ext:
+        return _failed_result(pair_index, f"File type mismatch. Source: {source_ext}, Destination: {dest_ext}")
+
+    try:
+        df_src = read_to_dataframe(source_file, source_ext, f"pair-{pair_index}")
+        df_dst = read_to_dataframe(dest_file, dest_ext, f"pair-{pair_index}")
+
+        if df_src.empty or df_dst.empty:
+            raise ValueError("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()
+
+        schema_report = compare_schemas(df_src, df_dst)
+        status = "MATCH" if schema_report["fully_match"] else "SCHEMA_MISMATCH"
+
+        common_cols = df_src.columns.intersection(df_dst.columns).tolist()
+
+        _, _, missing_rows, extra_rows = compare_rows(df_src, df_dst, common_cols)
+        if (missing_rows > 0 or extra_rows > 0) and status == "MATCH":
+            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)
+        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)
+
+        return {
+            "pair_index": pair_index,
+            "status": status,
+            "error": None,
+            "schema_report": schema_report,
+            "columns": col_reports,
+            "numeric_columns": num_reports,
+            "date_columns": date_reports,
+            "duplicates": dup_report,
+            "missing_values": missing_report,
+        }
+    except Exception as e:
+        _logger.exception(f"Pair {pair_index} failed during processing.")
+        return _failed_result(pair_index, str(e))
+
+
+@router.post(
+    "/reconcile/files",
+    response_model=ReconciliationResponse,
+    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,
+    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),
+    ]
+
+    pairs: List[tuple] = []
+    for idx, (src, dst) in enumerate(files, 1):
+        if src is None and dst is None:
+            continue
+        if src is None or dst is None:
+            raise HTTPException(status_code=400, detail={"success": False, "message": f"Pair {idx}: Both source and destination files required."})
+        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))
+
+    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 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)
+
+    results = await asyncio.gather(*[process_file_pair(*p) for p in pairs])
+    return _build_response(results, start_time)
+
+
+@router.post(
+    "/reconcile/urls",
+    response_model=ReconciliationResponse,
+    summary="Reconcile file URL pairs (up to 10 pairs)",
+)
+async def reconcile_urls(
+    body: ReconciliationUrlRequest,
+    token: str = Depends(require_auth),
+) -> ReconciliationResponse:
+    start_time = time.time()
+
+    if len(body.pairs) > _MAX_PAIRS:
+        raise HTTPException(status_code=400, detail={"success": False, "message": f"Maximum {_MAX_PAIRS} pairs per request."})
+    if len(body.pairs) < 1:
+        raise HTTPException(status_code=400, detail={"success": False, "message": "At least 1 pair is required."})
+
+    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 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),
+                    )
+                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")
+
+            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)])
+    return _build_response(results, start_time)

app/api/v1/router.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from fastapi import APIRouter
 
-from app.api.v1 import batch, code_executor, convert, database, embeddings, system
+from app.api.v1 import batch, code_executor, convert, database, embeddings, reconcile, system
 from app.api.verify import router as verify_router
 
 api_v1_router = APIRouter()
@@ -13,3 +13,4 @@ api_v1_router.include_router(database.router, tags=["Database"])
 api_v1_router.include_router(embeddings.router, tags=["Embeddings"])
 api_v1_router.include_router(code_executor.router, tags=["Code Executor"])
 api_v1_router.include_router(verify_router, prefix="/verify", tags=["Verify"])
+api_v1_router.include_router(reconcile.router, tags=["Reconcile"])

app/services/reconciliation_service.py

@@ -0,0 +1,376 @@
+from __future__ import annotations
+
+import asyncio
+import io
+import os
+import time
+import unicodedata
+from collections import Counter
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+import aiohttp
+import pandas as pd
+
+from app.config import get_settings
+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
+SUPPORTED_EXTENSIONS: Set[str] = {"csv", "xlsx", "xls", "tsv", "parquet"}
+_MAX_FILE_SIZE = _settings.max_upload_bytes
+
+DEFAULT_NUMERIC_KEYWORDS: Set[str] = {
+    "amount", "total", "debit", "credit", "tax",
+    "net amount", "gross amount", "balance", "quantity", "price", "rate", "value"
+}
+
+
+class ReconciliationError(Exception): pass
+class DownloadError(ReconciliationError): pass
+class FileTypeMismatchError(ReconciliationError): pass
+class SchemaMismatchError(ReconciliationError): pass
+class EmptyDatasetError(ReconciliationError): pass
+class UnsupportedFormatError(ReconciliationError): pass
+
+
+def extract_extension(url: str) -> str:
+    path = url.split("?")[0]
+    return os.path.splitext(path)[1].lstrip(".").lower()
+
+
+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})
+        elif pd.api.types.is_numeric_dtype(df[col]):
+            df[col] = df[col].replace({pd.NA: None})
+    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:
+            async with session.get(url, timeout=aiohttp.ClientTimeout(total=DOWNLOAD_TIMEOUT)) as response:
+                response.raise_for_status()
+                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}
+            )
+            await asyncio.sleep(wait_time)
+    raise DownloadError(f"Failed to download {url} after {DOWNLOAD_MAX_RETRIES} retries.")
+
+
+def read_to_dataframe(data: bytes, ext: str, correlation_id: str) -> pd.DataFrame:
+    try:
+        if ext == "csv":
+            return pd.read_csv(io.BytesIO(data))
+        elif ext == "tsv":
+            return pd.read_csv(io.BytesIO(data), sep="\t")
+        elif ext == "xlsx":
+            return pd.read_excel(io.BytesIO(data), engine="openpyxl")
+        elif ext == "xls":
+            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.")
+    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)}")
+
+
+def compare_schemas(df_src: pd.DataFrame, df_dst: pd.DataFrame) -> Dict[str, Any]:
+    src_cols = set(df_src.columns)
+    dst_cols = set(df_dst.columns)
+    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:
+            type_mismatches.append({
+                "column": col,
+                "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,
+        "source_columns": len(df_src.columns),
+        "destination_columns": len(df_dst.columns),
+        "missing_in_source": missing_in_src,
+        "missing_in_destination": missing_in_dst,
+        "type_mismatches": type_mismatches
+    }
+
+
+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())
+        status = "fully_match" if src_nn == dst_nn else "partial_match"
+        results.append({
+            "column_name": col,
+            "status": status,
+            "source_non_null": src_nn,
+            "destination_non_null": dst_nn,
+            "matching_values": 0,
+            "mismatching_values": 0,
+            "difference_count": abs(src_nn - dst_nn)
+        })
+    return results
+
+
+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
+    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)
+                })
+    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())
+            results.append({
+                "column_name": col,
+                "status": "fully_match" if df_src[col].equals(df_dst[col]) else "partial_match",
+                "source_non_null": src_nn,
+                "destination_non_null": dst_nn,
+                "matching_values": min(src_nn, dst_nn),
+                "mismatching_values": abs(src_nn - dst_nn),
+                "difference_count": abs(src_nn - dst_nn),
+                "source_min_date": str(df_src[col].min()),
+                "source_max_date": str(df_src[col].max()),
+                "destination_min_date": str(df_dst[col].min()),
+                "destination_max_date": str(df_dst[col].max()),
+                "missing_dates_source": int(df_src[col].isna().sum()),
+                "missing_dates_destination": int(df_dst[col].isna().sum())
+            })
+    return results
+
+
+def analyze_duplicates(df_src: pd.DataFrame, df_dst: pd.DataFrame) -> Dict[str, int]:
+    src_dups = int(df_src.duplicated().sum())
+    dst_dups = int(df_dst.duplicated().sum())
+    return {
+        "duplicate_rows_source": src_dups,
+        "duplicate_rows_destination": dst_dups,
+        "duplicate_difference": abs(src_dups - dst_dups)
+    }
+
+
+def analyze_missing_data(df_src: pd.DataFrame, df_dst: pd.DataFrame, common_cols: List[str]) -> List[Dict[str, Any]]:
+    results = []
+    for col in common_cols:
+        src_nulls = int(df_src[col].isna().sum())
+        dst_nulls = int(df_dst[col].isna().sum())
+        results.append({
+            "column_name": col,
+            "source_nulls": src_nulls,
+            "destination_nulls": dst_nulls,
+            "difference": abs(src_nulls - dst_nulls)
+        })
+    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]:
+    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": 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
+    }
+
+
+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]:
+    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 = df_src.infer_objects()
+        df_dst = 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))
+        
+        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,
+            "partially_matched_columns": part_cols,
+            "unmatched_columns": len(col_reports) - full_cols - part_cols,
+            "total_rows_source": len(df_src),
+            "total_rows_destination": len(df_dst),
+            "matching_rows": identical_rows,
+            "partial_rows": 0,
+            "unmatched_rows": missing_rows + extra_rows,
+            "duplicate_rows": dup_report["duplicate_difference"],
+            "overall_match_percentage": round(match_pct, 2),
+            "overall_status": status
+        }
+    except Exception as e:
+        _logger.exception(f"Job {job_id} failed during processing.", extra={"correlation_id": job_id})
+        status = "FAILED"
+        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 = {}, [], [], [], {}, []
+    
+    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),
+        "status": status,
+        "summary": summary,
+        "schema": schema_report,
+        "columns": col_reports,
+        "numeric_columns": num_reports,
+        "date_columns": date_reports,
+        "duplicates": dup_report,
+        "missing_values": missing_report,
+        "errors": errors,
+        "warnings": warnings
+    }
+
+
+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}"])
+    
+    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}"])
+    
+    async with aiohttp.ClientSession() as session:
+        try:
+            src_data, dst_data = await asyncio.gather(
+                download_file_with_retry(session, src_url, job_id),
+                download_file_with_retry(session, dst_url, job_id)
+            )
+        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)])
+    
+    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 await process_pair(src_url, dst_url, src_data, dst_data, src_ext, job_id, numeric_keywords)

requirements.txt

@@ -13,6 +13,8 @@ pypdfium2>=4.30.0
 pandas>=2.0.0
 sentence-transformers==5.6.0
 
+aiohttp>=3.9.0
+
 # FIXED: Downgraded to 4.49.0. Transformers 5.x breaks Nomic Vision's trust_remote_code architecture.
 transformers==4.49.0