200 lines
8.9 KiB
Python
200 lines
8.9 KiB
Python
"""
|
|
app.py — Flask application factory.
|
|
Web conversion of the desktop Bid Checker application.
|
|
Stack: Python 3.11+, Flask, MySQL (mysql-connector-python), Gunicorn, Nginx.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from flask import Flask
|
|
from config import initialize_database, db_log_handler, _stream_handler
|
|
|
|
# ─── Logging ──────────────────────────────────────────────────────────────────
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
handlers=[_stream_handler, db_log_handler],
|
|
)
|
|
logger = logging.getLogger("app")
|
|
|
|
|
|
def _run_reminder_job():
|
|
"""Background job: send shift-incomplete reminders if within the reminder window."""
|
|
from routes.admin_shifts import do_send_incomplete_reminders
|
|
try:
|
|
result = do_send_incomplete_reminders(triggered_by_user_id=None)
|
|
if result["sent"] > 0:
|
|
logger.info(f"[scheduler] Shift reminders sent: {result}")
|
|
except Exception as e:
|
|
logger.error(f"[scheduler] Shift reminder job error: {e}")
|
|
|
|
|
|
def _start_scheduler():
|
|
"""
|
|
Start a background scheduler that fires shift-incomplete reminder emails
|
|
every 5 minutes. Guards against double-start in Flask's debug reloader.
|
|
Uses daemon threads so the process exits cleanly even if the scheduler is running.
|
|
"""
|
|
if os.environ.get("WERKZEUG_RUN_MAIN") == "false":
|
|
# Flask reloader: skip in the parent monitor process; the child sets WERKZEUG_RUN_MAIN=true
|
|
return
|
|
try:
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
import atexit
|
|
|
|
scheduler = BackgroundScheduler(daemon=True)
|
|
scheduler.add_job(
|
|
func=_run_reminder_job,
|
|
trigger="interval",
|
|
minutes=5,
|
|
id="shift_incomplete_reminders",
|
|
max_instances=1,
|
|
misfire_grace_time=120,
|
|
)
|
|
scheduler.start()
|
|
atexit.register(lambda: scheduler.shutdown(wait=False))
|
|
logger.info("[scheduler] Background scheduler started — shift reminders every 5 min.")
|
|
except Exception as e:
|
|
logger.error(f"[scheduler] Failed to start background scheduler: {e}")
|
|
|
|
|
|
def create_app():
|
|
app = Flask(__name__)
|
|
|
|
# ── Secret key for session management ─────────────────────────────────────
|
|
secret_key = os.environ.get("SECRET_KEY")
|
|
if not secret_key:
|
|
logger.warning(
|
|
"SECRET_KEY not set in environment. Sessions will break across "
|
|
"Gunicorn workers. Set SECRET_KEY in .env before deploying."
|
|
)
|
|
app.secret_key = secret_key or os.urandom(32)
|
|
|
|
# ── Upload size limit (20 MB) ──────────────────────────────────────────────
|
|
app.config["MAX_CONTENT_LENGTH"] = 20 * 1024 * 1024
|
|
|
|
# ── Session timeout (30 minutes) ──────────────────────────────────────────
|
|
from datetime import timedelta
|
|
app.permanent_session_lifetime = timedelta(minutes=30)
|
|
|
|
# ── Database initialisation ────────────────────────────────────────────────
|
|
try:
|
|
initialize_database()
|
|
db_log_handler.install()
|
|
logger.info("Database initialised successfully.")
|
|
except Exception as e:
|
|
logger.error(f"Database initialisation failed: {e}")
|
|
|
|
# ── Register Blueprints ────────────────────────────────────────────────────
|
|
from routes.auth import auth_bp
|
|
from routes.admin_dashboard import admin_dashboard_bp
|
|
from routes.admin_users import admin_users_bp
|
|
from routes.admin_websites import admin_websites_bp
|
|
from routes.admin_shifts import admin_shifts_bp
|
|
from routes.admin_logs import admin_logs_bp
|
|
from routes.admin_reports import admin_reports_bp
|
|
from routes.admin_settings import admin_settings_bp
|
|
from routes.user_dashboard import user_dashboard_bp
|
|
from routes.ai_summary import ai_summary_bp
|
|
from routes.bid_tracker import bid_tracker_bp
|
|
from routes.internal import internal_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
app.register_blueprint(admin_dashboard_bp)
|
|
app.register_blueprint(admin_users_bp)
|
|
app.register_blueprint(admin_websites_bp)
|
|
app.register_blueprint(admin_shifts_bp)
|
|
app.register_blueprint(admin_logs_bp)
|
|
app.register_blueprint(admin_reports_bp)
|
|
app.register_blueprint(admin_settings_bp)
|
|
app.register_blueprint(user_dashboard_bp)
|
|
app.register_blueprint(ai_summary_bp)
|
|
app.register_blueprint(bid_tracker_bp)
|
|
app.register_blueprint(internal_bp)
|
|
|
|
if not os.environ.get("CRON_SECRET"):
|
|
logger.warning(
|
|
"CRON_SECRET not set in .env — the /internal/cron/* endpoints are disabled."
|
|
)
|
|
|
|
# ── Background scheduler: automatic shift-incomplete reminders ────────────
|
|
_start_scheduler()
|
|
|
|
# ── CSRF protection (Flask-WTF) ────────────────────────────────────────────
|
|
from flask_wtf.csrf import CSRFProtect
|
|
CSRFProtect(app)
|
|
|
|
# ── Security response headers ──────────────────────────────────────────────
|
|
from flask import session, redirect, url_for, g, request, jsonify, flash
|
|
import functools
|
|
|
|
@app.after_request
|
|
def set_security_headers(response):
|
|
response.headers["X-Frame-Options"] = "SAMEORIGIN"
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
|
response.headers["Content-Security-Policy"] = (
|
|
"default-src 'self'; "
|
|
"script-src 'self' 'unsafe-inline' https://cdnjs.cloudflare.com; "
|
|
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
|
|
"font-src 'self' https://fonts.gstatic.com; "
|
|
"img-src 'self' data:; "
|
|
"connect-src 'self'; "
|
|
"object-src 'none'; "
|
|
"base-uri 'self'; "
|
|
"form-action 'self'; "
|
|
"frame-ancestors 'self'"
|
|
)
|
|
return response
|
|
|
|
# ── 413 handler: file too large ────────────────────────────────────────────
|
|
from werkzeug.exceptions import RequestEntityTooLarge
|
|
|
|
@app.errorhandler(RequestEntityTooLarge)
|
|
def handle_too_large(e):
|
|
max_mb = app.config.get("MAX_CONTENT_LENGTH", 0) // (1024 * 1024)
|
|
msg = f"Upload too large. Maximum allowed size is {max_mb} MB."
|
|
if request.path.startswith("/ai-summary/"):
|
|
return jsonify({"error": msg}), 413
|
|
flash(msg, "danger")
|
|
return redirect(request.referrer or url_for("index")), 413
|
|
|
|
# ── Template context processors ────────────────────────────────────────────
|
|
|
|
# ── Custom template filter: timedelta/time → "HH:MM" ──────────────────────
|
|
import datetime as _dt
|
|
|
|
@app.template_filter('hhmm')
|
|
def hhmm_filter(value):
|
|
"""Render a MySQL TIME value (timedelta, time, or str) as HH:MM."""
|
|
if value is None:
|
|
return ''
|
|
if isinstance(value, _dt.timedelta):
|
|
total = int(value.total_seconds())
|
|
h, rem = divmod(abs(total), 3600)
|
|
m = rem // 60
|
|
return f"{h:02d}:{m:02d}"
|
|
if isinstance(value, _dt.time):
|
|
return value.strftime('%H:%M')
|
|
return str(value)[:5] # already a string like "08:00:00"
|
|
|
|
@app.context_processor
|
|
def inject_user():
|
|
return {"current_user": session.get("user")}
|
|
|
|
# ── Root redirect ──────────────────────────────────────────────────────────
|
|
@app.route("/")
|
|
def index():
|
|
if "user" in session:
|
|
user = session["user"]
|
|
if user["role"] == "admin":
|
|
return redirect(url_for("admin_dashboard.dashboard"))
|
|
return redirect(url_for("user_dashboard.my_shifts"))
|
|
return redirect(url_for("auth.login"))
|
|
|
|
return app
|
|
|
|
|
|
# ─── Dev entry point ──────────────────────────────────────────────────────────
|
|
if __name__ == "__main__":
|
|
application = create_app()
|
|
application.run(debug=False, host="0.0.0.0", port=5000) |