311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""
|
|
routes/bid_tracker.py — Bid / Opportunity tracker routes.
|
|
"""
|
|
|
|
import csv
|
|
import io
|
|
import logging
|
|
from flask import (Blueprint, render_template, request, redirect, url_for,
|
|
flash, session, jsonify, Response)
|
|
from models import (
|
|
get_all_bids, get_bid, create_bid, update_bid, delete_bid,
|
|
get_bid_updates, add_bid_update, delete_bid_update, BID_STATUSES,
|
|
log_action, get_bids_due_soon, get_admin_emails,
|
|
invalidate_admin_stats_cache,
|
|
)
|
|
from utils.decorators import login_required, admin_required
|
|
from utils.email import send_email
|
|
|
|
logger = logging.getLogger("routes.bid_tracker")
|
|
bid_tracker_bp = Blueprint("bid_tracker", __name__, url_prefix="/bids")
|
|
|
|
STATUS_LABELS = {
|
|
"open": "🟢 Open",
|
|
"monitoring": "🔵 Monitoring",
|
|
"awarded": "🏆 Awarded",
|
|
"no_bid": "⛔ No Bid",
|
|
"cancelled": "🚫 Cancelled",
|
|
}
|
|
|
|
|
|
def _ser(row):
|
|
"""Make a DB dict JSON-serialisable (dates/times → ISO strings)."""
|
|
return {k: v.isoformat() if hasattr(v, "isoformat") else v for k, v in row.items()}
|
|
|
|
|
|
@bid_tracker_bp.route("/")
|
|
@login_required
|
|
def bids_list():
|
|
status_filter = request.args.get("status", "")
|
|
bids = get_all_bids(status_filter=status_filter)
|
|
return render_template("bid_tracker.html",
|
|
bids=bids,
|
|
status_filter=status_filter,
|
|
bid_statuses=BID_STATUSES,
|
|
status_labels=STATUS_LABELS)
|
|
|
|
|
|
@bid_tracker_bp.route("/<int:bid_id>")
|
|
@login_required
|
|
def bid_detail(bid_id):
|
|
bid = get_bid(bid_id)
|
|
updates = get_bid_updates(bid_id)
|
|
if not bid:
|
|
flash("Bid not found.", "warning")
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
return render_template("bid_detail.html",
|
|
bid=bid, updates=updates,
|
|
bid_statuses=BID_STATUSES,
|
|
status_labels=STATUS_LABELS)
|
|
|
|
|
|
@bid_tracker_bp.route("/create", methods=["POST"])
|
|
@login_required
|
|
def create():
|
|
user = session["user"]
|
|
title = request.form.get("title", "").strip()
|
|
url = request.form.get("url", "").strip()
|
|
source = request.form.get("source", "").strip()
|
|
sol_no = request.form.get("solicitation_number", "").strip()
|
|
status = request.form.get("status", "open")
|
|
due_date = request.form.get("due_date") or None
|
|
notes = request.form.get("notes", "").strip()
|
|
|
|
if not title or not url:
|
|
flash("Title and URL are required.", "danger")
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
|
|
try:
|
|
bid_id = create_bid(user["id"], title, url, source, sol_no, status, due_date, notes)
|
|
log_action(user["id"], "CREATE_BID", "bid_tracker", bid_id,
|
|
f"Created bid '{title}' status='{status}'.")
|
|
invalidate_admin_stats_cache()
|
|
flash(f"Bid '{title}' added.", "success")
|
|
logger.info(f"Bid '{title}' created by user_id={user['id']}.")
|
|
except Exception as e:
|
|
logger.error(f"create_bid error: {e}")
|
|
flash(f"Error: {e}", "danger")
|
|
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
|
|
|
|
@bid_tracker_bp.route("/<int:bid_id>/edit", methods=["POST"])
|
|
@login_required
|
|
def edit(bid_id):
|
|
user = session["user"]
|
|
title = request.form.get("title", "").strip()
|
|
url = request.form.get("url", "").strip()
|
|
source = request.form.get("source", "").strip()
|
|
sol_no = request.form.get("solicitation_number", "").strip()
|
|
status = request.form.get("status", "open")
|
|
due_date = request.form.get("due_date") or None
|
|
notes = request.form.get("notes", "").strip()
|
|
|
|
try:
|
|
update_bid(user["id"], bid_id, title, url, source, sol_no, status, due_date, notes)
|
|
log_action(user["id"], "UPDATE_BID", "bid_tracker", bid_id,
|
|
f"Updated bid id={bid_id} status='{status}'.")
|
|
invalidate_admin_stats_cache()
|
|
flash(f"Bid '{title}' updated.", "success")
|
|
logger.info(f"Bid id={bid_id} updated by user_id={user['id']}.")
|
|
except Exception as e:
|
|
logger.error(f"update_bid error: {e}")
|
|
flash(f"Error: {e}", "danger")
|
|
|
|
return redirect(url_for("bid_tracker.bid_detail", bid_id=bid_id))
|
|
|
|
|
|
@bid_tracker_bp.route("/<int:bid_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete(bid_id):
|
|
user = session["user"]
|
|
try:
|
|
delete_bid(user["id"], bid_id)
|
|
log_action(user["id"], "DELETE_BID", "bid_tracker", bid_id,
|
|
f"Deleted bid id={bid_id}.")
|
|
invalidate_admin_stats_cache()
|
|
flash("Bid deleted.", "success")
|
|
logger.info(f"Bid id={bid_id} deleted by user_id={user['id']}.")
|
|
except Exception as e:
|
|
logger.error(f"delete_bid error: {e}")
|
|
flash(f"Error: {e}", "danger")
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
|
|
|
|
@bid_tracker_bp.route("/<int:bid_id>/updates", methods=["POST"])
|
|
@login_required
|
|
def add_update(bid_id):
|
|
user = session["user"]
|
|
content = request.form.get("content", "").strip()
|
|
if not content:
|
|
flash("Update content cannot be empty.", "warning")
|
|
else:
|
|
try:
|
|
upd_id = add_bid_update(user["id"], bid_id, content)
|
|
log_action(user["id"], "ADD_BID_UPDATE", "bid_updates", upd_id,
|
|
f"Posted update on bid_id={bid_id}.")
|
|
flash("Update posted.", "success")
|
|
logger.info(f"Bid update posted on bid_id={bid_id} by user_id={user['id']}.")
|
|
except Exception as e:
|
|
logger.error(f"add_bid_update error: {e}")
|
|
flash(f"Error: {e}", "danger")
|
|
return redirect(url_for("bid_tracker.bid_detail", bid_id=bid_id))
|
|
|
|
|
|
@bid_tracker_bp.route("/updates/<int:update_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete_update(update_id):
|
|
user = session["user"]
|
|
try:
|
|
delete_bid_update(user["id"], update_id)
|
|
log_action(user["id"], "DELETE_BID_UPDATE", "bid_updates", update_id,
|
|
f"Deleted bid update id={update_id}.")
|
|
flash("Update deleted.", "success")
|
|
logger.info(f"Bid update id={update_id} deleted by user_id={user['id']}.")
|
|
except Exception as e:
|
|
logger.error(f"delete_bid_update error: {e}")
|
|
flash(f"Error: {e}", "danger")
|
|
bid_id = request.form.get("bid_id")
|
|
if bid_id:
|
|
return redirect(url_for("bid_tracker.bid_detail", bid_id=int(bid_id)))
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
|
|
|
|
@bid_tracker_bp.route("/<int:bid_id>/json")
|
|
@login_required
|
|
def bid_json(bid_id):
|
|
"""Return bid detail + updates as JSON for the split-pane detail panel."""
|
|
user = session["user"]
|
|
bid = get_bid(bid_id)
|
|
if not bid:
|
|
return jsonify({"error": "Not found"}), 404
|
|
|
|
updates = get_bid_updates(bid_id)
|
|
|
|
is_owner = bid.get("added_by") == user["id"]
|
|
can_edit = user["role"] == "admin" or is_owner
|
|
|
|
return jsonify({
|
|
"bid": _ser(bid),
|
|
"updates": [_ser(u) for u in updates],
|
|
"can_edit": can_edit,
|
|
"user_id": user["id"],
|
|
"is_admin": user["role"] == "admin",
|
|
})
|
|
|
|
|
|
@bid_tracker_bp.route("/list/json")
|
|
@login_required
|
|
def list_json():
|
|
"""Return paginated bids as JSON for the AJAX list panel."""
|
|
status_filter = request.args.get("status", "")
|
|
try:
|
|
offset = int(request.args.get("offset", 0))
|
|
except (ValueError, TypeError):
|
|
offset = 0
|
|
try:
|
|
bids = get_all_bids(status_filter=status_filter, limit=50, offset=offset)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
return jsonify([_ser(b) for b in bids])
|
|
|
|
|
|
@bid_tracker_bp.route("/<int:bid_id>/updates/json", methods=["POST"])
|
|
@login_required
|
|
def add_update_json(bid_id):
|
|
"""Post a new update; return JSON so the page doesn't reload."""
|
|
user = session["user"]
|
|
content = request.form.get("content", "").strip()
|
|
if not content:
|
|
return jsonify({"error": "Update content cannot be empty."}), 400
|
|
try:
|
|
update_id = add_bid_update(user["id"], bid_id, content)
|
|
log_action(user["id"], "ADD_BID_UPDATE", "bid_updates", update_id,
|
|
f"Posted update on bid_id={bid_id}.")
|
|
logger.info(f"Bid update id={update_id} posted on bid_id={bid_id} by user_id={user['id']}.")
|
|
return jsonify({"ok": True, "update_id": update_id})
|
|
except Exception as e:
|
|
logger.error(f"add_update_json error: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@bid_tracker_bp.route("/updates/<int:update_id>/delete/json", methods=["POST"])
|
|
@login_required
|
|
def delete_update_json(update_id):
|
|
"""Delete an update; return JSON so the page doesn't reload."""
|
|
user = session["user"]
|
|
try:
|
|
delete_bid_update(user["id"], update_id)
|
|
log_action(user["id"], "DELETE_BID_UPDATE", "bid_updates", update_id,
|
|
f"Deleted bid update id={update_id}.")
|
|
logger.info(f"Bid update id={update_id} deleted by user_id={user['id']}.")
|
|
return jsonify({"ok": True})
|
|
except Exception as e:
|
|
logger.error(f"delete_update_json error: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@bid_tracker_bp.route("/remind", methods=["POST"])
|
|
@admin_required
|
|
def send_reminders():
|
|
"""Send an email digest of bids due within 7 days to all admin users with emails."""
|
|
user = session["user"]
|
|
bids = get_bids_due_soon(days=7)
|
|
recipients = get_admin_emails()
|
|
|
|
if not bids:
|
|
flash("No active bids due within 7 days.", "info")
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
|
|
if not recipients:
|
|
flash("No admin email addresses configured. Add them via Admin → Users.", "warning")
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
|
|
lines = [f"Bid Deadline Reminder — {len(bids)} bid(s) due within 7 days:\n"]
|
|
for b in bids:
|
|
due = str(b["due_date"])[:10] if b.get("due_date") else "N/A"
|
|
line = f" • {b['title']} — Due: {due} — Status: {b['status']}"
|
|
if b.get("solicitation_number"):
|
|
line += f" — Sol#: {b['solicitation_number']}"
|
|
lines.append(line)
|
|
lines.append("\nLog in to Website Checker for full details.")
|
|
body = "\n".join(lines)
|
|
|
|
try:
|
|
send_email(recipients, "Website Checker — Upcoming Bid Deadlines", body)
|
|
log_action(user["id"], "BID_REMIND", "bid_tracker", None,
|
|
f"Bid reminder email sent to {len(recipients)} admin(s), {len(bids)} bid(s) listed.")
|
|
flash(f"Reminder digest sent to {len(recipients)} recipient(s).", "success")
|
|
except Exception as e:
|
|
logger.error(f"send_reminders error: {e}")
|
|
flash(f"Failed to send email: {e}", "danger")
|
|
|
|
return redirect(url_for("bid_tracker.bids_list"))
|
|
|
|
|
|
@bid_tracker_bp.route("/export.csv")
|
|
@login_required
|
|
def export_csv():
|
|
status = request.args.get("status", "")
|
|
bids = get_all_bids(status_filter=status, limit=10000, offset=0)
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf)
|
|
writer.writerow([
|
|
"ID", "Title", "Source", "Solicitation #", "Status",
|
|
"Due Date", "URL", "Notes", "Added By", "Created At",
|
|
])
|
|
for b in bids:
|
|
writer.writerow([
|
|
b["id"], b.get("title", ""), b.get("source", ""),
|
|
b.get("solicitation_number", ""), b.get("status", ""),
|
|
str(b.get("due_date") or ""), b.get("url", ""),
|
|
b.get("notes", ""), b.get("added_by_username", ""),
|
|
str(b.get("created_at") or ""),
|
|
])
|
|
return Response(
|
|
buf.getvalue(),
|
|
mimetype="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=bids_export.csv"},
|
|
)
|