nngo/LT_Janitorial_Quality_Control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

05/11 Update: add contract & facility import function

Browse Source
This commit is contained in:
nngo
2026-05-11 13:05:40 -04:00
parent dc9bfb1674
commit b73448dfd0
3 changed files with 605 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/projects.py
+433 -1
View File
@@ -10,7 +10,7 @@ Access matrix:
"""
import logging
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, Response
from flask_login import login_required, current_user
from app import db
from app.models.project import Project, CustomerAssignment
@@ -234,3 +234,435 @@ def remove_assignment(assignment_id):
f'{username} → {project.name}')
flash(f'Assignment for "{username}" removed.', 'success')
return redirect(url_for('projects.view', project_id=project_id))
# ── Bulk Import — Excel template download ─────────────────────────────────────
@bp.route('/import/template')
@login_required
@supervisor_required
def import_template():
"""Download a blank .xlsx showing the expected import format."""
import io
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment
wb = Workbook()
# ── Sheet 1: Contracts ────────────────────────────────────────────────
ws_c = wb.active
ws_c.title = 'Contracts'
hdr_fill = PatternFill('solid', start_color='1F4E79')
hdr_font = Font(bold=True, color='FFFFFF')
hdr_align = Alignment(horizontal='center', vertical='center')
contract_headers = ['contract_name', 'description', 'active']
for col, h in enumerate(contract_headers, 1):
cell = ws_c.cell(row=1, column=col, value=h)
cell.font = hdr_font
cell.fill = hdr_fill
cell.alignment = hdr_align
# Example rows
ws_c.append(['Acme Corp - Downtown', 'Main office complex cleaning contract', 'yes'])
ws_c.append(['Acme Corp - Warehouse', '', 'yes'])
ws_c.column_dimensions['A'].width = 30
ws_c.column_dimensions['B'].width = 40
ws_c.column_dimensions['C'].width = 10
# ── Sheet 2: Facilities ───────────────────────────────────────────────
ws_f = wb.create_sheet('Facilities')
facility_headers = [
'contract_name', 'facility_name', 'address',
'contact_person', 'contact_phone', 'active',
]
for col, h in enumerate(facility_headers, 1):
cell = ws_f.cell(row=1, column=col, value=h)
cell.font = hdr_font
cell.fill = hdr_fill
cell.alignment = hdr_align
ws_f.append(['Acme Corp - Downtown', 'Tower A', '123 Main St, Suite 100', 'Jane Smith', '555-0101', 'yes'])
ws_f.append(['Acme Corp - Downtown', 'Parking Garage', '123 Main St, Level B1', '', '', 'yes'])
ws_f.append(['Acme Corp - Warehouse', 'Bay 1', '456 Industrial Blvd', 'Bob Jones', '555-0202', 'yes'])
ws_f.column_dimensions['A'].width = 30
ws_f.column_dimensions['B'].width = 25
ws_f.column_dimensions['C'].width = 35
ws_f.column_dimensions['D'].width = 20
ws_f.column_dimensions['E'].width = 15
ws_f.column_dimensions['F'].width = 10
buf = io.BytesIO()
wb.save(buf)
buf.seek(0)
return Response(
buf.getvalue(),
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
headers={'Content-Disposition': 'attachment; filename="contract_facility_import_template.xlsx"'},
)
# ── Bulk Import — upload → preview → confirm ──────────────────────────────────
@bp.route('/import', methods=['GET', 'POST'])
@login_required
@supervisor_required
def bulk_import():
"""Two-phase Excel import for Contracts (Projects) and Facilities.
Phase 1 (GET / POST with file):
Parse and validate the workbook, return a preview. No DB writes.
Phase 2 (POST with confirmed=1):
Write all validated rows to the database.
Excel format (two sheets)
--------------------------
Sheet "Contracts": contract_name*, description, active
Sheet "Facilities": contract_name*, facility_name*, address,
contact_person, contact_phone, active
* required columns.
- "active" column: any of yes/true/1 → True; blank defaults to True.
- Contracts that already exist (by name, case-insensitive) are reused,
not duplicated.
- Facilities that already exist (same name within same contract) are
skipped and reported.
"""
import io, json
from flask import session as _session
# ── Phase 2: commit ───────────────────────────────────────────────────
if request.method == 'POST' and request.form.get('confirmed') == '1':
rows_json = request.form.get('rows_json', '[]')
try:
rows = json.loads(rows_json)
except Exception:
flash('Import session expired. Please re-upload the file.', 'danger')
return redirect(url_for('projects.bulk_import'))
created_contracts = 0
reused_contracts = 0
created_facilities = 0
skipped_facilities = 0
# Cache contracts created/found in this batch
contract_cache = {} # lower-name → Project
for row in rows:
cname = row['contract_name']
cdesc = row.get('description') or None
c_active = row.get('contract_active', True)
fname = row.get('facility_name') or None
# Get or create contract
cache_key = cname.strip().lower()
project = contract_cache.get(cache_key)
if project is None:
project = Project.query.filter(
db.func.lower(Project.name) == cache_key
).first()
if project is None:
project = Project(
name = cname.strip(),
description = cdesc,
active = c_active,
)
db.session.add(project)
db.session.flush()
contract_cache[cache_key] = project
created_contracts += 1
log_action(ACTION_CREATE, 'Project', project.id, project.name,
f'active={c_active}; source=bulk_import')
logger.info('BULK IMPORT | contract_created | name=%s by=%s',
project.name, current_user.username)
else:
contract_cache[cache_key] = project
reused_contracts += 1
# Create facility if present in this row
if fname:
faddr = row.get('address') or None
fcp = row.get('contact_person') or None
fphone = row.get('contact_phone') or None
f_active = row.get('facility_active', True)
existing_fac = Facility.query.filter(
Facility.project_id == project.id,
db.func.lower(Facility.name) == fname.strip().lower(),
).first()
if existing_fac:
skipped_facilities += 1
else:
facility = Facility(
name = fname.strip(),
address = faddr,
contact_person = fcp,
contact_phone = fphone,
active = f_active,
project_id = project.id,
)
db.session.add(facility)
db.session.flush()
created_facilities += 1
log_action(ACTION_CREATE, 'Facility', facility.id, facility.name,
f'project_id={project.id}; source=bulk_import')
logger.info('BULK IMPORT | facility_created | name=%s project_id=%s by=%s',
facility.name, project.id, current_user.username)
db.session.commit()
logger.info(
'BULK IMPORT COMMITTED | by=%s | contracts_new=%s contracts_reused=%s '
'facilities_new=%s facilities_skipped=%s',
current_user.username, created_contracts, reused_contracts,
created_facilities, skipped_facilities,
)
parts = []
if created_contracts:
parts.append(f'{created_contracts} contract(s) created')
if reused_contracts:
parts.append(f'{reused_contracts} existing contract(s) reused')
if created_facilities:
parts.append(f'{created_facilities} facilit{"y" if created_facilities == 1 else "ies"} created')
if skipped_facilities:
parts.append(f'{skipped_facilities} duplicate facilit{"y" if skipped_facilities == 1 else "ies"} skipped')
flash('Import complete: ' + ', '.join(parts) + '.', 'success')
return redirect(url_for('projects.index'))
# ── Phase 1: parse and validate ───────────────────────────────────────
preview_rows = []
errors = []
raw_valid_rows = []
if request.method == 'POST':
file = request.files.get('xlsx_file')
if not file or not file.filename:
flash('Please select an Excel file to upload.', 'warning')
return render_template('projects/import.html')
if not file.filename.lower().endswith(('.xlsx', '.xlsm')):
flash('Only .xlsx / .xlsm files are accepted.', 'danger')
return render_template('projects/import.html')
try:
from openpyxl import load_workbook
wb = load_workbook(filename=io.BytesIO(file.stream.read()), data_only=True)
except Exception as exc:
flash(f'Could not open workbook: {exc}', 'danger')
return render_template('projects/import.html')
# ── Parse Contracts sheet ─────────────────────────────────────────
if 'Contracts' not in wb.sheetnames:
flash('Workbook is missing the "Contracts" sheet. Download the template and try again.', 'danger')
return render_template('projects/import.html')
ws_c = wb['Contracts']
c_rows = list(ws_c.iter_rows(values_only=True))
if not c_rows:
flash('"Contracts" sheet is empty.', 'danger')
return render_template('projects/import.html')
c_headers = [str(h).strip().lower() if h else '' for h in c_rows[0]]
if 'contract_name' not in c_headers:
flash('"Contracts" sheet is missing required column "contract_name".', 'danger')
return render_template('projects/import.html')
def col(headers, name):
try:
return headers.index(name)
except ValueError:
return None
c_name_idx = col(c_headers, 'contract_name')
c_desc_idx = col(c_headers, 'description')
c_active_idx = col(c_headers, 'active')
# name → {description, active} for valid contracts found in sheet
contract_sheet = {} # lower-name → dict
contract_errors = []
for i, row in enumerate(c_rows[1:], start=2):
cname = str(row[c_name_idx]).strip() if row[c_name_idx] is not None else ''
if not cname or cname.lower() == 'none':
contract_errors.append(f'Row {i}: contract_name is required')
continue
raw_active = row[c_active_idx] if c_active_idx is not None else None
c_active = _parse_bool(raw_active, default=True)
cdesc = str(row[c_desc_idx]).strip() if (c_desc_idx is not None and row[c_desc_idx] is not None) else ''
contract_sheet[cname.lower()] = {
'contract_name': cname,
'description': cdesc or None,
'contract_active': c_active,
}
# ── Parse Facilities sheet ────────────────────────────────────────
facility_sheet_rows = []
if 'Facilities' in wb.sheetnames:
ws_f = wb['Facilities']
f_rows = list(ws_f.iter_rows(values_only=True))
if f_rows:
f_headers = [str(h).strip().lower() if h else '' for h in f_rows[0]]
f_cname_idx = col(f_headers, 'contract_name')
f_fname_idx = col(f_headers, 'facility_name')
f_addr_idx = col(f_headers, 'address')
f_cp_idx = col(f_headers, 'contact_person')
f_phone_idx = col(f_headers, 'contact_phone')
f_active_idx = col(f_headers, 'active')
if f_cname_idx is None or f_fname_idx is None:
flash('"Facilities" sheet is missing required columns "contract_name" or "facility_name".', 'danger')
return render_template('projects/import.html')
for i, row in enumerate(f_rows[1:], start=2):
cname = str(row[f_cname_idx]).strip() if row[f_cname_idx] is not None else ''
fname = str(row[f_fname_idx]).strip() if row[f_fname_idx] is not None else ''
if not cname or cname.lower() == 'none':
continue # skip blank rows silently
if not fname or fname.lower() == 'none':
continue
raw_active = row[f_active_idx] if f_active_idx is not None else None
facility_sheet_rows.append({
'sheet_row': i,
'contract_name': cname,
'facility_name': fname,
'address': str(row[f_addr_idx]).strip() if (f_addr_idx is not None and row[f_addr_idx]) else '',
'contact_person': str(row[f_cp_idx]).strip() if (f_cp_idx is not None and row[f_cp_idx]) else '',
'contact_phone': str(row[f_phone_idx]).strip() if (f_phone_idx is not None and row[f_phone_idx]) else '',
'facility_active': _parse_bool(raw_active, default=True),
})
# ── Build preview rows ────────────────────────────────────────────
#
# Strategy: one preview row per (contract) from Contracts sheet,
# then one preview row per facility from Facilities sheet.
# Validation: facility's contract_name must appear in Contracts sheet.
existing_projects = {
p.name.strip().lower(): p
for p in Project.query.all()
}
row_num = 0
# Contract rows
for lower_name, cdata in contract_sheet.items():
row_num += 1
row_errors = []
if lower_name in existing_projects:
status = 'exists'
note = 'Contract already exists — will be reused'
else:
status = 'ok'
note = ''
preview_rows.append({
'row': row_num,
'sheet': 'Contracts',
'contract_name': cdata['contract_name'],
'facility_name': '',
'note': note,
'status': status,
'errors': row_errors,
})
if status != 'error':
raw_valid_rows.append({
'contract_name': cdata['contract_name'],
'description': cdata['description'],
'contract_active': cdata['contract_active'],
'facility_name': None,
})
# Facility rows
seen_facilities = set() # (lower_contract, lower_facility) within file
for frow in facility_sheet_rows:
row_num += 1
row_errors = []
lower_c = frow['contract_name'].lower()
lower_f = frow['facility_name'].lower()
if lower_c not in contract_sheet:
row_errors.append(
f'Contract "{frow["contract_name"]}" not found in the Contracts sheet'
)
dup_key = (lower_c, lower_f)
if dup_key in seen_facilities:
row_errors.append('Duplicate facility name within this contract in the file')
else:
seen_facilities.add(dup_key)
# Check DB for existing facility with same name in same contract
db_conflict = False
if not row_errors:
proj = existing_projects.get(lower_c)
if proj:
db_conflict = Facility.query.filter(
Facility.project_id == proj.id,
db.func.lower(Facility.name) == lower_f,
).first() is not None
if db_conflict:
status = 'exists'
note = 'Facility already exists in this contract — will be skipped'
elif row_errors:
status = 'error'
note = ''
else:
status = 'ok'
note = ''
preview_rows.append({
'row': frow['sheet_row'],
'sheet': 'Facilities',
'contract_name': frow['contract_name'],
'facility_name': frow['facility_name'],
'note': note,
'status': status,
'errors': row_errors,
})
if status == 'ok':
raw_valid_rows.append({
'contract_name': frow['contract_name'],
'description': None,
'contract_active': True,
'facility_name': frow['facility_name'],
'address': frow['address'] or None,
'contact_person': frow['contact_person'] or None,
'contact_phone': frow['contact_phone'] or None,
'facility_active': frow['facility_active'],
})
valid_count = sum(1 for r in preview_rows if r['status'] == 'ok')
has_errors = any(r['status'] == 'error' for r in preview_rows)
rows_json = json.dumps(raw_valid_rows)
# Contract-sheet parse errors shown as flash
for ce in contract_errors:
flash(ce, 'warning')
return render_template(
'projects/import.html',
preview_rows = preview_rows,
valid_count = valid_count,
has_errors = has_errors,
rows_json = rows_json,
)
return render_template('projects/import.html')
def _parse_bool(value, default=True):
"""Convert Excel cell value to Python bool for 'active' columns."""
if value is None:
return default
s = str(value).strip().lower()
if s in ('yes', 'true', '1', 'y'):
return True
if s in ('no', 'false', '0', 'n'):
return False
return default