ankitm/data_pipeline
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

3rd commit

Browse Source
This commit is contained in:
Ankit Malik
2026-06-12 15:39:39 +05:30
parent 80bb585cdb
commit 8aaae1e27d
12 changed files with 1280 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kpi/facts.py
+95 -88
View File
@@ -1,5 +1,5 @@
import os
import pyarrow
# import pyarrow
import sys
import logging
from datetime import date, timedelta
@@ -10,7 +10,7 @@ import clickhouse_connect
from dotenv import load_dotenv
from log import log
from clickhouse_task.create_table import create_clickhouse_table , check
from clickhouse_task.create_table import *
from db_con.connection import *
from mids import *
@@ -127,6 +127,7 @@ def fetch_SOS_OneApp(engine: Engine, mids: list[int]) -> pl.DataFrame:
) A
"""
log.info(f"Fetching data for {len(mids):,} MIDs")
df = pl.read_database(
@@ -134,6 +135,7 @@ def fetch_SOS_OneApp(engine: Engine, mids: list[int]) -> pl.DataFrame:
connection=engine
)
log.info(f"Fetched {len(df):,} rows from SQL Server")
return df
@@ -229,6 +231,7 @@ def fetch_OQaD(engine: Engine, mids: list[int]) -> pl.DataFrame:
connection=engine
)
log.info(f"Fetched {len(df):,} rows")
return df
@@ -316,6 +319,7 @@ def fetch_Survey(engine: Engine, mids: list[int]) -> pl.DataFrame:
ON Q.StoreId = SM.StoreId
"""
log.info(f"Fetching Survey data for {len(mids):,} MIDs")
df = pl.read_database(
@@ -323,6 +327,7 @@ def fetch_Survey(engine: Engine, mids: list[int]) -> pl.DataFrame:
connection=engine
)
log.info(f"Fetched {len(df):,} Survey rows")
return df
@@ -396,7 +401,7 @@ def fetch_additional_visibility(
AND TS.Present = 1
AND SC.MID IN ({mid_list})
"""
log.info(
f"Fetching Additional Visibility data for {len(mids):,} MIDs"
)
@@ -406,6 +411,7 @@ def fetch_additional_visibility(
connection=engine
)
log.info(
f"Fetched {len(df):,} Additional Visibility rows"
)
@@ -485,6 +491,7 @@ def fetch_Coverage(engine: Engine, mids: list[int]) -> pl.DataFrame:
AND Em.UserName NOT LIKE 'test%'
"""
log.info(f"Fetching coverage data for {len(mids):,} MIDs")
df = pl.read_database(
@@ -492,14 +499,91 @@ def fetch_Coverage(engine: Engine, mids: list[int]) -> pl.DataFrame:
connection=engine
)
log.info(f"Fetched {len(df):,} rows from SQL Server")
return df
def fetch_Login(engine: Engine , mids: list[int]) -> pl.DataFrame:
def fetch_stock_details(
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:
@@ -583,7 +667,7 @@ def fetch_stock_details(
WHERE EM.EmpName NOT LIKE 'test%'
AND SC.MID IN ({mid_list})
"""
log.info(
f"Fetching Stock Details data for {len(mids):,} MIDs"
)
@@ -678,88 +762,10 @@ def fetch_Attendance(
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_journey_plan(
def fetch_Journey_Plan(
engine: Engine,
report_date: date
) -> pl.DataFrame:
@@ -830,7 +836,7 @@ def fetch_coverage_remarks(engine: Engine , mids: list[int]) -> pl.DataFrame:
return df
def fetch_web_logins(
def fetch_Web_Logins(
engine: Engine,
run_date: date
) -> pl.DataFrame:
@@ -893,7 +899,7 @@ def fetch_web_logins(
WHERE CAST(AL.Date AS DATE) = '{run_date}'
"""
log.info(f"Fetching Web Login data for {run_date}")
df = pl.read_database(
@@ -901,6 +907,7 @@ def fetch_web_logins(
connection=engine
)
log.info(
f"Fetched {len(df):,} Web Login records"
)
@@ -910,7 +917,7 @@ def fetch_web_logins(
def fetch_promotion(
def fetch_Promotion(
engine: Engine,
mids: list[int]
) -> pl.DataFrame:
@@ -1061,7 +1068,7 @@ def fetch_promotion(
def fetch_paid_visibility(
def fetch_PaidVisibility(
engine: Engine,
mids: list[int]
) -> pl.DataFrame: