data_pipeline/kpi/facts.py

import os
# import pyarrow
import sys
import logging
from datetime import date, timedelta
import polars as pl
from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine, URL
import clickhouse_connect
from dotenv import load_dotenv

from log import log 
from clickhouse_task.create_table import *
from db_con.connection import *
from mids import *



#PROJECT_ID = 40148
p=40148




def fetch_SOS_OneApp(engine: Engine, mids: list[int]) -> pl.DataFrame:
    
    
    
    if not mids:
        log.warning("No MIDs — nothing to fetch.")
        return pl.DataFrame()

    mid_list = ",".join(str(mid) for mid in mids)

    sql = f"""   SELECT
        MID,
        EmpId AS employee_id,
        StoreId AS store_id,
        VisitDate AS visit_date,
        StoreTypeid AS storetype_id,
        ChannelId AS channel_id,
        SOSDefinitionName,
        SOSHeaderDeatils,
        SOSHeaderName,
        SOSHeaderID,
        SOSChildDeatils,
        SOSChildName,
        SOSChildID,
        SOSHeaderFacing,
        ChildTotalFacing,
        ChildSelfFacing,
        SOSTarget
    FROM
    (
        SELECT
            sm.CountryName,
            sc.MID,
            sm.RegionName,
            sm.StateName,
            sm.CityName,
            Em.SupervisorName,
            Em.EmpId,
            Em.EmpName AS EmployeeName,
            Em.DesignationName AS Designation,
            sm.StoreId,
            CAST(sc.VisitDate AS DATE) AS VisitDate,
            sm.StoreCode,
            sm.StoreName,
            sm.Address,
            sm.StoreTypeid,
            sm.ChannelId,
            sm.ChainName,

            MSD.SOSDefinitionName,

            CASE
                WHEN ISNULL(TS.SOSHeaderTable,'')='Master_Category' THEN 'Category'
                WHEN ISNULL(TS.SOSHeaderTable,'')='Master_SubCategory' THEN 'SubCategory'
                WHEN ISNULL(TS.SOSHeaderTable,'')='Master_Brand' THEN 'Brand'
                WHEN ISNULL(TS.SOSHeaderTable,'')='Master_SubBrand' THEN 'SubBrand'
            END AS SOSHeaderDeatils,

            TS.SOSHeaderName,
            TS.SOSHeaderValue AS SOSHeaderID,

            CASE
                WHEN ISNULL(TSC.SOSChildTable,'')='Master_Category' THEN 'Category'
                WHEN ISNULL(TSC.SOSChildTable,'')='Master_SubCategory' THEN 'SubCategory'
                WHEN ISNULL(TSC.SOSChildTable,'')='Master_Brand' THEN 'Brand'
                WHEN ISNULL(TSC.SOSChildTable,'')='Master_SubBrand' THEN 'SubBrand'
            END AS SOSChildDeatils,

            TSC.SOSChildName,
            TSC.SOSChildValue AS SOSChildID,
            TSC.ChildTotalFacing,
            TS.SOSHeaderFacing,
            TSC.ChildSelfFacing,

            (
                SELECT TOP 1 SOSTarget
                FROM OneApp_KelloggsMT.dbo.Mapping_StoreShareOfShelfTarget a
                WHERE a.SOSDefinitionId = MSD.SOSDefinitionId
                    AND a.StoreId = sm.StoreId
                    AND a.FromDate <= sc.VisitDate
                    AND a.ToDate >= sc.VisitDate
            ) AS SOSTarget

        FROM OneApp_KelloggsMT.dbo.T_ShareOfShelfHeader TS

        INNER JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage SC
            ON TS.MID = SC.MID

        INNER JOIN OneApp_KelloggsMT.dbo.vw_StoreDetail SM
            ON SC.StoreId = SM.StoreId

        INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail EM
            ON SC.EmpId = EM.EmpId

        INNER JOIN OneApp_KelloggsMT.dbo.T_ShareOfShelfChild TSC
            ON TS.SOSId = TSC.SOSId

        INNER JOIN OneApp_KelloggsMT.dbo.Master_ShareOfShelfDefinition MSD
            ON MSD.SOSDefinitionId = TSC.SOSDefinitionId

        WHERE EM.EmpName NOT LIKE 'test%'
            AND SC.MID IN ({mid_list})
    ) A
        """
    
    
    log.info(f"Fetching  data for {len(mids):,} MIDs")

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    
    log.info(f"Fetched {len(df):,} rows from SQL Server")

    return df

def fetch_OQaD(
    engine: Engine,
    run_date: date,
) -> pl.DataFrame:

    sql = f"""
    WITH MID_TABLE_COV1 AS
    (
        SELECT
            EmpId,
            CAST(VisitDate AS DATE) AS VisitDate
        FROM OneApp_KelloggsMT.dbo.T_OQAD
        WHERE CAST(CreateDate AS DATE) = '{run_date}'

        UNION

        SELECT
            EmpId,
            CAST(VisitDate AS DATE) AS VisitDate
        FROM OneApp_KelloggsMT.dbo.T_OQAD
        WHERE CAST(UpdateDate AS DATE) = '{run_date}'
    ),

    QUIZ AS
    (
        SELECT DISTINCT
            E.EmpId,
            CAST(DQ.VisitDate AS DATE) AS VisitDate,
            DQ.QuestionId,
            DQ.AnswerId,
            QC.QuestionCategoryId,
            QC.QuestionCategory

        FROM OneApp_KelloggsMT.dbo.T_OQAD DQ

        INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail E
            ON DQ.EmpId = E.EmpId

        INNER JOIN OneApp_KelloggsMT.dbo.Master_OQAD_Question QU
            ON DQ.QuestionId = QU.QuestionId

        INNER JOIN OneApp_KelloggsMT.dbo.Master_OQAD_Category QC
            ON QU.QuestionCategoryId = QC.QuestionCategoryId

        WHERE E.RightId = 6
          AND E.EmpName NOT LIKE 'test%'
          AND E.EmpName NOT LIKE '%TEST%'
          AND (
                E.ResignDate IS NULL
                OR E.ResignDate >= '{run_date}'
              )
          AND EXISTS
          (
              SELECT 1
              FROM MID_TABLE_COV1 A
              WHERE A.EmpId = DQ.EmpId
                AND A.VisitDate = CAST(DQ.VisitDate AS DATE)
          )
    )

    SELECT
        40148 AS project_id,
        Q.EmpId AS employee_id,
        0 AS process_id,
        Q.VisitDate AS visit_date,
        Q.QuestionCategoryId AS question_category_id,
        Q.QuestionCategory AS question_category,
        QM.QuestionId AS question_id,
        QM.Question AS question,
        ISNULL(QA.AnswerId, 0) AS answer_id,
        ISNULL(QA.Answer, '') AS answer,

        CASE
            WHEN QA.AnswerId IS NULL THEN 'Not Answer'
            WHEN QA.RightAnswer = 1 THEN 'Y'
            WHEN QA.RightAnswer IS NULL THEN 'Not Answer'
            ELSE 'N'
        END AS correct_answer,

        GETDATE() AS update_date,
        'SP-Pius' AS update_by

    FROM QUIZ Q

    INNER JOIN OneApp_KelloggsMT.dbo.Master_OQAD_Question QM
        ON Q.QuestionId = QM.QuestionId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_OQAD_Answer QA
        ON Q.AnswerId = QA.AnswerId
    """

    log.info("Fetching OQaD data for run_date=%s", run_date)

    df = pl.read_database(
        query=sql,
        connection=engine,
    )

    log.info("Fetched %s rows", len(df))

    return df


def fetch_Survey(engine: Engine, mids: list[int]) -> pl.DataFrame:
    if not mids:
        log.warning("No MIDs — nothing to fetch.")
        return pl.DataFrame()

    mid_list = ",".join(map(str, mids))

    sql = f"""
    WITH QUIZ AS
    (
        SELECT
            SC.MID,
            SC.StoreId,
            EM.EmpId,
            EM.SupervisorId,
            CAST(SC.VisitDate AS DATE) AS VisitDate,

            DQ.QuestionId,
            QU.Question,
            DQ.Answer,

            QC.CategoryId,
            QC.Category,

            MSS.SubCategoryId,
            MSS.SubCategory

        FROM OneApp_KelloggsMT.dbo.T_SURVEY DQ

        INNER JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage SC
            ON DQ.MID = SC.MID

        INNER JOIN OneApp_KelloggsMT.dbo.Master_SurveyQuestion QU
            ON DQ.QuestionId = QU.QuestionId

        INNER JOIN OneApp_KelloggsMT.dbo.Master_SurveySubCategory MSS
            ON QU.SubCategoryId = MSS.SubCategoryId

        INNER JOIN OneApp_KelloggsMT.dbo.Master_SurveyCategory QC
            ON MSS.CategoryId = QC.CategoryId

        INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail EM
            ON SC.EmpId = EM.EmpId

        INNER JOIN OneApp_KelloggsMT.dbo.Master_Survey MS
            ON DQ.SurveyId = MS.SurveyId
           AND QU.SurveyId = MS.SurveyId

        WHERE EM.EmpName NOT LIKE 'test%'
          AND SC.MID IN ({mid_list})
    )

    SELECT
        Q.MID,
        Q.SupervisorId,
        Q.EmpId,
        SM.StoreId,
        Q.VisitDate,

        SM.StoreTypeId,
        SM.ChainId,

        Q.CategoryId,
        Q.Category,

        Q.SubCategoryId,
        Q.SubCategory,

        Q.QuestionId,
        Q.Question,
        Q.Answer

    FROM QUIZ Q

    INNER JOIN OneApp_KelloggsMT.dbo.vw_StoreDetail SM
        ON Q.StoreId = SM.StoreId
    """

    
    log.info(f"Fetching Survey data for {len(mids):,} MIDs")

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    
    log.info(f"Fetched {len(df):,} Survey rows")

    return df


def fetch_additional_visibility(
    engine: Engine,
    mids: list[int]
) -> pl.DataFrame:

    if not mids:
        log.warning("No MIDs — nothing to fetch.")
        return pl.DataFrame()

    mid_list = ",".join(map(str, mids))

    sql = f"""
    SELECT
        TS.MID,

        EM.EmpId AS emp_id,

        SM.StoreId AS store_id,
        SM.StoreTypeId AS storetype_id,
        SM.ChannelId AS channel_id,
        SM.ChainId AS chain_id,

        CASE
            WHEN SM.CameraAllow = 1 THEN 'Y'
            ELSE 'N'
        END AS camera_allowed,

        CAST(SC.VisitDate AS DATE) AS visit_date,

        CASE
            WHEN TS.Present = 0 THEN 'N'
            ELSE 'Y'
        END AS is_present,

        ISNULL(MB.BrandId, 0) AS brand_id,
        ISNULL(MD.DisplayId, 0) AS display_id,

        ISNULL(TS.Remark, '') AS remarks,

        CASE
            WHEN ISNULL(TS.ImageUrl, '') = '' THEN ''
            ELSE CONCAT(
                'https://kimt1.parinaam.in/Upload/VisibilityImages/',
                TS.ImageUrl
            )
        END AS image_url

    FROM OneApp_KelloggsMT.dbo.T_AdditionalVisibility TS

    INNER JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage SC
        ON TS.MID = SC.MID

    INNER JOIN OneApp_KelloggsMT.dbo.vw_StoreDetail SM
        ON SC.StoreId = SM.StoreId

    INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail EM
        ON SC.EmpId = EM.EmpId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_Brand MB
        ON TS.BrandId = MB.BrandId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_Display MD
        ON TS.DisplayId = MD.DisplayId

    WHERE EM.EmpName NOT LIKE 'test%'
      AND TS.Present = 1
      AND SC.MID IN ({mid_list})
    """
    
    log.info(
        f"Fetching Additional Visibility data for {len(mids):,} MIDs"
    )

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    
    log.info(
        f"Fetched {len(df):,} Additional Visibility rows"
    )

    return df


def fetch_Coverage(engine: Engine, mids: list[int]) -> pl.DataFrame:
    
    
    
    if not mids:
        log.warning("No MIDs — nothing to fetch.")
        return pl.DataFrame()

    mid_list = ",".join(str(mid) for mid in mids)

    sql = f"""
       WITH SOS_BASE AS
(
    SELECT
        sm.CountryName,
        sc.MID,
        sm.RegionName,
        sm.StateName,
        sm.CityName,
        Em.SupervisorName,
        em.EmpId,
        Em.EmpName AS EmployeeName,
        Em.DesignationName AS Designation,
        sm.StoreId,
        CONVERT(varchar, sc.VisitDate, 101) AS VisitDate,
        sm.StoreCode,
        sm.StoreName,
        sm.Address,
        sm.StoreTypeid,
        sm.ChannelId,
        sm.ChainName,

        MSD.SOSDefinitionName,

        CASE
            WHEN ISNULL(TS.SOSHeaderTable,'')='Master_Category' THEN 'Category'
            WHEN ISNULL(TS.SOSHeaderTable,'')='Master_SubCategory' THEN 'SubCategory'
            WHEN ISNULL(TS.SOSHeaderTable,'')='Master_Brand' THEN 'Brand'
            WHEN ISNULL(TS.SOSHeaderTable,'')='Master_SubBrand' THEN 'SubBrand'
        END AS SOSHeaderDeatils,

        TS.SOSHeaderName,
        TS.SOSHeaderValue AS SOSHeaderID,

        'Header_Image' AS HDR1,

        CASE
            WHEN ISNULL(TSC.SOSChildTable,'')='Master_Category' THEN 'Category'
            WHEN ISNULL(TSC.SOSChildTable,'')='Master_SubCategory' THEN 'SubCategory'
            WHEN ISNULL(TSC.SOSChildTable,'')='Master_Brand' THEN 'Brand'
            WHEN ISNULL(TSC.SOSChildTable,'')='Master_SubBrand' THEN 'SubBrand'
        END AS SOSChildDeatils,

        TSC.SOSChildName,
        TSC.SOSChildValue AS SOSChildID,
        TSC.ChildTotalFacing,
        TS.SOSHeaderFacing,
        TSC.ChildSelfFacing,

        (
            SELECT TOP 1 SOSTarget
            FROM OneApp_KelloggsMT.dbo.Mapping_StoreShareOfShelfTarget a
            WHERE a.SOSDefinitionId = MSD.SOSDefinitionId
              AND a.StoreId = sm.StoreId
              AND a.FromDate <= sc.VisitDate
              AND a.ToDate >= sc.VisitDate
        ) AS SOSTarget,

        CASE
            WHEN ISNULL(SHI.SOSHeaderImage,'') = ''
                THEN ''
            ELSE
                'https://kimt1.parinaam.in/Upload/SOSImages/'
                + SHI.SOSHeaderImage
        END AS SOSHeaderImg

    FROM OneApp_KelloggsMT.dbo.T_ShareOfShelfHeader ts

    INNER JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage sc
        ON ts.MID = sc.MID

    INNER JOIN OneApp_KelloggsMT.dbo.vw_StoreDetail sm
        ON sc.StoreId = sm.StoreId

    INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail Em
        ON sc.EmpId = Em.EmpId

    INNER JOIN OneApp_KelloggsMT.dbo.T_ShareOfShelfChild tsc
        ON ts.SOSId = tsc.SOSId

    INNER JOIN OneApp_KelloggsMT.dbo.Master_ShareOfShelfDefinition msd
        ON msd.SOSDefinitionId = tsc.SOSDefinitionId

    LEFT JOIN OneApp_KelloggsMT.dbo.T_ShareOfShelfHeaderImages SHI
        ON ts.SOSId = SHI.SOSId
       AND ts.SOSHeaderValue = SHI.SOSHeaderValue

    LEFT JOIN OneApp_KelloggsMT.dbo.T_ShareOfShelfChildImages SCI
        ON tsc.SOSId = SCI.SOSId
       AND tsc.SOSChildValue = SCI.SOSChildValue

    WHERE Em.EmpName NOT LIKE 'test%'
      AND sc.MID IN ({mid_list})
),

SOS_PIVOT AS
(
    SELECT *
    FROM SOS_BASE
    PIVOT
    (
        MIN(SOSHeaderImg)
        FOR HDR1 IN ([Header_Image])
    ) pvt
)

SELECT
    '40148' AS ProjectId,
    MID,
    EmpId AS employee_id,
    StoreId AS store_id,
    VisitDate AS visit_date,
    StoreTypeid AS storetype_id,
    ChannelId AS channel_id,
    SOSDefinitionName,
    SOSHeaderDeatils,
    SOSHeaderName,
    SOSHeaderID,
    SOSChildDeatils,
    SOSChildName,
    SOSChildID,
    SOSHeaderFacing,
    ChildTotalFacing,
    ChildSelfFacing,
    SOSTarget,
    GETDATE() AS update_date,
    'SP-Pius' AS update_by

FROM SOS_PIVOT

GROUP BY
    CountryName,
    MID,
    RegionName,
    StateName,
    CityName,
    SupervisorName,
    EmpId,
    EmployeeName,
    Designation,
    StoreId,
    VisitDate,
    StoreCode,
    StoreName,
    Address,
    StoreTypeid,
    ChannelId,
    ChainName,
    SOSDefinitionName,
    SOSHeaderDeatils,
    SOSHeaderName,
    SOSHeaderID,
    SOSChildDeatils,
    SOSChildName,
    SOSChildID,
    ChildTotalFacing,
    SOSHeaderFacing,
    ChildSelfFacing,
    SOSTarget

ORDER BY
    RegionName,
    StateName,
    CityName,
    VisitDate;
    """
    
    
    log.info(f"Fetching coverage data for {len(mids):,} MIDs")

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    
    log.info(f"Fetched {len(df):,} rows from SQL Server")

    return df



def fetch_Login(engine: Engine , mids: list[int]) -> pl.DataFrame:

    sql = """
    WITH login_data AS
    (
        SELECT
            UD.EmpId,
            CAST(UD.LoginDate AS DATE) AS LoginDate,
            CONVERT(VARCHAR(8), UD.InTime, 108) AS LoginTime,

            ROW_NUMBER() OVER
            (
                PARTITION BY
                    UD.EmpId,
                    CAST(UD.LoginDate AS DATE)
                ORDER BY UD.LoginDate
            ) AS rn

        FROM OneApp_KelloggsMT.dbo.T_DeviceLogin UD
        INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail EM
            ON UD.EmpId = EM.EmpId

        WHERE CAST(UD.LoginDate AS DATE) =
              CAST(DATEADD(DAY,-1,GETDATE()) AS DATE)

          AND EM.RightId = 6
          AND EM.EmpName NOT LIKE '%test%'
          AND (
                EM.ResignDate IS NULL
                OR CAST(EM.ResignDate AS DATE) >=
                   CAST(DATEADD(DAY,-1,GETDATE()) AS DATE)
              )
    )

    SELECT
        LD.EmpId AS employee_id,
        LD.LoginDate AS login_date,
        LD.LoginTime AS login_time,

        (
            SELECT MIN(CONVERT(VARCHAR(8), SC.InTime, 108))
            FROM OneApp_KelloggsMT.dbo.T_StoreCoverage SC
            WHERE SC.IsDel = 0
              AND SC.EmpId = LD.EmpId
              AND CAST(SC.VisitDate AS DATE) = LD.LoginDate
        ) AS first_store_in_time,

        (
            SELECT MAX(CONVERT(VARCHAR(8), SC.OutTime, 108))
            FROM OneApp_KelloggsMT.dbo.T_StoreCoverage SC
            WHERE SC.IsDel = 0
              AND SC.EmpId = LD.EmpId
              AND CAST(SC.VisitDate AS DATE) = LD.LoginDate
        ) AS last_store_out_time,

        CONCAT('40148','_',CAST(LD.EmpId AS VARCHAR(50))) AS unique_id

    FROM login_data LD
    WHERE LD.rn = 1
    """

    
    log.info("Fetching Login data for yesterday")

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    
    log.info(f"Fetched {len(df):,} Login rows")

    return df




def fetch_Stock_Details(
    engine: Engine,
    mids: list[int]
) -> pl.DataFrame:

    if not mids:
        log.warning("No MIDs — nothing to fetch.")
        return pl.DataFrame()

    mid_list = ",".join(map(str, mids))

    sql = f"""
    SELECT
        SC.MID,

        EM.SupervisorId AS supervisor_id,
        EM.EmpId AS employee_id,

        SM.StoreId AS store_id,

        CAST(SC.VisitDate AS DATE) AS visitdate,

        SC.StoreTypeId,
        SM.StoreCategoryId,

        VP.ProductId AS product_id,

        TS.MSL,
        MP.MBQ,

        ISNULL(TS.OpeningStock, 0)
        + ISNULL(TS.MidDayStock, 0) AS stock_qty,

        0 AS damagedstock,
        0 AS loststock,
        0 AS expirystock,

        CASE
            WHEN ISNULL(TS.OpeningStock, 0)
               + ISNULL(TS.MidDayStock, 0) >= 1
            THEN 'Y'
            ELSE 'N'
        END AS skuavailability,

        'Parinaam' AS stocktype

    FROM OneApp_KelloggsMT.dbo.T_Stock TS

    INNER JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage SC
        ON TS.MID = SC.MID

    INNER JOIN OneApp_KelloggsMT.dbo.vw_StoreDetail SM
        ON SC.StoreId = SM.StoreId

    INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail EM
        ON SC.EmpId = EM.EmpId

    INNER JOIN OneApp_KelloggsMT.dbo.vw_Product VP
        ON TS.ProductId = VP.ProductId

    INNER JOIN
    (
        SELECT
            MBQ,
            ProductId,
            StateId,
            ChainId,
            StoreTypeId,
            StoreCategoryId,
            StoreClassId
        FROM OneApp_KelloggsMT.dbo.Mapping_ProductAssortment
        WHERE FromDate <= CAST(GETDATE() AS DATE)
          AND ToDate >= CAST(GETDATE() AS DATE)
    ) MP
        ON MP.StateId = SM.StateId
       AND MP.ChainId = SM.ChainId
       AND MP.StoreTypeId = SM.StoreTypeId
       AND MP.StoreCategoryId = SM.StoreCategoryId
       AND MP.StoreClassId = SM.StoreClassId
       AND MP.ProductId = TS.ProductId

    WHERE EM.EmpName NOT LIKE 'test%'
      AND SC.MID IN ({mid_list})
    """
    
    log.info(
        f"Fetching Stock Details data for {len(mids):,} MIDs"
    )

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    log.info(
        f"Fetched {len(df):,} Stock Details rows"
    )

    return df





def fetch_Attendance(
    engine: Engine,
    end_date: date | None = None,
    days_back: int = 15
) -> pl.DataFrame:
    """
    Fetch attendance source data.

    Default:
        end_date = yesterday
        start_date = yesterday - 15 days
    """

    if end_date is None:
        end_date = date.today() - timedelta(days=1)

    start_date = end_date - timedelta(days=days_back)

    sql = f"""
    SELECT
        JP.EmpId AS employee_id,
        JP.StoreId AS store_id,
        CAST(JP.VisitDate AS DATE) AS visit_date,

        EM.ManagerId AS supervisor_id,
        EM.JoinDate AS date_of_join,
        EM.ResignDate AS date_of_resign,
        EM.LegacyCode,

        SC.ReasonId,
        SC.InTime,
        SC.OutTime,

        MPU.PositionId,
        MP.PositionCode

    FROM OneApp_KelloggsMT.dbo.Mapping_JourneyPlan JP

    INNER JOIN OneApp_KelloggsMT.dbo.AspNetUsers EM
        ON JP.EmpId = EM.Id

    LEFT JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage SC
        ON JP.EmpId = SC.EmpId
       AND JP.StoreId = SC.StoreId
       AND JP.VisitDate = SC.VisitDate

    LEFT JOIN OneApp_KelloggsMT.dbo.Mapping_PositionUser MPU
        ON JP.EmpId = MPU.EmpId
       AND JP.VisitDate BETWEEN MPU.FromDate AND MPU.ToDate

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_Position MP
        ON MPU.PositionId = MP.PositionId

    WHERE JP.VisitDate BETWEEN '{start_date}'
                          AND '{end_date}'
      AND EM.RightId = 6
      AND EM.EmployeeName NOT LIKE '%test%'
    """

    log.info(
        f"Fetching Attendance data from {start_date} to {end_date}"
    )

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    log.info(
        f"Fetched {len(df):,} attendance rows "
        f"for {df['employee_id'].n_unique():,} employees"
    )

    return df




def fetch_Journey_Plan(
    engine: Engine,
    report_date: date
) -> pl.DataFrame:

    sql = f"""
    SELECT
        StoreId AS store_id,
        EmpId AS employee_id,
        CAST(VisitDate AS DATE) AS visit_date,
        Deviation AS process_id

    FROM OneApp_KelloggsMT.dbo.Mapping_JourneyPlan

    WHERE MONTH(VisitDate) = {report_date.month}
      AND YEAR(VisitDate) = {report_date.year}

      AND EmpId NOT IN (
            SELECT ID
            FROM OneApp_KelloggsMT.dbo.AspNetUsers
            WHERE UserName LIKE 'TEST%'
      )
    """

    log.info(
        f"Fetching Journey Plan for {report_date:%Y-%m}"
    )

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    log.info(
        f"Fetched {len(df):,} Journey Plan records"
    )
    return df





def fetch_Web_Logins(
    engine: Engine,
    run_date: date
) -> pl.DataFrame:
    """
    Source:
        T_User_Activity_Log

    Target:
        Web Logins
    """

    sql = f"""
      SELECT DISTINCT
        40148 AS project_id,
        EM1.Id AS supervisor_id,
        EM1.EmployeeName AS supervisor_name,
        EM.Id AS emp_id,
        EM.EmployeeName AS employee_name,
        DM.DesignationName AS designation,
        CAST(AL.Date AS DATE) AS date,
        CONVERT(VARCHAR(8), AL.Date, 108) AS time,
        AL.Thread AS activity_name,
        AL.Level AS activity_type,
        R.RightName AS right_name,
        GETDATE() AS CreateDate,
        'Pius' AS CreateBy
    FROM OneApp_KelloggsMT.dbo.T_User_Activity_Log AL
    LEFT JOIN OneApp_KelloggsMT.dbo.AspNetUsers EM
        ON AL.Logger = EM.USERNAME
    INNER JOIN OneApp_KelloggsMT.dbo.AspNetUsers EM1
        ON EM1.Id = EM.ManagerId
    INNER JOIN OneApp_KelloggsMT.dbo.AspNetUsers EM2
        ON EM2.Id = EM1.ManagerId
    INNER JOIN OneApp_KelloggsMT.dbo.Master_Designation DM
        ON EM.DesignationId = DM.DesignationId
    INNER JOIN OneApp_KelloggsMT.dbo.Master_City CM
        ON EM.CityId = CM.CityId
    INNER JOIN OneApp_KelloggsMT.dbo.Master_State ST
        ON ST.StateId = CM.StateId
    INNER JOIN OneApp_KelloggsMT.dbo.Master_Region RM
        ON RM.RegionId = ST.RegionId
    INNER JOIN OneApp_KelloggsMT.dbo.Right_Master R
        ON EM.RightId = R.RightId
    WHERE CAST(AL.Date AS DATE) = '{run_date}'
    """


    log.info(f"Fetching Web Login data for {run_date}")

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    
    log.info(
        f"Fetched {len(df):,} Web Login records"
    )

    return df




def fetch_Promotion(
    engine: Engine,
    mids: list[int]
) -> pl.DataFrame:

    if not mids:
        log.warning("No MIDs supplied.")
        return pl.DataFrame()

    mid_list = ",".join(map(str, mids))

    sql = f"""
    SELECT
        SC.MID,

        40148 AS project_id,

        SC.StoreId AS store_id,
        EM.EmpId AS employee_id,
        SC.VisitDate AS visit_date,

        EM.SupervisorId AS supervisor_id,

        SM.ChannelId AS channel_id,
        SM.ChainId AS chain_id,
        SM.StoreTypeId AS storetype_id,

        MSD.PromoDefinitionId AS promo_definition_id,
        MSD.PromoDefinitionName AS promo_definition_name,

        CASE
            WHEN ISNULL(TS.PromoTable,'')='Master_Category'
                THEN 'Category'
            WHEN ISNULL(TS.PromoTable,'')='Master_SubCategory'
                THEN 'SubCategory'
            WHEN ISNULL(TS.PromoTable,'')='Master_Brand'
                THEN 'Brand'
            WHEN ISNULL(TS.PromoTable,'')='Master_SubBrand'
                THEN 'SubBrand'
        END AS promotion_details,

        TS.PromoValue AS promotion_details_id,

        CASE
            WHEN ISNULL(TS.PromoTable,'')='Master_Category'
                THEN (
                    SELECT CategoryName
                    FROM OneApp_KelloggsMT.dbo.Master_Category A
                    WHERE A.CategoryId = TS.PromoValue
                )

            WHEN ISNULL(TS.PromoTable,'')='Master_SubCategory'
                THEN (
                    SELECT SubCategoryName
                    FROM OneApp_KelloggsMT.dbo.Master_SubCategory A
                    WHERE A.SubCategoryId = TS.PromoValue
                )

            WHEN ISNULL(TS.PromoTable,'')='Master_Brand'
                THEN (
                    SELECT BrandName
                    FROM OneApp_KelloggsMT.dbo.Master_Brand A
                    WHERE A.BrandId = TS.PromoValue
                )

            WHEN ISNULL(TS.PromoTable,'')='Master_SubBrand'
                THEN (
                    SELECT SubBrandName
                    FROM OneApp_KelloggsMT.dbo.Master_SubBrand A
                    WHERE A.SubBrandId = TS.PromoValue
                )
        END AS promotion_value_name,

        CASE
            WHEN TS.Present = 0 THEN 'N'
            ELSE 'Y'
        END AS present,

        CASE
            WHEN TS.Present = 1 THEN ''
            ELSE ISNULL(MNP.PromoReason,'')
        END AS reason,

        ISNULL(MPQ.PromoQuestionName,'') AS promo_question,
        ISNULL(TPQ.PromoAnswerName,'') AS promo_answer,

        CASE
            WHEN ISNULL(SHI.PromoImage1,'') = ''
                THEN ''
            ELSE
                'https://kimt1.parinaam.in/Upload/PromotionImages/'
                + SHI.PromoImage1
        END AS image1,

        CASE
            WHEN ISNULL(SHI.PromoImage2,'') = ''
                THEN ''
            ELSE
                'https://kimt1.parinaam.in/Upload/PromotionImages/'
                + SHI.PromoImage2
        END AS image2

    FROM OneApp_KelloggsMT.dbo.T_Promotion TS

    INNER JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage SC
        ON TS.MID = SC.MID

    INNER JOIN OneApp_KelloggsMT.dbo.vw_StoreDetail SM
        ON SC.StoreId = SM.StoreId

    INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail EM
        ON SC.EmpId = EM.EmpId

    INNER JOIN OneApp_KelloggsMT.dbo.Master_PromotionDefinition MSD
        ON MSD.PromoDefinitionId = TS.PromoDefinitionId

    LEFT JOIN OneApp_KelloggsMT.dbo.T_PromotionImages SHI
        ON TS.PId = SHI.PId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_PromotionReason MNP
        ON TS.PromoReasonId = MNP.PromoReasonId

    LEFT JOIN OneApp_KelloggsMT.dbo.T_PromotionQuestion TPQ
        ON TS.PId = TPQ.PId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_PromotionQuestion MPQ
        ON TPQ.PromoQuestionId = MPQ.PromoQuestionId

    WHERE EM.EmpName NOT LIKE 'test%'
      AND SC.MID IN ({mid_list})
    """

    log.info(
        f"Fetching Promotion data for {len(mids):,} MIDs"
    )

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    log.info(
        f"Fetched {len(df):,} Promotion records"
    )

    return df





def fetch_PaidVisibility(
    engine: Engine,
    mids: list[int]
) -> pl.DataFrame:

    if not mids:
        log.warning("No MIDs supplied.")
        return pl.DataFrame()

    mid_list = ",".join(map(str, mids))

    sql = f"""
    SELECT
        SC.MID,

        40148 AS project_id,

        SC.StoreId AS store_id,
        EM.EmpId AS employee_id,
        SC.VisitDate AS visit_date,

        EM.SupervisorId AS supervisor_id,

        SM.ChannelId AS channel_id,
        SM.ChainId AS chain_id,
        SM.StoreTypeId AS storetype_id,

        0 AS menu_id,
        '' AS menu_name,

        TS.VisibilityId AS visibility_id,
        MV.VisibilityName AS visibility_name,

        MSD.VisibilityDefinitionId AS visibility_definition_id,
        MSD.VisibilityDefinitionName AS visibility_definition_name,

        CASE
            WHEN ISNULL(TS.VisibilityTable,'')='Master_Category'
                THEN 'Category'
            WHEN ISNULL(TS.VisibilityTable,'')='Master_SubCategory'
                THEN 'SubCategory'
            WHEN ISNULL(TS.VisibilityTable,'')='Master_Brand'
                THEN 'Brand'
            WHEN ISNULL(TS.VisibilityTable,'')='Master_SubBrand'
                THEN 'SubBrand'
        END AS visibility_details,

        TS.VisibilityValue AS visibility_details_id,

        CASE
            WHEN ISNULL(TS.VisibilityTable,'')='Master_Category'
                THEN (
                    SELECT CategoryName
                    FROM OneApp_KelloggsMT.dbo.Master_Category A
                    WHERE A.CategoryId = TS.VisibilityValue
                )

            WHEN ISNULL(TS.VisibilityTable,'')='Master_SubCategory'
                THEN (
                    SELECT SubCategoryName
                    FROM OneApp_KelloggsMT.dbo.Master_SubCategory A
                    WHERE A.SubCategoryId = TS.VisibilityValue
                )

            WHEN ISNULL(TS.VisibilityTable,'')='Master_Brand'
                THEN (
                    SELECT BrandName
                    FROM OneApp_KelloggsMT.dbo.Master_Brand A
                    WHERE A.BrandId = TS.VisibilityValue
                )

            WHEN ISNULL(TS.VisibilityTable,'')='Master_SubBrand'
                THEN (
                    SELECT SubBrandName
                    FROM OneApp_KelloggsMT.dbo.Master_SubBrand A
                    WHERE A.SubBrandId = TS.VisibilityValue
                )
        END AS visibility_value_name,

        CASE
            WHEN TS.Present = 0 THEN 'N'
            ELSE 'Y'
        END AS present,

        CASE
            WHEN TS.Present = 0
                THEN TS.VisibilityReasonId
            ELSE NULL
        END AS reason_id,

        CASE
            WHEN TS.Present = 0
                THEN ISNULL(MNP.VisibilityReason,'')
            ELSE ''
        END AS reason,

        ISNULL(MVQ.VisibilityQuestionName,'') AS visibility_question,
        ISNULL(TVQ.VisibilityAnswerName,'') AS visibility_answer,

        CASE
            WHEN TS.Present = 0 THEN ''
            ELSE
                CASE
                    WHEN ISNULL(SHI.VisibilityImage1,'') = ''
                        THEN ''
                    ELSE
                        'https://kimt1.parinaam.in/Upload/PaidVisibilityImages/'
                        + SHI.VisibilityImage1
                END
        END AS image1,

        CASE
            WHEN ISNULL(SHI.VisibilityImage2,'') = ''
                THEN ''
            ELSE
                'https://kimt.parinaam.in/Upload/PaidVisibilityImages/'
                + SHI.VisibilityImage2
        END AS image2

    FROM OneApp_KelloggsMT.dbo.T_Visibility TS WITH (NOLOCK)

    INNER JOIN OneApp_KelloggsMT.dbo.T_StoreCoverage SC
        ON TS.MID = SC.MID

    INNER JOIN OneApp_KelloggsMT.dbo.vw_StoreDetail SM
        ON SC.StoreId = SM.StoreId

    INNER JOIN OneApp_KelloggsMT.dbo.vw_Employee_Detail EM
        ON SC.EmpId = EM.EmpId

    INNER JOIN OneApp_KelloggsMT.dbo.Master_VisibilityDefinition MSD
        ON MSD.VisibilityDefinitionId = TS.VisibilityDefinitionId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_Visibility MV
        ON TS.VisibilityId = MV.VisibilityId

    LEFT JOIN OneApp_KelloggsMT.dbo.T_VisibilityImages SHI
        ON TS.VId = SHI.VId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_VisibilityReason MNP
        ON TS.VisibilityReasonId = MNP.VisibilityReasonId

    LEFT JOIN OneApp_KelloggsMT.dbo.T_VisibilityStock TVS
        ON TS.VId = TVS.VId

    LEFT JOIN OneApp_KelloggsMT.dbo.T_VisibilityQuestion TVQ
        ON TS.VId = TVQ.VId

    LEFT JOIN OneApp_KelloggsMT.dbo.Master_VisibilityQuestion MVQ
        ON TVQ.VisibilityQuestionId = MVQ.VisibilityQuestionId

    WHERE EM.EmpName NOT LIKE 'test%'
      AND SC.MID IN ({mid_list})
    """

    log.info(
        f"Fetching Paid Visibility data for {len(mids):,} MIDs"
    )

    df = pl.read_database(
        query=sql,
        connection=engine
    )

    log.info(
        f"Fetched {len(df):,} Paid Visibility records"
    )

    return df