pjpatil12/Comparison_Project
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

update code of dash and report of cont_client and devlp abstract of qty

Browse Source
This commit is contained in:
Pankaj J Patil
2026-07-30 14:38:37 +05:30
parent 12b2032430
commit bf1df3a817
42 changed files with 3066 additions and 952 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
app/routes/auth.py
View File

@@ -1,48 +1,97 @@
from flask import Blueprint, render_template, request, redirect, url_for, flash, session
from flask import (Blueprint, render_template, request, redirect, url_for, flash, session, current_app)
from app.services.user_service import UserService
from app.constants.messages import SuccessMessage, ErrorMessage
from app.constants.http_status import HTTPStatus
auth_bp = Blueprint("auth", __name__)
# ==========================
# LOGIN
# ==========================
@auth_bp.route("/login", methods=["GET", "POST"])
def login():
if session.get("user_id"):
current_app.logger.info("User already logged in.")
return redirect(url_for("dashboard.dashboard"))
if request.method == "POST":
email = request.form.get("email")
password = request.form.get("password")
user = UserService.validate_login(email, password)
if user:
session["user_id"] = user.id
session["user_name"] = user.name
flash("Login successful", "success")
return redirect(url_for("dashboard.dashboard"))
try:
email = request.form.get("email", "").strip()
password = request.form.get("password", "")
flash("Invalid email or password", "danger")
if not email or not password:
flash(ErrorMessage.INVALID_REQUEST, "danger")
current_app.logger.warning("Login failed. Email or password missing.")
return render_template("login.html", title="Login")
user = UserService.validate_login(email, password)
if user:
session.clear()
session["user_id"] = user.id
session["user_name"] = user.name
session.permanent = True
current_app.logger.info(f"Login successful. User={user.name}")
flash(SuccessMessage.LOGIN, "success")
return redirect(url_for("dashboard.dashboard"))
current_app.logger.warning(f"Invalid login attempt. Email={email}")
flash(ErrorMessage.LOGIN_FAILED,"danger")
except Exception as e:
current_app.logger.exception("Login Error" )
flash(ErrorMessage.INTERNAL_SERVER_ERROR,"danger")
return render_template("login.html", title="Login")
# ==========================
# LOGOUT
# ==========================
@auth_bp.route("/logout")
def logout():
username = session.get("user_name", "Unknown")
session.clear()
flash("Logged out successfully", "info")
current_app.logger.info(f"Logout successful. User={username}")
flash(SuccessMessage.LOGOUT,"info")
return redirect(url_for("auth.login"))
# ==========================
# REGISTER
# ==========================
@auth_bp.route("/register", methods=["GET", "POST"])
def register():
if request.method == "POST":
name = request.form.get("name")
email = request.form.get("email")
password = request.form.get("password")
try:
name = request.form.get("name", "").strip()
email = request.form.get("email", "").strip()
password = request.form.get("password", "")
user = UserService.register_user(name, email, password)
if not user:
flash("Email already exists", "danger")
return redirect(url_for("auth.register"))
if not name or not email or not password:
flash(ErrorMessage.INVALID_REQUEST,"danger")
return redirect(url_for("auth.register"))
flash("User registered successfully", "success")
return redirect(url_for("auth.login"))
user = UserService.register_user(name, email, password)
return render_template("register.html", title="Register")
if not user:
current_app.logger.warning(f"Duplicate Registration: {email}")
flash(ErrorMessage.DUPLICATE_ENTRY,"danger")
return redirect(url_for("auth.register"))
current_app.logger.info(f"New user registered: {email}")
flash(SuccessMessage.CREATED,"success")
return redirect(url_for("auth.login"))
except Exception:
current_app.logger.exception("User Registration Failed" )
flash(ErrorMessage.INTERNAL_SERVER_ERROR,"danger")
return render_template("register.html",title="Register")

274
app/routes/dashboard.py
View File

@@ -6,14 +6,21 @@ import matplotlib.pyplot as plt
import io
import base64
from app.utils.plot_utils import plot_to_base64
from app.utils.helpers import login_required
from app.services.dashboard_service import DashboardService
from sqlalchemy import func
from app import db
# Subcontractor models import
from app.models.subcontractor_model import Subcontractor
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.laying_model import Laying
from app.models.subcontractor_model import Subcontractor
# client models import
from app.models.tr_ex_client_model import TrenchExcavationClient
dashboard_bp = Blueprint("dashboard", __name__, url_prefix="/dashboard")
@@ -26,6 +33,7 @@ def dashboard():
@dashboard_bp.route("/api/live-stats")
@login_required
def live_stats():
try:
# 1. Overall Volume
@@ -64,6 +72,7 @@ def live_stats():
# subcontractor dashboard
@dashboard_bp.route("/subcontractor_dashboard")
@login_required
def subcontractor_dashboard():
if not session.get("user_id"):
@@ -76,8 +85,9 @@ def subcontractor_dashboard():
subcontractors=subcontractors
)
# ✅ API: Get Unique RA Bills
# API: Get Unique RA Bills
@dashboard_bp.route("/api/get-ra-bills")
@login_required
def get_ra_bills():
subcontractor_id = request.args.get("subcontractor")
@@ -96,149 +106,161 @@ def get_ra_bills():
).distinct().order_by(TrenchExcavation.RA_Bill_No).all()
# (Add others same pattern later)
case _:
results = []
case "manhole_excavation":
results = db.session.query(
ManholeExcavation.RA_Bill_No
).filter(
ManholeExcavation.subcontractor_id == subcontractor_id
).distinct().order_by(ManholeExcavation.RA_Bill_No).all()
case "Manhole_Domestic_Chamber":
results = db.session.query(
ManholeDomesticChamber.RA_Bill_No
).filter(
ManholeDomesticChamber.subcontractor_id == subcontractor_id
).distinct().order_by(ManholeDomesticChamber.RA_Bill_No).all()
case "Laying":
results = db.session.query(
Laying.RA_Bill_No
).filter(
Laying.subcontractor_id == subcontractor_id
).distinct().order_by(Laying.RA_Bill_No).all()
ra_bills = [r[0] for r in results if r[0]]
print(results)
return {"ra_bills": ra_bills}
# API FOR CHART DATA
# @dashboard_bp.route("/api/subcontractor-chart")
# def subcontractor_chart():
# subcontractor_id = request.args.get("subcontractor")
# category = request.args.get("category")
# ra_bill = request.args.get("ra_bill")
# labels = []
# values = []
# match category:
# # ✅ Trench
# case "trench_excavation":
# query = db.session.query(
# TrenchExcavation.excavation_category,
# func.sum(TrenchExcavation.Total)
# )
# if subcontractor_id:
# query = query.filter(TrenchExcavation.subcontractor_id == subcontractor_id)
# if ra_bill:
# query = query.filter(TrenchExcavation.RA_Bill_No == ra_bill)
# results = query.group_by(TrenchExcavation.excavation_category).all()
# # ✅ Manhole
# case "manhole_excavation":
# query = db.session.query(
# ManholeExcavation.excavation_category,
# func.sum(ManholeExcavation.Total)
# )
# if subcontractor_id:
# query = query.filter(ManholeExcavation.subcontractor_id == subcontractor_id)
# if ra_bill:
# query = query.filter(ManholeExcavation.RA_Bill_No == ra_bill)
# results = query.group_by(ManholeExcavation.excavation_category).all()
# case _:
# results = []
# for r in results:
# labels.append(r[0])
# values.append(float(r[1] or 0))
# return jsonify({
# "labels": labels,
# "values": values
# })
def total(records, field):
return float(sum(getattr(r, field) or 0 for r in records))
@dashboard_bp.route("/api/trench-analysis")
def trench_analysis():
subcontractor_id = request.args.get("subcontractor")
ra_bill = request.args.get("ra_bill")
subcontractor_id = request.args.get("subcontractor", "").strip()
ra_bill = request.args.get("ra_bill", "").strip()
query = TrenchExcavation.query
# Convert "1,2,3" -> ["1", "2", "3"]
ra_bill_list = []
if ra_bill:
ra_bill_list = [x.strip() for x in ra_bill.split(",") if x.strip()]
# Subcontractor Query
sub_query = TrenchExcavation.query
if subcontractor_id:
query = query.filter(TrenchExcavation.subcontractor_id == subcontractor_id)
if ra_bill:
query = query.filter(TrenchExcavation.RA_Bill_No == ra_bill)
data = query.all()
result = {
"Soft Murum": {"depth": 0, "qty": 0},
"Hard Murum": {"depth": 0, "qty": 0},
"Soft Rock": {"depth": 0, "qty": 0},
"Hard Rock": {"depth": 0, "qty": 0},
}
for r in data:
# Soft Murum
result["Soft Murum"]["depth"] += (
(r.Soft_Murum_0_to_1_5 or 0) +
(r.Soft_Murum_1_5_to_3_0 or 0) +
(r.Soft_Murum_3_0_to_4_5 or 0)
)
result["Soft Murum"]["qty"] += (
(r.Soft_Murum_0_to_1_5_total or 0) +
(r.Soft_Murum_1_5_to_3_0_total or 0) +
(r.Soft_Murum_3_0_to_4_5_total or 0)
sub_query = sub_query.filter(
TrenchExcavation.subcontractor_id == int(subcontractor_id)
)
# Hard Murum
result["Hard Murum"]["depth"] += (
(r.Hard_Murum_0_to_1_5 or 0) +
(r.Hard_Murum_1_5_to_3_0 or 0)
)
result["Hard Murum"]["qty"] += (
(r.Hard_Murum_0_to_1_5_total or 0) +
(r.Hard_Murum_1_5_and_above_total or 0)
if ra_bill_list:
sub_query = sub_query.filter(
TrenchExcavation.RA_Bill_No.in_(ra_bill_list)
)
# Soft Rock
result["Soft Rock"]["depth"] += (
(r.Soft_Rock_0_to_1_5 or 0) +
(r.Soft_Rock_1_5_to_3_0 or 0)
)
result["Soft Rock"]["qty"] += (
(r.Soft_Rock_0_to_1_5_total or 0) +
(r.Soft_Rock_1_5_and_above_total or 0)
tr_sub = [r.serialize() for r in sub_query.all()]
# Client Query
client_query = TrenchExcavationClient.query
if ra_bill_list:
client_query = client_query.filter(
TrenchExcavationClient.RA_Bill_No.in_(ra_bill_list)
)
# Hard Rock
result["Hard Rock"]["depth"] += (
(r.Hard_Rock_0_to_1_5 or 0) +
(r.Hard_Rock_1_5_to_3_0 or 0) +
(r.Hard_Rock_3_0_to_4_5 or 0) +
(r.Hard_Rock_4_5_to_6_0 or 0) +
(r.Hard_Rock_6_0_to_7_5 or 0)
)
result["Hard Rock"]["qty"] += (
(r.Hard_Rock_0_to_1_5_total or 0) +
(r.Hard_Rock_1_5_to_3_0_total or 0) +
(r.Hard_Rock_3_0_to_4_5_total or 0) +
(r.Hard_Rock_4_5_to_6_0_total or 0) +
(r.Hard_Rock_6_0_to_7_5_total or 0)
)
tr_client = [r.serialize() for r in client_query.all()]
labels = list(result.keys())
depth = [result[k]["depth"] for k in labels]
qty = [result[k]["qty"] for k in labels]
chart_data = [
{
"label": "Marshi 0 to 1.5",
"client": total(client_query.all(), "Marshi_Muddy_Slushy_0_to_1_5_total"),
"sub": 0
},
{
"label": "Marshi 1.5 to 3.0",
"client": total(client_query.all(), "Marshi_Muddy_Slushy_1_5_to_3_0_total"),
"sub": 0
},
{
"label": "Marshi 3.0-4.5",
"client": total(client_query.all(), "Marshi_Muddy_Slushy_3_0_to_4_5_total"),
"sub": 0
},
{
"label": "Soft Murum 0-1.5",
"client": total(client_query.all(), "Soft_Murum_0_to_1_5_total"),
"sub": total(sub_query.all(), "Soft_Murum_0_to_1_5_total")
},
{
"label": "Soft Murum 1.5-3.0",
"client": total(client_query.all(), "Soft_Murum_1_5_to_3_0_total"),
"sub": total(sub_query.all(), "Soft_Murum_1_5_to_3_0_total")
},
{
"label": "Soft Murum 3.0-4.5",
"client": total(client_query.all(), "Soft_Murum_3_0_to_4_5_total"),
"sub": total(sub_query.all(), "Soft_Murum_3_0_to_4_5_total")
},
{
"label": "Hard Murum 0-1.5",
"client": total(client_query.all(), "Hard_Murum_0_to_1_5_total"),
"sub": total(sub_query.all(), "Hard_Murum_0_to_1_5_total")
},
{
"label": "Hard Murum 1.5+",
"client": total(client_query.all(), "Hard_Murum_1_5_to_3_0_total"),
"sub": total(sub_query.all(), "Hard_Murum_1_5_and_above_total")
},
{
"label": "Soft Rock 0-1.5",
"client": total(client_query.all(), "Soft_Rock_0_to_1_5_total"),
"sub": total(sub_query.all(), "Soft_Rock_0_to_1_5_total")
},
{
"label": "Soft Rock 1.5+",
"client": total(client_query.all(), "Soft_Rock_1_5_to_3_0_total"),
"sub": total(sub_query.all(), "Soft_Rock_1_5_and_above_total")
},
{
"label": "Hard Rock 0-1.5",
"client": total(client_query.all(), "Hard_Rock_0_to_1_5_total"),
"sub": total(sub_query.all(), "Hard_Rock_0_to_1_5_total")
},
{
"label": "Hard Rock 1.5-3.0",
"client": total(client_query.all(), "Hard_Rock_1_5_to_3_0_total"),
"sub": total(sub_query.all(), "Hard_Rock_1_5_to_3_0_total")
},
{
"label": "Hard Rock 3.0-4.5",
"client": total(client_query.all(), "Hard_Rock_3_0_to_4_5_total"),
"sub": total(sub_query.all(), "Hard_Rock_3_0_to_4_5_total")
},
{
"label": "Hard Rock 4.5-6.0",
"client": total(client_query.all(), "Hard_Rock_4_5_to_6_0_total"),
"sub": total(sub_query.all(), "Hard_Rock_4_5_to_6_0_total")
},
{
"label": "Hard Rock 6.0-7.5",
"client": total(client_query.all(), "Hard_Rock_6_0_to_7_5_total"),
"sub": total(sub_query.all(), "Hard_Rock_6_0_to_7_5_total")
}
]
return jsonify({
"labels": labels,
"depth": depth,
"qty": qty
})
"labels": [x["label"] for x in chart_data],
"client_qty": [x["client"] for x in chart_data],
"sub_qty": [x["sub"] for x in chart_data]
})

16
app/routes/engineering.py Normal file
View File

@@ -0,0 +1,16 @@
from flask import Blueprint
from app.models.width_model import Width
engi_bp = Blueprint("engineering",__name__, url_prefix="/engi")
@engi_bp.route("/add")
def add_width_md():
return True
@engi_bp.route("/list")
def display_list():
return True

504
app/routes/file_report.py
View File

@@ -1,10 +1,12 @@
import pandas as pd
import io
from flask import Blueprint, render_template, request, send_file, flash
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
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
@@ -15,10 +17,320 @@ 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")
# ---------------- 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-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
# ---------------- 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):
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()
print("location::",search)
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()
]
# 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"]
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)
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("/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()
tables = None
abstract_html = ""
selected_sc_id = None
ra_bill_no = ""
location = ""
category = ""
# Search or load data
if request.method == "POST":
# get from data
subcontractor_id = request.form.get("subcontractor_id")
ra_bill_no = request.form.get("ra_bill_no", "").strip()
location = request.form.get("location", "").strip()
category = request.form.get("category", "")
action = request.form.get("action", "preview")
if not subcontractor_id:
flash("Select Subcontractor", "danger")
return render_template(
"subcontractor_report.html",
subcontractors=subcontractors
)
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)
# -----------------------------------------
# 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()
# ===================================================
# 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)
bill.df_tr.to_excel(writer,sheet_name="Tr.Ex",index=False)
bill.df_mh.to_excel(writer,sheet_name="Mh.Ex",index=False)
bill.df_dc.to_excel(writer,sheet_name="MH & DC",index=False)
bill.df_laying.to_excel(writer,sheet_name="Pipe Laying",index=False)
writer.close()
output.seek(0)
return send_file(
output,
download_name= "subcontractor_Report.xlsx",
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")
# 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": bill.df_tr.to_html(classes=table_class, index=False, escape=False),
"mh": bill.df_mh.to_html(classes=table_class, index=False, escape=False),
"dc": bill.df_dc.to_html(classes=table_class, index=False, escape=False ),
"laying": bill.df_laying.to_html(classes=table_class, index=False, escape=False)
}
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,
tables=tables,
abstract_html=abstract_html
)
# --- Client class ---
class ClientBill:
def __init__(self):
@@ -43,105 +355,8 @@ class ClientBill:
if not df.empty:
df.drop(columns=drop_cols, errors="ignore", inplace=True)
# --- Subcontractor class ---
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()
def Fetch(self, RA_Bill_No=None, subcontractor_id=None):
filters = {}
if subcontractor_id:
filters["subcontractor_id"] = subcontractor_id
if RA_Bill_No:
filters["RA_Bill_No"] = RA_Bill_No
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()
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)
# --- subcontractor report only ---
@file_report_bp.route("/Subcontractor_report", methods=["GET", "POST"])
@login_required
def report_file():
subcontractors = Subcontractor.query.all()
tables = None
selected_sc_id = None
ra_bill_no = None
download_all = False
if request.method == "POST":
subcontractor_id = request.form.get("subcontractor_id")
ra_bill_no = request.form.get("ra_bill_no")
download_all = request.form.get("download_all") == "true"
action = request.form.get("action")
if not subcontractor_id:
flash("Please select a subcontractor.", "danger")
return render_template("subcontractor_report.html", subcontractors=subcontractors)
subcontractor = Subcontractor.query.get(subcontractor_id)
bill_gen = SubcontractorBill()
if download_all:
bill_gen.Fetch(subcontractor_id=subcontractor_id)
file_name = f"{subcontractor.subcontractor_name}_ALL_BILLS.xlsx"
else:
if not ra_bill_no:
flash("Please enter an RA Bill Number.", "danger")
return render_template("subcontractor_report.html", subcontractors=subcontractors)
bill_gen.Fetch(RA_Bill_No=ra_bill_no, subcontractor_id=subcontractor_id)
file_name = f"{subcontractor.subcontractor_name}_RA_{ra_bill_no}_Report.xlsx"
if bill_gen.df_tr.empty and bill_gen.df_mh.empty and bill_gen.df_dc.empty:
flash("No data found for this selection.", "warning")
return render_template("subcontractor_report.html", subcontractors=subcontractors)
# If download is clicked, return file immediately
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="Tr.Ex.")
bill_gen.df_mh.to_excel(writer, index=False, sheet_name="MH.Ex.")
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=file_name, as_attachment=True)
# We add bootstrap classes directly to the pandas output
table_classes = "table table-bordered table-striped table-hover table-sm mb-0"
tables = {
"tr": bill_gen.df_tr.to_html(classes=table_classes, index=False),
"mh": bill_gen.df_mh.to_html(classes=table_classes, index=False),
"dc": bill_gen.df_dc.to_html(classes=table_classes, index=False),
"laying": bill_gen.df_laying.to_html(classes=table_classes, index=False)
}
selected_sc_id = subcontractor_id
return render_template(
"subcontractor_report.html",
subcontractors=subcontractors,
tables=tables,
selected_sc_id=selected_sc_id,
ra_bill_no=ra_bill_no,
download_all=download_all
)
# --- CLIENT REPORT (PREVIEW + DOWNLOAD) ---
@file_report_bp.route("/client_report", methods=["GET", "POST"])
@login_required
@@ -202,4 +417,93 @@ def client_report():
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)
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

14
app/routes/generate_comparison_report.py
View File

@@ -2,27 +2,27 @@ from flask import Blueprint, render_template, request, send_file, flash
from collections import defaultdict
import pandas as pd
import io
from app.utils.helpers import login_required
from app.utils.regex_utils import RegularExpression
# Contractor models import
from app.models.subcontractor_model import Subcontractor
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.laying_model import Laying
# Client models import
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.models.laying_client_model import LayingClient
from app.utils.helpers import login_required
from app.utils.regex_utils import RegularExpression
generate_report_bp = Blueprint("generate_report", __name__, url_prefix="/report")
# NORMALIZER
def normalize_key(value):
if value is None:
@@ -241,7 +241,7 @@ def comparison_report():
write_sheet(writer, df_dc, "MH & DC", subcontractor.subcontractor_name)
write_sheet(writer, df_lay, "Laying", subcontractor.subcontractor_name)
output.seek(0)
output.seek(0)
return send_file(
output,
as_attachment=True,

13
app/routes/user.py
View File

@@ -1,13 +0,0 @@
from flask import Blueprint, render_template
from app.services.user_service import UserService
from app.utils.helpers import login_required
from flask import current_app
user_bp = Blueprint("user", __name__, url_prefix="/user")
@user_bp.route("/list")
@login_required
def list_users():
current_app.logger.info("User list viewed")
users = UserService.get_all_users()
return render_template("users.html", users=users, title="Users")

50
app/routes/user_routes.py Normal file
View File

@@ -0,0 +1,50 @@
# from flask import Blueprint, render_template
# from app.services.user_service import UserService
# from app.utils.helpers import login_required
# from flask import current_app
# user_bp = Blueprint("user", __name__, url_prefix="/user")
# @user_bp.route("/list")
# @login_required
# def list_users():
# current_app.logger.info("User list viewed")
# users = UserService.get_all_users()
# return render_template("/user/users.html", users=users, title="Users | List")
from flask import (Blueprint,render_template,current_app,flash)
from app.services.user_service import UserService
from app.utils.helpers import login_required
from app.constants.messages import SuccessMessage, ErrorMessage
from app.constants.http_status import HTTPStatus
user_bp = Blueprint("user", __name__, url_prefix="/user")
# ==================================================
# User List
# ==================================================
@user_bp.route("/list", methods=["GET"])
@login_required
def list_users():
try:
current_app.logger.info("Fetching user list.")
users = UserService.get_all_users()
current_app.logger.info(f"User list loaded successfully. Total Users: {len(users)}")
return render_template("user/users.html", users=users, title="Users | List")
except Exception as e:
current_app.logger.exception("Failed to load user list.")
flash(ErrorMessage.INTERNAL_SERVER_ERROR,"danger")
return render_template("user/users.html",users=[],title="Users | List"), HTTPStatus.INTERNAL_SERVER_ERROR