import os import secrets from datetime import date, datetime, timedelta from functools import wraps from flask import Flask, jsonify, redirect, render_template, request, session, url_for from sqlalchemy import text from werkzeug.security import check_password_hash, generate_password_hash from models import InternetConnection, Phone, User, db DATA_DIR = os.environ.get("TELCO_DATA_DIR", os.path.join(os.path.dirname(__file__), "..", "data")) os.makedirs(DATA_DIR, exist_ok=True) DB_PATH = os.path.join(DATA_DIR, "telco.db") SECRET_KEY_PATH = os.path.join(DATA_DIR, "secret_key.txt") # Seeded once, only if the users table is empty (first run against a fresh database). # Passwords are hashed before they ever touch the database - see seed_users(). SEED_USERS = [ ("MBelan", "KFC@99", "admin"), ("AMaar", "KFC@99", "admin"), ("CGasser", "KFC@99", "admin"), ("RSimko", "KFC@99", "admin"), ("TPanakova", "KFC@99", "admin"), ("JLapinova", "KFC@11", "viewer"), ("JTrnovska", "KFC@11", "viewer"), ] def get_or_create_secret_key(): """Persist a random session-signing key in the data dir so it survives container restarts and is shared by every gunicorn worker (each worker imports this module independently, so a key generated only in memory would differ per worker and invalidate sessions randomly).""" try: with open(SECRET_KEY_PATH, "r") as f: existing = f.read().strip() if existing: return existing except FileNotFoundError: pass key = secrets.token_hex(32) tmp_path = f"{SECRET_KEY_PATH}.tmp.{os.getpid()}" try: with open(tmp_path, "w") as f: f.write(key) os.replace(tmp_path, SECRET_KEY_PATH) except OSError: pass # Re-read in case another worker won the race and wrote its own key first. try: with open(SECRET_KEY_PATH, "r") as f: existing = f.read().strip() if existing: return existing except FileNotFoundError: pass return key def create_app(): app = Flask(__name__) app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{DB_PATH}" app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False app.config["JSON_SORT_KEYS"] = False app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(days=30) app.secret_key = os.environ.get("TELCO_SECRET_KEY") or get_or_create_secret_key() db.init_app(app) with app.app_context(): db.create_all() run_migrations() seed_users() register_routes(app) return app def run_migrations(): """Lightweight in-place migrations for columns added after the DB was first created. db.create_all() only creates missing tables, not missing columns on existing tables.""" existing_columns = {row[1] for row in db.session.execute(text("PRAGMA table_info(phones)")).fetchall()} if "comment" not in existing_columns: db.session.execute(text("ALTER TABLE phones ADD COLUMN comment TEXT")) db.session.commit() if "verified" not in existing_columns: db.session.execute(text("ALTER TABLE phones ADD COLUMN verified VARCHAR(8)")) db.session.commit() def seed_users(): """First-run only: create the initial login accounts if no users exist yet at all.""" if User.query.count() > 0: return for username, password, role in SEED_USERS: db.session.add(User(username=username, password_hash=generate_password_hash(password), role=role)) db.session.commit() def parse_date(value): """Parse an ISO date string ('YYYY-MM-DD') from a form/JSON payload. Empty -> None.""" if not value: return None if isinstance(value, date): return value value = value.strip() if not value: return None return datetime.strptime(value, "%Y-%m-%d").date() VENUE_OPTIONS = ["QWE Headoffice", "KFC Restaurant"] VERIFIED_OPTIONS = ["Yes", "No"] ROLE_OPTIONS = ["admin", "viewer"] def login_required(view): @wraps(view) def wrapped(*args, **kwargs): if "user_id" not in session: if request.path.startswith("/api/"): return jsonify({"error": "Login required."}), 401 return redirect(url_for("login", next=request.path)) return view(*args, **kwargs) return wrapped def admin_required(view): @wraps(view) def wrapped(*args, **kwargs): if "user_id" not in session: if request.path.startswith("/api/"): return jsonify({"error": "Login required."}), 401 return redirect(url_for("login", next=request.path)) if session.get("role") != "admin": if request.path.startswith("/api/"): return jsonify({"error": "Admin access required."}), 403 return render_template("403.html"), 403 return view(*args, **kwargs) return wrapped def register_routes(app): # --------------------------------------------------------------- # Auth # --------------------------------------------------------------- @app.get("/login") def login(): if "user_id" in session: return redirect(url_for("index")) return render_template("login.html", error=None, next=request.args.get("next", "")) @app.post("/login") def login_submit(): username = (request.form.get("username") or "").strip() password = request.form.get("password") or "" next_url = request.form.get("next") or "" user = User.query.filter(db.func.lower(User.username) == username.lower()).first() if not user or not check_password_hash(user.password_hash, password): return render_template("login.html", error="Invalid username or password.", next=next_url), 401 session.clear() session.permanent = True session["user_id"] = user.id session["username"] = user.username session["role"] = user.role if next_url.startswith("/") and not next_url.startswith("//"): return redirect(next_url) return redirect(url_for("index")) @app.get("/logout") def logout(): session.clear() return redirect(url_for("login")) @app.get("/") @login_required def index(): return render_template("index.html") # --------------------------------------------------------------- # Mobile phones # --------------------------------------------------------------- SORTABLE_PHONE_COLUMNS = { "phone_number": Phone.phone_number, "plan": Phone.plan, "contract_expiration": Phone.contract_expiration, "eligible_for_replacement": Phone.eligible_for_replacement, "current_owner": Phone.current_owner, "venue": Phone.venue, "verified": Phone.verified, "comment": Phone.comment, } @app.get("/phones") @login_required def phones_page(): sort = request.args.get("sort", "phone_number") direction = request.args.get("dir", "asc") if sort not in SORTABLE_PHONE_COLUMNS: sort = "phone_number" if direction not in ("asc", "desc"): direction = "asc" column = SORTABLE_PHONE_COLUMNS[sort] order = column.desc() if direction == "desc" else column.asc() # Secondary sort by phone number keeps ties in a stable, predictable order. phones = Phone.query.order_by(order, Phone.phone_number.asc()).all() today = date.today() plan_options = [ row[0] for row in db.session.query(Phone.plan).filter(Phone.plan.isnot(None)).distinct().order_by(Phone.plan).all() ] headoffice_count = sum(1 for p in phones if p.venue == "QWE Headoffice") restaurant_count = sum(1 for p in phones if p.venue == "KFC Restaurant") unassigned_count = len(phones) - headoffice_count - restaurant_count verified_count = sum(1 for p in phones if p.verified == "Yes") not_verified_count = len(phones) - verified_count stats = { "total": len(phones), "expired": sum(1 for p in phones if p.contract_expiration and p.contract_expiration < today), "expiring_soon": sum( 1 for p in phones if p.contract_expiration and p.contract_expiration >= today and (p.contract_expiration - today).days <= 60 ), "eligible_now": sum( 1 for p in phones if not p.eligible_for_replacement or p.eligible_for_replacement <= today ), "headoffice": headoffice_count, "restaurant": restaurant_count, "unassigned": unassigned_count, "headoffice_pct": round(100 * headoffice_count / len(phones)) if phones else 0, "restaurant_pct": round(100 * restaurant_count / len(phones)) if phones else 0, "verified": verified_count, "not_verified": not_verified_count, } return render_template( "phones.html", phones=phones, today=today, sort=sort, direction=direction, plan_options=plan_options, venue_options=VENUE_OPTIONS, verified_options=VERIFIED_OPTIONS, stats=stats, active="phones", ) @app.get("/api/phones/") @login_required def get_phone(phone_id): phone = Phone.query.get_or_404(phone_id) return jsonify(phone.to_dict()) @app.post("/api/phones") @admin_required def create_phone(): payload = request.get_json(force=True) phone = Phone( phone_number=(payload.get("phone_number") or "").strip(), plan=(payload.get("plan") or "").strip() or None, contract_expiration=parse_date(payload.get("contract_expiration")), eligible_for_replacement=parse_date(payload.get("eligible_for_replacement")), current_owner=(payload.get("current_owner") or "").strip() or None, venue=(payload.get("venue") or "").strip() or None, verified=(payload.get("verified") or "").strip() or None, comment=(payload.get("comment") or "").strip() or None, ) if not phone.phone_number: return jsonify({"error": "Phone number is required."}), 400 db.session.add(phone) db.session.commit() return jsonify(phone.to_dict()), 201 @app.put("/api/phones/") @admin_required def update_phone(phone_id): phone = Phone.query.get_or_404(phone_id) payload = request.get_json(force=True) # Phone Number and Plan are set once at creation and are not editable afterward - # ignore any values sent for them here (enforced server-side, not just hidden in the UI). phone.contract_expiration = parse_date(payload.get("contract_expiration")) phone.eligible_for_replacement = parse_date(payload.get("eligible_for_replacement")) phone.current_owner = (payload.get("current_owner") or "").strip() or None phone.venue = (payload.get("venue") or "").strip() or None phone.verified = (payload.get("verified") or "").strip() or None phone.comment = (payload.get("comment") or "").strip() or None db.session.commit() return jsonify(phone.to_dict()) @app.delete("/api/phones/") @admin_required def delete_phone(phone_id): phone = Phone.query.get_or_404(phone_id) db.session.delete(phone) db.session.commit() return jsonify({"ok": True}) # --------------------------------------------------------------- # Internet connectivity # --------------------------------------------------------------- @app.get("/internet") @login_required def internet_page(): connections = InternetConnection.query.order_by(InternetConnection.venue.asc()).all() return render_template("internet.html", connections=connections, active="internet") @app.get("/api/internet/") @login_required def get_connection(conn_id): conn = InternetConnection.query.get_or_404(conn_id) return jsonify(conn.to_dict()) @app.post("/api/internet") @admin_required def create_connection(): payload = request.get_json(force=True) conn = InternetConnection( venue=(payload.get("venue") or "").strip(), isp_provider=(payload.get("isp_provider") or "").strip() or None, plan=(payload.get("plan") or "").strip() or None, contract_expiration=parse_date(payload.get("contract_expiration")), notes=(payload.get("notes") or "").strip() or None, ) if not conn.venue: return jsonify({"error": "Venue is required."}), 400 db.session.add(conn) db.session.commit() return jsonify(conn.to_dict()), 201 @app.put("/api/internet/") @admin_required def update_connection(conn_id): conn = InternetConnection.query.get_or_404(conn_id) payload = request.get_json(force=True) conn.venue = (payload.get("venue") or conn.venue).strip() conn.isp_provider = (payload.get("isp_provider") or "").strip() or None conn.plan = (payload.get("plan") or "").strip() or None conn.contract_expiration = parse_date(payload.get("contract_expiration")) conn.notes = (payload.get("notes") or "").strip() or None db.session.commit() return jsonify(conn.to_dict()) @app.delete("/api/internet/") @admin_required def delete_connection(conn_id): conn = InternetConnection.query.get_or_404(conn_id) db.session.delete(conn) db.session.commit() return jsonify({"ok": True}) # --------------------------------------------------------------- # User accounts (admin only) - login/password management # --------------------------------------------------------------- @app.get("/users") @admin_required def users_page(): users = User.query.order_by(User.username.asc()).all() admin_count = sum(1 for u in users if u.role == "admin") viewer_count = len(users) - admin_count return render_template( "users.html", users=users, role_options=ROLE_OPTIONS, admin_count=admin_count, viewer_count=viewer_count, active="users", ) @app.post("/api/users") @admin_required def create_user(): payload = request.get_json(force=True) username = (payload.get("username") or "").strip() password = payload.get("password") or "" role = (payload.get("role") or "").strip() if not username or not password: return jsonify({"error": "Username and password are required."}), 400 if role not in ROLE_OPTIONS: return jsonify({"error": "Role must be admin or viewer."}), 400 if User.query.filter(db.func.lower(User.username) == username.lower()).first(): return jsonify({"error": "That username already exists."}), 400 user = User(username=username, password_hash=generate_password_hash(password), role=role) db.session.add(user) db.session.commit() return jsonify(user.to_dict()), 201 @app.put("/api/users/") @admin_required def update_user(user_id): user = User.query.get_or_404(user_id) payload = request.get_json(force=True) username = (payload.get("username") or "").strip() password = payload.get("password") or "" role = (payload.get("role") or "").strip() if not username: return jsonify({"error": "Username is required."}), 400 if role not in ROLE_OPTIONS: return jsonify({"error": "Role must be admin or viewer."}), 400 clash = User.query.filter( db.func.lower(User.username) == username.lower(), User.id != user_id ).first() if clash: return jsonify({"error": "That username already exists."}), 400 if user.role == "admin" and role != "admin": remaining_admins = User.query.filter(User.role == "admin", User.id != user_id).count() if remaining_admins == 0: return jsonify({"error": "At least one admin account must remain."}), 400 user.username = username user.role = role if password: user.password_hash = generate_password_hash(password) db.session.commit() # Keep the active session in sync if an admin edits their own account. if user.id == session.get("user_id"): session["username"] = user.username session["role"] = user.role return jsonify(user.to_dict()) @app.delete("/api/users/") @admin_required def delete_user(user_id): user = User.query.get_or_404(user_id) if user.id == session.get("user_id"): return jsonify({"error": "You can't delete your own account while logged in."}), 400 if user.role == "admin": remaining_admins = User.query.filter(User.role == "admin", User.id != user_id).count() if remaining_admins == 0: return jsonify({"error": "At least one admin account must remain."}), 400 db.session.delete(user) db.session.commit() return jsonify({"ok": True}) app = create_app() if __name__ == "__main__": app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8000)), debug=True)