import pandas as pd import io from flask import Blueprint, render_template, request, send_file, flash, jsonify,redirect, url_for from app.utils.helpers import login_required from app.utils.regex_utils import RegularExpression from app import db from sqlalchemy import func import re from app.models.subcontractor_model import Subcontractor from app.models.manhole_excavation_model import ManholeExcavation from app.models.trench_excavation_model import TrenchExcavation from app.models.manhole_domestic_chamber_model import ManholeDomesticChamber from app.models.laying_model import Laying from app.models.mh_ex_client_model import ManholeExcavationClient from app.models.tr_ex_client_model import TrenchExcavationClient from app.models.mh_dc_client_model import ManholeDomesticChamberClient from app.models.laying_client_model import LayingClient from app.services.abstract_service import AbstractReportService # --- BLUEPRINT DEFINITION --- file_report_bp = Blueprint("file_report", __name__, url_prefix="/file") # ---------------- LOCATION HELPERS ---------------- WORK_MODELS = [TrenchExcavation, ManholeExcavation, ManholeDomesticChamber, Laying] def get_distinct_locations(): """Union of distinct, non-empty Location values across all 4 work tables.""" locations = set() for Model in WORK_MODELS: rows = db.session.query(Model.Location).distinct().all() for (loc,) in rows: if loc and loc.strip(): locations.add(loc.strip()) return sorted(locations) def get_subcontractors_for_location(location): """Return Subcontractor objects that have at least one work record (in any of the 4 category tables) at the given location. If no location is given, returns every subcontractor. Shared by the AJAX endpoint below and by the server-rendered dropdown so the list is correct even before/without JS running (e.g. on page reload or a validation-error re-render). Matching is case- and whitespace-insensitive, since Location is a free-text field and stored values can drift ("Pune" vs "PUNE " etc.) even though the dropdown options themselves come from a distinct query and look identical.""" location = (location or "").strip() if not location: return Subcontractor.query.order_by(Subcontractor.subcontractor_name).all() target = location.upper() sc_ids = set() for Model in WORK_MODELS: rows = ( db.session.query(Model.subcontractor_id) .filter(func.upper(func.trim(Model.Location)) == target) .distinct() .all() ) for (sid,) in rows: if sid: sc_ids.add(sid) if not sc_ids: return [] return ( Subcontractor.query.filter(Subcontractor.id.in_(sc_ids)) .order_by(Subcontractor.subcontractor_name) .all() ) @file_report_bp.route("/get_subcontractors_by_location") @login_required def get_subcontractors_by_location(): """AJAX endpoint: return subcontractors that have at least one work record (in any of the 4 category tables) at the given location.""" location = request.args.get("location", "").strip() subs = get_subcontractors_for_location(location) return jsonify([{"id": s.id, "name": s.subcontractor_name} for s in subs]) # ---------------- ACTION COLUMN ---------------- def add_action_columns(df, model_key): if df.empty: return df df.insert(0, "Select", df["Id"].apply( lambda x: f'' )) df["Update"] = df["Id"].apply( lambda x: f' Edit' ) df["Delete"] = df["Id"].apply( lambda x: f'' ) return df # ---------------- SELECT-ALL HEADER ---------------- def add_select_all_header(table_html, model_key): """Swap pandas' plain 'Select' column header for a select-all checkbox scoped to this table (via data-model), so checking it only toggles rows in this table - not the other 3 category tables on the page.""" return table_html.replace( "Select", f'', 1 ) # ---------------- TABLE OR EMPTY-STATE ---------------- def add_data_field_attrs(table_html, raw_fields): """Tag each editable data with a data-field attribute naming its underlying database column, so bulk-edit mode on the frontend knows exactly which field to submit for each input it creates. The 'id' column is deliberately skipped - it's the primary key and must never be editable. Uses match spans (not string search-and-replace) to rebuild each row, because a naive .replace() on duplicate cell text (e.g. two cells that both just say "0.00") would edit the wrong cell.""" total_cols = 1 + len(raw_fields) + 2 # Select + data columns + Update + Delete def process_row(m): row_html = m.group(0) matches = list(re.finditer(r".*?", row_html, flags=re.S)) if len(matches) != total_cols: return row_html # shape mismatch - leave untouched rather than guess pieces = [] last_end = 0 for i, mt in enumerate(matches): pieces.append(row_html[last_end:mt.start()]) cell = mt.group(0) if 1 <= i <= len(raw_fields): field = raw_fields[i - 1] if field.lower() != "id": cell = f'' + cell[len(""):] pieces.append(cell) last_end = mt.end() pieces.append(row_html[last_end:]) return "".join(pieces) return re.sub(r".*?", process_row, table_html, flags=re.S) def render_table_or_empty(df, model_key, table_class, raw_fields=None): """Render a table, or a friendly placeholder if there's no data. IMPORTANT: pandas' to_html() on a fully empty DataFrame (0 rows AND 0 columns, which is what we get when a category has no matching records) still emits a - just with a header row that has zero
cells. jQuery DataTables then tries to initialize on a table with no columns and throws, which (since the init code runs as one synchronous block) silently kills every bit of JS registered after it - including the select-all checkbox handler for the OTHER tables on the page. Returning a plain message instead of an empty avoids ever handing DataTables something it can't initialize.""" if df.empty: return '
No records found.
' html = df.to_html(classes=table_class, index=False, escape=False) html = add_select_all_header(html, model_key) html = add_data_field_attrs(html, raw_fields or []) return html # ---------------- FETCH ---------------- class SubcontractorBill: def __init__(self): self.df_tr = pd.DataFrame() self.df_mh = pd.DataFrame() self.df_dc = pd.DataFrame() self.df_laying = pd.DataFrame() # self.df_abstract = pd.DataFrame() # NEW def Fetch(self, RA_Bill_No=None, subcontractor_id=None, location=None, mh_no=None): filters = {} if subcontractor_id: filters["subcontractor_id"] = subcontractor_id if RA_Bill_No: filters["RA_Bill_No"] = RA_Bill_No # Fetch data in database trench = TrenchExcavation.query.filter_by(**filters).all() mh = ManholeExcavation.query.filter_by(**filters).all() dc = ManholeDomesticChamber.query.filter_by(**filters).all() lay = Laying.query.filter_by(**filters).all() # LOCATION FILTER if location: search = location.strip().lower() trench = [ t for t in trench if search in (t.Location or "").strip().lower() ] mh = [ t for t in mh if search in (t.Location or "").strip().lower() ] dc = [ t for t in dc if search in (t.Location or "").strip().lower() ] lay = [ t for t in lay if search in (t.Location or "").strip().lower() ] # MH NO FILTER if mh_no: mh_search = mh_no.strip().lower() trench = [ t for t in trench if mh_search in (t.MH_NO or "").strip().lower() ] mh = [ t for t in mh if mh_search in (t.MH_NO or "").strip().lower() ] dc = [ t for t in dc if mh_search in (t.MH_NO or "").strip().lower() ] lay = [ t for t in lay if mh_search in (t.MH_NO or "").strip().lower() ] # Set dataframe self.df_tr = pd.DataFrame([c.serialize() for c in trench]) self.df_mh = pd.DataFrame([c.serialize() for c in mh]) self.df_dc = pd.DataFrame([c.serialize() for c in dc]) self.df_laying = pd.DataFrame([c.serialize() for c in lay]) drop_cols = ["11", "_sa_instance_state", "subcontractor_id" , "created_at"] # Raw (pre-format_column_names) column names for each table, e.g. # "MH_NO" rather than the display label "MH No". Bulk-edit mode # needs these to know which real database column each input maps # to, since the table only shows the prettified header text. self.tr_fields = [] self.mh_fields = [] self.dc_fields = [] self.laying_fields = [] for df, attr in [ (self.df_tr, "tr_fields"), (self.df_mh, "mh_fields"), (self.df_dc, "dc_fields"), (self.df_laying, "laying_fields"), ]: if not df.empty: df.drop(columns=drop_cols, errors="ignore", inplace=True) setattr(self, attr, list(df.columns)) format_column_names(df) name = "" if subcontractor_id: sc = Subcontractor.query.get(subcontractor_id) if sc: name = sc.subcontractor_name # ---------------- DELETE ---------------- @file_report_bp.route("/delete_records", methods=["POST"]) @login_required def delete_records(): data = request.json or {} model = data.get("model") ids = data.get("ids", []) model_map = { "tr": TrenchExcavation, "mh": ManholeExcavation, "dc": ManholeDomesticChamber, "laying": Laying } ModelClass = model_map.get(model) # validate model BEFORE using it if not ModelClass: return jsonify({"status": "error", "message": f"Invalid model '{model}'"}), 400 if not ids: return jsonify({"status": "error", "message": "No IDs provided"}), 400 try: for record_id in ids: obj = ModelClass.query.get(record_id) if obj: db.session.delete(obj) db.session.commit() return jsonify({"status": "success"}) except Exception as e: db.session.rollback() return jsonify({"status": "error", "message": str(e)}), 500 @file_report_bp.route("/bulk_update", methods=["POST"]) @login_required def bulk_update(): """Bulk-edit save endpoint. Expects JSON shaped like: { "tr": { "5": {"MH_NO": "12A", "Location": "Pune"}, ... }, "mh": {...}, ... } Field names are validated against each model's real table columns server-side - the frontend sending a field name is not enough on its own to permit writing it; id/subcontractor_id/created_at are always refused regardless of what's submitted. """ data = request.json or {} model_map = { "tr": TrenchExcavation, "mh": ManholeExcavation, "dc": ManholeDomesticChamber, "laying": Laying } PROTECTED_FIELDS = {"id", "subcontractor_id", "created_at"} updated = 0 errors = [] try: for model_key, records in data.items(): ModelClass = model_map.get(model_key) if not ModelClass: errors.append(f"Unknown table '{model_key}'") continue valid_columns = {c.name for c in ModelClass.__table__.columns} - PROTECTED_FIELDS for record_id, fields in (records or {}).items(): obj = ModelClass.query.get(record_id) if not obj: errors.append(f"{model_key} #{record_id}: record not found") continue for field, value in (fields or {}).items(): if field not in valid_columns: errors.append(f"{model_key} #{record_id}: '{field}' is not editable") continue setattr(obj, field, value) updated += 1 db.session.commit() except Exception as e: db.session.rollback() return jsonify({"status": "error", "message": str(e)}), 500 return jsonify({"status": "success", "updated": updated, "errors": errors}) @file_report_bp.route("/edit//", methods=["GET", "POST"]) @login_required def edit_record(model, record_id): model_map = { "tr": TrenchExcavation, "mh": ManholeExcavation, "dc": ManholeDomesticChamber, "laying": Laying } ModelClass = model_map.get(model) if not ModelClass: flash("Invalid Model.", "danger") return redirect(url_for("file_report.report_file")) record = ModelClass.query.get_or_404(record_id) if request.method == "POST": # Update all fields except id for column in record.__table__.columns: if column.name == "id": continue if column.name in request.form: setattr(record, column.name, request.form.get(column.name)) try: db.session.commit() flash("Record updated successfully.", "success") # ✅ fixed: correct blueprint name return redirect(url_for("file_report.report_file")) except Exception as e: db.session.rollback() flash(str(e), "danger") return render_template( "edit_record.html", record=record, model=model ) @file_report_bp.route("/Subcontractor_report", methods=["GET", "POST"]) @login_required def report_file(): # get all subcontractor data subcontractors = Subcontractor.query.all() locations = get_distinct_locations() tables = None abstract_html = "" selected_sc_id = None has_data = {"tr": False, "mh": False, "dc": False, "laying": False} ra_bill_no = "" location = "" mh_no = "" category = "" # Search or load data if request.method == "POST": # get from data subcontractor_id = request.form.get("subcontractor_id") or None ra_bill_no = request.form.get("ra_bill_no", "").strip() location = request.form.get("location", "").strip() mh_no = request.form.get("mh_no", "").strip() category = request.form.get("category", "") action = request.form.get("action", "preview") # Keep the subcontractor dropdown scoped to the chosen location # even on a plain (non-JS) page render. subcontractors = get_subcontractors_for_location(location) # Subcontractor is now optional - at least one other filter must # be given so the search isn't a "return everything" query. if not subcontractor_id and not location and not ra_bill_no and not mh_no: flash("Enter at least a Location, RA Bill No, MH No, or Subcontractor to search", "danger") return render_template( "subcontractor_report.html", subcontractors=subcontractors, locations=locations, has_data=has_data ) selected_sc_id = subcontractor_id bill = SubcontractorBill() if action == "excel_all": bill.Fetch(subcontractor_id=subcontractor_id) else: bill.Fetch(ra_bill_no, subcontractor_id, location, mh_no) # --------------------------------------------------------- if ( bill.df_tr.empty and bill.df_mh.empty and bill.df_dc.empty and bill.df_laying.empty ): flash( f"No records found for RA Bill No '{ra_bill_no}'. " "Please check the RA Bill No and try again." if ra_bill_no else "No records found for the selected filters.", "warning" ) return render_template( "subcontractor_report.html", subcontractors=subcontractors, selected_sc_id=selected_sc_id, selected_ra_bill=ra_bill_no, selected_location=location, selected_category=category ) # ----------------------------------------- # Generate Abstract Report for Web # ----------------------------------------- abstract_service = AbstractReportService( subcontractor_id=subcontractor_id, ra_bill_no=ra_bill_no ) abstract_html = abstract_service.generate_html() # ---------------- CATEGORY FILTER ---------------- if category == "tr": bill.df_mh = bill.df_dc = bill.df_laying = pd.DataFrame() elif category == "mh": bill.df_tr = bill.df_dc = bill.df_laying = pd.DataFrame() elif category == "dc": bill.df_tr = bill.df_mh = bill.df_laying = pd.DataFrame() elif category == "laying": bill.df_tr = bill.df_mh = bill.df_dc = pd.DataFrame() if ( category in ("tr", "mh", "dc", "laying") and bill.df_tr.empty and bill.df_mh.empty and bill.df_dc.empty and bill.df_laying.empty ): category_labels = { "tr": "Trench Excavation", "mh": "Manhole Excavation", "dc": "Domestic Chamber", "laying": "Pipe Laying", } flash( f"No {category_labels[category]} records found for the " "selected filters.", "warning" ) return render_template( "subcontractor_report.html", subcontractors=subcontractors, selected_sc_id=selected_sc_id, selected_ra_bill=ra_bill_no, selected_location=location, selected_category=category ) # =================================================== # DOWNLOAD EXCEL # =================================================== if action in ["excel", "excel_all"]: output = io.BytesIO() with pd.ExcelWriter(output,engine="xlsxwriter") as writer: workbook = writer.book abstract = AbstractReportService(subcontractor_id=subcontractor_id,ra_bill_no=ra_bill_no) abstract.generate(workbook) sheet_map = [ (bill.df_tr, "Tr.Ex"), (bill.df_mh, "Mh.Ex"), (bill.df_dc, "MH & DC"), (bill.df_laying, "Pipe Laying"), ] for df, sheet_name in sheet_map: if not df.empty: df.to_excel(writer, sheet_name=sheet_name, index=False) writer.close() output.seek(0) sc_obj = next( (s for s in subcontractors if str(s.id) == str(subcontractor_id)), None ) sc_name = sc_obj.subcontractor_name if sc_obj else "Subcontractor" sc_name = re.sub(r'[^A-Za-z0-9_-]+', '_', sc_name).strip('_') name_parts = [sc_name] if ra_bill_no: name_parts.append(f"RA{re.sub(r'[^A-Za-z0-9_-]+', '_', ra_bill_no)}") if location: name_parts.append(re.sub(r'[^A-Za-z0-9_-]+', '_', location).strip('_')) if category and category != "all": name_parts.append(category.upper()) if action == "excel_all": name_parts.append("All") filename = "_".join(name_parts) + "_Report.xlsx" return send_file( output, download_name=filename, as_attachment=True ) # =================================================== # PDF # =================================================== if action == "pdf": flash( "PDF Export Coming Soon.", "info" ) # =================================================== # ADD ACTIONS # =================================================== bill.df_tr = add_action_columns(bill.df_tr, "tr") bill.df_mh = add_action_columns(bill.df_mh, "mh") bill.df_dc = add_action_columns(bill.df_dc, "dc") bill.df_laying = add_action_columns(bill.df_laying, "laying") # Used by the template to hide Delete Selected / Bulk Edit for a # category that has no rows to act on. has_data = { "tr": not bill.df_tr.empty, "mh": not bill.df_mh.empty, "dc": not bill.df_dc.empty, "laying": not bill.df_laying.empty, } # this are html classes # table_class = ( "table " "table-bordered" "table-hover " "table-striped " "table-sm " "align-middle " "datatable " "mb-0") table_class = ( "table " "table-bordered " "table-hover " "table-striped " "table-sm " "align-middle " "datatable " "text-nowrap " "mb-0" ) # This are showing on web tables tables = { "tr": render_table_or_empty(bill.df_tr, "tr", table_class, bill.tr_fields), "mh": render_table_or_empty(bill.df_mh, "mh", table_class, bill.mh_fields), "dc": render_table_or_empty(bill.df_dc, "dc", table_class, bill.dc_fields), "laying": render_table_or_empty(bill.df_laying, "laying", table_class, bill.laying_fields) } return render_template( "subcontractor_report.html", subcontractors=subcontractors, locations=locations, selected_sc_id=selected_sc_id, selected_ra_bill=ra_bill_no, selected_location=location, selected_mh_no=mh_no, selected_category=category, tables=tables, abstract_html=abstract_html, has_data=has_data ) # --- Client class --- class ClientBill: def __init__(self): self.df_tr = pd.DataFrame() self.df_mh = pd.DataFrame() self.df_dc = pd.DataFrame() self.df_laying = pd.DataFrame() def Fetch(self, RA_Bill_No): trench = TrenchExcavationClient.query.filter_by(RA_Bill_No=RA_Bill_No).all() mh = ManholeExcavationClient.query.filter_by(RA_Bill_No=RA_Bill_No).all() dc = ManholeDomesticChamberClient.query.filter_by(RA_Bill_No=RA_Bill_No).all() lay = LayingClient.query.filter_by(RA_Bill_No=RA_Bill_No).all() self.df_tr = pd.DataFrame([c.serialize() for c in trench]) self.df_mh = pd.DataFrame([c.serialize() for c in mh]) self.df_dc = pd.DataFrame([c.serialize() for c in dc]) self.df_laying = pd.DataFrame([c.serialize() for c in lay]) drop_cols = ["id", "created_at", "_sa_instance_state"] for df in [self.df_tr, self.df_mh, self.df_dc, self.df_laying]: if not df.empty: df.drop(columns=drop_cols, errors="ignore", inplace=True) # --- CLIENT REPORT (PREVIEW + DOWNLOAD) --- @file_report_bp.route("/client_report", methods=["GET", "POST"]) @login_required def client_report(): tables = {"tr": None, "mh": None, "dc": None, "laying": None} ra_val = "" if request.method == "POST": # ⚠ MUST match HTML name RA_Bill_No = request.form.get("RA_Bill_No") action = request.form.get("action") ra_val = RA_Bill_No if not RA_Bill_No: flash("Please enter RA Bill No.", "danger") return render_template("client_report.html", tables=tables, ra_val=ra_val) # -------- FETCH CLIENT DATA -------- bill_gen = ClientBill() bill_gen.Fetch(RA_Bill_No) # If no data if ( bill_gen.df_tr.empty and bill_gen.df_mh.empty and bill_gen.df_dc.empty and bill_gen.df_laying.empty ): flash(f"No Client records found for RA Bill {RA_Bill_No}", "warning") return render_template("client_report.html", tables=tables, ra_val=ra_val) # -------- DOWNLOAD -------- if action == "download": output = io.BytesIO() with pd.ExcelWriter(output, engine="xlsxwriter") as writer: bill_gen.df_tr.to_excel(writer, index=False, sheet_name="Trench") bill_gen.df_mh.to_excel(writer, index=False, sheet_name="MH") bill_gen.df_dc.to_excel(writer, index=False, sheet_name="MH & DC") bill_gen.df_laying.to_excel(writer, index=False, sheet_name="Laying") output.seek(0) return send_file( output, download_name=f"Client_RA_{RA_Bill_No}_Report.xlsx", as_attachment=True ) # -------- PREVIEW -------- table_class = "table table-bordered table-striped table-hover table-sm" tables["tr"] = bill_gen.df_tr.to_html(classes=table_class, index=False) tables["mh"] = bill_gen.df_mh.to_html(classes=table_class, index=False) tables["dc"] = bill_gen.df_dc.to_html(classes=table_class, index=False) tables["laying"] = bill_gen.df_laying.to_html(classes=table_class, index=False) return render_template("client_report.html", tables=tables, ra_val=ra_val) def format_column_names(df): if df.empty: return df new_columns = [] for col in df.columns: # ---------------------------------------- # Pipe columns # pipe_150_mm -> Pipe 150 MM # ---------------------------------------- if RegularExpression.PIPE_MM_PATTERN.match(col): m = re.match(r"pipe_(\d+)_mm", col) new_columns.append(f"Pipe {m.group(1)} MM") continue # ---------------------------------------- # Domestic Chamber # d_0_to_0_75 -> 0.00 To 0.75 # d_1_5_to_3_0 -> 1.50 To 3.00 # ---------------------------------------- if RegularExpression.D_RANGE_PATTERN.match(col): value = col[2:] # remove d_ value = re.sub( r'(\d+)_(\d+)', lambda m: f"{m.group(1)}.{m.group(2)}", value ) value = value.replace("_to_", " To ") new_columns.append(value) continue # ---------------------------------------- # Total columns # Soft_Murum_0_to_1_5_total # -> # Soft Murum 0 To 1.5 Total # ---------------------------------------- if RegularExpression.STR_TOTAL_PATTERN.match(col): value = col[:-6] # remove _total value = re.sub( r'(\d+)_(\d+)', lambda m: f"{m.group(1)}.{m.group(2)}", value ) value = value.replace("_to_", " To ") value = value.replace("_", " ") new_columns.append(value.title() + " Total") continue # ---------------------------------------- # General columns # ---------------------------------------- value = col.replace("_", " ").title() replacements = { "Mh No": "MH No", "Ra Bill No": "RA Bill No", "Cc Length": "CC Length", "Id Of Mh M": "ID of MH (m)", "Pipe Dia Mm": "Pipe Dia (MM)", "Mh Top Level": "MH Top Level", "Upto Il Depth": "Upto IL Depth", "Actual Trench Length": "Actual Trench Length", "Ground Level": "Ground Level", "Invert Level": "Invert Level", "Ex Dia Of Manhole": "External Dia of Manhole", "Area Of Manhole": "Area of Manhole", "Depth Of Mh": "Depth of MH", } value = replacements.get(value, value) new_columns.append(value) df.columns = new_columns return df