

import math
import re
from datetime import date, datetime
import time
from typing import Dict, Literal, Optional, Tuple, Union
from fastapi import HTTPException, logger
import pandas as pd
from sqlalchemy import UUID, delete, func, select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
import numpy as np
from src.constants import PeriodType
from src.HRAndCRM.schema import CRMBase, HRBase
from src.HRAndCRM.models import CRM,CRMIntegrationHistory, HRIntegrationHistory, Employee
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

import pandas as pd
import numpy as np
import re
from datetime import datetime
from sqlalchemy import select, delete, func
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional, Dict

import hashlib
import logging

from src.financial.service import FinancialReportError

logger = logging.getLogger(__name__)


class HRDataHandler:

    required_fields = [
        'employee_id', 'name', 'email', 'job_title', 'department',
        'start_date', 'base_salary', 'status'
    ]

    


    @staticmethod
    def validate_dataframe(df: pd.DataFrame) -> pd.Series:
        """
        Validate the dataframe rows and return a boolean mask
        indicating valid rows.
        """

        # Example required columns
        required_columns = ['name', 'email', 'department', 'employee_id']
        for col in required_columns:
            if col not in df.columns:
                logger.error(f"Missing required column: {col}")
                # If required column missing, return all False
                return pd.Series([False] * len(df))

        # Check non-empty required columns
        non_empty_required = df[required_columns].notnull().all(axis=1)

        # Email format regex validation
        email_regex = r"^[\w\.-]+@[\w\.-]+\.\w+$"
        email_valid = df['email'].astype(str).str.match(email_regex)

        # Example: employee_id must be numeric string of length 6
        emp_id_valid = df['employee_id'].astype(str).str.match(r'^\d{6}$')

        # Combine all validations
        valid_mask = non_empty_required & email_valid & emp_id_valid

        logger.debug(f"HR validation: {valid_mask.sum()} valid out of {len(df)} rows")
        return valid_mask


    @staticmethod
    async def file_hr_data_upload(df: pd.DataFrame, column_mapping: dict, db: AsyncSession, user_id: int) -> dict:
        """
        Process and upload HR data after validation.
        Only rows passing validation should be uploaded.
        """

        valid_mask = HRDataHandler.validate_dataframe(df)
        if not valid_mask.any():
            return {"success": False, "detail": "No valid rows to upload."}

        # Filter valid rows
        df_valid = df.loc[valid_mask].copy()

        # Map columns based on mapping provided
        df_valid = df_valid.rename(columns=column_mapping)

        # Example: convert dataframe to list of dict for DB insert
        data_to_insert = df_valid.to_dict(orient='records')

        # Now save to DB (async)
        try:
            
                for record in data_to_insert:
                    # Construct your ORM object here, e.g.:
                    # hr_record = HRModel(**record, created_by=user_id)
                    # db.add(hr_record)

                    # Placeholder: simulate async DB add
                    await db.execute("INSERT INTO hr_table (...) VALUES (...)", params=record)

                await db.commit()
                logger.info(f"Uploaded {len(data_to_insert)} valid HR records for user {user_id}")

                return {"success": True, "uploaded_rows": len(data_to_insert)}

        except Exception as e:
            logger.exception("Failed to upload HR data")
            return {"success": False, "detail": f"DB upload error: {str(e)}"}

    @staticmethod
    def clean_nan_in_obj(obj):
        if isinstance(obj, dict):
            return {k: HRDataHandler.clean_nan_in_obj(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [HRDataHandler.clean_nan_in_obj(i) for i in obj]
        elif isinstance(obj, float) and (np.isnan(obj) or obj in [np.inf, -np.inf]):
            return None
        else:
            return obj

    

    @staticmethod
    async def file_crm_data_upload(df: pd.DataFrame, column_mapping: dict | None, db: AsyncSession, user_id: str) -> dict:
        from .models import CRM
        from .service import IntegrationService
        import hashlib
        import numpy as np
        import pandas as pd

        # Rename columns if mapping provided
        if column_mapping:
            df = df.rename(columns=column_mapping)

        df.columns = df.columns.str.strip()

        # Convert expected_close_date to datetime safely
        df['expected_close_date'] = pd.to_datetime(df['expected_close_date'], errors='coerce')

        def compute_row_hash(row):
            close_date_str = row['expected_close_date'].strftime('%Y-%m-%d') if pd.notnull(row['expected_close_date']) else ''
            row_str = "|".join(str(row.get(col, "")) for col in [
                'opportunity_id', 'customer_name', 'contact_person', 'email', 'product',
                'deal_amount', 'stage', 'probability', close_date_str, 'region'])
            return hashlib.md5(row_str.encode('utf-8')).hexdigest()

        df['row_hash'] = df.apply(compute_row_hash, axis=1)

        # Example validation: check required columns non-null and not empty
        required_cols = ['opportunity_id', 'customer_name', 'email']
        valid_mask = df[required_cols].notnull().all(axis=1) & (df['opportunity_id'].astype(str).str.strip() != '')
        valid_df = df[valid_mask]
        invalid_df = df[~valid_mask]

        if valid_df.empty:
            return {"success": False, "detail": "No valid data to insert after validation."}

        # Check existing hashes in DB
        incoming_hashes = valid_df['row_hash'].tolist()
        query = select(CRM.row_hash).where(CRM.row_hash.in_(incoming_hashes))
        result = await db.execute(query)
        existing_hashes = {row[0] for row in result.fetchall() if row[0] is not None}

        # Separate duplicates and insertable
        duplicate_df = valid_df[valid_df['row_hash'].isin(existing_hashes)]
        insertable_df = valid_df[~valid_df['row_hash'].isin(existing_hashes)]

        if insertable_df.empty:
            return {
                "success": False,
                "detail": "All provided rows are duplicates of existing data. Nothing to insert.",
                "duplicate_count": len(duplicate_df),
                "invalid_count": len(invalid_df)
            }

        # Create integration history
        integration_history = await IntegrationService.create_integration_history(
            integration_type="crm",
            db=db,
            source_type="file",
            status="success" if len(invalid_df) == 0 else "partial_success",
            uploaded_user_id=user_id,
            source_identifier=None,
            message=f"{len(invalid_df)} invalid, {len(duplicate_df)} duplicates skipped"
        )
        integration_id = integration_history.get("id")

        records_to_insert = []
        for _, row in insertable_df.iterrows():
            # Build the CRM record
            crm_record = CRM(
                opportunity_id=row['opportunity_id'],
                customer_name=row['customer_name'],
                contact_person=row['contact_person'],
                email=row['email'],
                product=row['product'],
                deal_amount=row['deal_amount'],
                stage=row['stage'],
                probability=int(row['probability']),
                expected_close_date=row['expected_close_date'].to_pydatetime() if pd.notnull(row['expected_close_date']) else None,
                region=row['region'],
                extra_data={},  # optionally handle extra_data here if you want
                user_id=str(user_id),
                row_hash=row['row_hash'],
                integration_id=integration_id
            )
            records_to_insert.append(crm_record)

        db.add_all(records_to_insert)

        try:
            await db.commit()
        except Exception as e:
            await db.rollback()
            return {"success": False, "detail": f"DB upload error: {str(e)}"}

        # Helper to clean NaNs for JSON serialization
        def clean_nan(df):
            return df.replace([np.inf, -np.inf], np.nan).where(pd.notnull(df), None)

        return {
            "success": True,
            "inserted_count": len(insertable_df),
            "duplicate_count": len(duplicate_df),
            "invalid_count": len(invalid_df),
            "skipped_duplicates": [row for row in clean_nan(duplicate_df).to_dict(orient="records")],
            "invalid_rows": [row for row in clean_nan(invalid_df).to_dict(orient="records")]
        }

    # Other methods (get_all_employees_paginated, get_employee_by_id, delete_employee, update_employee)
    # remain unchanged as you provided


   
    @staticmethod
    async def get_all_employees_paginated(db: AsyncSession, page: int = 1, limit: int = 10):
        from .models import Employee
        from .schema import HRBase

        offset = (page - 1) * limit

        total_query = await db.execute(select(func.count()).select_from(Employee))
        total_items = total_query.scalar_one()

        result = await db.execute(select(Employee).offset(offset).limit(limit))
        items = result.scalars().all()

        total_pages = (total_items + limit - 1) // limit
        has_next = page < total_pages

        return {
            "page": page,
            "limit": limit,
            "total_items": total_items,
            "total_pages": total_pages,
            "has_next": has_next,
            "items": [HRBase.from_orm(emp) for emp in items]
        }

    @staticmethod
    async def get_employee_by_id(employee_id: str, db: AsyncSession):
        from .models import Employee
        result = await db.execute(select(Employee).where(Employee.employee_id == employee_id))
        return result.scalar_one_or_none()

    @staticmethod
    async def delete_employee(employee_id: str, db: AsyncSession):
        from .models import Employee
        await db.execute(delete(Employee).where(Employee.employee_id == employee_id))
        await db.commit()

    @staticmethod
    async def update_employee(employee_id: str, update_data: dict, db: AsyncSession):
        from .models import Employee
        print("heello")
        result = await db.execute(select(Employee).where(Employee.employee_id == employee_id))
        employee = result.scalar_one_or_none()

        if not employee:
            raise ValueError("Employee not found")

        for key, value in update_data.items():
            if hasattr(employee, key):
                setattr(employee, key, value)

        await db.commit()
        return employee


class CRMDataHandler:
    required_fields = [
        'opportunity_id', 'customer_name', 'contact_person', 'email',
        'product', 'deal_amount', 'stage', 'probability',
        'expected_close_date', 'region'
    ]

    valid_stages = {
        "Prospecting", "Proposal", "Negotiation",
        "Closed Won", "Closed Lost"
    }

    
    
    @staticmethod
    def validate_dataframe(df: pd.DataFrame) -> pd.Series:
        # Required fields for validation
        required_fields = [
    'opportunity_id', 'customer_name', 'contact_person', 'email',
    'product', 'deal_amount', 'stage', 'probability',
    'expected_close_date', 'region'
]


        # Check missing columns
        for col in required_fields:
            if col not in df.columns:
                logger.error(f"Missing required column: {col}")
                return pd.Series([False] * len(df))

        # Check non-empty required columns
        non_empty = df[required_fields].notnull().all(axis=1)

        # Email regex validation
        email_regex = r"^[\w\.-]+@[\w\.-]+\.\w+$"
        email_valid = df['email'].astype(str).str.match(email_regex)

        # Phone validation (if present)
        if 'phone_number' in df.columns:
            phone_valid = df['phone_number'].astype(str).str.match(r'^\+?\d{10,15}$')
        else:
            phone_valid = pd.Series([True] * len(df))  # Assume valid if not present

        valid_mask = non_empty & email_valid & phone_valid

        logger.debug(f"CRM validation: {valid_mask.sum()} valid out of {len(df)} rows")
        return valid_mask

    @staticmethod
    def clean_nan_in_obj(obj):
        if isinstance(obj, dict):
            return {k: CRMDataHandler.clean_nan_in_obj(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [CRMDataHandler.clean_nan_in_obj(i) for i in obj]
        elif isinstance(obj, float) and np.isnan(obj):
            return None
        else:
            return obj
    @staticmethod
    async def file_crm_data_upload(
        df: pd.DataFrame,
        column_mapping: dict | None,
        db: AsyncSession,
        user_id: str
    ) -> dict:
        from .models import CRM
        from .service import IntegrationService
        import hashlib
        import numpy as np
        import pandas as pd

        # 1. Rename columns if mapping provided
        if column_mapping:
            df = df.rename(columns=column_mapping)
        df.columns = df.columns.str.strip()

        # 2. Convert date field
        df['expected_close_date'] = pd.to_datetime(df['expected_close_date'], errors='coerce')

        # 3. Compute row hash
        def compute_row_hash(row):
            close_date_str = row['expected_close_date'].strftime('%Y-%m-%d') if pd.notnull(row['expected_close_date']) else ''
            row_str = "|".join(str(row.get(col, "")) for col in [
                'opportunity_id', 'customer_name', 'contact_person', 'email', 'product',
                'deal_amount', 'stage', 'probability', close_date_str, 'region'])
            return hashlib.md5(row_str.encode('utf-8')).hexdigest()

        df['row_hash'] = df.apply(compute_row_hash, axis=1)

        # 4. Basic validation
        required_cols = ['opportunity_id', 'customer_name', 'email']
        valid_mask = df[required_cols].notnull().all(axis=1) & (df['opportunity_id'].astype(str).str.strip() != '')
        valid_df = df[valid_mask]
        invalid_df = df[~valid_mask]

        if valid_df.empty:
            return {"success": False, "detail": "No valid data to insert after validation."}

        # 5. Check duplicates
        incoming_hashes = valid_df['row_hash'].tolist()
        query = select(CRM.row_hash).where(CRM.row_hash.in_(incoming_hashes))
        result = await db.execute(query)
        existing_hashes = {row[0] for row in result.fetchall() if row[0] is not None}

        duplicate_df = valid_df[valid_df['row_hash'].isin(existing_hashes)]
        insertable_df = valid_df[~valid_df['row_hash'].isin(existing_hashes)]

        if insertable_df.empty:
            return {
                "success": False,
                "detail": "All provided rows are duplicates of existing data. Nothing to insert.",
                "duplicate_count": len(duplicate_df),
                "invalid_count": len(invalid_df)
            }

        # 6. Create integration history
        try:
            integration_history = await IntegrationService.create_integration_history(
                integration_type="crm",
                db=db,
                source_type="file",
                status="success" if len(invalid_df) == 0 else "partial_success",
                uploaded_user_id=user_id,
                source_identifier=None,
                message=f"{len(invalid_df)} invalid, {len(duplicate_df)} duplicates skipped"
            )
            integration_id = integration_history.get("id")
        except Exception as e:
            await db.rollback()
            return {"success": False, "detail": f"Failed to create integration history: {str(e)}"}

        # 7. Insert records
        try:
            records_to_insert = []
            for _, row in insertable_df.iterrows():
                crm_record = CRM(
                    opportunity_id=row['opportunity_id'],
                    customer_name=row['customer_name'],
                    contact_person=row['contact_person'],
                    email=row['email'],
                    product=row['product'],
                    deal_amount=row['deal_amount'],
                    stage=row['stage'],
                    probability=int(row['probability']),
                    expected_close_date=row['expected_close_date'].to_pydatetime() if pd.notnull(row['expected_close_date']) else None,
                    region=row['region'],
                    extra_data={},
                    user_id=str(user_id),
                    row_hash=row['row_hash'],
                    integration_id=integration_id
                )
                records_to_insert.append(crm_record)

            db.add_all(records_to_insert)
            await db.commit()
            logger.info(f"Inserted {len(records_to_insert)} CRM records for user {user_id}")

        except Exception as e:
            await db.rollback()
            logger.exception("Error inserting CRM data")
            return {"success": False, "detail": f"DB upload error: {str(e)}"}

        # 8. Sanitize for return
        def clean_nan(df):
            return df.replace([np.inf, -np.inf], np.nan).where(pd.notnull(df), None)

        return {
            "success": True,
            "inserted_count": len(insertable_df),
            "duplicate_count": len(duplicate_df),
            "invalid_count": len(invalid_df),
            "skipped_duplicates": clean_nan(duplicate_df).to_dict(orient="records"),
            "invalid_rows": clean_nan(invalid_df).to_dict(orient="records")
        }

    
    @staticmethod
    async def get_all_opportunities_paginated(db: AsyncSession, page: int = 1, limit: int = 10):
        from .models import CRM

        offset = (page - 1) * limit

        total_query = await db.execute(select(func.count()).select_from(CRM))
        total_items = total_query.scalar_one()

        result = await db.execute(select(CRM).offset(offset).limit(limit))
        orm_items = result.scalars().all()

        
        items = [CRMBase.model_validate(item) for item in orm_items]

        total_pages = (total_items + limit - 1) // limit
        has_next = page < total_pages

        return {
            "page": page,
            "limit": limit,
            "total_items": total_items,
            "total_pages": total_pages,
            "has_next": has_next,
            "items": items,
        }

    @staticmethod
    async def get_CRM_by_id(opportunity_id: str, db: AsyncSession):
        from .models import CRM
        result = await db.execute(select(CRM).where(CRM.opportunity_id == opportunity_id))
        return result.scalar_one_or_none()

    @staticmethod
    async def delete_CRM(opportunity_id: str, db: AsyncSession):
        from .models import CRM
        await db.execute(delete(CRM).where(CRM.opportunity_id == opportunity_id))
        await db.commit()

    @staticmethod
    async def update_CRM(opportunity_id: str, update_data: dict, db: AsyncSession):
        from .models import CRM
        result = await db.execute(select(CRM).where(CRM.opportunity_id == opportunity_id))
        crm = result.scalar_one_or_none()

        if not crm:
            raise ValueError("CRM not found")

        for key, value in update_data.items():
            if hasattr(crm, key):
                setattr(crm, key, value)

        await db.commit()
        return crm



# src/services/integration_service.py


import time
import logging
from typing import Dict, Union, Optional, Literal
from uuid import UUID
from datetime import datetime

from fastapi import HTTPException
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)


class IntegrationService:

    # ---- CRM ----

    @staticmethod
    async def get_crm_history(page: int, limit: int, db: AsyncSession):
        offset = (page - 1) * limit
        logger.info(f"Fetching CRM integration history: page={page}, limit={limit}")

        try:
            total_query = await db.execute(select(func.count()).select_from(CRMIntegrationHistory))
            total_records = total_query.scalar()

            result = await db.execute(
                select(CRMIntegrationHistory)
                .order_by(CRMIntegrationHistory.created_at.desc())
                .offset(offset)
                .limit(limit)
            )
            records = result.scalars().all()

            logger.info(f"Fetched {len(records)} CRM history records")

            return {
                "page": page,
                "limit": limit,
                "total_records": total_records,
                "has_next": page * limit < total_records,
                "records": [r.as_dict() for r in records]
            }

        except Exception as e:
            logger.error(f"Error fetching CRM history: {e}", exc_info=True)
            raise HTTPException(status_code=500, detail="Error retrieving CRM history.")

   

    @staticmethod
    async def get_crm_entries_paginated(
        integration_id: str,
        page: int,
        limit: int,
        db: AsyncSession,
    ) -> Dict:
        logger.info(f"Fetching CRM entries - integration_id={integration_id}, page={page}, limit={limit}")
        start = time.time()

        offset = (page - 1) * limit

        try:
            # Get total count
            count_query = await db.execute(
                select(func.count()).select_from(CRM).where(CRM.integration_id == integration_id)
            )
            total_entries = count_query.scalar()

            # Fetch paginated entries
            result = await db.execute(
                select(CRM)
                .where(CRM.integration_id == integration_id)
                .offset(offset)
                .limit(limit)
            )
            entries = result.scalars().all()

            # Serialize safely
            serialized_entries = [
                {k: v for k, v in entry.__dict__.items() if not k.startswith("_")}
                for entry in entries
            ]

            duration = time.time() - start
            logger.info(f"CRM entries fetched in {duration:.2f}s")

            return {
                "integration_id": integration_id,
                "page": page,
                "limit": limit,
                "total_entries": total_entries,
                "has_next": page * limit < total_entries,
                "entries": serialized_entries
            }

        except Exception as e:
            logger.error(f"Error fetching paginated CRM entries: {e}", exc_info=True)
            raise HTTPException(status_code=500, detail="Failed to fetch CRM entries.")

    @staticmethod
    async def delete_crm_history(integration_id: str, db: AsyncSession):
        logger.info(f"Deleting CRM integration history with id={integration_id}")

        try:
            result = await db.execute(select(CRMIntegrationHistory).where(CRMIntegrationHistory.id == integration_id))
            record = result.scalar_one_or_none()

            if not record:
                raise HTTPException(status_code=404, detail="CRM integration history not found.")

            await db.delete(record)
            await db.commit()

            logger.info("CRM integration history deleted successfully")
            return {"detail": "CRM integration and related entries deleted."}

        except Exception as e:
            await db.rollback()
            logger.error(f"Error deleting CRM integration history: {e}", exc_info=True)
            raise HTTPException(status_code=500, detail=f"Failed to delete integration: {str(e)}")

    # ---- HR ----

    @staticmethod
    async def get_hr_history(page: int, limit: int, db: AsyncSession):
        logger.info(f"Fetching HR integration history: page={page}, limit={limit}")
        start_time = time.time()

        try:
            offset = (page - 1) * limit

            total_query = await db.execute(select(func.count()).select_from(HRIntegrationHistory))
            total_records = total_query.scalar()
            logger.info(f"Total HR history records found: {total_records}")

            result = await db.execute(
                select(HRIntegrationHistory)
                .order_by(HRIntegrationHistory.created_at.desc())
                .offset(offset)
                .limit(limit)
            )
            records = result.scalars().all()

            duration = time.time() - start_time
            logger.info(f"Fetched {len(records)} HR records in {duration:.2f}s")

            return {
                "page": page,
                "limit": limit,
                "total_records": total_records,
                "has_next": page * limit < total_records,
                "records": [r.as_dict() for r in records]
            }

        except Exception as e:
            logger.error(f"Error fetching HR integration history: {e}", exc_info=True)
            raise HTTPException(status_code=500, detail="Error retrieving HR history.")

    

    @staticmethod
    async def get_hr_entries_paginated(
        integration_id: str,
        page: int,
        limit: int,
        db: AsyncSession,
    ) -> Dict:
        logger.info(f"Fetching HR entries - integration_id={integration_id}, page={page}, limit={limit}")
        start = time.time()

        offset = (page - 1) * limit

        try:
            # Get total count
            count_query = await db.execute(
                select(func.count()).select_from(Employee).where(Employee.integration_id == integration_id)
            )
            total_entries = count_query.scalar()

            # Fetch paginated entries
            result = await db.execute(
                select(Employee)
                .where(Employee.integration_id == integration_id)
                .offset(offset)
                .limit(limit)
            )
            entries = result.scalars().all()

            # Serialize safely
            serialized_entries = [
                {k: v for k, v in entry.__dict__.items() if not k.startswith("_")}
                for entry in entries
            ]

            duration = time.time() - start
            logger.info(f"HR entries fetched in {duration:.2f}s")

            return {
                "integration_id": integration_id,
                "page": page,
                "limit": limit,
                "total_entries": total_entries,
                "has_next": page * limit < total_entries,
                "entries": serialized_entries
            }

        except Exception as e:
            logger.error(f"Error fetching paginated HR entries: {e}", exc_info=True)
            raise HTTPException(status_code=500, detail="Failed to fetch HR entries.")

    @staticmethod
    async def delete_hr_history(integration_id: str, db: AsyncSession):
        logger.info(f"Deleting HR integration history with id={integration_id}")

        try:
            result = await db.execute(select(HRIntegrationHistory).where(HRIntegrationHistory.id == integration_id))
            record = result.scalar_one_or_none()

            if not record:
                raise HTTPException(status_code=404, detail="HR integration history not found.")

            await db.delete(record)
            await db.commit()

            logger.info("HR integration history deleted successfully")
            return {"detail": "HR integration and related entries deleted."}

        except Exception as e:
            await db.rollback()
            logger.error(f"Error deleting HR integration history: {e}", exc_info=True)
            raise HTTPException(status_code=500, detail=f"Failed to delete integration: {str(e)}")

    @staticmethod
    async def create_integration_history(
        integration_type: Literal["crm", "hr"],
        db: AsyncSession,
        source_type: str,
        status: str,
        uploaded_user_id: Optional[UUID] = None,
        source_identifier: Optional[str] = None,
        message: Optional[str] = None,
    ) -> Dict[str, Union[str, datetime]]:
        logger.info(f"Creating integration history: type={integration_type}, status={status}")

        if integration_type == "crm":
            history_model = CRMIntegrationHistory
        elif integration_type == "hr":
            history_model = HRIntegrationHistory
        else:
            raise HTTPException(status_code=400, detail="Invalid integration type.")

        record = history_model(
            source_type=source_type,
            source_identifier=source_identifier,
            status=status,
            message=message,
            uploaded_user_id=uploaded_user_id,
        )

        db.add(record)
        try:
            await db.commit()
            await db.refresh(record)
            logger.info(f"Integration history created with ID={record.id}")
            return record.as_dict()
        except Exception as e:
            await db.rollback()
            logger.error(f"Error creating integration history: {e}", exc_info=True)
            raise HTTPException(status_code=500, detail=f"Failed to create integration history: {str(e)}")

 

def _validate_date_range(start_date: datetime, end_date: datetime) -> None:
        """Validate that start_date is before end_date"""
        if start_date > end_date:
            raise FinancialReportError("start_date must be before end_date")

def get_date_range(
        period: str, 
        year: Optional[int], 
        start_date: Optional[datetime], 
        end_date: Optional[datetime]
    ) -> Tuple[datetime, datetime]:
        """Get date range based on period and year"""
        try:
            if period == "yearly" and year:
                return datetime(year, 1, 1), datetime(year, 12, 31)
            elif start_date and end_date:
                _validate_date_range(start_date, end_date)
                return start_date, end_date
            else:
                raise FinancialReportError("Either year or both start_date and end_date must be provided")
        except Exception as e:
            logger.error(f"Failed to get date range: {e}")
            raise FinancialReportError(f"Failed to get date range: {e}")

import pandas as pd
import calendar
from datetime import datetime, date
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, and_
from dateutil.parser import ParserError
async def get_payroll_data(
    db: AsyncSession,
    year: int,
    period: PeriodType,
    custom_start: Optional[datetime],
    custom_end: Optional[datetime],
    entity: Optional[str],
    hierarchy: Optional[str]
):
    from sqlalchemy import and_, or_
    from datetime import datetime, date
    import pandas as pd
    import calendar

    # 1. Set date range
    start_date, end_date = get_date_range(period, year, custom_start, custom_end)
    start_date = start_date.date() if isinstance(start_date, datetime) else start_date
    end_date = end_date.date() if isinstance(end_date, datetime) else end_date

    # 2. Build dynamic DB query
    query = select(
        Employee.employee_id,
        Employee.base_salary,
        Employee.bonus,
        Employee.start_date,
        Employee.termination_date,
        Employee.extra_data,
        Employee.location,
        Employee.department,
    )

    filters = []

    # Filters for entity and hierarchy
    if entity:
        if hierarchy == "department":
            filters.append(Employee.department == entity)
        elif hierarchy == "product":
            filters.append(Employee.product == entity)
        else:
            filters.append(Employee.location == entity)

    # Add active employees filter directly in SQL:
    # Include those with termination_date NULL (still active)
    filters.append(
        and_(
            Employee.start_date <= end_date,
            or_(
                Employee.termination_date == None,
                Employee.termination_date >= start_date,
                Employee.status == "active"
            ),
        )
    )

    if filters:
        query = query.where(and_(*filters))

    # Execute query
    result = await db.execute(query)
    rows = result.all()

    if not rows:
        return {
            "monthly": {m: {"payroll": 0, "benefits": 0, "taxes": 0, "actual_disbursed": 0} for m in range(1, 13)},
            "quarterly": {q: {"payroll": 0, "benefits": 0, "taxes": 0, "actual_disbursed": 0} for q in range(1, 5)},
            "yearly": {"payroll": 0, "benefits": 0, "taxes": 0, "actual_disbursed": 0},
        }

    # 3. Load into DataFrame
    df = pd.DataFrame(rows, columns=[
        "id", "base_salary", "bonus", "start_date", "termination_date",
        "extra_data", "location", "department"
    ])

    # 4. Normalize the `extra_data` JSON into separate columns if needed
    # (optional - you can skip this if you prefer original method)
    extra_df = pd.json_normalize(df["extra_data"]).fillna(0)
    extra_df.columns = [f"actual_{col}" for col in extra_df.columns]
    df = df.drop(columns=["extra_data"]).join(extra_df)

    # 5. Normalize dates and fill missing termination_date to end of year
    df["start_date"] = pd.to_datetime(df["start_date"]).dt.date
    df["termination_date"] = pd.to_datetime(
        df["termination_date"].fillna(pd.Timestamp(f"{year}-12-31"))
    ).dt.date

    # 6. Calculate payroll components
    df["monthly_payroll"] = df["base_salary"] / 12
    df["monthly_benefits"] = df["bonus"] / 12
    df["monthly_tax"] = 0.25 * (df["monthly_payroll"] + df["monthly_benefits"])

    # 7. Process actual disbursed values from extra columns (if normalized)
    # Sum up actual disbursed per month by aggregating columns prefixed with 'actual_'
    monthly_actual = {m: 0 for m in range(1, 13)}
    for col in extra_df.columns:
        # Expecting keys like actual_YYYY-MM or actual_YYYY-MM-DD
        try:
            # parse month from column name e.g., actual_2023-01-15 -> 1
            date_part = col.replace("actual_", "")
            dt = pd.to_datetime(date_part)
            month = dt.month
            monthly_actual[month] += df[col].sum()
        except Exception:
            continue

    # 8. Initialize results
    monthly_results = {m: {"payroll": 0, "benefits": 0, "taxes": 0, "actual_disbursed": 0} for m in range(1, 13)}
    quarterly_results = {q: {"payroll": 0, "benefits": 0, "taxes": 0, "actual_disbursed": 0} for q in range(1, 5)}
    yearly_results = {"payroll": 0, "benefits": 0, "taxes": 0, "actual_disbursed": 0}

    # 9. Aggregate monthly data
    for month in range(1, 13):
        month_start = date(year, month, 1)
        month_end = date(year, month, calendar.monthrange(year, month)[1])
        active_employees = df[(df["start_date"] <= month_end) & (df["termination_date"] >= month_start)]

        if active_employees.empty:
            continue

        payroll_sum = active_employees["monthly_payroll"].sum()
        benefits_sum = active_employees["monthly_benefits"].sum()
        taxes_sum = active_employees["monthly_tax"].sum()
        actual_sum = monthly_actual.get(month, 0)

        monthly_results[month] = {
            "payroll": payroll_sum,
            "benefits": benefits_sum,
            "taxes": taxes_sum,
            "actual_disbursed": actual_sum,
        }

        quarter = (month - 1) // 3 + 1
        quarterly_results[quarter]["payroll"] += payroll_sum
        quarterly_results[quarter]["benefits"] += benefits_sum
        quarterly_results[quarter]["taxes"] += taxes_sum
        quarterly_results[quarter]["actual_disbursed"] += actual_sum

        yearly_results["payroll"] += payroll_sum
        yearly_results["benefits"] += benefits_sum
        yearly_results["taxes"] += taxes_sum
        yearly_results["actual_disbursed"] += actual_sum

    return {
        "monthly": monthly_results,
        "quarterly": quarterly_results,
        "yearly": yearly_results,
    }


def prepare_payroll_report(payroll_data, months=12):
    import calendar
    
    monthly_data = payroll_data["monthly"]
    quarterly_data = payroll_data["quarterly"]
    yearly_data = payroll_data["yearly"]

    # Monthly Summary Table
    monthly_summary = []
    total_payroll_monthly = 0
    total_benefits_monthly = 0
    total_taxes_monthly = 0
    total_actual_disbursed_monthly = 0

    for month in range(1, months + 1):
        p = monthly_data[month]
        payroll_k = round(p["payroll"] / 1000, 1)
        benefits_k = round(p["benefits"] / 1000, 1)
        taxes_k = round(p["taxes"] / 1000, 1)
        actual_disbursed_k = round(p.get("actual_disbursed", 0) / 1000, 1)
        total_k = payroll_k + benefits_k + taxes_k + actual_disbursed_k

        monthly_summary.append({
            "period": calendar.month_abbr[month],
            "payroll": f"{payroll_k}",
            "benefits": f"{benefits_k}",
            "taxes": f"{taxes_k}",
            "actual_disbursed": f"{actual_disbursed_k}",
            "total": f"{round(total_k, 1)}",
        })

        total_payroll_monthly += p["payroll"]
        total_benefits_monthly += p["benefits"]
        total_taxes_monthly += p["taxes"]
        total_actual_disbursed_monthly += p.get("actual_disbursed", 0)

    # Quarterly Summary Table
    quarterly_summary = []
    total_actual_disbursed_quarterly = 0
    for quarter in range(1, 5):
        q = quarterly_data[quarter]
        payroll_k = round(q["payroll"] / 1000, 1)
        benefits_k = round(q["benefits"] / 1000, 1)
        taxes_k = round(q["taxes"] / 1000, 1)
        actual_disbursed_k = round(q.get("actual_disbursed", 0) / 1000, 1)
        total_k = payroll_k + benefits_k + taxes_k + actual_disbursed_k

        quarterly_summary.append({
            "period": f"Q{quarter}",
            "payroll": f"{payroll_k}",
            "benefits": f"{benefits_k}",
            "taxes": f"{taxes_k}",
            "actual_disbursed": f"{actual_disbursed_k}",
            "total": f"{round(total_k, 1)}",
        })

        total_actual_disbursed_quarterly += q.get("actual_disbursed", 0)

    # Yearly Summary
    yearly_payroll_k = round(yearly_data["payroll"] / 1000, 1)
    yearly_benefits_k = round(yearly_data["benefits"] / 1000, 1)
    yearly_taxes_k = round(yearly_data["taxes"] / 1000, 1)
    yearly_actual_disbursed_k = round(yearly_data.get("actual_disbursed", 0) / 1000, 1)
    yearly_total_k = yearly_payroll_k + yearly_benefits_k + yearly_taxes_k + yearly_actual_disbursed_k

    yearly_summary = {
        "period": "Yearly",
        "payroll": f"{yearly_payroll_k}",
        "benefits": f"{yearly_benefits_k}",
        "taxes": f"{yearly_taxes_k}",
        "actual_disbursed": f"{yearly_actual_disbursed_k}",
        "total": f"{round(yearly_total_k, 1)}",
    }

    grand_total = yearly_data["payroll"] + yearly_data["benefits"] + yearly_data["taxes"] + yearly_data.get("actual_disbursed", 0)
    average_per_month = grand_total / 12

    # Monthly Reconciliation Table
    monthly_reconciliation = []
    total_gross = 0
    total_taxes_withheld = 0
    total_benefits_deducted = 0
    total_net_pay_calc = 0
    total_amount_disbursed = 0
    total_variance = 0

    for month in range(1, months + 1):
        p = monthly_data[month]
        payroll = p["payroll"]
        benefits = p["benefits"]
        taxes = p["taxes"]
        actual_disbursed = p.get("actual_disbursed", 0)

        gross_wages = payroll + benefits
        taxes_withheld = taxes
        benefits_deducted = benefits

        net_pay_calc = round(gross_wages - taxes_withheld - benefits_deducted)
        amount_disbursed = round(actual_disbursed) if actual_disbursed else net_pay_calc
        variance = net_pay_calc - amount_disbursed

        monthly_reconciliation.append({
            "period": calendar.month_abbr[month],
            "gross_wages": f"{int(gross_wages):,}",
            "taxes_withheld": f"{int(taxes_withheld):,}",
            "benefits_deducted": f"{int(benefits_deducted):,}",
            "net_pay_calculated": f"{int(net_pay_calc):,}",
            "amount_disbursed": f"{int(amount_disbursed):,}",
            "variance": f"{int(variance):,}" if variance != 0 else 0,
        })

        total_gross += gross_wages
        total_taxes_withheld += taxes_withheld
        total_benefits_deducted += benefits_deducted
        total_net_pay_calc += net_pay_calc
        total_amount_disbursed += amount_disbursed
        total_variance += variance

    # Quarterly Reconciliation
    quarterly_reconciliation = []
    for quarter in range(1, 5):
        q = quarterly_data[quarter]
        payroll = q["payroll"]
        benefits = q["benefits"]
        taxes = q["taxes"]
        actual_disbursed = q.get("actual_disbursed", 0)

        gross_wages = payroll + benefits
        taxes_withheld = taxes
        benefits_deducted = benefits

        net_pay_calc = round(gross_wages - taxes_withheld - benefits_deducted)
        amount_disbursed = round(actual_disbursed) if actual_disbursed else net_pay_calc
        variance = net_pay_calc - amount_disbursed

        quarterly_reconciliation.append({
            "period": f"Q{quarter}",
            "gross_wages": f"{int(gross_wages):,}",
            "taxes_withheld": f"{int(taxes_withheld):,}",
            "benefits_deducted": f"{int(benefits_deducted):,}",
            "net_pay_calculated": f"{int(net_pay_calc):,}",
            "amount_disbursed": f"{int(amount_disbursed):,}",
            "variance": f"{int(variance):,}" if variance != 0 else 0,
        })

    reconciliation_insights = (
        f"A minor variance of ${int(total_variance):,}" if total_variance != 0 else "No variance detected."
    )

    return {
        "monthly": {
            "summary_table": monthly_summary,
            "reconciliation_table": monthly_reconciliation
        },
        "quarterly": {
            "summary_table": quarterly_summary,
            "reconciliation_table": quarterly_reconciliation
        },
        "yearly": {
            "summary": yearly_summary,
            "key_metrics": {
                "total_payroll": f"{int(yearly_data['payroll']):,}",
                "total_benefits": f"{int(yearly_data['benefits']):,}",
                "total_taxes": f"{int(yearly_data['taxes']):,}",
                "total_actual_disbursed": f"{int(yearly_data.get('actual_disbursed', 0)):,}",
                "grand_total": f"{int(grand_total):,}",
                "average_per_month": f"{int(average_per_month):,}",
            },
            "reconciliation_key_metrics": {
                "total_gross_payroll": f"{int(total_gross):,}",
                "total_disbursed": f"{int(total_amount_disbursed):,}",
                "total_reconciliation_variance": f"{int(total_variance):,}",
            },
            "reconciliation_insights": reconciliation_insights,
        }
    }


# Example usage:
async def generate_complete_payroll_report(
    db: AsyncSession,
    year: int,
    period: PeriodType = PeriodType.YEARLY,
    custom_start: Optional[datetime] = None,
    custom_end: Optional[datetime] = None,
    entity: Optional[str] = None,
    hierarchy: Optional[str] = None
):
    """Complete payroll report with monthly, quarterly, and yearly data"""
    try:
        # Get payroll data
        payroll_data = await get_payroll_data(
            db, year, period, custom_start, custom_end, entity, hierarchy
        )
        
        # Prepare comprehensive report
        report = prepare_payroll_report(payroll_data)
        
        return report
        
    except Exception as e:
        logger.error(f"Failed to generate payroll report: {e}")
        raise FinancialReportError(f"Failed to generate payroll report: {e}")