# routes/assets.py from __future__ import annotations from datetime import datetime, date from flask import ( Blueprint, render_template, request, redirect, url_for, flash, current_app, make_response ) from flask_login import login_required, current_user from sqlalchemy import or_, func from sqlalchemy.orm import joinedload from extensions import db from models import ( Asset, AssetCategory, AssetVendor, AssetAllocation, AssetLossDamage, Employee, AssetEventLog, AssetSubCategory, ) import os import uuid import csv import io from werkzeug.utils import secure_filename from sqlalchemy.exc import IntegrityError assets_bp = Blueprint("assets", __name__, url_prefix="/assets") # ---------- Helpers ---------- ALLOWED_ASSET_IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "webp"} ALLOWED_ASSET_UPLOAD_EXTENSIONS = {"csv"} # keep it simple; can extend to xlsx later def _allowed_image(filename: str) -> bool: return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_ASSET_IMAGE_EXTENSIONS def _allowed_asset_file(filename: str) -> bool: return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_ASSET_UPLOAD_EXTENSIONS def _parse_date(val: str): if not val: return None try: return datetime.strptime(val, "%Y-%m-%d").date() except Exception: return None def _safe_int(val, default: int) -> int: """ Convert to int without treating 0 as "missing". - None / '' => default - 0 stays 0 """ if val is None: return default if isinstance(val, str) and val.strip() == "": return default try: return int(val) except Exception: return default def _save_asset_image(file_storage, asset_tag: str) -> tuple[str, str]: """ Returns (relative_path, original_filename) Saves under: static/uploads/assets/ """ upload_dir = os.path.join(current_app.root_path, "static", "uploads", "assets") os.makedirs(upload_dir, exist_ok=True) original = secure_filename(file_storage.filename or "") ext = original.rsplit(".", 1)[1].lower() fname = f"{secure_filename(asset_tag)}_{uuid.uuid4().hex}.{ext}" abs_path = os.path.join(upload_dir, fname) file_storage.save(abs_path) rel_path = f"uploads/assets/{fname}" # relative to /static/ return rel_path, original def _log_asset_event( asset: Asset, action: str, from_employee_id: int | None = None, to_employee_id: int | None = None, notes: str | None = None, ): """ Stores event in AssetEventLog. Supports different column schemas safely. """ if not asset or not getattr(asset, "id", None): return msg_parts = [] if notes: msg_parts.append(notes) if from_employee_id: msg_parts.append(f"from_emp_id={from_employee_id}") if to_employee_id: msg_parts.append(f"to_emp_id={to_employee_id}") log = AssetEventLog( asset_id=asset.id, event_type=action, event_time=datetime.utcnow(), message=" | ".join(msg_parts), created_by_id=getattr(current_user, "id", None), ) # If your model has employee_id, keep using it for quick filtering if hasattr(log, "employee_id"): setattr(log, "employee_id", to_employee_id or from_employee_id) # If your model has explicit from/to columns, set them too if hasattr(log, "from_employee_id"): setattr(log, "from_employee_id", from_employee_id) if hasattr(log, "to_employee_id"): setattr(log, "to_employee_id", to_employee_id) db.session.add(log) # ---------- Asset Master ---------- @assets_bp.route("/", methods=["GET", "POST"]) @login_required def list_assets(): """ Asset master page: - GET: list assets + filters - POST: create / update asset (from modal) """ if request.method == "POST": mode = request.form.get("mode") # "create" or "edit" asset_id = request.form.get("asset_id") asset_code = (request.form.get("asset_code") or "").strip() asset_tag = (request.form.get("asset_tag") or "").strip() name = (request.form.get("name") or "").strip() category_id = request.form.get("category_id") or None subcategory_id = request.form.get("subcategory_id") or None vendor_id = request.form.get("vendor_id") or None serial_number = (request.form.get("serial_number") or "").strip() purchase_date = _parse_date(request.form.get("purchase_date")) warranty_expiry = _parse_date(request.form.get("warranty_expiry")) status = (request.form.get("status") or "in_stock").strip() location = (request.form.get("location") or "").strip() remarks = (request.form.get("remarks") or "").strip() image_file = request.files.get("image") qty_total_raw = (request.form.get("quantity_total") or "").strip() try: quantity_total = int(qty_total_raw) if qty_total_raw != "" else 1 except ValueError: quantity_total = 1 if quantity_total < 1: quantity_total = 1 if not asset_tag or not name: flash("Asset Tag and Name are required.", "danger") return redirect(url_for("assets.list_assets")) try: if mode == "edit" and asset_id: asset = Asset.query.get_or_404(int(asset_id)) old_status = asset.status or "in_stock" is_new = False else: asset = Asset() db.session.add(asset) old_status = None is_new = True # Ensure asset_code is always non-null if asset_code: asset.asset_code = asset_code elif not getattr(asset, "asset_code", None): asset.asset_code = asset_tag asset.asset_tag = asset_tag # ---- Image upload ---- if image_file and image_file.filename: if not _allowed_image(image_file.filename): raise ValueError("Invalid image type. Allowed: png, jpg, jpeg, webp") rel_path, original_name = _save_asset_image(image_file, asset_tag) old_rel = getattr(asset, "image_path", None) if old_rel: try: old_abs = os.path.join(current_app.root_path, "static", old_rel) if os.path.exists(old_abs): os.remove(old_abs) except Exception: current_app.logger.warning("Could not delete old asset image", exc_info=True) asset.image_path = rel_path asset.image_name = original_name asset.name = name asset.category_id = int(category_id) if category_id else None asset.subcategory_id = int(subcategory_id) if subcategory_id else None asset.vendor_id = int(vendor_id) if vendor_id else None asset.serial_number = serial_number asset.purchase_date = purchase_date asset.warranty_expiry = warranty_expiry asset.status = status asset.location = location asset.remarks = remarks # ---- Quantity handling (DO NOT reset availability wrongly) ---- old_total = _safe_int(getattr(asset, "quantity_total", None), 1) old_avail_raw = getattr(asset, "quantity_available", None) old_avail = _safe_int(old_avail_raw, old_total) # 0 stays 0 now old_allocated = max(old_total - old_avail, 0) # Validate subcategory belongs to category if asset.subcategory_id: sc = AssetSubCategory.query.get(asset.subcategory_id) if not sc or (asset.category_id and sc.category_id != asset.category_id): asset.subcategory_id = None if is_new: asset.quantity_total = quantity_total asset.quantity_available = quantity_total else: if quantity_total < old_allocated: raise ValueError( f"Quantity total cannot be less than already allocated ({old_allocated})." ) asset.quantity_total = quantity_total asset.quantity_available = max(quantity_total - old_allocated, 0) # ---- Logging ---- if is_new: _log_asset_event(asset, "new_asset", notes="New asset created via Asset Master.") if status == "in_stock": _log_asset_event(asset, "inventory", notes="Asset added to inventory.") else: if old_status != status: if status == "in_stock": _log_asset_event(asset, "inventory", notes=f"Status changed from {old_status} to in_stock.") elif status == "under_repair": _log_asset_event(asset, "under_repair", notes=f"Status changed from {old_status} to under_repair.") elif status == "retired": _log_asset_event(asset, "retired", notes=f"Status changed from {old_status} to retired.") else: _log_asset_event(asset, "status_change", notes=f"Status changed from {old_status} to {status}.") db.session.commit() flash("Asset saved successfully.", "success") except Exception as e: current_app.logger.exception("Failed to save asset") db.session.rollback() flash(f"Error saving asset: {e}", "danger") return redirect(url_for("assets.list_assets")) # ---- GET: filters + listing ---- q = (request.args.get("q") or "").strip() status = (request.args.get("status") or "").strip() category_id = request.args.get("category_id") or None subcategory_id = request.args.get("subcategory_id") or None vendor_id = request.args.get("vendor_id") or None page = request.args.get("page", 1, type=int) per_page = request.args.get("per_page", 25, type=int) per_page = max(10, min(per_page, 200)) query = ( Asset.query .options( joinedload(Asset.category), # backref relationship ✅ joinedload(Asset.subcategory), # explicit relationship ✅ joinedload(Asset.vendor), # backref relationship ✅ ) ) if category_id: query = query.filter(Asset.category_id == int(category_id)) if subcategory_id: query = query.filter(Asset.subcategory_id == int(subcategory_id)) if vendor_id: query = query.filter(Asset.vendor_id == int(vendor_id)) if q: like = f"%{q}%" query = query.filter(or_( Asset.asset_tag.ilike(like), Asset.asset_code.ilike(like), Asset.name.ilike(like), Asset.serial_number.ilike(like), )) if status: query = query.filter(Asset.status == status) assets_pagination = ( query.order_by(Asset.asset_tag.asc()) .paginate(page=page, per_page=per_page, error_out=False) ) assets = assets_pagination.items categories = AssetCategory.query.order_by(AssetCategory.name.asc()).all() subcategories = [] if category_id: try: subcategories = ( AssetSubCategory.query .filter(AssetSubCategory.category_id == int(category_id)) .order_by(AssetSubCategory.name.asc()) .all() ) except ValueError: subcategories = [] vendors = AssetVendor.query.order_by(AssetVendor.name.asc()).all() employees = Employee.query.order_by(Employee.full_name.asc()).all() pending_ld_asset_ids = { aid for (aid,) in db.session.query(AssetLossDamage.asset_id) .filter(AssetLossDamage.status == "pending") .distinct() } blocked_for_allocation = set(pending_ld_asset_ids) # IMPORTANT: don't treat 0 as missing for a in assets: total = _safe_int(getattr(a, "quantity_total", None), 1) avail = _safe_int(getattr(a, "quantity_available", None), total) if avail <= 0: blocked_for_allocation.add(a.id) return render_template( "assets/asset_master.html", assets=assets, pagination=assets_pagination, per_page=per_page, page=page, categories=categories, subcategories=subcategories, vendors=vendors, vendor_id=vendor_id, q=q, status=status, category_id=category_id, subcategory_id=subcategory_id, employees=employees, now=datetime.utcnow, blocked_for_allocation=blocked_for_allocation, ) # ---------- Allocation / Return ---------- @assets_bp.route("/allocate", methods=["POST"]) @login_required def allocate_asset(): """ Allocate asset to an employee. Prevents: - allocating when available == 0 - allocating with qty <= 0 - the '0 resets to total' bug Uses row lock to avoid double-allocations in concurrent requests. """ asset_id = request.form.get("asset_id") if not asset_id: flash("Missing asset id for allocation.", "danger") return redirect(url_for("assets.list_assets")) employee_id = request.form.get("employee_id") if not employee_id: flash("Please choose an employee to allocate.", "danger") return redirect(url_for("assets.list_assets")) emp = Employee.query.get(int(employee_id)) if not emp: flash("Invalid employee selected.", "danger") return redirect(url_for("assets.list_assets")) allocated_on = _parse_date(request.form.get("allocated_on")) or datetime.utcnow().date() expected_return_date = _parse_date(request.form.get("due_back_on")) remarks = (request.form.get("remarks") or "").strip() condition_on_issue = (request.form.get("condition_on_issue") or "").strip() qty_raw = (request.form.get("quantity") or "").strip() if qty_raw == "": qty = 1 else: try: qty = int(qty_raw) except ValueError: flash("Invalid quantity.", "danger") return redirect(url_for("assets.list_assets")) # ✅ Critical fix: do NOT silently coerce 0 to 1 if qty <= 0: flash("Quantity must be at least 1.", "danger") return redirect(url_for("assets.list_assets")) try: # Lock asset row to prevent concurrent double allocations asset = ( db.session.query(Asset) .filter(Asset.id == int(asset_id)) .with_for_update() .first() ) if not asset: flash("Asset not found.", "danger") return redirect(url_for("assets.list_assets")) # Block allocating certain statuses if (asset.status or "").lower() in {"lost", "retired", "damaged"}: flash(f"Cannot allocate asset in status: {asset.status}", "danger") return redirect(url_for("assets.list_assets")) asset_total = _safe_int(getattr(asset, "quantity_total", None), 1) asset_avail = _safe_int(getattr(asset, "quantity_available", None), asset_total) # 0 stays 0 if asset_avail <= 0: flash("This asset is out of stock.", "danger") return redirect(url_for("assets.list_assets")) if qty > asset_avail: flash(f"Only {asset_avail} quantity available for this asset.", "danger") return redirect(url_for("assets.list_assets")) # find previous active allocation (if any) for logs only prev_alloc = ( AssetAllocation.query .filter( AssetAllocation.asset_id == asset.id, AssetAllocation.status == "allocated", ) .order_by(AssetAllocation.allocated_on.desc()) .first() ) prev_emp_id = prev_alloc.employee_id if prev_alloc else None allocation = AssetAllocation( asset_id=asset.id, employee_id=emp.id, allocated_on=allocated_on, expected_return_date=expected_return_date, remarks=remarks, condition_on_issue=condition_on_issue, quantity=qty, status="allocated", ) # Reduce available quantity asset.quantity_available = asset_avail - qty # Status: allocated if fully consumed, else in_stock asset.status = "allocated" if asset.quantity_available <= 0 else "in_stock" db.session.add(allocation) # ---- Logging ---- if prev_emp_id and prev_emp_id != emp.id: _log_asset_event( asset, "reallocated", from_employee_id=prev_emp_id, to_employee_id=emp.id, notes=f"Reallocated qty={qty} to {emp.employee_code} - {emp.full_name}.", ) else: _log_asset_event( asset, "allocated", to_employee_id=emp.id, notes=f"Allocated qty={qty} to {emp.employee_code} - {emp.full_name}.", ) db.session.commit() flash(f"Asset {asset.asset_tag} allocated to {emp.full_name}.", "success") except Exception as e: db.session.rollback() current_app.logger.exception("Failed to allocate asset") flash(f"Error allocating asset: {e}", "danger") return redirect(url_for("assets.list_assets")) @assets_bp.route("/return/", methods=["POST"]) @login_required def return_asset(allocation_id): alloc = AssetAllocation.query.get_or_404(allocation_id) asset = alloc.asset ret_date = _parse_date(request.form.get("returned_on")) or datetime.utcnow().date() remarks = (request.form.get("return_remarks") or "").strip() try: alloc.actual_return_date = ret_date alloc.condition_on_return = remarks alloc.status = "return_pending" db.session.commit() flash( f"Return request raised for asset {asset.asset_tag}. " "It will be updated in stock after approval.", "success", ) except Exception as e: db.session.rollback() current_app.logger.exception("Failed to create return request") flash(f"Error creating return request: {e}", "danger") return redirect(url_for("assets.allocations")) # ---------- Allocation History List ---------- @assets_bp.route("/allocations") @login_required def allocations(): q = (request.args.get("q") or "").strip() status = (request.args.get("status") or "").strip() query = ( AssetAllocation.query .join(Asset, AssetAllocation.asset_id == Asset.id) .join(Employee, AssetAllocation.employee_id == Employee.id) ) if q: like = f"%{q}%" query = query.filter(or_( Asset.asset_tag.ilike(like), Asset.name.ilike(like), Employee.full_name.ilike(like), Employee.employee_code.ilike(like), )) if status: query = query.filter(AssetAllocation.status == status) allocations_list = query.order_by(AssetAllocation.allocated_on.desc()).all() return render_template( "assets/allocations.html", allocations=allocations_list, q=q, status=status, ) # ---------- Loss / Damage ---------- @assets_bp.route("/loss-damage", methods=["GET", "POST"]) @login_required def loss_damage(): if request.method == "POST": asset_id = request.form.get("asset_id") incident_date = _parse_date(request.form.get("reported_on")) or datetime.utcnow().date() incident_type = (request.form.get("type") or "loss").strip() description = (request.form.get("description") or "").strip() estimated_cost = request.form.get("estimated_cost") or None if estimated_cost: try: estimated_cost = float(estimated_cost) except ValueError: estimated_cost = None if not asset_id: flash("Please select an asset.", "danger") return redirect(url_for("assets.loss_damage")) asset = Asset.query.get_or_404(int(asset_id)) record = AssetLossDamage( asset_id=asset.id, allocation_id=None, employee_id=current_user.id, incident_date=incident_date, incident_type=incident_type, description=description, estimated_amount=estimated_cost, status="pending", created_by_id=current_user.id, ) db.session.add(record) db.session.commit() flash("Loss / Damage report submitted.", "success") return redirect(url_for("assets.loss_damage")) q = (request.args.get("q") or "").strip() status = (request.args.get("status") or "").strip() query = ( AssetLossDamage.query .join(Asset, AssetLossDamage.asset_id == Asset.id) .outerjoin(Employee, AssetLossDamage.employee_id == Employee.id) ) if q: like = f"%{q}%" query = query.filter(or_( Asset.asset_tag.ilike(like), Asset.name.ilike(like), Employee.full_name.ilike(like), )) if status: query = query.filter(AssetLossDamage.status == status) records = query.order_by(AssetLossDamage.created_at.desc()).all() assets = Asset.query.order_by(Asset.asset_tag.asc()).all() return render_template( "assets/loss_damage.html", records=records, assets=assets, q=q, status=status, ) # ---------- Category Master ---------- @assets_bp.route("/categories", methods=["GET", "POST"]) @login_required def categories(): if request.method == "POST": mode = request.form.get("mode") cat_id = request.form.get("category_id") name = (request.form.get("name") or "").strip() description = (request.form.get("description") or "").strip() if not name: flash("Category name is required.", "danger") return redirect(url_for("assets.categories")) try: if mode == "edit" and cat_id: cat = AssetCategory.query.get_or_404(int(cat_id)) else: cat = AssetCategory() db.session.add(cat) cat.name = name cat.description = description db.session.commit() flash("Category saved.", "success") except Exception as e: db.session.rollback() flash(f"Error saving category: {e}", "danger") return redirect(url_for("assets.categories")) q = (request.args.get("q") or "").strip() query = AssetCategory.query if q: like = f"%{q}%" query = query.filter(AssetCategory.name.ilike(like)) categories_list = query.order_by(AssetCategory.name.asc()).all() return render_template("assets/categories.html", categories=categories_list, q=q) @assets_bp.post("/categories//delete") @login_required def delete_category(cat_id): cat = AssetCategory.query.get_or_404(cat_id) try: db.session.delete(cat) db.session.commit() flash("Category deleted.", "success") except Exception as e: db.session.rollback() flash(f"Cannot delete category: {e}", "danger") return redirect(url_for("assets.categories")) # ---------- Sub Category Master ---------- @assets_bp.route("/subcategories", methods=["GET", "POST"]) @login_required def subcategories(): if request.method == "POST": mode = request.form.get("mode") subcat_id = request.form.get("subcat_id") category_id = request.form.get("category_id") name = (request.form.get("name") or "").strip() description = (request.form.get("description") or "").strip() if not category_id: flash("Please select a Category.", "danger") return redirect(url_for("assets.subcategories")) if not name: flash("Sub Category name is required.", "danger") return redirect(url_for("assets.subcategories")) try: if mode == "edit" and subcat_id: sc = AssetSubCategory.query.get_or_404(int(subcat_id)) else: sc = AssetSubCategory() db.session.add(sc) sc.category_id = int(category_id) sc.name = name sc.description = description db.session.commit() flash("Sub Category saved.", "success") except Exception as e: db.session.rollback() flash(f"Error saving sub category: {e}", "danger") return redirect(url_for("assets.subcategories")) q = (request.args.get("q") or "").strip() category_id = request.args.get("category_id") or "" query = AssetSubCategory.query.join(AssetCategory, AssetSubCategory.category_id == AssetCategory.id) if q: like = f"%{q}%" query = query.filter(AssetSubCategory.name.ilike(like)) if category_id: try: query = query.filter(AssetSubCategory.category_id == int(category_id)) except ValueError: pass subcategories_list = query.order_by(AssetCategory.name.asc(), AssetSubCategory.name.asc()).all() categories_list = AssetCategory.query.order_by(AssetCategory.name.asc()).all() return render_template( "assets/subcategories.html", subcategories=subcategories_list, categories=categories_list, q=q, category_id=category_id, ) @assets_bp.post("/subcategories//delete") @login_required def delete_subcategory(subcat_id): sc = AssetSubCategory.query.get_or_404(subcat_id) try: db.session.delete(sc) db.session.commit() flash("Sub Category deleted.", "success") except Exception as e: db.session.rollback() flash(f"Cannot delete sub category: {e}", "danger") return redirect(url_for("assets.subcategories")) @assets_bp.get("/api/subcategories") @login_required def api_subcategories(): category_id = request.args.get("category_id") if not category_id: return {"items": []} try: cid = int(category_id) except ValueError: return {"items": []} items = ( AssetSubCategory.query .filter(AssetSubCategory.category_id == cid) .order_by(AssetSubCategory.name.asc()) .all() ) return {"items": [{"id": x.id, "name": x.name} for x in items]} # ---------- Vendor Master ---------- @assets_bp.route("/vendors", methods=["GET", "POST"]) @login_required def vendors(): if request.method == "POST": mode = request.form.get("mode") vendor_id = request.form.get("vendor_id") name = (request.form.get("name") or "").strip() contact_person = (request.form.get("contact_person") or "").strip() phone = (request.form.get("phone") or "").strip() email = (request.form.get("email") or "").strip() address = (request.form.get("address") or "").strip() if not name: flash("Vendor name is required.", "danger") return redirect(url_for("assets.vendors")) try: if mode == "edit" and vendor_id: v = AssetVendor.query.get_or_404(int(vendor_id)) else: v = AssetVendor() db.session.add(v) v.name = name v.contact_person = contact_person v.phone = phone v.email = email v.address = address db.session.commit() flash("Vendor saved.", "success") except Exception as e: db.session.rollback() flash(f"Error saving vendor: {e}", "danger") return redirect(url_for("assets.vendors")) q = (request.args.get("q") or "").strip() query = AssetVendor.query if q: like = f"%{q}%" query = query.filter(AssetVendor.name.ilike(like)) vendors_list = query.order_by(AssetVendor.name.asc()).all() return render_template("assets/vendors.html", vendors=vendors_list, q=q) @assets_bp.post("/vendors//delete") @login_required def delete_vendor(vendor_id): v = AssetVendor.query.get_or_404(vendor_id) try: db.session.delete(v) db.session.commit() flash("Vendor deleted.", "success") except Exception as e: db.session.rollback() flash(f"Cannot delete vendor: {e}", "danger") return redirect(url_for("assets.vendors")) # ---------- Return / Loss-Damage Approvals ---------- @assets_bp.route("/approvals") @login_required def approvals(): pending_returns = ( AssetAllocation.query .join(Asset, AssetAllocation.asset_id == Asset.id) .join(Employee, AssetAllocation.employee_id == Employee.id) .filter(AssetAllocation.status == "return_pending") .order_by(AssetAllocation.actual_return_date.desc()) .all() ) pending_ld = ( AssetLossDamage.query .join(Asset, AssetLossDamage.asset_id == Asset.id) .join(Employee, AssetLossDamage.employee_id == Employee.id) .filter(AssetLossDamage.status == "pending") .order_by(AssetLossDamage.incident_date.desc()) .all() ) return render_template( "assets/approvals.html", pending_returns=pending_returns, pending_ld=pending_ld, ) @assets_bp.post("/approvals/return//") @login_required def approve_return(allocation_id, action): alloc = AssetAllocation.query.get_or_404(allocation_id) asset = alloc.asset if alloc.status != "return_pending": flash("This allocation is not pending for return.", "warning") return redirect(url_for("assets.approvals")) try: if action == "approve": alloc.status = "returned" qty = _safe_int(getattr(alloc, "quantity", None), 1) total = _safe_int(getattr(asset, "quantity_total", None), 1) avail = _safe_int(getattr(asset, "quantity_available", None), total) asset.quantity_available = min(avail + qty, total) asset.status = "in_stock" if asset.quantity_available > 0 else "allocated" _log_asset_event( asset, "returned", from_employee_id=alloc.employee_id, notes=f"Returned qty={qty} by employee ID {alloc.employee_id}.", ) _log_asset_event(asset, "inventory", notes="Asset moved back to inventory after return.") flash(f"Return approved for asset {asset.asset_tag}.", "success") else: alloc.status = "allocated" alloc.actual_return_date = None alloc.condition_on_return = None flash(f"Return request rejected for asset {asset.asset_tag}.", "info") db.session.commit() except Exception as e: db.session.rollback() current_app.logger.exception("Error processing return approval") flash(f"Error processing return approval: {e}", "danger") return redirect(url_for("assets.approvals")) @assets_bp.post("/approvals/loss-damage//") @login_required def approve_loss_damage(record_id, action): rec = AssetLossDamage.query.get_or_404(record_id) asset = rec.asset if rec.status != "pending": flash("This loss/damage record is not pending.", "warning") return redirect(url_for("assets.approvals")) try: if action == "approve": rec.status = "approved" rec.approved_by_id = current_user.id rec.approved_at = datetime.utcnow() if rec.incident_type == "loss": asset.status = "lost" _log_asset_event(asset, "lost", from_employee_id=rec.employee_id, notes=f"Loss approved. Estimated amount: {rec.estimated_amount or 0}.") else: asset.status = "damaged" _log_asset_event(asset, "damaged", from_employee_id=rec.employee_id, notes=f"Damage approved. Estimated amount: {rec.estimated_amount or 0}.") flash(f"{rec.incident_type.title()} record approved for asset {asset.asset_tag}.", "success") else: rec.status = "rejected" rec.approved_by_id = current_user.id rec.approved_at = datetime.utcnow() flash(f"{rec.incident_type.title()} record rejected for asset {asset.asset_tag}.", "info") db.session.commit() except Exception as e: db.session.rollback() current_app.logger.exception("Error processing loss/damage approval") flash(f"Error processing loss/damage approval: {e}", "danger") return redirect(url_for("assets.approvals")) # ---------- Dashboard / My Assets ---------- @assets_bp.route("/dashboard") @login_required def asset_dashboard(): total_assets = db.session.query(func.count(Asset.id)).scalar() or 0 status_counts = dict( db.session.query(Asset.status, func.count(Asset.id)) .group_by(Asset.status) .all() ) def sc(key): return int(status_counts.get(key, 0) or 0) in_stock = sc("in_stock") allocated = sc("allocated") under_repair = sc("under_repair") retired = sc("retired") lost = sc("lost") damaged = sc("damaged") pending_returns = ( db.session.query(func.count(AssetAllocation.id)) .filter(AssetAllocation.status == "return_pending") .scalar() or 0 ) pending_loss_damage = ( db.session.query(func.count(AssetLossDamage.id)) .filter(AssetLossDamage.status == "pending") .scalar() or 0 ) recent_allocations = ( AssetAllocation.query .join(Asset, AssetAllocation.asset_id == Asset.id) .join(Employee, AssetAllocation.employee_id == Employee.id) .order_by(AssetAllocation.allocated_on.desc()) .limit(5) .all() ) recent_incidents = ( AssetLossDamage.query .join(Asset, AssetLossDamage.asset_id == Asset.id) .outerjoin(Employee, AssetLossDamage.employee_id == Employee.id) .order_by(AssetLossDamage.incident_date.desc()) .limit(5) .all() ) return render_template( "assets/dashboard.html", total_assets=total_assets, in_stock=in_stock, allocated=allocated, under_repair=under_repair, retired=retired, lost=lost, damaged=damaged, pending_returns=pending_returns, pending_loss_damage=pending_loss_damage, recent_allocations=recent_allocations, recent_incidents=recent_incidents, ) @assets_bp.route("/my-assets") @login_required def my_assets(): my_allocations = ( AssetAllocation.query .join(Asset, AssetAllocation.asset_id == Asset.id) .filter(AssetAllocation.employee_id == current_user.id) .order_by(AssetAllocation.allocated_on.desc()) .all() ) return render_template( "assets/my_assets.html", allocations=my_allocations, today=date.today(), ) # ---------- Bulk Upload Assets ---------- @assets_bp.route("/bulk-template") @login_required def bulk_template(): output = io.StringIO() writer = csv.writer(output) writer.writerow([ "Asset Tag", "Asset Code (optional)", "Name", "Category", "Sub Category", "Vendor", "Serial Number", "Purchase Date (YYYY-MM-DD)", "Warranty Expiry (YYYY-MM-DD)", "Status (in_stock/allocated/under_repair/retired/lost/damaged)", "Location", "Remarks", "Quantity", ]) writer.writerow([ "LAP-001", "", "Laptop Dell", "Laptop", "Ultrabook", "Dell", "SN12345", "2025-01-15", "2027-01-15", "in_stock", "Mumbai Office", "Imported via bulk upload", "1", ]) output.seek(0) resp = make_response(output.getvalue()) resp.headers["Content-Type"] = "text/csv" resp.headers["Content-Disposition"] = "attachment; filename=asset_bulk_template.csv" return resp @assets_bp.route("/bulk-upload", methods=["GET", "POST"]) @login_required def bulk_upload_assets(): if request.method == "GET": return render_template("assets/bulk_upload.html") file = request.files.get("file") if not file or file.filename == "": flash("Please choose a CSV file to upload.", "danger") return redirect(url_for("assets.bulk_upload_assets")) if not _allowed_asset_file(file.filename): flash("Only .csv files are supported for asset bulk upload.", "danger") return redirect(url_for("assets.bulk_upload_assets")) filename = secure_filename(file.filename) try: text = file.stream.read().decode("utf-8-sig") stream = io.StringIO(text, newline="") except UnicodeDecodeError: flash("Unable to read file. Please ensure it is a UTF-8 encoded CSV.", "danger") return redirect(url_for("assets.bulk_upload_assets")) reader = csv.DictReader(stream) if reader.fieldnames: reader.fieldnames = [fn.strip().lstrip("\ufeff") for fn in reader.fieldnames] required_cols = {"Asset Tag", "Name"} missing_required = [c for c in required_cols if c not in (reader.fieldnames or [])] if missing_required: flash(f"Missing required column(s): {', '.join(missing_required)}", "danger") return redirect(url_for("assets.bulk_upload_assets")) created = 0 updated = 0 skipped = 0 errors: list[str] = [] row_num = 1 for row in reader: row_num += 1 try: with db.session.begin_nested(): asset_tag = (row.get("Asset Tag") or "").strip() name = (row.get("Name") or "").strip() if not asset_tag or not name: skipped += 1 errors.append(f"Row {row_num}: Missing Asset Tag or Name – skipped.") continue asset_code = (row.get("Asset Code (optional)") or "").strip() category_name = (row.get("Category") or "").strip() subcategory_name = (row.get("Sub Category") or "").strip() vendor_name = (row.get("Vendor") or "").strip() serial_number = (row.get("Serial Number") or "").strip() purchase_date_str = (row.get("Purchase Date (YYYY-MM-DD)") or "").strip() warranty_expiry_str = (row.get("Warranty Expiry (YYYY-MM-DD)") or "").strip() status = (row.get("Status (in_stock/allocated/under_repair/retired/lost/damaged)") or "").strip() location = (row.get("Location") or "").strip() remarks = (row.get("Remarks") or "").strip() qty_str = (row.get("Quantity") or "").strip() try: qty_total = int(qty_str) if qty_str != "" else 1 except ValueError: qty_total = 1 if qty_total < 1: qty_total = 1 purchase_date = _parse_date(purchase_date_str) warranty_expiry = _parse_date(warranty_expiry_str) if not status: status = "in_stock" status = status.lower() with db.session.no_autoflush: if serial_number: existing_sn = Asset.query.filter(Asset.serial_number == serial_number).first() if existing_sn and (existing_sn.asset_tag != asset_tag): skipped += 1 errors.append( f"Row {row_num}: Serial Number '{serial_number}' already exists " f"(asset_tag={existing_sn.asset_tag}) – skipped." ) continue asset = Asset.query.filter_by(asset_tag=asset_tag).first() is_new = asset is None if is_new: asset = Asset() asset.asset_tag = asset_tag asset.asset_code = asset_code or asset_tag db.session.add(asset) created += 1 else: updated += 1 category_id = None if category_name: cat = AssetCategory.query.filter(func.lower(AssetCategory.name) == category_name.lower()).first() if not cat: cat = AssetCategory(name=category_name) db.session.add(cat) db.session.flush([cat]) category_id = cat.id subcategory_id = None if category_id and subcategory_name: sc = ( AssetSubCategory.query .filter( AssetSubCategory.category_id == category_id, func.lower(AssetSubCategory.name) == subcategory_name.lower() ) .first() ) if not sc: sc = AssetSubCategory(category_id=category_id, name=subcategory_name) db.session.add(sc) db.session.flush([sc]) subcategory_id = sc.id vendor_id = None if vendor_name: v = AssetVendor.query.filter(func.lower(AssetVendor.name) == vendor_name.lower()).first() if not v: v = AssetVendor(name=vendor_name) db.session.add(v) db.session.flush([v]) vendor_id = v.id asset.asset_code = asset_code or (asset.asset_code or asset_tag) asset.asset_tag = asset_tag asset.name = name asset.category_id = category_id asset.subcategory_id = subcategory_id asset.vendor_id = vendor_id asset.serial_number = serial_number or None asset.purchase_date = purchase_date asset.warranty_expiry = warranty_expiry asset.status = status asset.location = location asset.remarks = remarks if is_new: asset.quantity_total = qty_total asset.quantity_available = qty_total else: old_total = _safe_int(getattr(asset, "quantity_total", None), 1) old_avail = _safe_int(getattr(asset, "quantity_available", None), old_total) old_allocated = max(old_total - old_avail, 0) if qty_total < old_allocated: raise ValueError(f"Quantity cannot be less than already allocated ({old_allocated}).") asset.quantity_total = qty_total asset.quantity_available = max(qty_total - old_allocated, 0) db.session.flush() except IntegrityError as ie: db.session.rollback() skipped += 1 errors.append(f"Row {row_num}: DB error: {ie.orig}") except Exception as e: db.session.rollback() skipped += 1 errors.append(f"Row {row_num}: {e}") try: db.session.commit() except Exception as e: db.session.rollback() current_app.logger.exception("Bulk asset upload failed at commit") flash(f"Error committing bulk upload: {e}", "danger") return redirect(url_for("assets.bulk_upload_assets")) msg = f"Bulk upload complete. Created: {created}, Updated: {updated}, Skipped: {skipped}." if errors: msg += " Some rows had issues (see details below)." flash(msg, "warning") else: flash(msg, "success") return render_template( "assets/bulk_upload.html", errors=errors, created=created, updated=updated, skipped=skipped, filename=filename, ) # ---------- Logs ---------- @assets_bp.route("/logs") @login_required def asset_logs(): q = (request.args.get("q") or "").strip() employee_id = request.args.get("employee_id") or "" action = (request.args.get("action") or "").strip() # Join employee via employee_id (since _log_asset_event sets employee_id if present) query = ( AssetEventLog.query .join(Asset, AssetEventLog.asset_id == Asset.id) .outerjoin(Employee, AssetEventLog.employee_id == Employee.id) ) if q: like = f"%{q}%" # If employee join not present, employee filters won't apply - still ok query = query.filter(or_( Asset.asset_tag.ilike(like), Asset.name.ilike(like), )) if employee_id and hasattr(AssetEventLog, "employee_id"): try: eid = int(employee_id) query = query.filter(AssetEventLog.employee_id == eid) except ValueError: pass if action: query = query.filter(AssetEventLog.event_type == action) logs = query.order_by(AssetEventLog.event_time.desc()).limit(500).all() employees = Employee.query.order_by(Employee.full_name.asc()).all() actions = [ ("", "All Actions"), ("new_asset", "New Asset"), ("inventory", "In Inventory"), ("allocated", "Allocated"), ("reallocated", "Allocated to Another Employee"), ("returned", "Returned"), ("under_repair", "Under Repair"), ("retired", "Retired"), ("lost", "Lost"), ("damaged", "Damaged"), ("status_change", "Status Change"), ] return render_template( "assets/logs.html", logs=logs, q=q, employee_id=employee_id, employees=employees, action=action, actions=actions, )