254 lines
9.1 KiB
Python
254 lines
9.1 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
CREATED: Mon, 23rd Feb, 2026
|
|
UPDATED: Mon, 23rd Feb, 2026
|
|
|
|
OBJECTIVE:
|
|
|
|
To manually sync attendance for a specific time period.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# For system-level activities:
|
|
import os
|
|
import copy
|
|
|
|
# To work with date and time:
|
|
import time
|
|
import datetime
|
|
|
|
# To work with tabulate data:
|
|
import pandas as pd
|
|
|
|
# Cosec-related:
|
|
from cosec_web.cosec_web import CosecWeb
|
|
|
|
# TCAOFF-related:
|
|
from tcaoff.async_tcaoff import AsyncTheCAOffice
|
|
|
|
# My utils:
|
|
from utils_v2.system import files
|
|
from utils_v2.string import json
|
|
from utils_v2.string import regex
|
|
from utils_v2.date_time import date_time
|
|
|
|
# To work with datatypes:
|
|
from typing import List, Dict, Any
|
|
from collections import defaultdict
|
|
|
|
# To run a cron-like scheduler:
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
|
|
# For async activities:
|
|
import asyncio
|
|
|
|
# For debugging:
|
|
from icecream import IceCreamDebugger
|
|
|
|
# Common and Shared:
|
|
from scripts import common
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# Debugging:
|
|
printer = IceCreamDebugger(prefix = "Manual | ", includeContext = True)
|
|
err_printer = IceCreamDebugger(prefix = "[ERR] Manual | ", includeContext = True)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
async def manual_attendance(
|
|
cosec_creds: dict,
|
|
tcaoff_client: AsyncTheCAOffice,
|
|
from_date: date_time.datetime = None,
|
|
to_date: date_time.datetime = None,
|
|
test_mode: bool = False,
|
|
) -> None:
|
|
|
|
printer("MANUAL ATTENDANCE")
|
|
|
|
# Try the whole process once:
|
|
try:
|
|
|
|
# Log in to TCAOFF:
|
|
success = await tcaoff_client.login()
|
|
if not success: raise RuntimeError("TCAOFF Login Failed!")
|
|
|
|
# Try the part that needs COSEC:
|
|
try:
|
|
|
|
# Figure out the dates:
|
|
now = date_time.get_current_ist_date_time()
|
|
if not from_date: from_date = now - datetime.timedelta(days = 1)
|
|
if not to_date: to_date = now
|
|
|
|
# Set the correct sequence for the dates:
|
|
if from_date > to_date: from_date, to_date = to_date, from_date
|
|
|
|
# Get the previous day's In/Out Summary and then wait
|
|
# for the driver's resources to get freed:
|
|
success = common.get_in_out_summary(
|
|
cosec_creds = cosec_creds,
|
|
from_dt = from_date,
|
|
to_dt = to_date,
|
|
cache_file = common.MANUAL_SUMMARY_CACHE_FILE,
|
|
test_mode = test_mode
|
|
)
|
|
|
|
# Sync data between Cosec and TCAOFF:
|
|
if success:
|
|
printer("IN/OUT SUMMARY: Sync'ing with TCAOFF")
|
|
await common.sync_attendance_to_tcaoff(
|
|
tcaoff_client = tcaoff_client,
|
|
cosec_in_out_summary = json.from_file(common.MANUAL_SUMMARY_CACHE_FILE),
|
|
chunk_size = 10
|
|
)
|
|
|
|
# If something goes wrong in the COSEC step:
|
|
except Exception as exception:
|
|
err_printer(exception)
|
|
if test_mode: raise
|
|
|
|
# Log out of TCAOFF:
|
|
success = await tcaoff_client.logout()
|
|
if not success: raise RuntimeError("TCAOFF Logout Failed!")
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
err_printer(exception)
|
|
await tcaoff_client.logout()
|
|
if test_mode: raise
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
# Parse args:
|
|
import argparse
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument(
|
|
"--test", "--test-mode",
|
|
action = "store_true",
|
|
default = False,
|
|
)
|
|
ap.add_argument(
|
|
"--verbose",
|
|
action = "store_true",
|
|
default = False,
|
|
)
|
|
ap.add_argument(
|
|
"--from-date",
|
|
help = "The date from which you want to sync attendance.",
|
|
default = date_time.get_current_ist_date_time()
|
|
)
|
|
ap.add_argument(
|
|
"--to-date",
|
|
help = "The date till when you want to sync attendance.",
|
|
default = date_time.get_current_ist_date_time()
|
|
)
|
|
args = ap.parse_args()
|
|
|
|
# Explicitly mention the expected file paths for other devs to maintain:
|
|
print("PROJ. DIR. :", common.PROJ_DIR)
|
|
print("COSEC CREDS :", common.COSEC_CREDS_FILE)
|
|
print("TCAOFF CREDS:", common.TCAOFF_CREDS_FILE)
|
|
|
|
# Read required credentials:
|
|
cosec_creds = json.from_file(common.COSEC_CREDS_FILE)
|
|
tcaoff_creds = json.from_file(common.TCAOFF_CREDS_FILE)
|
|
|
|
# Date-handling:
|
|
args.from_date = date_time.parse_date_time(args.from_date, timezone = date_time.TIMEZONE_IST)
|
|
args.to_date = date_time.parse_date_time(args.to_date, timezone = date_time.TIMEZONE_IST)
|
|
if args.from_date > args.to_date: args.from_date, args.to_date = args.to_date, args.from_date
|
|
print("From Date:", args.from_date)
|
|
print(" To Date:", args.to_date)
|
|
# dates_list = [args.from_date]
|
|
# while args.from_date < args.to_date:
|
|
# args.from_date = args.from_date + datetime.timedelta(days = 1)
|
|
# dates_list.append(args.from_date)
|
|
# print(f"DATES ({len(dates_list)}):", dates_list)
|
|
|
|
# Create the clients:
|
|
tcaoff_client = AsyncTheCAOffice(
|
|
username = tcaoff_creds["creds"]["username"],
|
|
password = tcaoff_creds["creds"]["password"],
|
|
debug = True,
|
|
debug_only_errors = False if args.verbose else True,
|
|
)
|
|
|
|
# Schedule the activities:
|
|
asyncio.run(
|
|
manual_attendance(
|
|
cosec_creds = cosec_creds,
|
|
tcaoff_client = tcaoff_client,
|
|
from_date = args.from_date,
|
|
to_date = args.to_date,
|
|
test_mode = args.test,
|
|
)
|
|
)
|