842 lines
29 KiB
Python
842 lines
29 KiB
Python
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'<input type="checkbox" class="row-check" data-model="{model_key}" data-id="{x}">'
|
|
))
|
|
|
|
df["Update"] = df["Id"].apply(
|
|
lambda x: f'<a href="/file/edit/{model_key}/{x}" class="btn btn-sm btn-warning edit-btn"><i class="bi bi-pencil-square"></i> Edit</a>'
|
|
)
|
|
|
|
df["Delete"] = df["Id"].apply(
|
|
lambda x: f'<button class="btn btn-sm btn-danger delete-btn" data-id="{x}" data-model="{model_key}">Delete</button>'
|
|
)
|
|
|
|
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(
|
|
"<th>Select</th>",
|
|
f'<th><input type="checkbox" class="select-all-checkbox" '
|
|
f'data-model="{model_key}" title="Select All"></th>',
|
|
1
|
|
)
|
|
|
|
|
|
# ---------------- TABLE OR EMPTY-STATE ----------------
|
|
def add_data_field_attrs(table_html, raw_fields):
|
|
"""Tag each editable data <td> 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"<td>.*?</td>", 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'<td data-field="{field}">' + cell[len("<td>"):]
|
|
pieces.append(cell)
|
|
last_end = mt.end()
|
|
pieces.append(row_html[last_end:])
|
|
return "".join(pieces)
|
|
|
|
return re.sub(r"<tr>.*?</tr>", 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 <table class="datatable"> - just with a
|
|
header row that has zero <th> 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 <table class="datatable"> avoids ever handing DataTables
|
|
something it can't initialize."""
|
|
if df.empty:
|
|
return '<div class="alert alert-info mb-0">No records found.</div>'
|
|
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/<string:model>/<int:record_id>", 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 |