import pandas as pd
import glob
import os
from sqlalchemy import create_engine
from urllib.parse import quote_plus

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

user = "postgres"
password = quote_plus("Krasava10&")
host = "postgres"
port = "5432"
database = "employee_db"

engine = create_engine(
    f"postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}"
)


def load():
    os.makedirs(TRANSFORM_DIR, exist_ok=True)
    batch_files = glob.glob(os.path.join(TRANSFORM_DIR, "trans_batch_*.csv"))

    if not batch_files:
        print("No transformed batches found")
        return

    for batch_file in batch_files:
        print("Loading batch:", batch_file)
        df = pd.read_csv(batch_file, parse_dates=['Hire_Date'])
        df.to_sql('employee_performance', engine,
                  if_exists='append', index=False)
        print(f"Batch {batch_file} loaded into PostgreSQL")


if __name__ == "__main__":
    load()