107 lines
4.2 KiB
Python
107 lines
4.2 KiB
Python
import os
|
|
import pandas as pd
|
|
from flask import request, redirect, url_for, current_app
|
|
from flask_login import current_user
|
|
from app import db
|
|
from app.models import Task, WorkDetail
|
|
from app.service.logger import log_activity
|
|
|
|
# keep helper inside controller
|
|
def to_2_decimal(value):
|
|
try:
|
|
if value is None or value == "":
|
|
return None
|
|
return round(float(value), 2)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
def upload_controller():
|
|
if 'file' not in request.files:
|
|
return "No file part"
|
|
|
|
file = request.files['file']
|
|
|
|
if file.filename == '':
|
|
return "No selected file"
|
|
|
|
filepath = os.path.join(current_app.config['UPLOAD_FOLDER'], file.filename)
|
|
file.save(filepath)
|
|
|
|
log_activity(current_user.username, "File Upload", f"Uploaded file: {file.filename}")
|
|
|
|
work_details_data = pd.read_excel(filepath, nrows=11, header=None, dtype=str)
|
|
work_details_dict = {
|
|
"name_of_work": work_details_data.iloc[0, 1],
|
|
"cover_agreement_no": work_details_data.iloc[1, 1],
|
|
"name_of_contractor": work_details_data.iloc[2, 1],
|
|
"name_of_tpi_agency": work_details_data.iloc[3, 1],
|
|
"name_of_division": work_details_data.iloc[4, 1],
|
|
"name_of_village": work_details_data.iloc[5, 1],
|
|
"block": work_details_data.iloc[6, 1],
|
|
"scheme_id": work_details_data.iloc[7, 1],
|
|
"date_of_billing": work_details_data.iloc[8, 1],
|
|
"measurement_book": work_details_data.iloc[9, 1],
|
|
"district": work_details_data.iloc[10, 1]
|
|
}
|
|
work_details_dict = {k: (None if pd.isna(v) else v) for k, v in work_details_dict.items()}
|
|
work_detail = WorkDetail(**work_details_dict)
|
|
db.session.add(work_detail)
|
|
data = pd.read_excel(filepath, skiprows=10)
|
|
data = data.astype(object).where(pd.notna(data), None)
|
|
expected_columns = [
|
|
"serial_number", "task_name", "unit", "qty", "rate", "boq_amount",
|
|
"previous_billed_qty", "previous_billing_amount",
|
|
"in_this_ra_bill_qty", "in_this_ra_billing_amount",
|
|
"cumulative_billed_qty", "cumulative_billed_amount",
|
|
"variation_qty", "variation_amount", "remark"
|
|
]
|
|
if data.shape[1] == len(expected_columns):
|
|
data.columns = expected_columns
|
|
else:
|
|
data.columns = expected_columns[:data.shape[1]]
|
|
current_main_task_serial = None
|
|
current_main_task_name = None
|
|
for _, row in data.iterrows():
|
|
task_name = str(row["task_name"]) if row["task_name"] else ""
|
|
serial_number = str(row["serial_number"]) if row["serial_number"] else None
|
|
if serial_number:
|
|
current_main_task_serial = serial_number
|
|
current_main_task_name = task_name
|
|
parent_id = None
|
|
else:
|
|
parent_id = current_main_task_serial
|
|
task = Task(
|
|
district=work_details_dict.get("district"),
|
|
block_name=work_details_dict["block"],
|
|
village_name=work_details_dict["name_of_village"],
|
|
serial_number=serial_number,
|
|
task_name=task_name,
|
|
unit=row["unit"],
|
|
qty=to_2_decimal(row["qty"]),
|
|
rate=to_2_decimal(row["rate"]),
|
|
boq_amount=to_2_decimal(row["boq_amount"]),
|
|
previous_billed_qty=to_2_decimal(row["previous_billed_qty"]),
|
|
previous_billing_amount=to_2_decimal(row["previous_billing_amount"]),
|
|
in_this_ra_bill_qty=to_2_decimal(row["in_this_ra_bill_qty"]),
|
|
in_this_ra_billing_amount=to_2_decimal(row["in_this_ra_billing_amount"]),
|
|
cumulative_billed_qty=to_2_decimal(row["cumulative_billed_qty"]),
|
|
cumulative_billed_amount=to_2_decimal(row["cumulative_billed_amount"]),
|
|
variation_qty=to_2_decimal(row["variation_qty"]),
|
|
variation_amount=to_2_decimal(row["variation_amount"]),
|
|
parent_id=parent_id,
|
|
parent_task_name=current_main_task_name if not serial_number else None,
|
|
remark=row["remark"]
|
|
)
|
|
db.session.add(task)
|
|
db.session.commit()
|
|
|
|
log_activity(
|
|
current_user.username,
|
|
"Database Insert",
|
|
f"Inserted work details and tasks from {file.filename}"
|
|
)
|
|
|
|
return redirect(url_for('main.display_tasks'))
|
|
|
|
|