import pandas as pd
import os
import glob

BASE_DIR = "/opt/airflow"
BATCH_DIR = os.path.join(BASE_DIR, "batches")
TRANSFORM_DIR = os.path.join(BASE_DIR, "transformed_batches")


def transform():
    os.makedirs(TRANSFORM_DIR, exist_ok=True)
    files = glob.glob(os.path.join(BATCH_DIR, "batch_*.csv"))

    if not files:
        print("No batches found")
        return

    for file in files:
        print("Processing:", file)
        df = pd.read_csv(file)

        df = df.drop_duplicates()
        df = df.dropna(subset=[
            "Employee_ID",
            "Department",
            "Age",
            "Monthly_Salary"
        ])

        df["Hire_Date"] = pd.to_datetime(df["Hire_Date"])
        df["Hire_Year"] = df['Hire_Date'].dt.year
        df["Hire_Month"] = df['Hire_Date'].dt.month

        df['Salary_Per_Year'] = df["Monthly_Salary"]*12
        df['Overtime_Ratio'] = df["Overtime_Hours"]/df["Work_Hours_Per_Week"]

        df["is_manager"] = df["Job_Title"].str.contains(
            "Manager", case=False, na=False).astype(int)

        text_cols = ["Department", "Gender", "Job_Title", "Education_Level"]
        for col in text_cols:
            df[col] = df[col].str.strip().str.lower()

        name = os.path.basename(file)
        out = os.path.join(TRANSFORM_DIR, f"trans_{name}")
        df.to_csv(out, index=False)
        print("Saved:", out)


if __name__ == "__main__":
    transform()