Comparison_Project/app/services/file_service.py

import os
import pandas as pd
from werkzeug.utils import secure_filename

from app.config import Config
from app import db

from app.models.trench_excavation_model import TrenchExcavation
from app.models.manhole_excavation_model import ManholeExcavation
from app.models.manhole_domestic_chamber_model import ManholeDomesticChamber

from app.models.tr_ex_client_model import TrenchExcavationClient
from app.models.mh_ex_client_model import ManholeExcavationClient
from app.models.mh_dc_client_model import ManholeDomesticChamberClient

from app.utils.file_utils import ensure_upload_folder


class FileService:

    def allowed_file(self, filename):
        return (
            "." in filename
            and filename.rsplit(".", 1)[1].lower() in Config.ALLOWED_EXTENSIONS
        )

    # -------------------------- SUBCONTRACTORFILE UPLOAD ------------------------- 
    # SUBCONTRACTOR FILE UPLOAD
    def handle_file_upload(self, file, subcontractor_id, RA_Bill_No):

        if not subcontractor_id:
            return False, "Please select subcontractor."

        if not RA_Bill_No:
            return False, "Please Enter RA Bill No."

        if not file or file.filename == "":
            return False, "No file selected."

        if not self.allowed_file(file.filename):
            return False, "Invalid file type! Allowed: CSV, XLSX, XLS"

        ensure_upload_folder()

        folder = os.path.join(Config.UPLOAD_FOLDER, f"sub_{subcontractor_id}")
        os.makedirs(folder, exist_ok=True)

        filename = secure_filename(file.filename)
        filepath = os.path.join(folder, filename)
        file.save(filepath)

        try:
            df_tr_ex = pd.read_excel(filepath, sheet_name="Tr.Ex.", header=12)
            df_mh_ex = pd.read_excel(filepath, sheet_name="MH Ex.", header=12)
            df_mh_dc = pd.read_excel(filepath, sheet_name="MH & DC", header=11)

            self.process_trench_excavation(df_tr_ex, subcontractor_id, RA_Bill_No)
            self.process_manhole_excavation(df_mh_ex, subcontractor_id, RA_Bill_No)
            self.process_manhole_domestic_chamber(df_mh_dc, subcontractor_id, RA_Bill_No)

            return True, "SUBCONTRACTOR File uploaded successfully."

        except Exception as e:
            db.session.rollback()
            return False, f"Processing failed(SUBCONTRACTOR): {e}"

    # Trench Excavation (Subcontractor)
    def process_trench_excavation(self, df, subcontractor_id, RA_Bill_No):

        df.columns = (
            df.columns.astype(str)
            .str.strip()
            .str.replace(r"[^\w]", "_", regex=True)
            .str.replace("__+", "_", regex=True)
            .str.strip("_")
        )

        df = df.dropna(how="all")

        if "Location" in df.columns:
            df["Location"] = df["Location"].ffill()

        try:
            for _, row in df.iterrows():
                location = row.get("Location")
                mh_no = row.get("MH_NO")

                if (
                    pd.isna(location)
                    or str(location).strip() == ""
                    or pd.isna(mh_no)
                    or str(mh_no).strip() == ""
                ):
                    continue

                record_data = {}
                for col in df.columns:
                    if hasattr(TrenchExcavation, col):
                        val = row[col]
                        if pd.isna(val) or str(val).strip() in ["", "-", "—", "nan"]:
                            val = None
                        record_data[col] = val

                record = TrenchExcavation(
                    subcontractor_id=subcontractor_id,
                    RA_Bill_No=RA_Bill_No,
                    **record_data,
                )
                db.session.add(record)

            db.session.commit()
            return True, "Trench Excavation saved successfully."

        except Exception as e:
            db.session.rollback()
            raise e

    # Manhole Excavation (Subcontractor)
    def process_manhole_excavation(self, df, subcontractor_id, RA_Bill_No):

        df.columns = (
            df.columns.astype(str)
            .str.strip()
            .str.replace(r"[^\w]", "_", regex=True)
            .str.replace("__+", "_", regex=True)
            .str.strip("_")
        )

        df = df.dropna(how="all")

        if "Location" in df.columns:
            df["Location"] = df["Location"].ffill()

        try:
            for _, row in df.iterrows():
                location = row.get("Location")
                mh_no = row.get("MH_NO")

                if (
                    pd.isna(location)
                    or str(location).strip() == ""
                    or pd.isna(mh_no)
                    or str(mh_no).strip() == ""
                ):
                    continue

                record_data = {}
                for col in df.columns:
                    if hasattr(ManholeExcavation, col):
                        val = row[col]
                        if pd.isna(val) or str(val).strip() in ["", "-", "—", "nan"]:
                            val = None
                        record_data[col] = val

                record = ManholeExcavation(
                    subcontractor_id=subcontractor_id,
                    RA_Bill_No=RA_Bill_No,
                    **record_data,
                )
                db.session.add(record)

            db.session.commit()
            return True, "Manhole Excavation saved successfully."

        except Exception as e:
            db.session.rollback()
            raise e

    
    # Manhole & Domestic Chamber (Subcontractor)
    def process_manhole_domestic_chamber(self, df, subcontractor_id, RA_Bill_No):

        df.columns = (
            df.columns.astype(str)
            .str.strip()
            .str.replace(r"[^\w]", "_", regex=True)
            .str.replace("__+", "_", regex=True)
            .str.strip("_")
        )

        df = df.dropna(how="all")

        if "Location" in df.columns:
            df["Location"] = df["Location"].ffill()

        try:
            for _, row in df.iterrows():
                location = row.get("Location")
                node_no = row.get("Node_No")

                if (
                    pd.isna(location)
                    or str(location).strip() == ""
                    or pd.isna(node_no)
                    or str(node_no).strip() == ""
                ):
                    continue

                record_data = {}
                for col in df.columns:
                    if hasattr(ManholeDomesticChamber, col):
                        val = row[col]
                        if pd.isna(val) or str(val).strip() in ["", "-", "—", "nan"]:
                            val = None
                        record_data[col] = val

                record = ManholeDomesticChamber(
                    subcontractor_id=subcontractor_id,
                    RA_Bill_No=RA_Bill_No,
                    **record_data,
                )
                db.session.add(record)

            db.session.commit()
            return True, "Manhole Domestic Chamber saved successfully."

        except Exception as e:
            db.session.rollback()
            raise e

   
    # ------------------- CLIENT FILE UPLOAD --------------------------
    def handle_client_file_upload(self, file, RA_Bill_No):

        if not RA_Bill_No:
            return False, "Please Enter RA Bill No."

        if not file or file.filename == "":
            return False, "No file selected."

        if not self.allowed_file(file.filename):
            return False, "Invalid file type! Allowed: CSV, XLSX, XLS"

        ensure_upload_folder()

        folder = os.path.join(Config.UPLOAD_FOLDER, f"Client_Bill_{RA_Bill_No}")
        os.makedirs(folder, exist_ok=True)

        filename = secure_filename(file.filename)
        filepath = os.path.join(folder, filename)
        file.save(filepath)

        try:
            df_tr_ex = pd.read_excel(filepath, sheet_name="Tr.Ex.", header=4)
            df_mh_ex = pd.read_excel(filepath, sheet_name="MH Ex.", header=4)
            df_mh_dc = pd.read_excel(filepath, sheet_name="MH & DC", header=3)

            print("------------------")
            print("df_tr_ex :",df_tr_ex)
            print("------------------")
            print("df_mh_ex :",df_mh_ex)
            print("------------------")
            print("df_mh_dc :",df_mh_dc)
            print("------------------")

            self.save_client_data(df_tr_ex, TrenchExcavationClient, RA_Bill_No)
            self.save_client_data(df_mh_ex, ManholeExcavationClient, RA_Bill_No)
            self.save_client_data(df_mh_dc, ManholeDomesticChamberClient, RA_Bill_No)

            db.session.commit()
            return True, "Client file uploaded and all data saved successfully."

        except Exception as e:
            db.session.rollback()
            return False, f"Client data save failed: {e}"

    # Client all model Save Method
    def save_client_data(self, df, model, RA_Bill_No):

        df.columns = [str(c).strip() for c in df.columns]

        if "Location" in df.columns:
            df["Location"] = df["Location"].ffill()

        df = df.dropna(how="all")

        if "Location" in df.columns:
            missing_loc = df[
                df["Location"].isna()
                | (df["Location"].astype(str).str.strip() == "")
            ]
            if not missing_loc.empty:
                raise Exception(
                    f"{model.__name__}: Empty Location at rows {missing_loc.index.tolist()}"
                )

        for _, row in df.iterrows():
            record_data = {}

            for col in df.columns:
                if hasattr(model, col):
                    val = row[col]
                    if pd.isna(val) or str(val).strip() in ["", "-", "—", "nan", "NaN"]:
                        val = None
                    record_data[col] = val

            record = model(RA_Bill_No=RA_Bill_No, **record_data)
            db.session.add(record)