2nd commit

2026-06-12 12:30:34 +05:30
parent 7fbbffec65
commit 80bb585cdb
5 changed files with 595 additions and 109 deletions
@@ -753,120 +753,307 @@ def fetch_login(engine: Engine , mids: list[int]) -> pl.DataFrame:
 def fetch_Promotion(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)
+
 def fetch_journey_plan(
    engine: Engine,
    report_date: date
 ) -> pl.DataFrame:
    sql = f"""
    SELECT
-            sc.MID                                          AS mid,
+        StoreId AS store_id,
-            {p}                                   AS project_id,
+        EmpId AS employee_id,
-            sc.StoreId                                     AS store_id,
+        CAST(VisitDate AS DATE) AS visit_date,
-            Em.EmpId                                       AS employee_id,
+        Deviation AS process_id
            CONVERT(date, 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,
-            -- Resolve which dimension level the promo targets
+    FROM OneApp_KelloggsMT.dbo.Mapping_JourneyPlan
            -- (Category / SubCategory / Brand / SubBrand)
            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'
                ELSE ''
            END                                            AS promotion_details,
-            ts.PromoValue                                  AS promotion_details_id,
+    WHERE MONTH(VisitDate) = {report_date.month}
      AND YEAR(VisitDate) = {report_date.year}
-            -- Resolve the actual name of that dimension item
+      AND EmpId NOT IN (
-            -- SQL Server evaluates exactly ONE of these subqueries per row
+            SELECT ID
-            CASE
+            FROM OneApp_KelloggsMT.dbo.AspNetUsers
-                WHEN ISNULL(ts.PromoTable,'') = 'Master_Category'
+            WHERE UserName LIKE 'TEST%'
-                    THEN (SELECT a.CategoryName
+      )
                          FROM OneApp_KelloggsMT.dbo.Master_Category a
                          WHERE a.CategoryId = ts.PromoValue)
                WHEN ISNULL(ts.PromoTable,'') = 'Master_SubCategory'
                    THEN (SELECT a.SubCategoryName
                          FROM OneApp_KelloggsMT.dbo.Master_SubCategory a
                          WHERE a.SubCategoryId = ts.PromoValue)
                WHEN ISNULL(ts.PromoTable,'') = 'Master_Brand'
                    THEN (SELECT a.BrandName
                          FROM OneApp_KelloggsMT.dbo.Master_Brand a
                          WHERE a.BrandId = ts.PromoValue)
                WHEN ISNULL(ts.PromoTable,'') = 'Master_SubBrand'
                    THEN (SELECT a.SubBrandName
                          FROM OneApp_KelloggsMT.dbo.Master_SubBrand a
                          WHERE a.SubBrandId = ts.PromoValue)
            END                                            AS promo_value_name,
            -- Present flag: 0 → 'N', anything else → 'Y'
            CASE WHEN ts.Present = 0 THEN 'N' ELSE 'Y' END AS is_present,
            -- Reason only populated when promo is ABSENT
            CASE
                WHEN ts.Present = 1 THEN ''
                ELSE ISNULL(mnp.PromoReason, '')
            END                                            AS reason,
            ISNULL(mpq.PromoQuestionName, '')              AS question,
            ISNULL(tpq.PromoAnswerName,   '')              AS answer,
            -- Image URLs — empty string when no image
            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,
            GETDATE()                                      AS update_date,
            'ETL-SQLAlchemy'                               AS update_by
        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 Coverage data for {len(mids):,} MIDs")
+    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):,} rows from SQL Server")
+    log.info(
        f"Fetched {len(df):,} Journey Plan records"
    )
    return df
 def fetch_coverage_remarks(engine: Engine , mids: list[int]) -> pl.DataFrame:
    """
    Source:
        OneApp_KelloggsMT.dbo.Master_NonWorkingReason
    Target:
        coverage_remarks
    """
    sql = """
    SELECT
        40148 AS project_id,
        ReasonId AS reason_id,
        Reason AS reason_remarks
    FROM OneApp_KelloggsMT.dbo.Master_NonWorkingReason
    """
    log.info("Fetching Coverage Remarks")
    df = pl.read_database(
        query=sql,
        connection=engine
    )
    log.info(
        f"Fetched {len(df):,} Coverage Remark 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 login_date,
        CONVERT(VARCHAR, AL.Date, 108) AS login_time,
        AL.Thread AS activity_name,
        AL.Level AS activity_type,
        R.RightName AS right_name,
        GETDATE() AS create_date,
        'Pius' AS create_by
    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
@@ -874,3 +1061,172 @@ def fetch_Promotion(engine: Engine, mids: list[int]) -> pl.DataFrame:
 def fetch_paid_visibility(
    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
@@ -80,17 +80,20 @@ def main():
        log.info(table_name, operation)
        fn=f"fetch_{table_name}"
        list=["Attendance", "Journey_Plan", "Web_Logins"]
        if table_type =="FACT" :
-            if table_name == "Attendance" :
+            if table_name in list :
                df =  globals()[fn](sql_engine, run_date)
            else:
                df =  globals()[fn](sql_engine, mids)
        elif table_type =="BRIDGE" :
            df =  globals()[fn](sql_engine, run_date)
        else:
@@ -0,0 +1,64 @@
 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 create_clickhouse_table , check
 from db_con.connection import *
 from mids import *
 def fetch_mapping_store_visibility(
    engine: Engine,
    run_date: date
 ) -> pl.DataFrame:
    sql = f"""
    SELECT DISTINCT
        40148 AS project_id,
        Z.StoreId AS store_id,
        Z.VisibilityDefinitionId AS visibility_definition_id,
        Z.FromDate AS from_date,
        Z.ToDate AS to_date
    FROM OneApp_KelloggsMT.dbo.Mapping_StoreVisibility Z
    WHERE CAST(Z.FromDate AS DATE) <= '{run_date}'
      AND CAST(Z.ToDate AS DATE) >= '{run_date}'
      AND Z.VisibilityDefinitionId IN
      (
          SELECT DISTINCT VisibilityDefinitionId
          FROM OneApp_KelloggsMT.dbo.Master_VisibilityDefinition
          WHERE MenuId = 22
      )
    """
    log.info(
        f"Fetching Mapping Store Visibility for {run_date}"
    )
    df = pl.read_database(
        query=sql,
        connection=engine
    )
    log.info(
        f"Fetched {len(df):,} Mapping Store Visibility records"
    )
    return df
@@ -192,10 +192,6 @@ def fetch_display_master(engine: Engine) -> pl.DataFrame:
    return df
 import polars as pl
 from sqlalchemy.engine import Engine
 from loguru import logger as log
 def fetch_employee_master(engine: Engine) -> pl.DataFrame:
    """
@@ -360,3 +356,70 @@ def fetch_employee_master(engine: Engine) -> pl.DataFrame:
    return df
 def fetch_master_visibility_reason(engine: Engine) -> pl.DataFrame:
    """
    Source:
        Master_VisibilityReason
    Target:
        Master_VisibilityReason
    """
    sql = """
    SELECT DISTINCT
        40148 AS project_id,
        MenuId AS menu_id,
        VisibilityReasonId AS reason_id,
        VisibilityReason AS reason
    FROM OneApp_KelloggsMT.dbo.Master_VisibilityReason
    """
    log.info("Fetching Master Visibility Reason data")
    df = pl.read_database(
        query=sql,
        connection=engine
    )
    log.info(
        f"Fetched {len(df):,} Master Visibility Reason records"
    )
    return df
 def fetch_master_visibility_definition(engine: Engine) -> pl.DataFrame:
    """
    Source:
        OneApp_KelloggsMT.dbo.Master_VisibilityDefinition
    Target:
        Master_VisibilityDefinition
    """
    sql = """
    SELECT DISTINCT
        40148 AS project_id,
        VisibilityDefinitionId AS visibility_definition_id,
        VisibilityDefinitionName AS visibility_definition_name,
        GETDATE() AS create_date,
        'SP-Pius' AS create_by
    FROM OneApp_KelloggsMT.dbo.Master_VisibilityDefinition
    """
    log.info("Fetching Master Visibility Definition data")
    df = pl.read_database(
        query=sql,
        connection=engine
    )
    log.info(
        f"Fetched {len(df):,} Master Visibility Definition records"
    )
    return df
@@ -67,7 +67,7 @@ tables:
    type: DIMENSION
    operation: INSERT
-  - name: Web Logins
+  - name: Web_Logins
    type: FACT
    operation: INSERT