
import calendar
from datetime import datetime, timedelta,timezone
import copy
from decimal import Decimal
import decimal
from io import BytesIO
import json
import math
import uuid
from fastapi import HTTPException
from fastapi.responses import JSONResponse
import pandas as pd
from sqlalchemy import UUID, and_, extract, func, insert, select, text, tuple_, update,desc
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
from src.financial.constants import COLUMN_ALIASES
from src.Service.companyService import CompanyService
from src.constants import DataIntegrationSourceEnum,FinancialDashboardWidgetNameEnum
from src.financial.utils import clean_df_for_json, make_merge_key, to_camel_case
from src.financial.model import Attribute, Entity, FileUploadHistory, GeneralLedger, GeneralLedgerMaster, IntegrationHistory,FinancialWidget, TrackedJSONKey, Value
from src.financial.schemas import GeneralLedgerEntryUpdateRequest, GlMasterData,GeneralLedgerMasterDataRequest,WidgetSchema,KPIWidgetUpdateSchema
from sqlalchemy.exc import IntegrityError
from collections import Counter, OrderedDict, defaultdict, namedtuple
from typing import Tuple, Optional
from .model import FPNAYearlyReport , FPNAYearlyReport1

from .utils import make_row_hash
import re 
import numpy as np 
from typing import List,Dict,Optional, Set
import logging

# Setup logger
log_filename = "gl_upload_log.txt"  # Define log file location
logging.basicConfig(
    filename=log_filename,  # File to write logs
    level=logging.DEBUG,  # Log all levels of messages
    format='%(asctime)s - %(levelname)s - %(message)s'
)



logger = logging.getLogger()

# You can also log to console by adding a StreamHandler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)

# Function to log data (for convenience)
def log_dataframe(name,df):
    # Check if df is a list, and convert it to a DataFrame if so
    if isinstance(df, list):
        df = pd.DataFrame(df)  # Convert list to DataFrame
    logger.info(f"\n{name}:\n{df.to_string(index=False)}\n")





import time
import functools
import asyncio

def log_execution_time(func):
    if asyncio.iscoroutinefunction(func):
        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = await func(*args, **kwargs)
            elapsed = time.perf_counter() - start
            logger.info(f"{func.__name__} took {elapsed:.3f} seconds")
            return result
        return async_wrapper
    else:
        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = func(*args, **kwargs)
            elapsed = time.perf_counter() - start
            logger.info(f"{func.__name__} took {elapsed:.3f} seconds")
            return result
        return sync_wrapper



#saves and process the pandas dataframes olny
import pandas as pd
import uuid
import asyncio
from fastapi import HTTPException
from typing import Optional, Dict
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.exc import IntegrityError

from typing import Optional, Literal, Union

ColumnType = Optional[Literal["column", "jsonb", None]]
#. column checking weather is present or not

def get_column_type(field_name: str) -> ColumnType:
    """
    Determine whether the given field name corresponds to a normal column
    or a key in the `extra_data` JSONB column in the GeneralLedger model.

    Returns:
        "column" → if it's a standard column in GeneralLedger
        "jsonb"  → if it's a key inside `extra_data`
        None     → if the field is not found
    """
    # Check if it's a normal SQLAlchemy column
    if field_name in GeneralLedger.__table__.columns:
        return "column"

    # Check if it's in extra_data keys — optional: use known keys or skip
    # Here, we assume anything not in table is JSONB key (if you want)
    return "jsonb"

class AddGLData:
    import re

    @staticmethod
    @log_execution_time
    def clean_string_columns(df: pd.DataFrame) -> pd.DataFrame:
        for col in df.select_dtypes(include=["object", "string"]).columns:
            df[col] = (
                df[col]
                .astype(str)
                .str.strip()
                .str.lower()
                .str.replace(r"[^a-z0-9 ]", "", regex=True)  # keep only a-z, 0-9, space
                .str.replace(r"\s+", " ", regex=True)  # normalize multiple spaces
            )
        return df

    @staticmethod
    @log_execution_time
    def validate_and_rename_columns(df: pd.DataFrame, column_mapping: Optional[Dict[str, str]] = None) -> pd.DataFrame:
        print("validate_and_rename_columns start :",len(df))
        if column_mapping:
            df = df.rename(columns=column_mapping)

        df['date'] = pd.to_datetime(df['date'], errors='coerce')

        if "single_col_debit_credit" in df.columns:
            if "amount" not in df.columns:
                raise HTTPException(status_code=400, detail="'amount' column is missing.")
            df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
            df["debit"] = df["credit"] = 0.0
            debit_mask = df["single_col_debit_credit"].str.lower() == "debit"
            credit_mask = df["single_col_debit_credit"].str.lower() == "credit"
            df.loc[debit_mask, "debit"] = df["amount"]
            df.loc[credit_mask, "credit"] = df["amount"]
            df.drop(columns=["single_col_debit_credit", "amount"], inplace=True)

        required_cols = {"date", "general_ledger_code", "account_name", "category", "debit", "credit"}
        for col in required_cols:
            if col not in df.columns:
                df[col] = 0.0 if col in {"debit", "credit"} else None

        df = AddGLData.clean_string_columns(df)

        print("validate_and_rename_columns end :",len(df))
        return df

    @staticmethod
    @log_execution_time
    def validate_row_nulls(df: pd.DataFrame):
        print("validate_row_nulls start :",len(df))
        invalid_rows = df[
            df[["general_ledger_code", "account_name"]]
            .apply(lambda x: x.isnull() | (x.astype(str).str.strip() == ""), axis=1)
            .any(axis=1)
        ]
        print("validate_row_nulls end :",len(df))
        if not invalid_rows.empty:
            raise HTTPException(status_code=400, detail=f"Null or empty values in rows: {invalid_rows.index.tolist()}")

    @staticmethod
    @log_execution_time
    async def validate_master_gl_rows(df: pd.DataFrame, db):
        print("validate_master_gl_rows start :",len(df))
        df["account_name"] = df["account_name"].astype(str).str.strip().str.lower()
        df["general_ledger_code"] = df["general_ledger_code"].astype(int)
        print("validate master_glrow :",len(df))
        file_rows = df[["general_ledger_code", "account_name"]].drop_duplicates()
        print("validate master_glrow :",len(file_rows))
        stmt = select(GeneralLedgerMaster)
        result = await db.execute(stmt)
        master = result.scalars().all()
        master_df = pd.DataFrame([m.as_dict() for m in master])

        if master_df.empty:
            raise HTTPException(status_code=400, detail="GL master table is empty.")

        master_df["account_name"] = master_df["account_name"].str.lower()
        master_df["general_ledger_code"] = master_df["general_ledger_code"].astype(int)

        diff = file_rows.merge(
            master_df[["general_ledger_code", "account_name"]],
            on=["general_ledger_code", "account_name"],
            how="left",
            indicator=True
        ).query('_merge == "left_only"').drop(columns=["_merge"])
        print("validate_master_gl_rows end :",len(df))
        if not diff.empty:
            raise HTTPException(status_code=400, detail={
                "error": "Invalid GL rows",
                "invalid_rows": diff.to_dict(orient="records")
            })

    @staticmethod
    @log_execution_time
    async def add_new_tracked_json_keys(db, extra_columns: dict, table_name: str, column_name: str):
        stmt = select(TrackedJSONKey.key).where(
            (TrackedJSONKey.table_name == table_name) &
            (TrackedJSONKey.column_name == column_name)
        )
        existing = {row[0] for row in (await db.execute(stmt)).fetchall()}
        new_keys = set(extra_columns.keys()) - existing

        for key in new_keys:
            db.add(TrackedJSONKey(
                table_name=table_name,
                column_name=column_name,
                key=key,
                data_type=extra_columns[key]
            ))

        try:
            await db.commit()
        except IntegrityError:
            await db.rollback()

        return new_keys

    @staticmethod
    @log_execution_time
    async def check_duplicates_and_save(df: pd.DataFrame, db, source_identifier: str, uploaded_user_id: str):
        print("check_duplicates_and_save start:", len(df))
        batch_size = 1000

        integration_record = IntegrationHistory(
            source_type=DataIntegrationSourceEnum.FILE.value,
            source_identifier=source_identifier,
            status="success",
            message=f"Uploaded {df.shape[0]} entries",
            uploaded_user_id=uploaded_user_id
        )
        db.add(integration_record)
        await db.flush()

        df["id"] = [uuid.uuid4() for _ in range(len(df))]
        df["transaction_hash"] = df.apply(make_row_hash, axis=1)
        df["integration_id"] = integration_record.id
        df["entry_by"] = uploaded_user_id
        df["entry_system_source"] = "excel_upload"
        df["is_active"] = True
        df["date"] = pd.to_datetime(df["date"], errors="coerce")

        invalid_dates = df[df["date"].isnull()]
        if not invalid_dates.empty:
            raise HTTPException(status_code=400, detail={
                "error": "Invalid dates found.",
                "rows": invalid_dates.to_dict(orient="records")
            })

        base_cols = {
            'id', 'date', 'general_ledger_code', 'account_name', 'category',
            'description', 'debit', 'credit', 'is_active', 'entry_by',
            'entry_system_source', 'transaction_hash', 'integration_id'
        }
        extra_cols = set(df.columns) - base_cols

        def infer_dtype(value):
            if isinstance(value, int): return "integer"
            if isinstance(value, float): return "float"
            if isinstance(value, bool): return "boolean"
            return "string"

        extra_types = {
            col: infer_dtype(df[col].dropna().iloc[0]) if not df[col].dropna().empty else "string"
            for col in extra_cols
        }

        df["extra_data"] = df[list(extra_cols)].to_dict(orient="records") if extra_cols else [{} for _ in range(len(df))]

        await AddGLData.add_new_tracked_json_keys(db, extra_types, "general_ledger", "extra_data")

        inserted_hashes = set()
        all_hashes = set(df["transaction_hash"])

        skipped_hashes_total = set()
        duplicate_in_input_total = set()
        missing_hash_rows_total = []

        from sqlalchemy.dialects.postgresql import insert

        for i in range(0, len(df), batch_size):
            batch = df.iloc[i:i+batch_size].copy()
            if "description" not in batch:
                batch["description"] = ""

            batch["description"] = batch["description"].fillna("").astype(str)
            insert_data = batch[list(base_cols) + ["extra_data"]].to_dict(orient="records")

            stmt = insert(GeneralLedger).values(insert_data)
            stmt = stmt.on_conflict_do_nothing(index_elements=["transaction_hash"]).returning(GeneralLedger.transaction_hash)
            result = await db.execute(stmt)
            batch_inserted = {row[0] for row in result.fetchall()}
            inserted_hashes.update(batch_inserted)

            # Check skipped in this batch
            batch_all_hashes = {row["transaction_hash"] for row in insert_data if "transaction_hash" in row}
            skipped_hashes = batch_all_hashes - batch_inserted
            skipped_hashes_total.update(skipped_hashes)

            # Duplicate hashes in this batch input
            counter = Counter([row["transaction_hash"] for row in insert_data])
            duplicates = {h for h, c in counter.items() if c > 1}
            duplicate_in_input_total.update(duplicates)

            # Missing hashes in this batch input
            missing_hashes = [row for row in insert_data if not row.get("transaction_hash")]
            missing_hash_rows_total.extend(missing_hashes)

        await db.commit()

        if skipped_hashes_total or missing_hash_rows_total or duplicate_in_input_total:
            dup_df = df[df["transaction_hash"].isin(skipped_hashes_total)]
            dup_df_str = dup_df.astype(str).to_dict(orient="records")

            error_details = {
                "inserted_count": len(inserted_hashes),
                "skipped_count": len(skipped_hashes_total),
                "skipped_transaction_hashes": list(skipped_hashes_total),
                "duplicate_in_input": list(duplicate_in_input_total),
                "missing_transaction_hash_rows": missing_hash_rows_total,
                "skipped_rows_data": dup_df_str
            }

            from fastapi.responses import JSONResponse
            from starlette.status import HTTP_207_MULTI_STATUS

            return JSONResponse(
                status_code=HTTP_207_MULTI_STATUS,
                content={
                    "error": "Some rows were not inserted.",
                    "details": error_details
                }
            )


        print("check_duplicates_and_save end:", len(df))

        return {
            "inserted_count": len(inserted_hashes),
            "skipped_count": len(skipped_hashes_total),
            "skipped_transaction_hashes": list(skipped_hashes_total)
        }

        
    @staticmethod
    @log_execution_time
    async def save_gl_data_to_db(
        df: pd.DataFrame,
        column_mapping: Optional[Dict[str, str]],
        uploaded_user_id: str,
        db
    ):
        try:
            df = AddGLData.validate_and_rename_columns(df, column_mapping)
            AddGLData.validate_row_nulls(df)
            await AddGLData.validate_master_gl_rows(df, db)
            
            result = await AddGLData.check_duplicates_and_save(
                df,
                db,
                source_identifier="file",
                uploaded_user_id=uploaded_user_id
            )

            # ✅ Check if it's a JSONResponse (partial failure case)
            if isinstance(result, JSONResponse):
                return result

            print(result)
            if result.get("inserted_count", 0) > 0:
                await FinancialWidgetManagement.set_widget(db)

            return {
                "detail": f"Successfully uploaded {result['inserted_count']} entries. Duplicated: {result['skipped_count']}, Duplicates: {result['skipped_transaction_hashes']}"
            }

        except HTTPException as e:
            print("HTTPException:", e)
            raise e
        except Exception as e:
            print("Unhandled Exception:", e)
            raise HTTPException(status_code=500, detail=f"Unexpected error: {str(e)}")



class FinancialManagement(CompanyService):
    @staticmethod
    async def add_gl_master_account(GLData: GeneralLedgerMasterDataRequest, db: AsyncSession):
        """
        Process a list of master GL account entries:
        - Skip accounts that already exist in the database
        - Create new master GL accounts

        Arguments:
            - GLData: List of general_ledger_code, account_name, category, and optional tags
            - db: Async database session

        Returns:
            - message
            - list of created account UUIDs
            - list of skipped accounts with reasons
        """

        created_accounts = []
        skipped_accounts = []

        for entry in GLData.data:
            # Check for existing GL code
            result = await db.execute(
                select(GeneralLedgerMaster).where(
                    GeneralLedgerMaster.general_ledger_code == entry.general_ledger_code
                )
            )
            existing = result.scalars().first()

            if existing:
                skipped_accounts.append({
                    "general_ledger_code": entry.general_ledger_code,
                    "reason": "GL code already exists"
                })
                continue

            gl_account = GeneralLedgerMaster(
                id=uuid.uuid4(),
                general_ledger_code=entry.general_ledger_code,
                account_name=to_camel_case(entry.account_name),
                category=to_camel_case(entry.category),
                tags=entry.tags 
            )

            db.add(gl_account)
            created_accounts.append(gl_account)
            await db.commit()

        # Refresh all newly created accounts
        for account in created_accounts:
            await db.refresh(account)

        return {
            "message": f"{len(created_accounts)} GL account(s) created successfully. {len(skipped_accounts)} skipped.",
            "created_ids": [str(account.id) for account in created_accounts],
            "skipped": skipped_accounts
        }
    
    @staticmethod
    async def update_gl_master_account(general_ledger_code: int,update_data: GlMasterData,db: AsyncSession):
        trigger_gl_update = False
        original_code = general_ledger_code

        # Fetch the existing GL master record with a row-level lock
        result = await db.execute(
            select(GeneralLedgerMaster)
            .where(GeneralLedgerMaster.general_ledger_code == general_ledger_code)
            .with_for_update()
        )
        existing = result.scalars().first()

        print("-----------------------------**")


        if not existing:
            raise HTTPException(status_code=404, detail="GL account not found.")

        # Check if no change is requested
        if not any([
            update_data.account_name,
            update_data.category,
            update_data.tags,
            (update_data.general_ledger_code and update_data.general_ledger_code != general_ledger_code)
        ]):
            return {
                "message": "No changes detected.",
                "record": existing.as_dict()
            }

        print("-----------------------------")

        # Uniqueness check and apply GL code update
        if update_data.general_ledger_code and update_data.general_ledger_code != general_ledger_code:
            duplicate_check = await db.execute(
                select(GeneralLedgerMaster).where(
                    GeneralLedgerMaster.general_ledger_code == update_data.general_ledger_code
                )
            )
            if duplicate_check.scalars().first():
                raise HTTPException(status_code=400, detail="Updating general_ledger_code already exists.")

            existing.general_ledger_code = update_data.general_ledger_code
            trigger_gl_update = True

        # Update other fields
        if update_data.account_name is not None:
            existing.account_name = to_camel_case(update_data.account_name)
            trigger_gl_update = True

        if update_data.category is not None:
            existing.category = to_camel_case(update_data.category)
            trigger_gl_update = True

        if update_data.tags is not None:
            existing.tags = [tag.strip().lower() for tag in update_data.tags if tag.strip()]

        # Sync changes to GeneralLedger table
        old_gl_rows = []
        new_gl_rows = []
        if trigger_gl_update:
        # Fetch old rows
            old_gl_result = await db.execute(
                select(GeneralLedger)
                .where(GeneralLedger.general_ledger_code == original_code)
            )
            old_gl_rows = [row.as_dict() for row in old_gl_result.scalars().all()]

            # Run update
            update_result = await db.execute(
                update(GeneralLedger)
                .where(GeneralLedger.general_ledger_code == original_code)
                .values(
                    account_name=existing.account_name,
                    category=existing.category,
                    general_ledger_code=existing.general_ledger_code
                )
            )
            affected_rows = update_result.rowcount
        
        await db.commit()

        # Fetch updated rows (new values)
        new_gl_result = await db.execute(
            select(GeneralLedger)
            .where(GeneralLedger.general_ledger_code == existing.general_ledger_code)
        )
        new_gl_rows = [row.as_dict() for row in new_gl_result.scalars().all()]
        
        return {
        "message": "GL account updated successfully.",
        # "affected_rows_in_gl_data": affected_rows,
        # "general_ledger_updates": {
        #     "old_rows": old_gl_rows,
        #     "new_rows": new_gl_rows
        # }
    }


    @staticmethod
    async def get_all_master_gl_codes(db:AsyncSession):

        stmt = select(GeneralLedgerMaster)
        result = await db.execute(stmt)
        master_gl_info = result.scalars().all()
        
        return {
            "master_gl_info": master_gl_info,
            "count": len(master_gl_info)
        }
    
    @staticmethod
    async def delete_master_gl_account(gl_code:int,db:AsyncSession):

        # Fetch the account by GL code
        result = await db.execute(
            select(GeneralLedgerMaster).where(
                GeneralLedgerMaster.general_ledger_code == gl_code
            )
        )
        account = result.scalars().first()

        if not account:
            raise HTTPException(status_code=404, detail=f"GL account with code '{gl_code}' not found.")

        # Delete the account
        await db.delete(account)
        await db.commit()

        return {"message": f"GL account with code '{gl_code}' has been deleted successfully."}
    
    @staticmethod
    def file_gl_upload_background(contents, column_mapping, user_id,db:AsyncSession):
        df = pd.read_excel(
            BytesIO(contents),
            engine="calamine"
        )
        asyncio.run(AddGLData.save_gl_data_to_db(df, column_mapping, user_id,db))
   

    @staticmethod
    async def get_gl_entries(stmt,page:int,limit:int,db:AsyncSession):
        # Count total for pagination
        offset = (page - 1) * limit

        count_stmt = select(func.count()).select_from(stmt.subquery())
        total_result = await db.execute(count_stmt)
        total = total_result.scalar_one()
        print(total)
        # Apply pagination
        stmt = stmt.order_by(GeneralLedger.date.desc()).offset(offset).limit(limit)
        result = await db.execute(stmt)
        entries = result.scalars().all()
        #extra = entry.extra_data if isinstance(entry.extra_data, dict) else {}
    
        # Merge extra_data into base dict
        for entry in entries:
    

            extra = entry.extra_data if isinstance(entry.extra_data, dict) else {}
            #merged = {**base, **extra}

            print(extra)
        #print(entries.extra)
        total_pages = (total + limit - 1) // limit
        has_next = page * limit < total

        return {
            "page": page,
            "limit": limit,
            "total": total,
            "total_pages": total_pages,
            "has_next": has_next,
            "data": [entry.as_dict() for entry in entries]
            
        }
    
    @staticmethod
    async def delete_gl_entries_by_code_and_date(gl_code:int,date_from,date_to,db:AsyncSession):
         #filters
        
        stmt = select(GeneralLedger).where(
            GeneralLedger.general_ledger_code == gl_code,
            GeneralLedger.date >= date_from,
            GeneralLedger.date <= date_to
        )
        
        result = await db.execute(stmt)
        entries = result.scalars().all()

        if not entries:
            raise HTTPException(status_code=404, detail="No GL entries found for given criteria.")

        for entry in entries:
            await db.delete(entry)

        await db.commit()

        return {
            "message": f"Deleted {len(entries)} GL entries with code {gl_code} between {date_from.isoformat()} and {date_to.isoformat()}"
    }

    @staticmethod
    async def delete_gl_entry_by_id(entry_id:UUID,db:AsyncSession):
        
        stmt = select(GeneralLedger).where(GeneralLedger.id == entry_id)
        result = await db.execute(stmt)
        entry = result.scalars().first()

        if not entry:
            raise HTTPException(status_code=404, detail=f"No GL entry found with id: {entry_id}")

        await db.delete(entry)
        await db.commit()

        return {"message": f"Deleted GL entry with id {entry_id}"}
    
    @staticmethod
    async def update_gl_entry(
        entry_id: UUID,
        update_dict: dict,
        db: AsyncSession
    ):
        # Fetch the entry
        stmt = select(GeneralLedger).where(GeneralLedger.id == entry_id)
        result = await db.execute(stmt)
        entry = result.scalars().first()

        if not entry:
            raise HTTPException(status_code=404, detail=f"No GL entry found with id: {entry_id}")

        # Update fields dynamically
        for field, value in update_dict.items():
            if field == "extra_data":
                current_extra = getattr(entry, "extra_data", {}) or {}
                if isinstance(value, dict):
                    merged = {**current_extra, **value}
                    setattr(entry, "extra_data", merged)
                else:
                    setattr(entry, "extra_data", value)

            elif field == "date" and isinstance(value, datetime):
                # If datetime is timezone aware, convert to naive
                if value.tzinfo is not None:
                    value = value.astimezone(tz=None).replace(tzinfo=None)
                setattr(entry, field, value)

            elif hasattr(entry, field):
                setattr(entry, field, value)

        # Save changes
        try:
            db.add(entry)
            await db.commit()
            await db.refresh(entry)
        except Exception as e:
            print("error ",e)


        return {
            "message": f"GL entry with id {entry_id} updated successfully.",
            "updated_entry": entry.as_dict()
        }

    async def get_integration_history(page:int,limit:int,db: AsyncSession):
        offset = (page - 1) * limit

        # Count total records
        total_result = await db.execute(select(IntegrationHistory))
        total_records = len(total_result.scalars().all())

        # Fetch paginated data
        result = await db.execute(
            select(IntegrationHistory)
            .order_by(IntegrationHistory.created_at.desc())
            .offset(offset)
            .limit(limit)
        )
        records = result.scalars().all()
        has_next = page * limit < total_records
        return {
            "page": page,
            "limit": limit,
            "total_pages": total_records,
            "has_next":has_next,
            "records": [record.as_dict() for record in records]
        }
    
    async def get_gl_entries_by_integration_id(
            integration_id: str,
            page: int,
            limit: int,
            date: datetime,
            date_from: datetime,
            date_to: datetime,
            general_ledger_code: int,
            account_name:str,
            debit_min:str,
            debit_max:str,
            credit_min: str,
            credit_max:str,
            db:AsyncSession
    ):
        
        offset = (page - 1) * limit

        filters = [GeneralLedger.integration_id == integration_id]

        if date:
            filters.append(GeneralLedger.date == date)
        if date_from:
            filters.append(GeneralLedger.date >= date_from)
        if date_to:
            filters.append(GeneralLedger.date <= date_to)
        if general_ledger_code:
            filters.append(GeneralLedger.general_ledger_code == general_ledger_code)
        if account_name:
            filters.append(GeneralLedger.account_name.ilike(f"%{account_name}%"))
        if debit_min is not None:
            filters.append(GeneralLedger.debit >= debit_min)
        if debit_max is not None:
            filters.append(GeneralLedger.debit <= debit_max)
        if credit_min is not None:
            filters.append(GeneralLedger.credit >= credit_min)
        if credit_max is not None:
            filters.append(GeneralLedger.credit <= credit_max)

        # Get total filtered count
        total_result = await db.execute(
            select(func.count()).where(and_(*filters))
        )
        total_records = total_result.scalar()

        if total_records == 0:
            raise HTTPException(status_code=404, detail="No GL entries found with the given filters.")

        # Fetch filtered entries for current page
        result = await db.execute(
            select(GeneralLedger)
            .where(and_(*filters))
            .offset(offset)
            .limit(limit)
        )
        entries = result.scalars().all()

        return {
            "page": page,
            "limit": limit,
            "current_records": len(entries),
            "total_records": total_records,
            "has_next_page": (offset + limit) < total_records,
            "data": [entry.as_dict() for entry in entries]
        }
    
    async def delete_integration_history(integration_id: str, db: AsyncSession):

        # Fetch the integration record
        result = await db.execute(
            select(IntegrationHistory).where(IntegrationHistory.id == integration_id)
        )
        integration_record = result.scalar_one_or_none()

        if not integration_record:
            raise HTTPException(status_code=404, detail="IntegrationHistory not found.")
        print("result",result)
        # Delete the integration (GL entries will be deleted via cascade)
        await db.delete(integration_record)

        try:
            await db.commit()
            return {"detail": f"Integration record {integration_id} and its related GL entries have been deleted."}
        except Exception as e:
            await db.rollback()
            raise HTTPException(status_code=500, detail=f"Failed to delete integration: {str(e)}")

class FinancialWidgetManagement:
    @staticmethod
    async def set_widget(db: AsyncSession):
        
            position=1

            
            
           
            stmt = select(FinancialWidget)
            result = await db.execute(stmt)
            row = result.first()

            if  not row:
                for name in FinancialDashboardWidgetNameEnum:
                    new_widget = FinancialWidget(
                    name=name.value,
                    category="financial dashboard",
                    position=position
                    )
                    db.add(new_widget)
                    #widgets.append( WidgetSchema(Name=category.value,category="financial dashboard", Position=position))
                    position += 1
                for name in KPIWidgetNameEnum:
        
                    new_widget = FinancialWidget(
                    name=name,
                    category="KPI dashboard",
                    position=position
                    )
                    db.add(new_widget)
                    #widgets.append( WidgetSchema(Name=category.value,category="financial dashboard", Position=position))
                    position += 1
                try:
                    await db.commit()
                    print(" Inserted successfully.")
                except IntegrityError:
                    await db.rollback()
                    print(" Integrity error: possibly a duplicate.")

    @staticmethod
    def widget_to_dict(widget):
        """
Converts a FinancialWidget SQLAlchemy model instance into a dictionary format.

Parameters:
    widget (FinancialWidget): The widget object to convert.

Returns:
    dict: The widget's data as a dictionary, suitable for serialization.
"""

        return {
        "id": widget.id,
        "name": widget.name,
        "category": widget.category,
        "Type": widget.chart_Type.value if widget.chart_Type else None,
        "position": widget.position,
        "enable_or_disable": widget.enable_or_disable,
        "created_at": widget.created_at.isoformat() if widget.created_at else None,
        "updated_at": widget.updated_at.isoformat() if widget.updated_at else None,
    }
    @staticmethod       
    async def getwidgets(db: AsyncSession , year, actual_vs_budget):
        """
Fetches all widget records from the database and groups them by their category.

Returns:
    dict: A mapping of categories to lists of widgets under each category."""
        
        try:
            stmt = select(FPNAYearlyReport1).where(
                FPNAYearlyReport1.year == year
            )

            result = await db.execute(stmt)
            report = result.scalar_one_or_none()

            if not report:
                raise HTTPException(status_code=404, detail="Year not found")

            metric_trend = report.metric_trend or []

            req_list = []

            for trend_obj in metric_trend:

                for trend_name, trend_values in trend_obj.items():

                    if not isinstance(trend_values, list):
                        continue

                    # find metadata inside trend list
                    for item in trend_values:
                        if isinstance(item, dict) and "enableordisable" in item:

                            if item["enableordisable"] is False:
                                req_list.append({trend_name: trend_values})

                            break  


            actual_vs_budget_list = []
            other_trends = []

            for trend in req_list:
                if "actual_vs_budget_trend" in trend:
                    actual_vs_budget_list.append(trend)
                else:
                    other_trends.append(trend)

    
            if actual_vs_budget == "actual_vs_budget":
                return actual_vs_budget_list
            else:
                return other_trends

        except Exception as e:
            import sys
            line_no = sys.exc_info()[-1].tb_lineno
            print(line_no, str(e))
            raise e


        # try:
        #     stmt = select(FPNAYearlyReport1).where(FPNAYearlyReport1.year == year)
        #     result = await db.execute(stmt)
        #     report = result.scalar_one_or_none() 

        #     req_list = []

        #     trend_fields = [
        #         "revenue_trend",
        #         "expense_trend",
        #         "gross_profit_trend",
        #         "net_profit_trend",
        #         "actual_vs_budget_trend"
        #     ]

        #     for trend in trend_fields:
        #         field_values = getattr(report, trend, None)

        #         if not field_values:
        #             continue

        #         for item in field_values:
        #             if isinstance(item, dict) and "enableordisable" in item:

        #                 if item["enableordisable"] == False:
        #                     req_list.append({trend:field_values}) 
        #     print(req_list , "req_list==============================")
        #     actual_vs_budget_list = []
        #     other_trends = []

        #     for trend in req_list:
        #         if "actual_vs_budget_trend" in trend:
        #             actual_vs_budget_list.append(trend)
        #         else:
        #             other_trends.append(trend)

        #     if  actual_vs_budget == "actual_vs_budget":
        #         return actual_vs_budget_list
        #     else:
        #         return other_trends

            # grouped = defaultdict(list)
            # data=[FinancialWidgetManagement.widget_to_dict(w) for w in widget]
            # for item in data:
            #     grouped[item["category"]].append(item)

            # # grouped is a dict where key = category, value = list of items

            # # Example: print each category and its items
            # for category, items in grouped.items():
            #     print(f"Category: {category}")
            #     for i in items:
            #         print(f"  - {i['name']}")
    
            # return {"message": grouped}
        except HTTPException as e:
            raise e
    @staticmethod
    async def UpdateWidget(widget ,db):
        """
        Update widget data in the database.

        Parameters:
            new_widget (WidgetSchema): Pydantic schema containing the updated widget information.

        Returns:
            dict: JSON response with a success message.
        """

        def get_trend_meta(trend_list: list) -> dict:
            """
            Each trend list contains one meta dict with Position, chart_type, enable_or_disable
            """
            try:
                for item in trend_list:
                    if isinstance(item, dict) and "position" in item:
                        return item
            except Exception as e:
                line_no = sys.exc_info()[-1].tb_lineno
                print(line_no , str(e))
                raise ValueError("Trend metadata not found")

        # def update_trend_config(report, payload):
        #     try:
        #         target_trend = payload.name.lower()

        #         trend_fields = [
        #             "revenue_trend",
        #             "expense_trend",
        #             "gross_profit_trend",
        #             "net_profit_trend" , 
        #             "actual_vs_budget_trend"]

        #         if target_trend not in trend_fields:
        #             raise ValueError(f"{target_trend} is not a valid trend")

        #         new_position = payload.position
        #         new_chart_type = payload.chart_Type.value if payload.chart_Type else None
        #         new_enable = payload.enableordisable

        #         trend_metas = {}

        #         for trend in trend_fields:
        #             print(trend)
        #             trend_list = getattr(report, trend) 
        #             if trend_list:
        #                 trend_metas[trend] = get_trend_meta(trend_list)

        #         target_meta = trend_metas[target_trend]
  
        #         old_position = target_meta["position"]

        #         if new_position is not None and new_position != old_position:
        #             for trend_name, meta in trend_metas.items():
        #                 if trend_name != target_trend and meta["position"] == new_position:
        #                     meta["position"] = old_position
        #                     flag_modified(report, trend_name)   # 🔥 REQUIRED
        #                     break

        #             target_meta["position"] = new_position
        #             flag_modified(report, target_trend)      # 🔥 REQUIRED

        #         # 🎨 update only selected trend
        #         if new_chart_type is not None:
        #             target_meta["chart_type"] = new_chart_type
        #             flag_modified(report, target_trend)

        #         if new_enable is not None:
        #             target_meta["enableordisable"] = new_enable
        #             flag_modified(report, target_trend)
        #     except Exception as e:
        #         import sys
        #         line_no = sys.exc_info()[-1].tb_lineno 
        #         print(line_no , str(e))
        #         raise e

        
        def update_trend_config(report, payload):

            try:
                target_trend = payload.name.lower()
                new_position = payload.position
                new_chart_type = payload.chart_Type.value if payload.chart_Type else None
                new_enable = payload.enableordisable

                metric_trend = report.metric_trend or []

                trend_map = {}
                trend_metas = {}

                for item in metric_trend:
                    for key, value in item.items():
                        trend_map[key.lower()] = item 

                        # find metadata inside value
                        if isinstance(value, list):
                            for entry in value:
                                if isinstance(entry, dict) and "position" in entry:
                                    trend_metas[key.lower()] = entry
                                    break

                if target_trend not in trend_map:
                    raise ValueError(f"{target_trend} not found in metric_trend")

                target_meta = trend_metas.get(target_trend)

                if not target_meta:
                    raise ValueError("Trend metadata not found")

                old_position = target_meta.get("position")

                # ---------------------------------------
                # Step 2: Handle Position Swap
                # ---------------------------------------
                if new_position is not None and new_position != old_position:

                    for trend_name, meta in trend_metas.items():
                        if trend_name != target_trend and meta.get("position") == new_position:
                            meta["position"] = old_position
                            break

                    target_meta["position"] = new_position

                # ---------------------------------------
                # Step 3: Update Chart Type
                # ---------------------------------------
                if new_chart_type is not None:
                    target_meta["chart_type"] = new_chart_type

                # ---------------------------------------
                # Step 4: Update Enable/Disable
                # ---------------------------------------
                if new_enable is not None:
                    target_meta["enableordisable"] = new_enable

                # ---------------------------------------
                # Step 5: Mark JSON field modified
                # ---------------------------------------
                flag_modified(report, "metric_trend")

            except Exception as e:
                import sys
                line_no = sys.exc_info()[-1].tb_lineno
                print(line_no, str(e))
                raise e

        try:
            stmt = select(FPNAYearlyReport1).where(FPNAYearlyReport1.year == widget.year)
            result = await db.execute(stmt)
            report = result.scalar_one_or_none()

            if not report:
                raise HTTPException(404, detail="Year not found")

            update_trend_config(report, widget)

            await db.commit()
            await db.refresh(report)

            return report 

        except Exception as e:
            import sys , os
            line_no = sys.exc_info()[-1].tb_lineno
            print(line_no , "tttttttttttttttttttttttttt")
            raise e


     

        
  
from datetime import datetime, timezone


from decimal import Decimal
import decimal
from io import BytesIO
import math
import uuid
from fastapi import HTTPException
import pandas as pd
from sqlalchemy import UUID, and_, extract, func, insert, select, text, tuple_, update,desc
from sqlalchemy.ext.asyncio import AsyncSession
from src.financial.constants import COLUMN_ALIASES
from src.Service.companyService import CompanyService
from src.constants import DataIntegrationSourceEnum,FinancialDashboardWidgetNameEnum,KPIWidgetNameEnum, ChartTypeEnum
from src.financial.utils import clean_df_for_json, make_merge_key, to_camel_case
from src.financial.model import Attribute, Entity, FileUploadHistory, GeneralLedger, GeneralLedgerMaster, IntegrationHistory,FinancialWidget, TrackedJSONKey, Value
from src.financial.schemas import GeneralLedgerEntryUpdateRequest, GlMasterData,GeneralLedgerMasterDataRequest,WidgetSchema
from sqlalchemy.exc import IntegrityError
from collections import OrderedDict, defaultdict, namedtuple
from typing import Tuple, Optional

from .utils import make_row_hash
import re 
import numpy as np 
from typing import List,Dict,Optional, Set
import logging

# Setup logger
log_filename = "gl_upload_log.txt"  # Define log file location
logging.basicConfig(
    filename=log_filename,  # File to write logs
    level=logging.DEBUG,  # Log all levels of messages
    format='%(asctime)s - %(levelname)s - %(message)s'
)


from datetime import datetime, timedelta
from decimal import Decimal
from typing import Optional, Dict, List, Any, Tuple, Set
from collections import defaultdict
from sqlalchemy import select, func, extract, and_, or_
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import HTTPException
import logging

logger = logging.getLogger(__name__)


class FirstFinancialReport:
    """Class for generating various financial reports from general ledger data."""
    
    BUDGET_FACTORS = {
        "revenue": Decimal("0.9"),   # 10% lower than actual
        "costs": Decimal("0.95"),    # 5% lower than actual
        "operational": Decimal("0.95"),
        "investing": Decimal("0.90"),
        "financing": Decimal("0.85"),

    }

      # Constants for tag categories
    ASSET_TAGS = {"current_asset", "fixed_asset"}
    LIABILITY_TAGS = {"current_liability"}
    EQUITY_TAGS = {"equity","financial"}
    DEPRECIATION_TAGS = {"non_cash expense", "operating_expense"}
    REVENUE_TAGS = {"income", "revenue"}
    OPERATING_EXPENSE_TAGS = {"operating_expense"}
    PL_CATEGORIES = {"revenue", "cogs", "expense"}

    COGS_TAGS = {"cogs", "cost_of_goods_sold", "cost_of_sales"}
    OPERATING_CASHIN={"operating","cash_in"}
    OPERATING_CASHOUT={"operating","cash_out"}
    INVESTING_CASHIN={"investing","cash_in"}
    INVESTING_CASHOUT={"investing","cash_out"}
    FINANCING_CASHIN={"financing","cash_in"}
    FINANCING_CASHOUT={"financing","cash_out"}
    MONTH_NAMES = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", 
                  "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]

    @staticmethod
    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")

    @staticmethod
    async def _execute_query(db: AsyncSession, stmt) -> List:
        """Execute a database query with error handling"""
        try:
            result = await db.execute(stmt)
            return result.all()
        except Exception as e:
            logger.error(f"Database query failed: {e}")
            raise FinancialReportError(f"Database query failed: {e}")

    @staticmethod
    def _parse_tags(tags: Union[str, List, None]) -> Set[str]:
        """Parse tags from various formats into a set of normalized strings"""
        if tags is None:
            return set()
            
        if isinstance(tags, str):
            # Handle string format like "{tag1, tag2, tag3}"
            if tags.startswith('{') and tags.endswith('}'):
                tags = tags[1:-1]
            tags_list = [t.strip().lower() for t in tags.split(",")]
        elif isinstance(tags, list):
            tags_list = [str(t).strip().lower() for t in tags]
        else:
            return set()
            
        return {tag for tag in tags_list if tag}

    @staticmethod
    def _tag_match(tags: Union[str, List, None], target_tags: Set[str]) -> bool:
        """Check if any of the tags match the target tags"""
        if not tags or not target_tags:
            return False
            
        parsed_tags = FinancialReport._parse_tags(tags)
        return bool(parsed_tags & target_tags)

    @staticmethod
    async def get_account_mapping(db: AsyncSession) -> pd.DataFrame:
        """Get account mapping from GeneralLedgerMaster"""
        try:
            stmt = select(
                func.lower(GeneralLedgerMaster.account_name).label("account_name"),
            func.lower(GeneralLedgerMaster.tags).label("tags"),
                GeneralLedgerMaster.general_ledger_code
            )
            rows = await FinancialReport._execute_query(db, stmt)
            
            mapping = pd.DataFrame(rows, columns=["account_name", "tags", "generalledgercode"])
            mapping["account_name"] = mapping["account_name"].str.strip().str.lower()
            return mapping.set_index("account_name")
        except Exception as e:
            logger.error(f"Failed to get account mapping: {e}")
            raise FinancialReportError(f"Failed to get account mapping: {e}")

    @staticmethod
    def get_accounts_by_tags(
        mapping_df: pd.DataFrame, 
        tag_names: Set[str], 
        tag_column: str = "tags"
    ) -> List[str]:
        """
        Returns account GL codes from a DataFrame where tags match the given tag_names set.

        Parameters:
            mapping_df: Input DataFrame with account mapping
            tag_names: Tags to match against
            tag_column: Column name in df containing tag strings

        Returns:
            List of matching GL codes
        """
        try:
            if mapping_df.empty:
                return []
                
            target_tags = {tag.lower().strip() for tag in tag_names}
            
            # Filter rows where tags match any of the target tags
            matched = mapping_df[mapping_df[tag_column].apply(
                lambda tags: FinancialReport._tag_match(tags, target_tags)
            )]
            
            return matched["generalledgercode"].tolist()
        except Exception as e:
            logger.error(f"Failed to get accounts by tags: {e}")
            raise FinancialReportError(f"Failed to get accounts by tags: {e}")

    @staticmethod
    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:
                FinancialReport._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}")

    @staticmethod
    async def load_ledger_data(
        db: AsyncSession, 
        start_date: datetime, 
        end_date: datetime,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> pd.DataFrame:
        """Load ledger data efficiently using SQL aggregation"""
        try:
            logger.debug(f"Loading ledger data from {start_date} to {end_date}")
            
            # Build filters
            filters = [
                GeneralLedger.date >= start_date,
                GeneralLedger.date <= end_date
            ]
            
            if entity:
                filters.append(GeneralLedger.entity.ilike(f"%{entity}%"))
            if hierarchy:
                filters.append(GeneralLedger.hierarchy.ilike(f"%{hierarchy}%"))
            
            # Use SQL aggregation to reduce data transfer
            stmt = select(
                GeneralLedger.account_name,
                func.sum(GeneralLedger.debit).label('total_debit'),
                func.sum(GeneralLedger.credit).label('total_credit'),
                func.count().label('entry_count')
            ).where(
                and_(*filters)
            ).group_by(
                GeneralLedger.account_name
            )
            
            rows = await FinancialReport._execute_query(db, stmt)
            
            # Convert to pandas DataFrame
            df = pd.DataFrame(rows, columns=['account_name', 'debit', 'credit', 'entry_count'])
            df['account_name'] = df['account_name'].str.strip().str.lower()
            
            logger.debug(f"Loaded {len(df)} aggregated ledger entries")
            return df
        except Exception as e:
            logger.error(f"Failed to load ledger data: {e}")
            raise FinancialReportError(f"Failed to load ledger data: {e}")

    @staticmethod
    def filter_entity_and_group(
        df: pd.DataFrame, 
        entity: Optional[str] = None, 
        hierarchy_level: str = "group", 
        filters: Optional[dict] = None
    ) -> pd.DataFrame:
        """
        Filter raw data by entity (if given), then apply hierarchy grouping and other filters.

        Args:
            df: raw data DataFrame (with extra_data column)
            entity: optional entity filter (filters extra_data["entity"])
            hierarchy_level: 'group', 'department', or 'product'
            filters: dict of other filters to apply during grouping (optional)

        Returns:
            Filtered and grouped DataFrame
        """
        try:
            if df.empty:
                return df

            # Step 1: Filter by entity if provided
            if entity and "extra_data" in df.columns:
                df = df[df["extra_data"].apply(lambda x: x.get("entity") == entity if x else False)]

            # Step 2: Apply other filters except entity
            if filters:
                for key, value in filters.items():
                    if value is None or key == "entity":
                        continue
                    if key in df.columns:
                        df = df[df[key] == value]
                    elif "extra_data" in df.columns:
                        df = df[df["extra_data"].apply(lambda x: x.get(key) == value if x else False)]

            # Step 3: Group by hierarchy level
            if hierarchy_level == "group":
                return df
            elif hierarchy_level == "department":
                if "extra_data" not in df.columns:
                    return df
                    
                df["department"] = df["extra_data"].apply(lambda x: x.get("department") if x else None)
                grouped = df.groupby("department", dropna=False).agg({
                    "debit": "sum",
                    "credit": "sum"
                }).reset_index()
                return grouped
            elif hierarchy_level == "product":
                if "extra_data" not in df.columns:
                    return df
                    
                def get_product(x):
                    if not x:
                        return None
                    return x.get("product_code") or x.get("project_code") or None

                df["product"] = df["extra_data"].apply(get_product)
                grouped = df.groupby("product", dropna=False).agg({
                    "debit": "sum",
                    "credit": "sum"
                }).reset_index()
                return grouped
            else:
                raise ValueError("Invalid hierarchy_level. Must be one of 'group', 'department', or 'product'.")
        except Exception as e:
            logger.error(f"Failed to filter and group data: {e}")
            raise FinancialReportError(f"Failed to filter and group data: {e}")

    @staticmethod
    async def _fetch_gl_data_by_codes(
        db_session: AsyncSession,
        gl_codes: List[str],
        start_date: datetime,
        end_date: datetime
    ) -> pd.DataFrame:
        """Fetch GL data for specific GL codes within a date range"""
        try:
            if not gl_codes:
                return pd.DataFrame()

            stmt = select(
                GeneralLedger.account_name,
                GeneralLedger.debit,
                GeneralLedger.credit,
                GeneralLedger.date,
                GeneralLedger.general_ledger_code,
                GeneralLedger.extra_data
            ).where(
                and_(
                    GeneralLedger.general_ledger_code.in_(gl_codes),
                    GeneralLedger.date >= start_date,
                    GeneralLedger.date <= end_date
                )
            )

            rows = await FinancialReport._execute_query(db_session, stmt)

            if not rows:
                return pd.DataFrame()

            return pd.DataFrame(
                rows, 
                columns=["account_name", "debit", "credit", "date", "general_ledger_code", "extra_data"]
            )
        except Exception as e:
            logger.error(f"Failed to fetch GL data by codes: {e}")
            raise FinancialReportError(f"Failed to fetch GL data by codes: {e}")

    @staticmethod
    async def _fetch_categorized_data(
        db: AsyncSession,
        category_mapping: Dict[str, List[str]],
        start_date: datetime,
        end_date: datetime
    ) -> pd.DataFrame:
        """Fetch data for multiple categories"""
        try:
            dfs = []
            
            for category, gl_codes in category_mapping.items():
                if not gl_codes:
                    continue
                    
                df = await FinancialReport._fetch_gl_data_by_codes(db, gl_codes, start_date, end_date)
                if not df.empty:
                    df["category"] = category
                    dfs.append(df)

            if dfs:
                return pd.concat(dfs, ignore_index=True)
            else:
                return pd.DataFrame()
        except Exception as e:
            logger.error(f"Failed to fetch categorized data: {e}")
            raise FinancialReportError(f"Failed to fetch categorized data: {e}")

    @staticmethod
    def _calculate_amount(row: pd.Series) -> float:
        """Calculate amount based on assigned category and debit/credit values."""
        try:
            debit = float(row.get("debit", 0) or 0)
            credit = float(row.get("credit", 0) or 0)
            category = str(row.get("category", "")).strip().lower()

            if category in {"operating_cashin", "investing_cashin", "financing_cashin"}:
                return credit - debit  # Inflow
            elif category in {"operating_cashout", "investing_cashout", "financing_cashout"}:
                return debit - credit  # Outflow
            elif category == "depreciation":
                return -(debit - credit)
            elif category == "asset":
                return debit - credit
            elif category in {"liability", "equity"}:
                return credit - debit
            elif category == "revenue":
                return credit - debit
            elif category in {"operating_expense", "expense"}:
                return debit - credit
            else:
                return 0.0
        except Exception as e:
            logger.error(f"Failed to calculate amount: {e}")
            return 0.0

    @staticmethod
    def _aggregate_data(
        df: pd.DataFrame, 
        freq: str, 
        key_format: str
    ) -> Dict[str, List[Dict[str, Any]]]:
        """Aggregate data by time period"""
        try:
            if df.empty:
                return {}
                
            grouped = df.groupby([pd.Grouper(freq=freq), "category", "account_name"])["amount"].sum().reset_index()
            grouped.rename(columns={"date": "period"}, inplace=True)

            result = defaultdict(list)
            for _, row in grouped.iterrows():
                key = row["period"].strftime(key_format)
                result[key].append({
                    "category": row["category"],
                    "account_name": row["account_name"],
                    "amount": float(row["amount"])
                })
                
            return dict(result)
        except Exception as e:
            logger.error(f"Failed to aggregate data: {e}")
            raise FinancialReportError(f"Failed to aggregate data: {e}")
    
    @staticmethod
    async def balance_sheet(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None,
        compare_previous: bool = True
    ) -> Dict[str, Any]:
        """Generate balance sheet report"""
        try:
            # 1. Set date range for current and previous periods
            current_start, current_end = FinancialReport.get_date_range(
                period, year, start_date, end_date
            )
            previous_start, previous_end = FinancialReport.get_date_range(
                period, year - 1 if year else None, previous_start, previous_end
            )

            FinancialReport._validate_date_range(current_start, current_end)
            FinancialReport._validate_date_range(previous_start, previous_end)

            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)

            # 3. Get accounts by category
            accounts = {
                "asset": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.ASSET_TAGS),
                "liability": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.LIABILITY_TAGS),
                "equity": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.EQUITY_TAGS),
                "depreciation": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.DEPRECIATION_TAGS),
            }

            # 4. Fetch data for current and previous periods
            current_df_raw = await FinancialReport._fetch_categorized_data(db, accounts, current_start, current_end)
            previous_df_raw = await FinancialReport._fetch_categorized_data(db, accounts, previous_start, previous_end)

            if current_df_raw.empty or previous_df_raw.empty:
                return {
                    "monthly_data": {},
                    "quarterly_data": {},
                    "yearly_data": {},
                    "previous_monthly_data": {},
                    "previous_quarterly_data": {},
                    "previous_yearly_data": {},
                    "key_metrics": {},
                    "date_range": {
                        "start_date": current_start.isoformat(),
                        "end_date": current_end.isoformat()
                    }
                }

            # 5. Filter and process current data
            current_df_filtered = FinancialReport.filter_entity_and_group(
                df=current_df_raw,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=None
            )
            current_df = pd.DataFrame(current_df_filtered, columns=["account_name", "debit", "credit", "date", "category"])
            current_df["date"] = pd.to_datetime(current_df["date"])
            current_df["account_name"] = current_df["account_name"].str.strip().str.lower()
            current_df["amount"] = current_df.apply(FinancialReport._calculate_amount, axis=1)
            current_df.set_index("date", inplace=True)

            # 6. Filter and process previous data
            previous_df_filtered = FinancialReport.filter_entity_and_group(
                df=previous_df_raw,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=None
            )
            previous_df = pd.DataFrame(previous_df_filtered, columns=["account_name", "debit", "credit", "date", "category"])
            previous_df["date"] = pd.to_datetime(previous_df["date"])
            previous_df["account_name"] = previous_df["account_name"].str.strip().str.lower()
            previous_df["amount"] = previous_df.apply(FinancialReport._calculate_amount, axis=1)
            previous_df.set_index("date", inplace=True)

            # 7. Aggregate by period
            monthly = FinancialReport._aggregate_data(current_df, "ME", "%b")
            quarterly = FinancialReport._aggregate_data(current_df, "QE", "Q%q")
            yearly = FinancialReport._aggregate_data(current_df, "YE", "%Y")

            prev_monthly = FinancialReport._aggregate_data(previous_df, "ME", "%b")
            prev_quarterly = FinancialReport._aggregate_data(previous_df, "QE", "Q%q")
            prev_yearly = FinancialReport._aggregate_data(previous_df, "YE", "%Y")

            def pct_change(current, previous):
                if previous == 0:
                    return None
                return round(((current - previous) / abs(previous)) * 100, 2)

            # 8. Calculate key metrics comparing current vs previous (using yearly aggregates)
            metrics = {}
            if yearly and prev_yearly:
                year_key = next(iter(yearly), None)
                prev_year_key = next(iter(prev_yearly), None)
                if year_key and prev_year_key:
                    df_year = pd.DataFrame(yearly[year_key])
                    df_prev_year = pd.DataFrame(prev_yearly[prev_year_key])

                    total_assets = df_year[df_year["category"] == "asset"]["amount"].sum()
                    total_liabilities = df_year[df_year["category"] == "liability"]["amount"].sum()
                    total_equity = df_year[df_year["category"] == "equity"]["amount"].sum()
                    total_depreciation = df_year[
                        (df_year["category"] == "depreciation") & (df_year["account_name"] == "depreciation")
                    ]["amount"].sum()

                    total_assets_prev = df_prev_year[df_prev_year["category"] == "asset"]["amount"].sum()
                    total_liabilities_prev = df_prev_year[df_prev_year["category"] == "liability"]["amount"].sum()
                    total_equity_prev = df_prev_year[df_prev_year["category"] == "equity"]["amount"].sum()
                    total_depreciation_prev = df_prev_year[
                        (df_prev_year["category"] == "depreciation") & (df_prev_year["account_name"] == "depreciation")
                    ]["amount"].sum()

                    asset_growth = (total_assets + total_depreciation) - (total_assets_prev + total_depreciation_prev)
                    liability_growth = total_liabilities - total_liabilities_prev

                    asset_growth_pct = pct_change(total_assets + total_depreciation, total_assets_prev + total_depreciation_prev)
                    liability_growth_pct = pct_change(total_liabilities, total_liabilities_prev)

                    current_ratio = round(total_assets / total_liabilities, 2) if total_liabilities else None
                    debt_to_equity = round(total_liabilities / total_equity, 2) if total_equity else None

                    metrics = {
                        "Total_Assets": total_assets + total_depreciation,
                        "Total_Liabilities": total_liabilities,
                        "Net_Equity": total_equity,
                        "Asset_Growth": asset_growth,
                        "Asset_Growth_%": asset_growth_pct,
                        "Liability_Growth": liability_growth,
                        "Liability_Growth_%": liability_growth_pct,
                        "Current_Ratio": current_ratio,
                        "Debt_to_Equity": debt_to_equity,
                        "Accounting_Equation_Check": round(total_assets - (total_liabilities + total_equity), 2)
                    }

            # 9. Return final data
            return {
                "monthly_data": monthly,
                "quarterly_data": quarterly,
                "yearly_data": yearly,
                "previous_monthly_data": prev_monthly,
                "previous_quarterly_data": prev_quarterly,
                "previous_yearly_data": prev_yearly,
                "key_metrics": metrics,
                "date_range": {
                    "start_date": current_start.isoformat(),
                    "end_date": current_end.isoformat()
                }
            }

        except Exception as e:
            logger.error(f"Failed to generate balance sheet: {e}")
            raise FinancialReportError(f"Failed to generate balance sheet: {e}")

    @staticmethod
    async def generate_report_data(db: AsyncSession, params: Dict[str, Any]) -> Dict[str, Any]:
        """Generate all reports with proper session handling."""
        try:
            today = datetime.now()
            year = params.get("year", today.year)
            period = params.get("period", "yearly")
            custom_start = params.get("custom_start")
            custom_end = params.get("custom_end")
            entity = params.get("entity")
            hierarchy = params.get("hierarchy")
            
            reports = {}
            
            reports["profit_and_loss"] = await FirstFinancialReport.profit_loss_statement(
                db=db,
                year=year,
                period=period,
                custom_start=custom_start,
                custom_end=custom_end,
                entity=entity,
                hierarchy=hierarchy
            )
            
            # reports["cashflow"] = await FirstFinancialReport.get_cashflow_statement(
            #     db=db,
            #     year=year,
            #     period=period,
            #     custom_start=custom_start,
            #     custom_end=custom_end,
            #     entity=entity,
            #     hierarchy=hierarchy
            # )
            
            # reports["accounts_receivable"] = await FirstFinancialReport.get_accounts_receivable(
            #     db=db,
            #     year=year,
            #     period=period,
            #     custom_start=custom_start,
            #     custom_end=custom_end,
            #     entity=entity,
            #     hierarchy=hierarchy
            # )
            
            # reports["accounts_payable"] = await FirstFinancialReport.get_accounts_payable(
            #     db=db,
            #     year=year,
            #     period=period,
            #     custom_start=custom_start,
            #     custom_end=custom_end,
            #     entity=entity,
            #     hierarchy=hierarchy
            # )
            
            # reports["product_revenue"] = await FirstFinancialReport.get_product_revenue_report(
            #     db=db,
            #     year=year,
            #     period=period,
            #     custom_start=custom_start,
            #     custom_end=custom_end,
            #     entity=entity,
            #     hierarchy=hierarchy
            # )
            
            # reports["operation_expense"] = await FirstFinancialReport.get_operation_expense_report(
            #     db=db,
            #     year=year,
            #     period=period,
            #     custom_start=custom_start,
            #     custom_end=custom_end,
            #     entity=entity,
            #     hierarchy=hierarchy
            # )
            
            # reports["fixed_assets"] = await FirstFinancialReport.get_fixed_asset_register(
            #     db=db,
            #     year=year,
            #     period=period,
            #     custom_start=custom_start,
            #     custom_end=custom_end,
            #     entity=entity,
            #     hierarchy=hierarchy
            # )
            
            reports["balance_sheet"] = await FirstFinancialReport.get_balance_sheet(
                db=db,
                year=year,
                period=period,
                start_date=custom_start,
                end_date=custom_end,
                entity=entity,
                hierarchy=hierarchy
            )
            
            return reports
            
        except Exception as e:
            logger.error(f"Error generating reports: {str(e)}", exc_info=True)
            raise HTTPException(status_code=500, detail=str(e))

   
    @staticmethod
    async def profit_loss_statement(
        db: AsyncSession,
        year: int,
        period: str = "yearly",  # "yearly", "quarterly", "monthly" or "custom"
        custom_start: Optional[datetime] = None,
        custom_end: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate profit and loss statement with both monthly and quarterly breakdowns."""
        try:
            # 1. Get date range
            start_date, end_date = FinancialReport.get_date_range(
                period, year, custom_start, custom_end
            )
            
            # 2. Load account mapping
            account_mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get GL codes for P&L categories using reusable function
            revenue_gl_codes = FinancialReport.get_accounts_by_tags(
                account_mapping, FinancialReport.REVENUE_TAGS
            )
            operating_expense_gl_codes = FinancialReport.get_accounts_by_tags(
                account_mapping, FinancialReport.OPERATING_EXPENSE_TAGS
            )
            
            # 4. Add COGS tags and get COGS GL codes
            
            cogs_gl_codes = FinancialReport.get_accounts_by_tags(
                account_mapping, FinancialReport.COGS_TAGS
            )
            
            
            # 5. Create category mapping for fetching data
            category_mapping = {
                "revenue": revenue_gl_codes,
                "operating_expense": operating_expense_gl_codes,
                "cogs": cogs_gl_codes
            }
            print(category_mapping)
            
            # 6. Fetch categorized data using reusable function
            categorized_df = await FinancialReport._fetch_categorized_data(
                db, category_mapping, start_date, end_date
            )
            print(categorized_df)
            if categorized_df.empty:
                return {
                    "monthly_data": {},
                    "quarterly_data": {},
                    "yearly_data": [],
                    "key_metrics": {
                        "Total_Revenue": 0.0,
                        "Total_COGS": 0.0,
                        "Total_Expenses": 0.0,
                        "Gross_Profit": 0.0,
                        "Net_Income": 0.0,
                        "Gross_Margin_Percentage": 0.0,
                        "Net_Margin_Percentage": 0.0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }
            
            # 7. Filter by entity and hierarchy using reusable function
            filtered_df = FinancialReport.filter_entity_and_group(
                df=categorized_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=None
            )
            print(filtered_df)
            
            # 8. Calculate amount for each row using reusable function
            # Update _calculate_amount to handle COGS category
            filtered_df["amount"] = filtered_df.apply(
                lambda row: FinancialReport._calculate_amount(row) if row["category"] != "cogs" 
                            else float(row.get("debit", 0) - row.get("credit", 0)),
                axis=1
            )
            
            # 9. Ensure date column is datetime type and set as index for aggregation
            filtered_df["date"] = pd.to_datetime(filtered_df["date"])
            filtered_df.set_index("date", inplace=True)
            
            # 10. Aggregate data by different time periods using reusable function
            monthly_data = FinancialReport._aggregate_data(
                filtered_df, "M", "%b"
            )
            
            quarterly_data = FinancialReport._aggregate_data(
                filtered_df, "Q", "Q%q"
            )
            
            # Reset index for yearly aggregation
            filtered_df_reset = filtered_df.reset_index()
            
            # 11. Calculate yearly summary
            yearly_summary = filtered_df_reset.groupby(["category", "account_name"])["amount"].sum().reset_index()
            
            # 12. Calculate key metrics
            revenue = yearly_summary[yearly_summary["category"] == "revenue"]["amount"].sum()
            cogs = yearly_summary[yearly_summary["category"] == "cogs"]["amount"].sum()
            operating_expenses = yearly_summary[yearly_summary["category"] == "operating_expense"]["amount"].sum()
            actual_profit=monthly_data[monthly_data["category"]=="revenue"]["amount"]-monthly_data[monthly_data["category"]=="cogs"]["amount"]
            total_actual_profit=sum(actual_profit)
            gross_profit = revenue - cogs

            net_income = gross_profit - operating_expenses
            gross_margin = (gross_profit / revenue * 100) if revenue != 0 else 0
            net_margin = (net_income / revenue * 100) if revenue != 0 else 0
            
            metrics = {
                "Total_Revenue": float(revenue),
                "total_actual_profit":float(total_actual_profit),
                "Total_COGS": float(cogs),
                "Total_Expenses": float(operating_expenses),
                "Gross_Profit": float(gross_profit),
                "Net_Income": float(net_income),
                "Gross_Margin_Percentage": float(gross_margin),
                "Net_Margin_Percentage": float(net_margin)
            }
            
            # 13. Return final report
            return {
                "monthly_data": monthly_data,
                "quarterly_data": quarterly_data,
                "yearly_data": yearly_summary.to_dict('records'),
                "actual_profit":actual_profit,
                "key_metrics": metrics,
                "date_range": {
                    "start_date": start_date.isoformat(),
                    "end_date": end_date.isoformat()
                }
            }
            
        except Exception as e:
            logger.error(f"Failed to generate profit loss statement: {e}")
            raise FinancialReportError(f"Failed to generate profit loss statement: {e}")
                
    @staticmethod
    async def get_cashflow_statement(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        try:
            # 1. Get date range
            start_date, end_date = FinancialReport.get_date_range(
                period, year, start_date, end_date
            )
            
            # 2. Load account mapping
            account_mapping = await FinancialReport.get_account_mapping(db)
            
            accounts={
            "Operating_cashin": FinancialReport.get_accounts_by_tags(account_mapping, FinancialReport.OPERATING_CASHIN),
                "Operating_cashout": FinancialReport.get_accounts_by_tags(account_mapping, FinancialReport.OPERATING_CASHOUT),
                "Investing_cashin": FinancialReport.get_accounts_by_tags(account_mapping, FinancialReport.INVESTING_CASHIN),
                "Investing_cashout": FinancialReport.get_accounts_by_tags(account_mapping, FinancialReport.INVESTING_CASHOUT),
                "Financing_cashin": FinancialReport.get_accounts_by_tags(account_mapping, FinancialReport.FINANCING_CASHIN),
                "Financing_cashout": FinancialReport.get_accounts_by_tags(account_mapping, FinancialReport.FINANCING_CASHOUT)
            }
            
            # 4. Add COGS tags and get COGS GL codes
            
            
            
            # 6. Fetch categorized data using reusable function
            categorized_df = await FinancialReport._fetch_categorized_data(
                db, accounts, start_date, end_date
            )
            #print(categorized_df["category"].unique())
            
            if categorized_df.empty:
                return {
                    "monthly_data": {},
                    "quarterly_data": {},
                    "yearly_data": [],
                    "key_metrics": {
                        "Total_Revenue": 0.0,
                        "Total_COGS": 0.0,
                        "Total_Expenses": 0.0,
                        "Gross_Profit": 0.0,
                        "Net_Income": 0.0,
                        "Gross_Margin_Percentage": 0.0,
                        "Net_Margin_Percentage": 0.0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }
            
            # 7. Filter by entity and hierarchy using reusable function
            filtered_df = FinancialReport.filter_entity_and_group(
                df=categorized_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=None
            )
            
            # 8. Calculate amount for each row using reusable function
            # Update _calculate_amount to handle COGS category
            filtered_df["amount"] = filtered_df.apply(
                lambda row: FinancialReport._calculate_amount(row),
                axis=1
            )
            
            
            
            # 9. Ensure date column is datetime type and set as index for aggregation
            filtered_df["date"] = pd.to_datetime(filtered_df["date"])
            filtered_df.set_index("date", inplace=True)
           
            # # 10. Aggregate data by different time periods using reusable function
            monthly_data = FinancialReport._aggregate_data(
                filtered_df, "M", "%b"
            )
            
            quarterly_data = FinancialReport._aggregate_data(
                filtered_df, "Q", "Q%q"
            )
            
            # # Reset index for yearly aggregation
            filtered_df_reset = filtered_df.reset_index()
            
            # # 11. Calculate yearly summary
            yearly_summary = filtered_df_reset.groupby(["category", "account_name"])["amount"].sum().reset_index()
            ##monthly cashflow
            monthly_Operating_cashin=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashin"]["amount"].sum()
            monthly_Operating_cashout=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashout"]["amount"].sum()
            monthly_Investing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashin"]["amount"].sum()
            monthly_Investing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashout"]["amount"].sum()
            monthly_Financing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashin"]["amount"].sum()
            monthly_Financing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashout"]["amount"].sum()
            monthly_operation_cashflow=monthly_Operating_cashin-monthly_Operating_cashout
            monthly_investing_cashflow=monthly_Investing_cashin-monthly_Investing_cashout
            monthly_financing_cashflow=monthly_Financing_cashin-monthly_Financing_cashout
            monthly_net_cashflow=monthly_operation_cashflow+monthly_investing_cashflow+monthly_financing_cashflow
            
            #quarterly cashflow
            quarterly_Operating_cashin=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashin"]["amount"].sum()

            quarterly_Operating_cashout=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashout"]["amount"].sum()
            quarterly_Investing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashin"]["amount"].sum()
            quarterly_Investing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashout"]["amount"].sum()
            quarterly_Financing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashin"]["amount"].sum()
            quarterly_Financing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashout"]["amount"].sum()
            quarterly_operation_cashflow=quarterly_Operating_cashin-quarterly_Operating_cashout
            quarterly_investing_cashflow=quarterly_Investing_cashin-quarterly_Investing_cashout
            quarterly_financing_cashflow=quarterly_Financing_cashin-quarterly_Financing_cashout
            quarterly_net_cashflow=quarterly_operation_cashflow+quarterly_investing_cashflow+quarterly_financing_cashflow
            #yearly cashflow
            yearly_Operating_cashin=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashin"]["amount"].sum()
            yearly_Operating_cashout=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashout"]["amount"].sum()
            yearly_Investing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashin"]["amount"].sum()
            yearly_Investing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashout"]["amount"].sum()
            yearly_Financing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashin"]["amount"].sum()
            yearly_Financing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashout"]["amount"].sum()
            yearly_operation_cashflow=yearly_Operating_cashin-yearly_Operating_cashout
            yearly_investing_cashflow=yearly_Investing_cashin-yearly_Investing_cashout
            yearly_financing_cashflow=yearly_Financing_cashin-yearly_Financing_cashout
            yearly_net_cashflow=yearly_operation_cashflow+yearly_investing_cashflow+yearly_financing_cashflow
            return {
                "monthly_data": monthly_data,
                "quarterly_data": quarterly_data,
                "yearly_data": yearly_summary.to_dict('records'),
                "monthly_cashflow":{
                    "monthly_Operating_cashin":monthly_Operating_cashin,
                    "monthly_Operating_cashout":monthly_Operating_cashout,
                    "monthly_Investing_cashin":monthly_Investing_cashin,
                    "monthly_Investing_cashout":monthly_Investing_cashout,
                    "monthly_Financing_cashin":monthly_Financing_cashin,
                    "monthly_Financing_cashout":monthly_Financing_cashout,
                    "monthly_operation_cashflow":monthly_operation_cashflow,
                    "monthly_investing_cashflow":monthly_investing_cashflow,
                    "monthly_financing_cashflow":monthly_financing_cashflow,
                    "monthly_net_cashflow":monthly_net_cashflow
                },
                "quarterly_cashflow":{
                    "quarterly_Operating_cashin":quarterly_Operating_cashin,
                    "quarterly_Operating_cashout":quarterly_Operating_cashout,
                    "quarterly_Investing_cashin":quarterly_Investing_cashin,
                    "quarterly_Investing_cashout":quarterly_Investing_cashout,
                    "quarterly_Financing_cashin":quarterly_Financing_cashin,
                    "quarterly_Financing_cashout":quarterly_Financing_cashout,
                    "quarterly_operation_cashflow":quarterly_operation_cashflow,
                    "quarterly_investing_cashflow":quarterly_investing_cashflow,
                    "quarterly_financing_cashflow":quarterly_financing_cashflow,
                    "quarterly_net_cashflow":quarterly_net_cashflow
                },
                "yearly_cashflow":{
                    "yearly_Operating_cashin":yearly_Operating_cashin,
                    "yearly_Operating_cashout":yearly_Operating_cashout,
                    "yearly_Investing_cashin":yearly_Investing_cashin,
                    "yearly_Investing_cashout":yearly_Investing_cashout,
                    "yearly_Financing_cashin":yearly_Financing_cashin,
                    "yearly_Financing_cashout":yearly_Financing_cashout,
                    "yearly_operation_cashflow":yearly_operation_cashflow,
                    "yearly_investing_cashflow":yearly_investing_cashflow,
                    "yearly_financing_cashflow":yearly_financing_cashflow,
                    "yearly_net_cashflow":yearly_net_cashflow
                },
                "date_range": {
                    "start_date": start_date.isoformat(),
                    "end_date": end_date.isoformat()
            }
            }
        except Exception as e:
            logger.error(f"Failed to generate profit loss statement: {e}")
            raise FinancialReportError(f"Failed to generate profit loss statement: {e}")


    # @staticmethod
    # async def get_accounts_receivable(
    #     db: AsyncSession,
    #     year: int,
    #     period: str = "yearly",
    #     custom_start: Optional[datetime] = None,
    #     custom_end: Optional[datetime] = None,
    #     entity: Optional[str] = None,
    #     hierarchy: Optional[str] = None
    # ) -> Dict[str, Any]:
    #     """Generate accounts receivable aging report."""
    #     try:
    #         start, end = FirstFinancialReport.get_date_range(period, year, custom_start, custom_end)
    #         filters = FirstFinancialReport.build_filters(start, end, entity, hierarchy)

    #         # Filter for receivable accounts via name or master category
    #         filters.append(
    #             or_(
    #                 GeneralLedger.account_name.ilike("%receivable%"),
    #                 GeneralLedger.general_ledger_code.in_(
    #                     select(GeneralLedgerMaster.general_ledger_code)
    #                     .where(GeneralLedgerMaster.category == "receivable")
    #                 )
    #             )
    #         )

    #         stmt = select(GeneralLedger).where(and_(*filters))
    #         result = await db.execute(stmt)
    #         gl_entries = result.scalars().all()

    #         aging_buckets = defaultdict(lambda: Decimal("0.0"))
    #         total_receivables = Decimal("0.0")
    #         total_weighted_days = Decimal("0.0")

    #         for entry in gl_entries:
    #             if not entry.date:
    #                 continue
    #             amount = FirstFinancialReport.safe_decimal(entry.debit or 0)
    #             if amount <= 0:
    #                 continue

    #             days = (end - entry.date).days
    #             if days < 0:
    #                 days = 0  # Future dated within the period, treat as current

    #             if days <= 0:
    #                 aging_buckets["current"] += amount
    #             elif 1 <= days <= 30:
    #                 aging_buckets["1_30"] += amount
    #             elif 31 <= days <= 60:
    #                 aging_buckets["31_60"] += amount
    #             elif 61 <= days <= 90:
    #                 aging_buckets["61_90"] += amount
    #             else:
    #                 aging_buckets["over_90"] += amount

    #             total_receivables += amount
    #             total_weighted_days += amount * Decimal(days)

    #         overdue = total_receivables - aging_buckets["current"]
    #         overdue_pct = (overdue / total_receivables * 100) if total_receivables else Decimal("0")
    #         avg_days = (total_weighted_days / total_receivables) if total_receivables else Decimal("0")

    #         aging_report = [
    #             {
    #                 "bucket": f"{k.replace('_', '-')} days",
    #                 "amount": FirstFinancialReport.format_money(v),
    #                 "percentage": f"{(v / total_receivables * 100).quantize(Decimal('0.1'))}%" if total_receivables else "0.0%"
    #             }
    #             for k, v in aging_buckets.items()
    #         ]

    #         return {
    #             "status": "success",
    #             "data": {
    #                 "aging_report": aging_report,
    #                 "summary": {
    #                     "total_receivables": FirstFinancialReport.format_money(total_receivables),
    #                     "overdue_amount": FirstFinancialReport.format_money(overdue),
    #                     "overdue_percentage": f"{overdue_pct.quantize(Decimal('0.1'))}%",
    #                     "average_days_outstanding": f"{avg_days.quantize(Decimal('0'))}"
    #                 }
    #             },
    #             "metadata": {
    #                 "period": period,
    #                 "start_date": start.isoformat() if start else None,
    #                 "end_date": end.isoformat() if end else None,
    #                 "generated_at": datetime.now().isoformat(),
    #                 "record_count": len(gl_entries)
    #             }
    #         }

    #     except Exception as e:
    #         logger.error(f"Error generating AR report: {str(e)}", exc_info=True)
    #         return {
    #             "status": "error",
    #             "error": str(e),
    #             "metadata": {
    #                 "period": period,
    #                 "failed_at": datetime.now().isoformat()
    #             }
    #         }

    # @staticmethod
    # async def get_accounts_payable(
    #     db: AsyncSession,
    #     year: int,
    #     period: str = "yearly",
    #     custom_start: Optional[datetime] = None,
    #     custom_end: Optional[datetime] = None,
    #     entity: Optional[str] = None,
    #     hierarchy: Optional[str] = None
    # ) -> Dict[str, Any]:
    #     """Generate accounts payable aging report."""
    #     try:
    #         start, end = FirstFinancialReport.get_date_range(period, year, custom_start, custom_end)
    #         filters = FirstFinancialReport.build_filters(start, end, entity, hierarchy)

    #         # Filter for payable accounts via name or master category
    #         filters.append(
    #             or_(
    #                 GeneralLedger.account_name.ilike("%payable%"),
    #                 GeneralLedger.general_ledger_code.in_(
    #                     select(GeneralLedgerMaster.general_ledger_code)
    #                     .where(GeneralLedgerMaster.category == "payable")
    #                 )
    #             )
    #         )

    #         stmt = select(GeneralLedger).where(and_(*filters))
    #         result = await db.execute(stmt)
    #         gl_entries = result.scalars().all()

    #         aging_buckets = defaultdict(lambda: Decimal("0.0"))
    #         total_payables = Decimal("0.0")
    #         total_weighted_days = Decimal("0.0")

    #         for entry in gl_entries:
    #             if not entry.date:
    #                 continue
    #             amount = FirstFinancialReport.safe_decimal(entry.credit or 0)
    #             if amount <= 0:
    #                 continue

    #             days = (end - entry.date).days
    #             if days < 0:
    #                 days = 0  # Future dated within period, treat as current

    #             if days <= 0:
    #                 aging_buckets["current"] += amount
    #             elif 1 <= days <= 30:
    #                 aging_buckets["1_30"] += amount
    #             elif 31 <= days <= 60:
    #                 aging_buckets["31_60"] += amount
    #             elif 61 <= days <= 90:
    #                 aging_buckets["61_90"] += amount
    #             else:
    #                 aging_buckets["over_90"] += amount

    #             total_payables += amount
    #             total_weighted_days += amount * Decimal(days)

    #         overdue = total_payables - aging_buckets["current"]
    #         overdue_pct = (overdue / total_payables * 100) if total_payables else Decimal("0")
    #         avg_days = (total_weighted_days / total_payables) if total_payables else Decimal("0")

    #         aging_report = [
    #             {
    #                 "bucket": f"{k.replace('_', '-')} days",
    #                 "amount": FirstFinancialReport.format_money(v),
    #                 "percentage": f"{(v / total_payables * 100).quantize(Decimal('0.1'))}%" if total_payables else "0.0%"
    #             }
    #             for k, v in aging_buckets.items()
    #         ]

    #         return {
    #             "status": "success",
    #             "data": {
    #                 "aging_report": aging_report,
    #                 "summary": {
    #                     "total_payables": FirstFinancialReport.format_money(total_payables),
    #                     "overdue_amount": FirstFinancialReport.format_money(overdue),
    #                     "overdue_percentage": f"{overdue_pct.quantize(Decimal('0.1'))}%",
    #                     "average_days_payable": f"{avg_days.quantize(Decimal('0'))}"
    #                 }
    #             },
    #             "metadata": {
    #                 "period": period,
    #                 "start_date": start.isoformat() if start else None,
    #                 "end_date": end.isoformat() if end else None,
    #                 "generated_at": datetime.now().isoformat(),
    #                 "record_count": len(gl_entries)
    #             }
    #         }

    #     except Exception as e:
    #         logger.error(f"Error generating AP report: {str(e)}", exc_info=True)
    #         return {
    #             "status": "error",
    #             "error": str(e),
    #             "metadata": {
    #                 "period": period,
    #                 "failed_at": datetime.now().isoformat()
    #             }
    #         }
   

   

    @staticmethod
    async def get_product_revenue_report(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate product revenue report with monthly, quarterly, and yearly data"""
        try:
            # 1. Set date range
            start_date, end_date = FinancialReport.get_date_range(period, year, start_date, end_date)
            FinancialReport._validate_date_range(start_date, end_date)
            
            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get revenue accounts
            revenue_accounts = {
                "revenue": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.REVENUE_TAGS)
            }
            
            # 4. Fetch revenue data
            revenue_df = await FinancialReport._fetch_categorized_data(db, revenue_accounts, start_date, end_date)
            
            if revenue_df.empty:
                return {
                    "monthly_revenue": {},
                    "quarterly_revenue": {},
                    "yearly_revenue": {},
                    "summary": {
                        "total_revenue": 0,
                        "top_product": None,
                        "top_product_revenue": 0,
                        "product_count": 0,
                        "average_revenue_per_product": 0,
                        "monthly_average": 0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }

            # 5. Filter and process data
            other_filter = None
            df = FinancialReport.filter_entity_and_group(
                df=revenue_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
            
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category"])
            df["date"] = pd.to_datetime(df["date"])
            df["account_name"] = df["account_name"].str.strip().str.lower()
            df["amount"] = df["credit"] - df["debit"]
            df.set_index("date", inplace=True)
            
            # 6. Aggregate data
            monthly_revenue = FinancialReport._aggregate_data(df, "ME", "%b")
            quarterly_revenue = FinancialReport._aggregate_data(df, "QE", "Q%q")
            yearly_revenue = FinancialReport._aggregate_data(df, "YE", "%Y")
            
            # 7. Calculate summary
            total_revenue = df["amount"].sum()
            yearly_summary = df.groupby("account_name")["amount"].sum().to_dict()
            product_count = df["account_name"].nunique()

            # 8. Return final report
            return {
                "monthly_revenue": monthly_revenue,
                "quarterly_revenue": quarterly_revenue,
                "yearly_revenue": yearly_revenue,
                "summary": {
                    "total_revenue": float(total_revenue),
                    "top_product": max(yearly_summary, key=yearly_summary.get) if yearly_summary else None,
                    "top_product_revenue": max(yearly_summary.values()) if yearly_summary else 0,
                    "product_count": product_count,
                    "average_revenue_per_product": float(total_revenue / product_count) if product_count > 0 else 0,
                    "monthly_average": float(total_revenue / 12)
                },
                "date_range": {
                    "start_date": start_date.strftime("%Y-%m-%d"),
                    "end_date": end_date.strftime("%Y-%m-%d")
                }
            }
        except Exception as e:
            logger.error(f"Failed to generate product revenue report: {e}")
            raise FinancialReportError(f"Failed to generate product revenue report: {e}")

    @staticmethod
    async def get_operation_expense_report(
        db: AsyncSession,
        year: int,
        period: str = "yearly",
        custom_start: Optional[datetime] = None,
        custom_end: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate operation expense report with monthly and quarterly breakdowns."""
        try:
            start, end = FirstFinancialReport.get_date_range(period, year, custom_start, custom_end)
            filters = FirstFinancialReport.build_filters(start, end, entity, hierarchy)
            
            # Filter for expense category accounts
            filters.append(
                GeneralLedger.general_ledger_code.in_(
                    select(GeneralLedgerMaster.general_ledger_code)
                    .where(func.lower(GeneralLedgerMaster.category) == "expense")
                )
            )
            
            # Get base data with month and quarter information
            stmt = select(
                GeneralLedger.account_name,
                extract('month', GeneralLedger.date).label("month"),
                extract('quarter', GeneralLedger.date).label("quarter"),
                func.sum(GeneralLedger.debit).label("amount")
            ).where(
                and_(*filters)
            ).group_by(
                GeneralLedger.account_name,
                extract('month', GeneralLedger.date),
                extract('quarter', GeneralLedger.date)
            )

            result = await db.execute(stmt)
            rows = result.all()

            # Initialize data structures
            monthly_data = defaultdict(lambda: defaultdict(lambda: Decimal("0.0")))  # expense -> month -> amount
            quarterly_data = defaultdict(lambda: [Decimal("0.0")] * 4)  # expense -> [Q1, Q2, Q3, Q4]
            custom_range_data = defaultdict(lambda: Decimal("0.0")) if period == "custom" else None
            
            month_names = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", 
                        "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
            quarter_names = ["Q1", "Q2", "Q3", "Q4"]

            for account_name, month, quarter, amount in rows:
                if not account_name or not amount:
                    continue
                    
                amount_decimal = FirstFinancialReport.safe_decimal(amount)
                
                # Monthly data
                month_key = month_names[int(month) - 1]
                monthly_data[account_name][month_key] += amount_decimal
                
                # Quarterly data
                q_index = int(quarter) - 1
                quarterly_data[account_name][q_index] += amount_decimal
                
                # Custom range data if applicable
                if period == "custom":
                    custom_range_data[account_name] += amount_decimal

            # Prepare reports
            def prepare_monthly_report():
                report = []
                for expense, months in monthly_data.items():
                    report.append({
                        "expense": expense.title(),
                        **{month: FirstFinancialReport.format_money(months[month]) for month in month_names},
                        "Total": FirstFinancialReport.format_money(sum(months.values()))
                    })
                return report

            def prepare_quarterly_report():
                report = []
                for expense, quarters in quarterly_data.items():
                    report.append({
                        "expense": expense.title(),
                        "Q1": FirstFinancialReport.format_money(quarters[0]),
                        "Q2": FirstFinancialReport.format_money(quarters[1]),
                        "Q3": FirstFinancialReport.format_money(quarters[2]),
                        "Q4": FirstFinancialReport.format_money(quarters[3]),
                        "Total": FirstFinancialReport.format_money(sum(quarters))
                    })
                return report

            def prepare_custom_report():
                if not custom_range_data:
                    return None
                report = []
                for expense, amount in custom_range_data.items():
                    report.append({
                        "expense": expense.title(),
                        "Amount": FirstFinancialReport.format_money(amount)
                    })
                return report

            # Calculate metrics
            def calculate_metrics(data):
                if not data:
                    return {}
                    
                total = Decimal("0.0")
                highest = ("", Decimal("0.0"))
                lowest = ("", Decimal("9999999999.0"))
                period_totals = {}
                
                if isinstance(data, list) and all(isinstance(item, dict) for item in data):
                    # For monthly/quarterly reports
                    for item in data:
                        expense_total = FirstFinancialReport.safe_decimal(item["Total"].replace(",", ""))
                        total += expense_total
                        
                        if expense_total > highest[1]:
                            highest = (item["expense"], expense_total)
                        if expense_total < lowest[1]:
                            lowest = (item["expense"], expense_total)
                            
                    # For monthly reports, calculate monthly metrics
                    if "Jan" in data[0]:
                        monthly_totals = [sum(
                            FirstFinancialReport.safe_decimal(item[month].replace(",", "")) 
                            for item in data
                        ) for month in month_names]
                        
                        avg_monthly = total / Decimal("12.0") if total else Decimal("0.0")
                        highest_month = month_names[monthly_totals.index(max(monthly_totals))] if monthly_totals else ""
                        lowest_month = month_names[monthly_totals.index(min(monthly_totals))] if monthly_totals else ""
                        
                        return {
                            "total_expenses": FirstFinancialReport.format_money(total),
                            "highest_expense": {
                                "name": highest[0],
                                "amount": FirstFinancialReport.format_money(highest[1])
                            },
                            "lowest_expense": {
                                "name": lowest[0],
                                "amount": FirstFinancialReport.format_money(lowest[1])
                            },
                            "average_monthly": FirstFinancialReport.format_money(avg_monthly),
                            "highest_month": {
                                "name": highest_month,
                                "amount": FirstFinancialReport.format_money(max(monthly_totals)) if monthly_totals else "0.00"
                            },
                            "lowest_month": {
                                "name": lowest_month,
                                "amount": FirstFinancialReport.format_money(min(monthly_totals)) if monthly_totals else "0.00"
                            }
                        }
                    else:
                        # For quarterly reports
                        quarterly_totals = [sum(
                            FirstFinancialReport.safe_decimal(item[f"Q{i+1}"].replace(",", "")) 
                            for item in data
                        ) for i in range(4)]
                        
                        avg_quarterly = total / Decimal("4.0") if total else Decimal("0.0")
                        highest_quarter = quarter_names[quarterly_totals.index(max(quarterly_totals))] if quarterly_totals else ""
                        lowest_quarter = quarter_names[quarterly_totals.index(min(quarterly_totals))] if quarterly_totals else ""
                        
                        return {
                            "total_expenses": FirstFinancialReport.format_money(total),
                            "highest_expense": {
                                "name": highest[0],
                                "amount": FirstFinancialReport.format_money(highest[1])
                            },
                            "lowest_expense": {
                                "name": lowest[0],
                                "amount": FirstFinancialReport.format_money(lowest[1])
                            },
                            "average_quarterly": FirstFinancialReport.format_money(avg_quarterly),
                            "highest_quarter": {
                                "name": highest_quarter,
                                "amount": FirstFinancialReport.format_money(max(quarterly_totals)) if quarterly_totals else "0.00"
                            },
                            "lowest_quarter": {
                                "name": lowest_quarter,
                                "amount": FirstFinancialReport.format_money(min(quarterly_totals)) if quarterly_totals else "0.00"
                            }
                        }
                else:
                    # For custom range reports
                    for expense, amount in data.items():
                        total += amount
                        if amount > highest[1]:
                            highest = (expense, amount)
                        if amount < lowest[1]:
                            lowest = (expense, amount)
                            
                    return {
                        "total_expenses": FirstFinancialReport.format_money(total),
                        "highest_expense": {
                            "name": highest[0],
                            "amount": FirstFinancialReport.format_money(highest[1])
                        },
                        "lowest_expense": {
                            "name": lowest[0],
                            "amount": FirstFinancialReport.format_money(lowest[1])
                        }
                    }

            monthly_report = prepare_monthly_report()
            quarterly_report = prepare_quarterly_report()
            custom_report = prepare_custom_report()

            # Determine which metrics to use
            if period == "custom":
                key_metrics = calculate_metrics(custom_range_data)
            else:
                key_metrics = calculate_metrics(monthly_report)

            return {
                "status": "success",
                "data": {
                    "monthly_report": monthly_report,
                    "quarterly_report": quarterly_report,
                    "custom_range_report": custom_report if period == "custom" else None,
                    "key_metrics": key_metrics
                },
                "metadata": {
                    "period": period,
                    "start_date": start.isoformat() if start else None,
                    "end_date": end.isoformat() if end else None,
                    "generated_at": datetime.now().isoformat(),
                    "record_count": len(rows)
                }
            }

        except Exception as e:
            logger.error(f"Error generating operation expense report: {str(e)}", exc_info=True)
            return {
                "status": "error",
                "error": str(e),
                "metadata": {
                    "period": period,
                    "failed_at": datetime.now().isoformat()
                }
            }
    @staticmethod
    async def get_fixed_asset_register(
        db: AsyncSession,
        year: int,
        period: str = "yearly",
        custom_start: Optional[datetime] = None,
        custom_end: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate fixed asset register report."""
        try:
            start, end = FirstFinancialReport.get_date_range(period, year, custom_start, custom_end)
            filters = FirstFinancialReport.build_filters(start, end, entity, hierarchy)

            # Only include fixed assets from the GeneralLedgerMaster
            fixed_asset_codes_stmt = select(GeneralLedgerMaster.general_ledger_code).where(
                func.lower(GeneralLedgerMaster.category) == "fixed_asset"
            )
            fixed_asset_codes = await db.execute(fixed_asset_codes_stmt)
            codes = [code[0] for code in fixed_asset_codes.fetchall()]
            
            if not codes:
                logger.warning("No fixed asset codes found in master data.")
                return {
                    "status": "success",
                    "data": {
                        "asset_register": [],
                        "summary": {
                            "total_assets": 0,
                            "total_value": "0.00",
                            "net_value": "0.00",
                            "total_depreciation": "0.00",
                            "asset_types": 0
                        }
                    },
                    "metadata": {
                        "period": period,
                        "start_date": start.isoformat() if start else None,
                        "end_date": end.isoformat() if end else None,
                        "generated_at": datetime.now().isoformat(),
                        "record_count": 0,
                        "warning": "No fixed asset general ledger codes found"
                    }
                }

            filters.append(GeneralLedger.general_ledger_code.in_(codes))

            stmt = select(
                GeneralLedger.account_name,
                GeneralLedger.date,
                GeneralLedger.debit,
                GeneralLedger.credit,
                GeneralLedger.extra_data
            ).where(
                and_(*filters)
            ).order_by(
                GeneralLedger.account_name,
                GeneralLedger.date
            )

            result = await db.execute(stmt)
            rows = result.all()

            asset_register = []
            total_value = Decimal("0.0")
            total_depreciation = Decimal("0.0")
            asset_names = set()

            for account_name, date, debit, credit, extra_data in rows:
                if not account_name:
                    continue

                raw_cost = FirstFinancialReport.safe_decimal(debit or credit)
                if raw_cost == Decimal("0.0"):
                    continue

                acquisition_date = date.strftime('%Y-%m-%d') if date else "Unknown"
                location = "Unknown"
                condition = "Good"
                depreciation = Decimal("0.0")

                if extra_data and isinstance(extra_data, dict):
                    location = extra_data.get("location", location)
                    condition = extra_data.get("condition", condition)
                    depreciation = FirstFinancialReport.safe_decimal(extra_data.get("depreciation", "0.0"))

                net_value = raw_cost - depreciation

                asset_data = {
                    "asset_name": account_name.title(),
                    "acquisition_date": acquisition_date,
                    "original_cost": FirstFinancialReport.format_money(raw_cost),
                    "current_value": FirstFinancialReport.format_money(net_value),
                    "location": location,
                    "condition": condition
                }

                asset_register.append(asset_data)
                total_value += raw_cost
                total_depreciation += depreciation
                asset_names.add(account_name.lower())

            return {
                "status": "success",
                "data": {
                    "asset_register": asset_register,
                    "summary": {
                        "total_assets": len(asset_register),
                        "total_value": FirstFinancialReport.format_money(total_value),
                        "net_value": FirstFinancialReport.format_money(total_value - total_depreciation),
                        "total_depreciation": FirstFinancialReport.format_money(total_depreciation),
                        "asset_types": len(asset_names)
                    }
                },
                "metadata": {
                    "period": period,
                    "start_date": start.isoformat() if start else None,
                    "end_date": end.isoformat() if end else None,
                    "generated_at": datetime.now().isoformat(),
                    "record_count": len(rows)
                }
            }

        except Exception as e:
            logger.error(f"Error generating fixed asset register: {str(e)}", exc_info=True)
            return {
                "status": "error",
                "error": str(e),
                "metadata": {
                    "period": period,
                    "failed_at": datetime.now().isoformat()
                }
            }

    @staticmethod
    async def get_balance_sheet(
        db: AsyncSession,
        year: int,
        period: str = "yearly",
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate balance sheet report with monthly and quarterly breakdowns."""
        try:
            # 1. Set date range
            current_start, current_end = FinancialReport.get_date_range(
                period, year, start_date, end_date
            )
            previous_start, previous_end = FinancialReport.get_date_range(period,year-1, start_date, end_date)
            FinancialReport._validate_date_range(current_start, current_end)
            FinancialReport._validate_date_range(previous_start, previous_end)

            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
           
            # 3. Get accounts by category
            accounts = {
                "asset": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.ASSET_TAGS),
                "liability": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.LIABILITY_TAGS),
                "equity": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.EQUITY_TAGS),
                "depreciation": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.DEPRECIATION_TAGS),
            }

            # 4. Fetch data for all categories
            asset_df = await FinancialReport._fetch_categorized_data(db, accounts, current_start, current_end)
            asset_df1 = await FinancialReport._fetch_categorized_data(db, accounts, previous_start, previous_end)
            # Process previous period DataFrame if not empty
            if not asset_df1.empty:
                df_prev = FinancialReport.filter_entity_and_group(
                    df=asset_df1,
                    entity=entity,
                    hierarchy_level=hierarchy or "group"
                )
                df_prev = pd.DataFrame(df_prev, columns=["account_name", "debit", "credit", "date", "category"])
                df_prev["date"] = pd.to_datetime(df_prev["date"])
                df_prev["account_name"] = df_prev["account_name"].str.strip().str.lower()
                df_prev["amount"] = df_prev.apply(FinancialReport._calculate_amount, axis=1)
                df_prev.set_index("date", inplace=True)

                # Aggregate previous period data
                yearly_prev = FinancialReport._aggregate_data(df_prev, "YE", "%Y")
            else:
                yearly_prev = {}

            if asset_df.empty:
                return {
                    "status": "success",
                    "data": {
                        "monthly_summary": {},
                        "quarterly_summary": {},
                        "custom_range_summary": {} if period == "custom" else None
                    },
                    "metadata": {
                        "period": period,
                        "start_date": current_start.isoformat() if current_start else None,
                        "end_date": current_end.isoformat() if current_end else None,
                        "record_count": 0,
                        "warning": "No data found"
                    }
                }
            # 5. Filter and process data
            other_filter = None
            df = FinancialReport.filter_entity_and_group(
                df=asset_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
            
            # Prepare DataFrame
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category"])
            df["date"] = pd.to_datetime(df["date"])
            df["account_name"] = df["account_name"].str.strip().str.lower()
            
            # 6. Calculate net amount
            df["amount"] = df.apply(FinancialReport._calculate_amount, axis=1)
            df.set_index("date", inplace=True)
            def compute_change(current_val, prev_val):
                change = round(current_val - prev_val, 2)
                pct_change = round((change / prev_val) * 100, 2) if prev_val else None
                return {
                    "current": current_val,
                    "previous": prev_val,
                    "change": change,
                    "change_pct": pct_change
    }
            def compare_periods(curr_data, prev_data):
                result = {}
                for period_key in curr_data:
                    if period_key not in prev_data:
                        continue  # skip unmatched periods

                    df_curr = pd.DataFrame(curr_data[period_key])
                    df_prev = pd.DataFrame(prev_data.get(period_key, []))

                    period_result = {}
                    for category in ["asset", "liability", "equity", "depreciation"]:
                        curr_sum = df_curr[df_curr["category"] == category]["amount"].sum()
                        prev_sum = df_prev[df_prev["category"] == category]["amount"].sum()
                        period_result[category] = compute_change(curr_sum, prev_sum)

                    result[period_key] = period_result
                return result

            # 7. Aggregate by period
            monthly = FinancialReport._aggregate_data(df, "ME", "%b")
            quarterly = FinancialReport._aggregate_data(df, "QE", "Q%q")
            yearly = FinancialReport._aggregate_data(df, "YE", "%Y")

            key_metric_changes = {}

            if yearly and yearly_prev:
                curr_year_key = next(iter(yearly))
                prev_year_key = next(iter(yearly_prev))

                df_curr = pd.DataFrame(yearly[curr_year_key])
                df_prev = pd.DataFrame(yearly_prev[prev_year_key])

                def sum_amount(df, category, account=None):
                    data = df[df["category"] == category]
                    if account:
                        data = data[data["account_name"] == account]
                    return data["amount"].sum()

                curr_assets = sum_amount(df_curr, "asset")
                prev_assets = sum_amount(df_prev, "asset")

                curr_liabilities = sum_amount(df_curr, "liability")
                prev_liabilities = sum_amount(df_prev, "liability")

                curr_equity = sum_amount(df_curr, "equity")
                prev_equity = sum_amount(df_prev, "equity")

                curr_depr = sum_amount(df_curr, "depreciation", "depreciation")
                prev_depr = sum_amount(df_prev, "depreciation", "depreciation")

                key_metric_changes = {
                    "Net_Equity": compute_change(curr_equity, prev_equity),
                    "Total_Assets": compute_change(curr_assets + curr_depr, prev_assets + prev_depr),
                    "Total_Liabilities": compute_change(curr_liabilities, prev_liabilities),
                    "Asset_Growth": compute_change(curr_assets, prev_assets),
                    "Liability_Growth": compute_change(curr_liabilities, prev_liabilities),
                    "Current_Ratio": compute_change(
                        round(curr_assets / curr_liabilities, 2) if curr_liabilities else 0,
                        round(prev_assets / prev_liabilities, 2) if prev_liabilities else 0
                    ),
                    "Debt_to_Equity": compute_change(
                        round(curr_liabilities / curr_equity, 2) if curr_equity else 0,
                        round(prev_liabilities / prev_equity, 2) if prev_equity else 0
                    ),
                    "Accounting_Equation_Check": compute_change(
                        round(curr_assets - (curr_liabilities + curr_equity), 2),
                        round(prev_assets - (prev_liabilities + prev_equity), 2)
                    )
                }

            # 9. Return final data
            return {
                "monthly_data": monthly,
                "quarterly_data": quarterly,
                "yearly_data": yearly,
                "previous _year":yearly_prev,
                "key_metrics": key_metric_changes,
                "date_range": {
                    "start_date": current_start.isoformat(),
                    "end_date": current_end.isoformat()
                }
            }
        except Exception as e:
            logger.error(f"Failed to generate balance sheet: {e}")
            raise FinancialReportError(f"Failed to generate balance sheet: {e}")

            
       

       


    

import logging
from collections import defaultdict
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Dict, Optional, Tuple

from sqlalchemy import and_, extract, func, select
from sqlalchemy.ext.asyncio import AsyncSession
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, Tuple, List, Union
from sqlalchemy import select, func, text, and_, extract
from sqlalchemy.ext.asyncio import AsyncSession
import logging
import json

logger = logging.getLogger(__name__)

from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from collections import defaultdict
import pandas as pd
import numpy as np
from sqlalchemy import and_, extract, func, select
from sqlalchemy.ext.asyncio import AsyncSession
import logging

logger = logging.getLogger(__name__)

class FinancialReportError(Exception):
    """Base exception for FinancialReport errors"""
    pass

class FinancialReport:
    # Constants for tag categories
    ASSET_TAGS = {"current_asset", "fixed_asset"}
    LIABILITY_TAGS = {"current_liability"}
    EQUITY_TAGS = {"equity","financial"}
    DEPRECIATION_TAGS = {"non_cash_expense", "operating_expense"}
    REVENUE_TAGS = {"income", "revenue"}
    OPERATING_EXPENSE_TAGS = {"operating_expense"}
    PL_CATEGORIES = {"revenue", "cogs", "expense"}

    COGS_TAGS = {"cogs", "cost_of_goods_sold", "cost_of_sales"}
    OPERATING_CASHIN={"operating","cash_in"}
    OPERATING_CASHOUT={"operating","cash_out"}
    INVESTING_CASHIN={"investing","cash_in"}
    INVESTING_CASHOUT={"investing","cash_out"}
    FINANCING_CASHIN={"financing","cash_in"}
    FINANCING_CASHOUT={"financing","cash_out"}
    MONTH_NAMES = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", 
                  "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]

    @staticmethod
    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")

    @staticmethod
    async def _execute_query(db: AsyncSession, stmt) -> List:
        """Execute a database query with error handling"""
        try:
            result = await db.execute(stmt)
            return result.all()
        except Exception as e:
            logger.error(f"Database query failed: {e}")
            raise FinancialReportError(f"Database query failed: {e}")

    @staticmethod
    def _parse_tags(tags: Union[str, List, None]) -> Set[str]:
        """Parse tags from various formats into a set of normalized strings"""
        if tags is None:
            return set()
            
        if isinstance(tags, str):
            # Handle string format like "{tag1, tag2, tag3}"
            if tags.startswith('{') and tags.endswith('}'):
                tags = tags[1:-1]
            tags_list = [t.strip().lower() for t in tags.split(",")]
        elif isinstance(tags, list):
            tags_list = [str(t).strip().lower() for t in tags]
        else:
            return set()
            
        return {tag for tag in tags_list if tag}
    @staticmethod
    def get_cashflow_accounts_by_tags(
        mapping_df: pd.DataFrame, 
        category: str,           # e.g. "Operating", "Investing", "Financing"
        direction: str,          # "cash_in" or "cash_out"
        tag_column: str = "tags"
    ) -> List[str]:
        """
        Return GL codes for cash flow accounts matching category and cash direction.

        Args:
            mapping_df: DataFrame with account data
            category: One of "Operating", "Investing", "Financing"
            direction: "cash_in" or "cash_out"
            tag_column: column containing tags info
        
        Returns:
            List of general ledger codes matching the category + cash direction tags
        """
        import ast

        if mapping_df.empty:
            return []

        # normalize tags to lowercase
        required_tags = {category.lower(), direction.lower()}

        def tags_match(tags):
            try:
                if isinstance(tags, str):
                    parsed = ast.literal_eval(tags)
                else:
                    parsed = tags
                parsed_set = {str(t).strip().lower() for t in parsed or []}
            except Exception:
                return False
            return required_tags.issubset(parsed_set)

        matched_rows = mapping_df[mapping_df[tag_column].apply(tags_match)]
        return matched_rows["generalledgercode"].tolist()

    @staticmethod
    def _tag_match(
        tags: Union[str, List, None], 
        target_tags: Set[str], 
        match_all_tags: bool = False
    ) -> bool:
        """
        Check if tags match the target_tags with either OR (any) or AND (all) logic.

        Parameters:
            tags: tags to check (string or list)
            target_tags: set of tags to match against
            match_all_tags: if True, all target_tags must be present in tags; if False, any tag matches

        Returns:
            bool: True if match, else False
        """
        if not tags or not target_tags:
            return False

        parsed_tags = FinancialReport._parse_tags(tags)

        if match_all_tags:
            # AND logic: all target_tags must be present in parsed_tags
            return target_tags.issubset(parsed_tags)
        else:
            # OR logic: any tag matches
            return bool(parsed_tags & target_tags)


    @staticmethod
    async def get_account_mapping(db: AsyncSession) -> pd.DataFrame:
        """Get account mapping from GeneralLedgerMaster"""
        try:
            stmt = select(
                func.lower(GeneralLedgerMaster.account_name),
                GeneralLedgerMaster.tags,
                GeneralLedgerMaster.general_ledger_code
            )
            rows = await FinancialReport._execute_query(db, stmt)
            
            mapping = pd.DataFrame(rows, columns=["account_name", "tags", "generalledgercode"])
            mapping["account_name"] = mapping["account_name"].str.strip().str.lower()
            return mapping.set_index("account_name")
        except Exception as e:
            logger.error(f"Failed to get account mapping: {e}")
            raise FinancialReportError(f"Failed to get account mapping: {e}")

    @staticmethod
    def get_accounts_by_tags(
        mapping_df: pd.DataFrame, 
        tag_names: Set[str], 
        tag_column: str = "tags", match_all_tags:bool =False
    ) -> List[str]:
        """
        Returns account GL codes from a DataFrame where tags match the given tag_names set.

        Parameters:
            mapping_df: Input DataFrame with account mapping
            tag_names: Tags to match against
            tag_column: Column name in df containing tag strings

        Returns:
            List of matching GL codes
        """
        try:
            if mapping_df.empty:
                return []
            #print(mapping_df,tag_names)
            target_tags = {tag.lower().strip() for tag in tag_names}
            
            # Filter rows where tags match any of the target tags
            matched = mapping_df[mapping_df[tag_column].apply(
                lambda tags: FinancialReport._tag_match(tags, target_tags, match_all_tags)
            )]
            #print(matched)
            
            return matched["generalledgercode"].tolist()
        except Exception as e:
            logger.error(f"Failed to get accounts by tags: {e}")
            raise FinancialReportError(f"Failed to get accounts by tags: {e}")

    @staticmethod
    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:
                FinancialReport._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}")

    @staticmethod
    async def load_ledger_data(
        db: AsyncSession, 
        start_date: datetime, 
        end_date: datetime,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> pd.DataFrame:
        """Load ledger data efficiently using SQL aggregation"""
        try:
            logger.debug(f"Loading ledger data from {start_date} to {end_date}")
            
            # Build filters
            filters = [
                GeneralLedger.date >= start_date,
                GeneralLedger.date <= end_date
            ]
            
            if entity:
                filters.append(GeneralLedger.entity.ilike(f"%{entity}%"))
            if hierarchy:
                filters.append(GeneralLedger.hierarchy.ilike(f"%{hierarchy}%"))
            
            # Use SQL aggregation to reduce data transfer
            stmt = select(
                GeneralLedger.account_name,
                func.sum(GeneralLedger.debit).label('total_debit'),
                func.sum(GeneralLedger.credit).label('total_credit'),
                func.count().label('entry_count')
            ).where(
                and_(*filters)
            ).group_by(
                GeneralLedger.account_name
            )
            
            rows = await FinancialReport._execute_query(db, stmt)
            
            # Convert to pandas DataFrame
            df = pd.DataFrame(rows, columns=['account_name', 'debit', 'credit', 'entry_count'])
            df['account_name'] = df['account_name'].str.strip().str.lower()
            
            logger.debug(f"Loaded {len(df)} aggregated ledger entries")
            return df
        except Exception as e:
            logger.error(f"Failed to load ledger data: {e}")
            raise FinancialReportError(f"Failed to load ledger data: {e}")

    @staticmethod
    def filter_entity_and_group(
        df: pd.DataFrame, 
        entity: Optional[str] = None, 
        hierarchy_level: str = "group", 
        filters: Optional[dict] = None
    ) -> pd.DataFrame:
        """
        Filter raw data by entity (if given), then apply hierarchy grouping and other filters.

        Args:
            df: raw data DataFrame (with extra_data column)
            entity: optional entity filter (filters extra_data["entity"])
            hierarchy_level: 'group', 'department', or 'product'
            filters: dict of other filters to apply during grouping (optional)

        Returns:
            Filtered and grouped DataFrame
        """
        try:
            if df.empty:
                return df

            # Step 1: Filter by entity if provided
            if entity and "extra_data" in df.columns:
                df = df[df["extra_data"].apply(lambda x: x.get("entity") == entity if x else False)]

            # Step 2: Apply other filters except entity
            if filters:
                for key, value in filters.items():
                    if value is None or key == "entity":
                        continue
                    if key in df.columns:
                        df = df[df[key] == value]
                    elif "extra_data" in df.columns:
                        df = df[df["extra_data"].apply(lambda x: x.get(key) == value if x else False)]

            # Step 3: Group by hierarchy level
            if hierarchy_level == "group":
                return df
            elif hierarchy_level == "department":
                if "extra_data" not in df.columns:
                    return df
                    
                df["department"] = df["extra_data"].apply(lambda x: x.get("department") if x else None)
                grouped = df.groupby("department", dropna=False).agg({
                    "debit": "sum",
                    "credit": "sum"
                }).reset_index()
                return grouped
            elif hierarchy_level == "product":
                if "extra_data" not in df.columns:
                    return df
                    
                def get_product(x):
                    if not x:
                        return None
                    return x.get("product_code") or x.get("project_code") or None

                df["product"] = df["extra_data"].apply(get_product)
                grouped = df.groupby("product", dropna=False).agg({
                    "debit": "sum",
                    "credit": "sum"
                }).reset_index()
                return grouped
            else:
                raise ValueError("Invalid hierarchy_level. Must be one of 'group', 'department', or 'product'.")
        except Exception as e:
            logger.error(f"Failed to filter and group data: {e}")
            raise FinancialReportError(f"Failed to filter and group data: {e}")

    @staticmethod
    async def _fetch_gl_data_by_codes(
        db_session: AsyncSession,
        gl_codes: List[str],
        start_date: datetime,
        end_date: datetime
    ) -> pd.DataFrame:
        """Fetch GL data for specific GL codes within a date range"""
        try:
            if not gl_codes:
                return pd.DataFrame()

            stmt = select(
                GeneralLedger.account_name,
                GeneralLedger.debit,
                GeneralLedger.credit,
                GeneralLedger.date,
                GeneralLedger.general_ledger_code,
                GeneralLedger.extra_data
            ).where(
                and_(
                    GeneralLedger.general_ledger_code.in_(gl_codes),
                    GeneralLedger.date >= start_date,
                    GeneralLedger.date <= end_date
                )
            )

            rows = await FinancialReport._execute_query(db_session, stmt)

            if not rows:
                return pd.DataFrame()

            return pd.DataFrame(
                rows, 
                columns=["account_name", "debit", "credit", "date", "general_ledger_code", "extra_data"]
            )
        except Exception as e:
            logger.error(f"Failed to fetch GL data by codes: {e}")
            raise FinancialReportError(f"Failed to fetch GL data by codes: {e}")

    @staticmethod
    async def _fetch_categorized_data(
        db: AsyncSession,
        category_mapping: Dict[str, List[str]],
        start_date: datetime,
        end_date: datetime
    ) -> pd.DataFrame:
        """Fetch data for multiple categories"""
        try:
            dfs = []
            
            for category, gl_codes in category_mapping.items():
                if not gl_codes:
                    continue
                    
                df = await FinancialReport._fetch_gl_data_by_codes(db, gl_codes, start_date, end_date)
                if not df.empty:
                    df["category"] = category
                    dfs.append(df)

            if dfs:
                return pd.concat(dfs, ignore_index=True)
            else:
                return pd.DataFrame()
        except Exception as e:
            logger.error(f"Failed to fetch categorized data: {e}")
            raise FinancialReportError(f"Failed to fetch categorized data: {e}")

    @staticmethod
    def _calculate_amount(row: pd.Series) -> float:
        try:
            debit = float(row.get("debit", 0) or 0)
            credit = float(row.get("credit", 0) or 0)
            category = str(row.get("category", "")).strip().lower()

            print(f"Calculating amount for category: {category}, debit: {debit}, credit: {credit}")

            if category in {"operating_cashin", "investing_cashin", "financing_cashin"}:
                return credit - debit  # Inflow
            elif category in {"operating_cashout", "investing_cashout", "financing_cashout"}:
                return debit - credit  # Outflow
            elif category == "depreciation":
                return -(debit - credit)
            elif category in {"asset", "current_assets"}:
                return debit - credit
            elif category in {"liability", "current_liabilities", "total_liabilities", "equity"}:
                return credit - debit
            elif category == "revenue":
                return credit - debit
            elif category in {"operating_expense", "expense","cogs"}:
                return debit - credit
            else:
                print(f"Unknown category: {category}")
                return 0.0
        except Exception as e:
            logger.error(f"Failed to calculate amount: {e}")
            return 0.0

    def split_customer_names(name_string):
                if pd.isna(name_string):
                    return []
                
                # Normalize: replace ' and ' with ',', then split
                name_string = name_string.replace(" and ", ",")
                
                # Now split by comma and clean each name
                names = [name.strip().title() for name in name_string.split(",") if name.strip()]
                
                return names

    # @staticmethod
    # def _aggregate_data(
    #     df: pd.DataFrame, 
    #     freq: str, 
    #     key_format: str
    # ) -> Dict[str, List[Dict[str, Any]]]:
    #     """Aggregate data by time period"""
    #     try:
    #         if df.empty:
    #             return {}
                
    #         grouped = df.groupby([pd.Grouper(freq=freq), "category", "account_name"])["amount"].sum().reset_index()
    #         grouped.rename(columns={"date": "period"}, inplace=True)

    #         result = defaultdict(list)
    #         for _, row in grouped.iterrows():
    #             key = row["period"].strftime(key_format)
    #             result[key].append({
    #                 "category": row["category"],
    #                 "account_name": row["account_name"],
    #                 "amount": float(row["amount"])
    #             })
                
    #         return dict(result)
    #     except Exception as e:
    #         logger.error(f"Failed to aggregate data: {e}")
    #         raise FinancialReportError(f"Failed to aggregate data: {e}")
    @staticmethod
    def _aggregate_data(
        df: pd.DataFrame, 
        freq: str, 
        key_format: str
    ) -> Dict[str, List[Dict[str, Any]]]:
        """Aggregate data by time period"""
        try:
            if df.empty:
                return {}

            grouped = df.groupby([pd.Grouper(freq=freq), "category", "account_name"])["amount"].sum().reset_index()
            grouped.rename(columns={"date": "period"}, inplace=True)

            result = defaultdict(list)
            for _, row in grouped.iterrows():
                if key_format == "Q%q":
                    # Handle quarterly format manually
                    quarter = (row["period"].month - 1) // 3 + 1
                    key = f"Q{quarter}"
                else:
                    key = row["period"].strftime(key_format)

                result[key].append({
                    "category": row["category"],
                    "account_name": row["account_name"],
                    "amount": float(row["amount"])
                })

            return dict(result)
        except Exception as e:
            logger.error(f"Failed to aggregate data: {e}")
            raise FinancialReportError(f"Failed to aggregate data: {e}")

    @staticmethod
    async def balance_sheet(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None,
        compare_previous: bool = True
    ) -> Dict[str, Any]:
        """Generate balance sheet report"""
        try:
            # 1. Set date range
            current_start, current_end = FinancialReport.get_date_range(
                period, year, start_date, end_date
            )
            FinancialReport._validate_date_range(current_start, current_end)

            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
           
            # 3. Get accounts by category
            # 3. Get accounts by category
            accounts = {
                "asset": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.ASSET_TAGS),
                "liability": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.LIABILITY_TAGS),
                "equity": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.EQUITY_TAGS),
                "depreciation": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.DEPRECIATION_TAGS),
            }


            # 4. Fetch data for all categories
            asset_df = await FinancialReport._fetch_categorized_data(db, accounts, current_start, current_end)
            
            if asset_df.empty:
                return {
                    "monthly_data": {},
                    "quarterly_data": {},
                    "yearly_data": {},
                    "key_metrics": {},
                    "date_range": {
                        "start_date": current_start.isoformat(),
                        "end_date": current_end.isoformat()
                    }
                }

            # 5. Filter and process data
            other_filter = None
            df = FinancialReport.filter_entity_and_group(
                df=asset_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
            
            # Prepare DataFrame
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category"])
            df["date"] = pd.to_datetime(df["date"])
            df["account_name"] = df["account_name"].str.strip().str.lower()
            
            # 6. Calculate net amount
            df["amount"] = df.apply(FinancialReport._calculate_amount, axis=1)
            df.set_index("date", inplace=True)
            
            # 7. Aggregate by period
            monthly = FinancialReport._aggregate_data(df, "ME", "%b")
            quarterly = FinancialReport._aggregate_data(df, "QE", "Q%q")
            yearly = FinancialReport._aggregate_data(df, "YE", "%Y")

            # 8. Calculate key metrics from the latest year
            metrics = {}
            if yearly:
                year_key = next(iter(yearly), None)
                if year_key:
                    df_year = pd.DataFrame(yearly[year_key])
                    total_assets = df_year[df_year["category"] == "asset"]["amount"].sum()
                    total_liabilities = df_year[df_year["category"] == "liability"]["amount"].sum()
                    total_equity = df_year[df_year["category"] == "equity"]["amount"].sum()
                    total_depreciation = df_year[
                        (df_year["category"] == "depreciation") & 
                        (df_year["account_name"] == "depreciation")
                    ]["amount"].sum()

                    metrics = {
                        "Total_Assets": total_assets + total_depreciation,
                        "Total_Liabilities": total_liabilities,
                        "Total_Equity": total_equity,
                        "Current_Ratio": round(total_assets / total_liabilities, 2) if total_liabilities else None,
                        "Debt_to_Equity": round(total_liabilities / total_equity, 2) if total_equity else None,
                        "Accounting_Equation_Check": round(total_assets - (total_liabilities + total_equity), 2)
                    }

            # 9. Return final data
            return {
                "monthly_data": monthly,
                "quarterly_data": quarterly,
                "yearly_data": yearly,
                "key_metrics": metrics,
                "date_range": {
                    "start_date": current_start.isoformat(),
                    "end_date": current_end.isoformat()
                }
            }
        except Exception as e:
            logger.error(f"Failed to generate balance sheet: {e}")
            raise FinancialReportError(f"Failed to generate balance sheet: {e}")

    @staticmethod
    async def get_product_revenue_report(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate product revenue report with monthly, quarterly, and yearly data"""
        try:
            # 1. Set date range
            start_date, end_date = FinancialReport.get_date_range(period, year, start_date, end_date)
            FinancialReport._validate_date_range(start_date, end_date)
            
            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get revenue accounts
            revenue_accounts = {
                "revenue": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.REVENUE_TAGS)
            }
            
            # 4. Fetch revenue data
            revenue_df = await FinancialReport._fetch_categorized_data(db, revenue_accounts, start_date, end_date)
            
            if revenue_df.empty:
                return {
                    "monthly_revenue": {},
                    "quarterly_revenue": {},
                    "yearly_revenue": {},
                    "summary": {
                        "total_revenue": 0,
                        "top_product": None,
                        "top_product_revenue": 0,
                        "product_count": 0,
                        "average_revenue_per_product": 0,
                        "monthly_average": 0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }

            # 5. Filter and process data
            other_filter = None
            df = FinancialReport.filter_entity_and_group(
                df=revenue_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
            
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category"])
            df["date"] = pd.to_datetime(df["date"])
            df["account_name"] = df["account_name"].str.strip().str.lower()
            df["amount"] = df["credit"] - df["debit"]
            df.set_index("date", inplace=True)
            
            # 6. Aggregate data
            monthly_revenue = FinancialReport._aggregate_data(df, "ME", "%b")
            quarterly_revenue = FinancialReport._aggregate_data(df, "QE", "Q%q")
            yearly_revenue = FinancialReport._aggregate_data(df, "YE", "%Y")
            
            # 7. Calculate summary
            total_revenue = df["amount"].sum()
            yearly_summary = df.groupby("account_name")["amount"].sum().to_dict()
            product_count = df["account_name"].nunique()

            # 8. Return final report
            return {
                "monthly_revenue": monthly_revenue,
                "quarterly_revenue": quarterly_revenue,
                "yearly_revenue": yearly_revenue,
                "summary": {
                    "total_revenue": float(total_revenue),
                    "top_product": max(yearly_summary, key=yearly_summary.get) if yearly_summary else None,
                    "top_product_revenue": max(yearly_summary.values()) if yearly_summary else 0,
                    "product_count": product_count,
                    "average_revenue_per_product": float(total_revenue / product_count) if product_count > 0 else 0,
                    "monthly_average": float(total_revenue / 12)
                },
                "date_range": {
                    "start_date": start_date.strftime("%Y-%m-%d"),
                    "end_date": end_date.strftime("%Y-%m-%d")
                }
            }
        except Exception as e:
            logger.error(f"Failed to generate product revenue report: {e}")
            raise FinancialReportError(f"Failed to generate product revenue report: {e}")

    @staticmethod
    async def get_operation_expense_report(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None,
        department: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate operational expense report with monthly, quarterly, and yearly data"""
        try:
            # 1. Set date range
            start_date, end_date = FinancialReport.get_date_range(period, year, start_date, end_date)
            FinancialReport._validate_date_range(start_date, end_date)
            
            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get operating expense accounts
            operating_expense_accounts = {
                "operating_expense": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.OPERATING_EXPENSE_TAGS)
            }
            print(operating_expense_accounts)
            
            # 4. Fetch operating expense data
            operating_expense_df = await FinancialReport._fetch_categorized_data(db, operating_expense_accounts, start_date, end_date)
            #print(operating_expense_df)
            if operating_expense_df.empty:
                return {
                    "monthly_expenses": {},
                    "quarterly_expenses": {},
                    "yearly_expenses": {},
                    "department_account_breakdown": [],
                    "summary": {
                        "total_expenses": 0,
                        "department_count": 0,
                        "top_department": None,
                        "top_department_expense": 0,
                        "average_expense_per_department": 0,
                        "monthly_average_expense": 0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }

            # 5. Filter and process data
            other_filter = {"department": department} if department else None
            df = FinancialReport.filter_entity_and_group(
                df=operating_expense_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
            #print(df)
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category", "department"])
            print(df)
            df["date"] = pd.to_datetime(df["date"])
            # FIX: Convert to string and handle NaN values before using .str accessor
            for column in ["category", "department", "account_name"]:
                if column in df.columns:
                    # Fill NaN with empty string and convert to string
                    df[column] = df[column].fillna('').astype(str)
                    # Now safely apply string operations
                    df[column] = df[column].str.strip().str.lower()
            
            df["amount"] = df["debit"] - df["credit"]
            df.set_index("date", inplace=True)
            print(df)
            # 6. Aggregate data
            monthly_expenses = FinancialReport._aggregate_data(df, "ME", "%b")
            quarterly_expenses = FinancialReport._aggregate_data(df, "QE", "Q%q")
            
            # 7. Calculate yearly summary
            yearly_summary = df.groupby('account_name')['amount'].sum().sort_values(ascending=False)
            department_account_summary = df.groupby(['department', 'account_name'])['amount'].sum().reset_index()
            
            total_expenses = yearly_summary.sum()
            department_count = len(yearly_summary)
            
            # 8. Return final report
            return {
                "monthly_expenses": monthly_expenses,
                "quarterly_expenses": quarterly_expenses,
                "yearly_expenses": {k: float(v) for k, v in yearly_summary.to_dict().items()},
                "department_account_breakdown": department_account_summary.to_dict('records'),
                "summary": {
                    "total_expenses": float(total_expenses),
                    "department_count": department_count,
                    "top_department": yearly_summary.index[0] if not yearly_summary.empty else None,
                    "top_department_expense": float(yearly_summary.iloc[0]) if not yearly_summary.empty else 0,
                    "average_expense_per_department": float(total_expenses / department_count) if department_count > 0 else 0,
                    "monthly_average_expense": float(total_expenses / 12)
                },
                "date_range": {
                    "start_date": start_date.isoformat(),
                    "end_date": end_date.isoformat()
                }
            }
        except Exception as e:
            logger.error(f"Failed to generate operation expense report: {e}")
            raise FinancialReportError(f"Failed to generate operation expense report: {e}")
    @staticmethod
    async def profit_loss_statement(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate comprehensive profit and loss statement with all time periods"""
        try:
            # 1. Get date range
            start_date, end_date = FinancialReport.get_date_range(
                period, year, start_date, end_date
            )
            
            # 2. Load account mapping
            account_mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get GL codes for P&L categories using reusable function
            revenue_gl_codes = FinancialReport.get_accounts_by_tags(
                account_mapping, FinancialReport.REVENUE_TAGS
            )
            operating_expense_gl_codes = FinancialReport.get_accounts_by_tags(
                account_mapping, FinancialReport.OPERATING_EXPENSE_TAGS
            )
            
            # 4. Add COGS tags and get COGS GL codes
            
            cogs_gl_codes = FinancialReport.get_accounts_by_tags(
                account_mapping, FinancialReport.COGS_TAGS
            )
            
            # 5. Create category mapping for fetching data
            category_mapping = {
                "revenue": revenue_gl_codes,
                "operating_expense": operating_expense_gl_codes,
                "cogs": cogs_gl_codes
            }
            
            # 6. Fetch categorized data using reusable function
            categorized_df = await FinancialReport._fetch_categorized_data(
                db, category_mapping, start_date, end_date
            )
            
            if categorized_df.empty:
                return {
                    "monthly_data": {},
                    "quarterly_data": {},
                    "yearly_data": [],
                    "key_metrics": {
                        "Total_Revenue": 0.0,
                        "Total_COGS": 0.0,
                        "Total_Expenses": 0.0,
                        "Gross_Profit": 0.0,
                        "Net_Income": 0.0,
                        "Gross_Margin_Percentage": 0.0,
                        "Net_Margin_Percentage": 0.0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }
            
            # 7. Filter by entity and hierarchy using reusable function
            filtered_df = FinancialReport.filter_entity_and_group(
                df=categorized_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=None
            )
            
            # 8. Calculate amount for each row using reusable function
            # Update _calculate_amount to handle COGS category
            filtered_df["amount"] = filtered_df.apply(
                lambda row: FinancialReport._calculate_amount(row) if row["category"] != "cogs" 
                            else float(row.get("debit", 0) - row.get("credit", 0)),
                axis=1
            )
            
            # 9. Ensure date column is datetime type and set as index for aggregation
            filtered_df["date"] = pd.to_datetime(filtered_df["date"])
            filtered_df.set_index("date", inplace=True)
            
            # 10. Aggregate data by different time periods using reusable function
            monthly_data = FinancialReport._aggregate_data(
                filtered_df, "M", "%b"
            )
            
            quarterly_data = FinancialReport._aggregate_data(
                filtered_df, "Q", "Q%q"
            )
            
            # Reset index for yearly aggregation
            filtered_df_reset = filtered_df.reset_index()
            
            # 11. Calculate yearly summary
            yearly_summary = filtered_df_reset.groupby(["category", "account_name"])["amount"].sum().reset_index()
            
            # 12. Calculate key metrics
            revenue = yearly_summary[yearly_summary["category"] == "revenue"]["amount"].sum()
            cogs = yearly_summary[yearly_summary["category"] == "cogs"]["amount"].sum()
            operating_expenses = yearly_summary[yearly_summary["category"] == "operating_expense"]["amount"].sum()
            
            gross_profit = revenue - cogs
            net_income = gross_profit - operating_expenses
            gross_margin = (gross_profit / revenue * 100) if revenue != 0 else 0
            net_margin = (net_income / revenue * 100) if revenue != 0 else 0
            
            metrics = {
                "Total_Revenue": float(revenue),
                "Total_COGS": float(cogs),
                "Total_Expenses": float(operating_expenses),
                "Gross_Profit": float(gross_profit),
                "Net_Income": float(net_income),
                "Gross_Margin_Percentage": float(gross_margin),
                "Net_Margin_Percentage": float(net_margin)
            }
            
            # 13. Return final report
            return {
                "monthly_data": monthly_data,
                "quarterly_data": quarterly_data,
                "yearly_data": yearly_summary.to_dict('records'),
                "key_metrics": metrics,
                "date_range": {
                    "start_date": start_date.isoformat(),
                    "end_date": end_date.isoformat()
                }
            }
            
        except Exception as e:
            logger.error(f"Failed to generate profit loss statement: {e}")
            raise FinancialReportError(f"Failed to generate profit loss statement: {e}")
    
    @staticmethod
    async def get_cashflow_statement(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
    ) -> Dict[str, Any]:
        try:
            # 1. Get date range
            start_date, end_date = FinancialReport.get_date_range(
                period, year, start_date, end_date
            )
            
            # 2. Load account mapping
            account_mapping = await FinancialReport.get_account_mapping(db)
            
            
            accounts = {
    "Operating_cashin": FinancialReport.get_cashflow_accounts_by_tags(account_mapping, "Operating", "cash_in"),
    "Operating_cashout": FinancialReport.get_cashflow_accounts_by_tags(account_mapping, "Operating", "cash_out"),
    "Investing_cashin": FinancialReport.get_cashflow_accounts_by_tags(account_mapping, "Investing", "cash_in"),
    "Investing_cashout": FinancialReport.get_cashflow_accounts_by_tags(account_mapping, "Investing", "cash_out"),
    "Financing_cashin": FinancialReport.get_cashflow_accounts_by_tags(account_mapping, "Financing", "cash_in"),
    "Financing_cashout": FinancialReport.get_cashflow_accounts_by_tags(account_mapping, "Financing", "cash_out"),


            }
            
            # 4. Add COGS tags and get COGS GL codes
            print(accounts)
            # 6. Fetch categorized data using reusable function
            categorized_df = await FinancialReport._fetch_categorized_data(
                db, accounts, start_date, end_date
            )
            #print(categorized_df["category"].unique())
            
            if categorized_df.empty:
                return {
                    "monthly_data": {},
                    "quarterly_data": {},
                    "yearly_data": [],
                    "key_metrics": {
                        "Total_Revenue": 0.0,
                        "Total_COGS": 0.0,
                        "Total_Expenses": 0.0,
                        "Gross_Profit": 0.0,
                        "Net_Income": 0.0,
                        "Gross_Margin_Percentage": 0.0,
                        "Net_Margin_Percentage": 0.0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }
            
            # 7. Filter by entity and hierarchy using reusable function
            filtered_df = FinancialReport.filter_entity_and_group(
                df=categorized_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=None
            )
            
            # 8. Calculate amount for each row using reusable function
            # Update _calculate_amount to handle COGS category
            filtered_df["amount"] = filtered_df.apply(
                lambda row: FinancialReport._calculate_amount(row),
                axis=1
            )
            
            
            
            # 9. Ensure date column is datetime type and set as index for aggregation
            filtered_df["date"] = pd.to_datetime(filtered_df["date"])
            filtered_df.set_index("date", inplace=True)
           
            # # 10. Aggregate data by different time periods using reusable function
            monthly_data = FinancialReport._aggregate_data(
                filtered_df, "M", "%b"
            )
            
            quarterly_data = FinancialReport._aggregate_data(
                filtered_df, "Q", "Q%q"
            )
            
            # # Reset index for yearly aggregation
            filtered_df_reset = filtered_df.reset_index()
            
            # # 11. Calculate yearly summary
            yearly_summary = filtered_df_reset.groupby(["category", "account_name"])["amount"].sum().reset_index()
            print(pd.concat([yearly_summary["category"], yearly_summary["account_name"]]).unique())

            ##monthly cashflow
            monthly_Operating_cashin=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashin"]["amount"].sum()
            monthly_Operating_cashout=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashout"]["amount"].sum()
            monthly_Investing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashin"]["amount"].sum()
            monthly_Investing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashout"]["amount"].sum()
            monthly_Financing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashin"]["amount"].sum()
            monthly_Financing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashout"]["amount"].sum()
            monthly_operation_cashflow=monthly_Operating_cashin-monthly_Operating_cashout
            monthly_investing_cashflow=monthly_Investing_cashin-monthly_Investing_cashout
            monthly_financing_cashflow=monthly_Financing_cashin-monthly_Financing_cashout
            monthly_net_cashflow=monthly_operation_cashflow+monthly_investing_cashflow+monthly_financing_cashflow
            
            #quarterly cashflow
            quarterly_Operating_cashin=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashin"]["amount"].sum()

            quarterly_Operating_cashout=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashout"]["amount"].sum()
            quarterly_Investing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashin"]["amount"].sum()
            quarterly_Investing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashout"]["amount"].sum()
            quarterly_Financing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashin"]["amount"].sum()
            quarterly_Financing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashout"]["amount"].sum()
            quarterly_operation_cashflow=quarterly_Operating_cashin-quarterly_Operating_cashout
            quarterly_investing_cashflow=quarterly_Investing_cashin-quarterly_Investing_cashout
            quarterly_financing_cashflow=quarterly_Financing_cashin-quarterly_Financing_cashout
            quarterly_net_cashflow=quarterly_operation_cashflow+quarterly_investing_cashflow+quarterly_financing_cashflow
            #yearly cashflow
            yearly_Operating_cashin=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashin"]["amount"].sum()
            yearly_Operating_cashout=filtered_df_reset[filtered_df_reset["category"]=="Operating_cashout"]["amount"].sum()
            yearly_Investing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashin"]["amount"].sum()
            yearly_Investing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Investing_cashout"]["amount"].sum()
            yearly_Financing_cashin=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashin"]["amount"].sum()
            yearly_Financing_cashout=filtered_df_reset[filtered_df_reset["category"]=="Financing_cashout"]["amount"].sum()
            yearly_operation_cashflow=yearly_Operating_cashin-yearly_Operating_cashout
            yearly_investing_cashflow=yearly_Investing_cashin-yearly_Investing_cashout
            yearly_financing_cashflow=yearly_Financing_cashin-yearly_Financing_cashout
            yearly_net_cashflow=yearly_operation_cashflow+yearly_investing_cashflow+yearly_financing_cashflow
            # Fetch balance sheet data for ratios
            balance_sheet_accounts = {
                "current_assets": FinancialReport.get_accounts_by_tags(account_mapping, {"current_asset"}),
                "current_liabilities": FinancialReport.get_accounts_by_tags(account_mapping, {"current_liability"}),
                "total_liabilities": FinancialReport.get_accounts_by_tags(account_mapping,{ "liability"}),
                "equity": FinancialReport.get_accounts_by_tags(account_mapping, {"equity"})
            }
            #print("balance_sheet_accounts:",balance_sheet_accounts)

            balance_sheet_df = await FinancialReport._fetch_categorized_data(
                db, balance_sheet_accounts, start_date, end_date
            )
            #print("balancehseet:",balance_sheet_df)
             # 7. Filter by entity and hierarchy using reusable function
            filtered_df = FinancialReport.filter_entity_and_group(
               
                df=balance_sheet_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=None
            )
            
            # 8. Calculate amount for each row using reusable function
            # Update _calculate_amount to handle COGS category
            filtered_df["amount"] = filtered_df.apply(
    lambda row: FinancialReport._calculate_amount(row),
    axis=1
)
           
            

            current_assets = filtered_df[filtered_df['category'] == 'current_assets']['amount'].sum()
            current_liabilities = filtered_df[filtered_df['category'] == 'current_liabilities']['amount'].sum()
            total_liabilities = filtered_df[filtered_df['category'] == 'total_liabilities']['amount'].sum()
            equity = filtered_df[filtered_df['category'] == 'equity']['amount'].sum()
            inventory = filtered_df[filtered_df['account_name'].str.contains('inventory', case=False, na=False)]['amount'].sum()

            current_ratio = current_assets / current_liabilities if current_liabilities else None
            quick_ratio = (current_assets - inventory) / current_liabilities if current_liabilities else None
            debt_to_equity = total_liabilities / equity if equity else None
            working_capital = current_assets - current_liabilities


            return {
                "monthly_data": monthly_data,
                "quarterly_data": quarterly_data,
                "yearly_data": yearly_summary.to_dict('records'),
                "financial_ratios": {
        "current_ratio": current_ratio,
        "quick_ratio": quick_ratio,
        "debt_to_equity": debt_to_equity,
        "working_capital": working_capital
    },
                "cashflow":{
                    "Operating_cashin":quarterly_Operating_cashin,
                    "Operating_cashout":quarterly_Operating_cashout,
                    "Investing_cashin":quarterly_Investing_cashin,
                    "Investing_cashout":quarterly_Investing_cashout,
                    "Financing_cashin":quarterly_Financing_cashin,
                    "Financing_cashout":quarterly_Financing_cashout,
                    "operation_cashflow":quarterly_operation_cashflow,
                    "investing_cashflow":quarterly_investing_cashflow,
                    "financing_cashflow":quarterly_financing_cashflow,
                    "net_cashflow":quarterly_net_cashflow
                },
                "yearly_cashflow":{
                    "yearly_Operating_cash":yearly_Operating_cashin-yearly_Operating_cashout,
                    "yearly_Investing_cash":yearly_Investing_cashin-yearly_Investing_cashout,
                    "yearly_Financing_cash":yearly_Financing_cashin-yearly_Financing_cashout,
                   
                    "yearly_net_cashflow":yearly_net_cashflow
                },
                "date_range": {
                    "start_date": start_date.isoformat(),
                    "end_date": end_date.isoformat()
            }
            }
        
            
        except Exception as e:
            logger.error(f"Failed to generate profit loss statement: {e}")
            raise FinancialReportError(f"Failed to generate profit loss statement: {e}")
    @staticmethod
    async def get_customer_profit_analysis(
  db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
            ) -> Dict[str, Any]:       
        
        try:
            # 1. Set date range
            start_date, end_date = FinancialReport.get_date_range(period, year, start_date, end_date)
            FinancialReport._validate_date_range(start_date, end_date)
            
            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get operating expense accounts
            profit_accounts = {
                "Revenue": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.REVENUE_TAGS),
                "cogs":FinancialReport.get_accounts_by_tags(mapping,FinancialReport.COGS_TAGS)
            }
            print(profit_accounts)
            
            # 4. Fetch operating expense data
            profit_accounts_df = await FinancialReport._fetch_categorized_data(db, profit_accounts, start_date, end_date)
            #print(operating_expense_df)
            if profit_accounts_df.empty:
                return {
                    "customer_data": {},
                    "total_cogs": {},
                    "total_revenue": {},
                    "total profit": {},
                    "summary": {
                        "total_expenses": 0,
                        "department_count": 0,
                        "top_department": None,
                        "top_department_expense": 0,
                        "average_expense_per_department": 0,
                        "monthly_average_expense": 0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }
            department=None
            # 5. Filter and process data
            other_filter = {"department": department} if department else None
            df = FinancialReport.filter_entity_and_group(
                df=profit_accounts_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
            #print(df)
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category", "extra_data"])
            print(df)
            df["date"] = pd.to_datetime(df["date"])
            # Convert `extra_data` (dict) column to a DataFrame
            extra_df = pd.json_normalize(df['extra_data'])

            # Combine with original DataFrame (excluding the old `extra_data` column)
            df_flat = pd.concat([df.drop(columns=['extra_data']), extra_df], axis=1)
            df["region"]= df_flat["entity"]
            if("customer" in df_flat.columns):
                df["customer"]= df_flat["customer"]
            else:
                df["customer"]= df_flat["vendor"]
            
            df["industry"]= df_flat["department"]
            
            # Apply the function to every row
            df["customer"] = df["customer"].apply(FinancialReport.split_customer_names)

            # Optional: drop rows where customer list is empty
            df = df[df["customer"].map(len) > 0]

            # Reset index after filtering
            df.reset_index(drop=True, inplace=True)

            
            

        #     # FIX: Convert to string and handle NaN values before using .str accessor
            for column in ["category", "region","customer" ,"industry","account_name"]:
                if column in df.columns:
                    # Fill NaN with empty string and convert to string
                    df[column] = df[column].fillna('').astype(str)
                    # Now safely apply string operations
                    df[column] = df[column].str.strip().str.lower()
                        # Convert debit and credit columns to numeric
            df["debit"] = pd.to_numeric(df["debit"], errors="coerce")
            df["credit"] = pd.to_numeric(df["credit"], errors="coerce")

            # Optional: If you already have columns like revenue/cogs
            if "revenue" in df.columns:
                df["revenue"] = pd.to_numeric(df["revenue"], errors="coerce")
            if "cogs" in df.columns:
                df["cogs"] = pd.to_numeric(df["cogs"], errors="coerce")

            df["amount"] = df.apply(
                lambda row: FinancialReport._calculate_amount(row),
                axis=1
            )
            print(df["category"].unique())
            # Explode rows with multiple customers
            df = df.explode("customer").reset_index(drop=True)
            df.set_index("date", inplace=True)
            # Separate revenue and cogs
            df["revenue"] = df.apply(lambda row: row["credit"] if row["category"].lower() == "revenue" else 0, axis=1)
            df["cogs"] = df.apply(lambda row: row["debit"] if row["category"].lower() == "cogs" else 0, axis=1)

            # Group by customer
            customer_summary = df.groupby("customer").agg({
                "revenue": "sum",
                "cogs": "sum"
            }).reset_index()

            # Calculate profit and margin
            customer_summary["profit"] = customer_summary["revenue"] - customer_summary["cogs"]
            customer_summary["profit_margin"] = customer_summary["profit"] / customer_summary["revenue"]
            customer_summary["profit_margin"] = customer_summary["profit_margin"].round(2)

            # Replace NaNs in margin (e.g., revenue=0) with 0
            customer_summary["profit_margin"].fillna(0, inplace=True)

            # Rank customers by revenue or profit (descending)
            customer_summary["rank_by_revenue"] = customer_summary["revenue"].rank(ascending=False, method="dense").astype(int)
            customer_summary["rank_by_profit"] = customer_summary["profit"].rank(ascending=False, method="dense").astype(int)

            total_revenue = customer_summary["revenue"].sum()
            total_cogs = customer_summary["cogs"].sum()
            total_profit = customer_summary["profit"].sum()
            profit_margin = (total_profit / total_revenue) if total_revenue != 0 else 0
            profit_margin = round(profit_margin, 2)
            # Safely get top customer by revenue
            top_customer_row = customer_summary[customer_summary["rank_by_revenue"] == 1]
            top_rank_customer = top_customer_row["customer"].iloc[0] if not top_customer_row.empty else None

            # Safely get bottom customer by revenue
            bottom_customer_row = customer_summary[
                customer_summary["rank_by_revenue"] == customer_summary["rank_by_revenue"].max()
            ]
            least_rank_customer = bottom_customer_row["customer"].iloc[0] if not bottom_customer_row.empty else None

            # Debug print
            print(total_revenue, total_cogs, total_profit, profit_margin, top_rank_customer, least_rank_customer)

            print(customer_summary)

        
            return {"customer" : customer_summary.to_dict(orient="records"),
                    "total_cogs":total_cogs,
                    "total_revenue":total_revenue,
                    "total profit":total_profit,
                    "profit_margin":profit_margin,
                    "top_rank_customer":top_rank_customer,
                    "least_rank_customer":least_rank_customer,
                    "summary": {
                        "total_customers": len(customer_summary),
                        "total_revenue": total_revenue,
                        "total_cogs": total_cogs,
                        "total_profit": total_profit,
                        "overall_profit_margin": profit_margin,
                        "top_rank_customer": top_rank_customer,
                        "least_rank_customer": least_rank_customer
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
            }


        except Exception as e:
             print(e)
             logger.error(f"Failed to generate operation expense report: {e}")
             raise FinancialReportError(f"Failed to generate operation expense report: {e}")

    @staticmethod
    async def get_fixed_asset_register(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
            ) -> Dict[str, Any]:
         
        try:
            # 1. Set date range
            start_date, end_date = FinancialReport.get_date_range(period, year, start_date, end_date)
            FinancialReport._validate_date_range(start_date, end_date)
            
            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get operating expense accounts
            profit_accounts = {
                "asset": FinancialReport.get_accounts_by_tags(mapping, FinancialReport.ASSET_TAGS),
                "depreciation":FinancialReport.get_accounts_by_tags(mapping,FinancialReport.DEPRECIATION_TAGS)
            }
            
            
            # 4. Fetch operating expense data
            profit_accounts_df = await FinancialReport._fetch_categorized_data(db, profit_accounts, start_date, end_date)
            #print(profit_accounts_df)
            if profit_accounts_df.empty:
                return {
                    "customer_data": {},
                    "total_cogs": {},
                    "total_revenue": {},
                    "total profit": {},
                    "summary": {
                        "total_expenses": 0,
                        "department_count": 0,
                        "top_department": None,
                        "top_department_expense": 0,
                        "average_expense_per_department": 0,
                        "monthly_average_expense": 0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }
            department=None
            # 5. Filter and process data
            other_filter = {"department": department} if department else None
            df = FinancialReport.filter_entity_and_group(
                df=profit_accounts_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
           # print(df)
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category"])
            print(df)
            df["date"] = pd.to_datetime(df["date"])
           

            

            #             # Normalize first
            df["category"] = df["category"].str.strip().str.lower()
            df["account_name"] = df["account_name"].str.strip().str.lower()

            # # Filter
            # # Filter assets and depreciation entries
            assets_df = df[df["category"].str.lower() == "asset"].copy()
            depreciation_df = df[df["account_name"].str.lower() == "depreciation"].copy()

            print(depreciation_df)

                        # CapEx asset grouping
            assets_grouped = assets_df.groupby("account_name")["debit"].sum().reset_index().rename(columns={"debit": "cost"})

            # Depreciation grouping
            depreciation_grouped = depreciation_df.groupby("account_name")["debit"].sum().reset_index().rename(columns={"debit": "depreciation"})

            # Merge & compute NBV
            asset_status = pd.merge(assets_grouped, depreciation_grouped, on="account_name", how="left")
            asset_status["depreciation"] = asset_status["depreciation"].fillna(0)
            asset_status["net_book_value"] = asset_status["cost"] - asset_status["depreciation"]
            asset_status["status"] = asset_status["net_book_value"].apply(lambda x: "active" if x > 0 else "fully depreciated")

            # Totals
            total_cost = asset_status["cost"].sum()
            total_depreciation = asset_status["depreciation"].sum()
            total_net_book_value = asset_status["net_book_value"].sum()

            # CapEx object
            capex = {
                "fixed_asset_register": asset_status.to_dict(orient="records"),
                "summary": {
                    "total_cost": float(total_cost),
                    "total_depreciation": float(total_depreciation),
                    "total_net_book_value": float(total_net_book_value),
                    "total_assets": int(len(asset_status))
                }
            }
            return capex


        except Exception as e:
             print(e)
             logger.error(f"Failed to generate operation expense report: {e}")
             raise FinancialReportError(f"Failed to generate operation expense report: {e}")

    @staticmethod
    async def get_project_costing_and_capexd_report(
        db: AsyncSession,
        period: str = "yearly",
        year: Optional[int] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        entity: Optional[str] = None,
        hierarchy: Optional[str] = None
            ) -> Dict[str, Any]:
         
        try:
            # 1. Set date range
            start_date, end_date = FinancialReport.get_date_range(period, year, start_date, end_date)
            FinancialReport._validate_date_range(start_date, end_date)
            
            # 2. Load mapping
            mapping = await FinancialReport.get_account_mapping(db)
            
            # 3. Get operating expense accounts
            profit_accounts = {
                "asset": FinancialReport.get_accounts_by_tags(mapping,{"fixed_asset", "investing"},match_all_tags=True),
                "operating_expense": FinancialReport.get_accounts_by_tags(mapping,{"operating_expense", "cash_out"},match_all_tags=True),
                "depreciation":FinancialReport.get_accounts_by_tags(mapping,FinancialReport.DEPRECIATION_TAGS)
            }
            #print(profit_accounts)
            
            
            # 4. Fetch operating expense data
            profit_accounts_df = await FinancialReport._fetch_categorized_data(db, profit_accounts, start_date, end_date)
            
            if profit_accounts_df.empty:
                return {
                    "customer_data": {},
                    "total_cogs": {},
                    "total_revenue": {},
                    "total profit": {},
                    "summary": {
                        "total_expenses": 0,
                        "department_count": 0,
                        "top_department": None,
                        "top_department_expense": 0,
                        "average_expense_per_department": 0,
                        "monthly_average_expense": 0
                    },
                    "date_range": {
                        "start_date": start_date.isoformat(),
                        "end_date": end_date.isoformat()
                    }
                }
            department=None
            # 5. Filter and process data
            other_filter = {"department": department} if department else None
            df = FinancialReport.filter_entity_and_group(
                df=profit_accounts_df,
                entity=entity,
                hierarchy_level=hierarchy or "group",
                filters=other_filter
            )
           # print(df)
            df = pd.DataFrame(df, columns=["account_name", "debit", "credit", "date", "category","extra_data"])
            print(df)
            df["date"] = pd.to_datetime(df["date"])
            # Convert `extra_data` (dict) column to a DataFrame
            extra_df = pd.json_normalize(df['extra_data'])
            # Combine with original DataFrame (excluding the old `extra_data` column)
            df_flat = pd.concat([df.drop(columns=['extra_data']), extra_df], axis=1)
            if("project_code" in df_flat.columns):
                df["project"]= df_flat["project_code"]
            else:
                df["project"]= df_flat["project_name"]
            # Normalize first
            df["category"] = df["category"].str.strip().str.lower()
            df["account_name"] = df["account_name"].str.strip().str.lower()
            df["project"] = df["project"].str.strip().str.lower()
            # # Filter
            # # Filter assets and depreciation entries
            # assets_df = df[df["category"].str.lower() == "asset"].copy()
            # operating_expense_df = df[df["category"].str.lower() == "operating_expense"].copy()
            # depreciation_df = df[df["account_name"].str.lower() == "depreciation"].copy()
            # print(depreciation_df[["account_name", "debit", "credit"]])
            # print(operating_expense_df[["account_name", "debit", "credit"]])
            # print(assets_df[["account_name", "debit", "credit"]])
             # Sum capital expenditures by project_code
            # Filter by category/account_name
            assets_df = df[df["category"].str.lower() == "asset"].copy()
            operating_expense_df = df[df["category"].str.lower() == "operating_expense"].copy()
            depreciation_df = df[df["account_name"].str.lower() == "depreciation"].copy()

            # Group and sum by project code
            capex = (
                assets_df.groupby("project")["debit"]
                .sum()
                .reset_index()
                .rename(columns={"debit": "capital_expenditure"})
            )

            depreciation = (
                depreciation_df.groupby("project")["debit"]
                .sum()
                .reset_index()
                .rename(columns={"debit": "depreciation"})
            )

            # Capital Expenditure Report: project, capital_expenditure, depreciation
            capex_report = capex.merge(depreciation, on="project", how="outer").fillna(0)

            # Operating expenses grouped
            opex = (
                operating_expense_df.groupby("project")["debit"]
                .sum()
                .reset_index()
                .rename(columns={"debit": "operating_expense"})
            )

            # Project Cost Report: merge capex, opex, depreciation
            project_cost_report = capex.merge(opex, on="project", how="outer") \
                                    .merge(depreciation, on="project", how="outer").fillna(0)

            # Calculate total project cost = capital_expenditure + operating_expense
            project_cost_report["total_project_cost"] = (
                project_cost_report["capital_expenditure"] + project_cost_report["operating_expense"]
            )

            # Return only necessary columns
            capex_report = capex_report[["project", "capital_expenditure", "depreciation"]]
            project_cost_report = project_cost_report[
                ["project", "capital_expenditure", "operating_expense", "depreciation", "total_project_cost"]
            ]

            return {"capital_expenditure_report": capex_report, "project_cost_report": project_cost_report}

        except Exception as e:
            print(e)
            # Assuming logger and FinancialReportError are defined somewhere
            logger.error(f"Failed to generate reports: {e}")
            raise FinancialReportError(f"Failed to generate reports: {e}")




################################################################

class KPIFinancialWidgetManagement:
    @staticmethod
    async def _ensure_defaults(db: AsyncSession) -> None:
        defaults = [
            {
                "name": KPIWidgetNameEnum.REVENUM_PER_EMPLOYEE.value,
                "chart_Type": ChartTypeEnum.LINE_CHART.value,
                "enableordisable": True,
                "position": 0,
            },
            {
                "name": KPIWidgetNameEnum.MARKETING_ROI.value,
                "chart_Type": ChartTypeEnum.LINE_CHART.value,
                "enableordisable": True,
                "position": 1,
            },
            {
                "name": KPIWidgetNameEnum.COST_PER_LEAD.value,
                "chart_Type": ChartTypeEnum.LINE_CHART.value,
                "enableordisable": True,
                "position": 2,
            },
            {
                "name": KPIWidgetNameEnum.PROFITABILITY_BY_SEGMENT.value,
                "chart_Type": ChartTypeEnum.BAR_CHART.value,
                "enableordisable": True,
                "position": 3,
            },
        ]

        await db.execute(text("CREATE SCHEMA IF NOT EXISTS fpna_config;"))

        await db.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS fpna_config.kpi_widget_settings (
                    name TEXT PRIMARY KEY,
                    chart_type TEXT NOT NULL,
                    enableordisable BOOLEAN NOT NULL DEFAULT TRUE,
                    position INT NOT NULL DEFAULT 0,
                    updated_at TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT NOW()
                );
                """
            )
        )

        for d in defaults:
            await db.execute(
                text(
                    """
                    INSERT INTO fpna_config.kpi_widget_settings (name, chart_type, enableordisable, position)
                    VALUES (:name, :chart_type, :enableordisable, :position)
                    ON CONFLICT (name) DO NOTHING;
                    """
                ),
                {
                    "name": d["name"],
                    "chart_type": d["chart_Type"],
                    "enableordisable": d["enableordisable"],
                    "position": d["position"],
                },
            )

        await db.commit()

    @staticmethod
    async def GetWidgets(db: AsyncSession) -> List[Dict[str, Any]]:
        await KPIFinancialWidgetManagement._ensure_defaults(db)

        rows = (
            await db.execute(
                text(
                    """
                    SELECT
                        name,
                        chart_type,
                        enableordisable,
                        position
                    FROM fpna_config.kpi_widget_settings
                    ORDER BY position ASC;
                    """
                )
            )
        ).mappings().all()

        result = []
        for r in rows:
            result.append(
                {
                    "name": r["name"],
                    "chart_Type": r["chart_type"],
                    "enableordisable": bool(r["enableordisable"]),
                    "position": int(r["position"]),
                }
            )
        return result

    @staticmethod
    async def UpdateWidget(widget: KPIWidgetUpdateSchema, db: AsyncSession) -> Dict[str, Any]:
        await KPIFinancialWidgetManagement._ensure_defaults(db)

        existing = (
            await db.execute(
                text(
                    """
                    SELECT
                        name,
                        chart_type,
                        enableordisable,
                        position
                    FROM fpna_config.kpi_widget_settings
                    WHERE name = :name;
                    """
                ),
                {"name": widget.name.value},
            )
        ).mappings().first()

        if not existing:
            raise Exception("Widget not found")

        new_chart_type = existing["chart_type"] if widget.chart_Type is None else widget.chart_Type.value
        new_enable = existing["enableordisable"] if widget.enableordisable is None else bool(widget.enableordisable)
        new_position = existing["position"] if widget.position is None else int(widget.position)

        if widget.position is not None:
            other = (
                await db.execute(
                    text(
                        """
                        SELECT name, position
                        FROM fpna_config.kpi_widget_settings
                        WHERE position = :pos
                          AND name <> :name
                        LIMIT 1;
                        """
                    ),
                    {"pos": new_position, "name": widget.name.value},
                )
            ).mappings().first()

            if other:
                await db.execute(
                    text(
                        """
                        UPDATE fpna_config.kpi_widget_settings
                        SET position = :pos, updated_at = NOW()
                        WHERE name = :name;
                        """
                    ),
                    {"pos": int(existing["position"]), "name": other["name"]},
                )

        await db.execute(
            text(
                """
                UPDATE fpna_config.kpi_widget_settings
                SET
                    chart_type = :chart_type,
                    enableordisable = :enableordisable,
                    position = :position,
                    updated_at = NOW()
                WHERE name = :name;
                """
            ),
            {
                "name": widget.name.value,
                "chart_type": new_chart_type,
                "enableordisable": new_enable,
                "position": new_position,
            },
        )

        await db.commit()

        final_row = (
            await db.execute(
                text(
                    """
                    SELECT
                        name,
                        chart_type,
                        enableordisable,
                        position
                    FROM fpna_config.kpi_widget_settings
                    WHERE name = :name;
                    """
                ),
                {"name": widget.name.value},
            )
        ).mappings().first()

        return {
            "detail": "Widget updated successfully",
            "widget": {
                "name": final_row["name"],
                "chart_Type": final_row["chart_type"],
                "enableordisable": bool(final_row["enableordisable"]),
                "position": int(final_row["position"]),
            },
        }
