import os import logging from logging.handlers import RotatingFileHandler from flask import Flask from werkzeug.middleware.proxy_fix import ProxyFix from flask_sqlalchemy import SQLAlchemy from flask_login import LoginManager from flask_mail import Mail from flask_migrate import Migrate from flask_socketio import SocketIO from flask_wtf.csrf import CSRFProtect from flask_limiter import Limiter from flask_limiter.util import get_remote_address from config.config import config db = SQLAlchemy() login_manager= LoginManager() mail = Mail() migrate = Migrate() socketio = SocketIO() csrf = CSRFProtect() # limiter is a module-level name so blueprints can do `from app import limiter`, # but the actual Limiter object is constructed inside create_app() — AFTER # gunicorn has called eventlet.monkey_patch() in run.py. Constructing it here # (at import time) creates a threading.RLock before the patch runs, which # triggers: "1 RLock(s) were not greened". limiter: Limiter = None # type: ignore[assignment] def create_app(config_name=None): if config_name is None: config_name = os.environ.get('FLASK_ENV', 'production') app = Flask(__name__, template_folder='templates', static_folder='static') app.config.from_object(config.get(config_name, config['default'])) # ── ProxyFix: trust one nginx proxy hop for real client IPs ─────────────── app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1) # ── Extensions ──────────────────────────────────────────────────────────── db.init_app(app) login_manager.init_app(app) mail.init_app(app) migrate.init_app(app, db) csrf.init_app(app) # Limiter is constructed here — NOT at module level — so that # eventlet.monkey_patch() (called in run.py before any imports) has already # replaced threading.RLock with a green-thread-safe version. Constructing # Limiter at module import time creates a real OS RLock before the patch # runs, which produces: "1 RLock(s) were not greened". global limiter limiter = Limiter( key_func = get_remote_address, default_limits = [], storage_uri = app.config.get('RATELIMIT_STORAGE_URI'), ) limiter.init_app(app) socketio.init_app( app, async_mode = 'eventlet', cors_allowed_origins = app.config.get('APP_BASE_URL', ''), # Ping settings: server sends a ping every 25s, client has 60s to respond. # This ensures dead connections are detected and closed cleanly rather than # being torn down by nginx timeouts, which causes [Errno 9] Bad file descriptor. ping_interval = 25, ping_timeout = 60, # Suppress eventlet's low-level socket error messages from flooding the log. # The errors still occur occasionally (it's an eventlet limitation) but are # benign — connections recover automatically via Socket.IO's reconnect logic. logger = False, engineio_logger = False, ) login_manager.login_view = 'auth.login' login_manager.login_message = 'Please log in to access this page.' login_manager.login_message_category = 'info' # ── Upload directory ────────────────────────────────────────────────────── upload_dir = app.config['UPLOAD_FOLDER'] # always absolute — set in config.py os.makedirs(upload_dir, exist_ok=True) # ── Logging ─────────────────────────────────────────────────────────────── if not app.debug: os.makedirs('logs', exist_ok=True) file_handler = RotatingFileHandler( 'logs/it_tickets.log', maxBytes=10_485_760, backupCount=10 ) file_handler.setFormatter(logging.Formatter( '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]' )) file_handler.setLevel(logging.INFO) app.logger.addHandler(file_handler) app.logger.setLevel(logging.INFO) app.logger.info('IT Ticket System startup') # ── Blueprints ──────────────────────────────────────────────────────────── from app.routes.auth import auth_bp from app.routes.tickets import tickets_bp from app.routes.admin import admin_bp from app.routes.api import api_bp from app.routes.chatbot import chatbot_bp app.register_blueprint(auth_bp) app.register_blueprint(tickets_bp) app.register_blueprint(admin_bp) app.register_blueprint(api_bp) app.register_blueprint(chatbot_bp) # ── User loader ─────────────────────────────────────────────────────────── from app.models import User @login_manager.user_loader def load_user(user_id): # db.session.get() is the SQLAlchemy 2.x successor to the deprecated # Model.query.get(). Both hit the identity map first (no SQL if the # object is already in session), then fall back to a SELECT by PK. return db.session.get(User, int(user_id)) # ── Timezone filter ────────────────────────────────────────────────────── # All datetimes in the DB are stored as UTC (naive). The localtime filter # converts them to the admin-configured display timezone for templates. # Python routes should call app_localtime(dt) when they need a local datetime. from zoneinfo import ZoneInfo, ZoneInfoNotFoundError def _get_tz(app_obj): """Return the configured ZoneInfo, falling back to UTC on bad input.""" from app.models import SystemSetting tz_name = SystemSetting.get('app_timezone', 'America/New_York') or 'America/New_York' try: return ZoneInfo(tz_name) except (ZoneInfoNotFoundError, KeyError): app_obj.logger.warning(f'[TZ] Unknown timezone {tz_name!r}, falling back to UTC') return ZoneInfo('UTC') def localtime_filter(dt, fmt='%b %d, %Y %H:%M %Z'): """Jinja2 filter: convert a naive UTC datetime to local display time.""" if dt is None: return '' from datetime import timezone as _tz tz = _get_tz(app) aware = dt.replace(tzinfo=_tz.utc) return aware.astimezone(tz).strftime(fmt) app.jinja_env.filters['localtime'] = localtime_filter # ── Context processors ──────────────────────────────────────────────────── @app.context_processor def inject_globals(): from flask import has_request_context from flask_login import current_user from app.models import SystemSetting unread = 0 # current_user is only resolvable inside an active request context. # When render_template_string() is called from a background job (e.g. # APScheduler email ingestion), there is no session to load from, so # Flask-Login returns None — causing AttributeError on .is_authenticated. # The has_request_context() guard makes the context processor safe for # both web requests and background callers. if has_request_context() and current_user.is_authenticated: from app.models import Notification unread = Notification.query.filter_by( user_id=current_user.id, is_read=False ).count() branding = { 'app_name' : SystemSetting.get('app_name', 'TechDesk'), 'app_subtitle' : SystemSetting.get('app_subtitle', 'IT Helpdesk System'), 'company_name' : SystemSetting.get('company_name', ''), 'logo_stored_name' : SystemSetting.get('logo_stored_name', ''), 'logo_initials' : SystemSetting.get('logo_initials', 'TD'), 'primary_color' : SystemSetting.get('primary_color', '#2563eb'), } from datetime import datetime as _dt, timezone as _tz from zoneinfo import ZoneInfo as _ZI from app.models import SystemSetting as _SS _tz_name = _SS.get('app_timezone', 'America/New_York') or 'America/New_York' try: _zone = _ZI(_tz_name) except Exception: _zone = _ZI('UTC') now_local = _dt.now(_tz.utc).astimezone(_zone) return dict(unread_notifications=unread, branding=branding, now_local=now_local, app_tz_name=_tz_name) # ── Security headers ────────────────────────────────────────────────────── @app.after_request def set_security_headers(response): response.headers.setdefault('X-Frame-Options', 'SAMEORIGIN') response.headers.setdefault('X-Content-Type-Options', 'nosniff') response.headers.setdefault('Referrer-Policy', 'strict-origin-when-cross-origin') response.headers.setdefault('X-XSS-Protection', '1; mode=block') response.headers.setdefault('Permissions-Policy', 'geolocation=(), microphone=(), camera=()') return response # ── DB initialisation (first run) ───────────────────────────────────────── with app.app_context(): db.create_all() _seed_admin(app) _seed_settings() _seed_ticket_templates() # ── Background schedulers (SLA + Email Ingestion) ─────────────────────────── # Both jobs share a single BackgroundScheduler instance to avoid duplicate # thread pools and simplify startup/shutdown. The scheduler is only started # in the main process — not during Flask's reloader child — to prevent # duplicate job execution. _start_background_schedulers(app) return app def _start_background_schedulers(app): """Start a single shared APScheduler instance hosting both the SLA breach check and the inbound email ingestion job. A single BackgroundScheduler is used rather than two separate instances to avoid creating duplicate thread pools under eventlet. Both jobs are registered here so they share one daemon thread pool. The scheduler is skipped in Flask's reloader subprocess (identified by the WERKZEUG_RUN_MAIN env var) to prevent double-registration on hot-reloads. Under gunicorn with preload_app=False, each worker calls create_app() once, giving exactly one scheduler per worker. """ import os as _os if _os.environ.get('WERKZEUG_RUN_MAIN') == 'true': return # skip reloader child process try: from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.cron import CronTrigger from app.services.sla_service import check_sla_breaches from app.services.email_ingestion_service import check_inbound_email from app.services.notification_service import send_weekly_digest from app.models import SystemSetting with app.app_context(): email_interval = int(SystemSetting.get('email_ingestion_interval', '5') or '5') scheduler = BackgroundScheduler(daemon=True) # SLA breach check — every 30 minutes scheduler.add_job( func = check_sla_breaches, trigger = IntervalTrigger(minutes=30), id = 'sla_check', name = 'SLA Breach Check', replace_existing = True, args = [app], ) # Inbound email ingestion — interval from SystemSetting scheduler.add_job( func = check_inbound_email, trigger = IntervalTrigger(minutes=email_interval), id = 'email_ingest', name = 'Inbound Email Ingestion', replace_existing = True, args = [app], ) # Weekly digest — every Monday at 08:00 UTC scheduler.add_job( func = send_weekly_digest, trigger = CronTrigger(day_of_week='mon', hour=8, minute=0), id = 'weekly_digest', name = 'Weekly IT Digest Email', replace_existing = True, args = [app], ) scheduler.start() app.logger.info('[SCHEDULER] APScheduler started — SLA check every 30 min, ' f'email ingestion every {email_interval} min, ' 'weekly digest every Monday 08:00 UTC') except Exception as exc: app.logger.error(f'[SCHEDULER] Failed to start APScheduler: {exc}') def _seed_ticket_templates(): """Seed a starter set of ticket templates if none exist.""" from app.models import TicketTemplate if TicketTemplate.query.first(): return # already seeded defaults = [ dict(name='VPN / Remote Access Issue', category='network', priority='high', icon='bi-shield-lock', title_hint='Cannot connect to VPN', description='Steps to reproduce:\n1. \n\nError message:\n\nOperating system:\n\nLast time it worked:', sort_order=1), dict(name='New Software Request', category='software', priority='low', icon='bi-box-arrow-in-down', title_hint='Software installation request — ', description='Software name and version:\n\nBusiness justification:\n\nApproved by (manager):', sort_order=2), dict(name='Password / Account Access', category='access', priority='medium', icon='bi-key', title_hint='Cannot log in to ', description='System / application:\n\nError message:\n\nLast successful login:', sort_order=3), dict(name='Hardware Issue', category='hardware', priority='medium', icon='bi-pc-display', title_hint='Hardware problem — ', description='Device type and asset tag:\n\nSymptoms:\n\nWhen did it start:', sort_order=4), dict(name='New Employee Onboarding', category='access', priority='high', icon='bi-person-plus', title_hint='New employee setup — ', description='Employee name:\nStart date:\nDepartment:\nManager:\n\nAccounts needed:\n- Email\n- VPN\n- Other:', sort_order=5), dict(name='Printer / Scanner Issue', category='printer', priority='low', icon='bi-printer', title_hint='Printer not working — ', description='Printer name / location:\n\nError message:\n\nComputer OS:', sort_order=6), ] for d in defaults: db.session.add(TicketTemplate(**d, is_active=True)) db.session.commit() def _seed_settings(): """Ensure all required system settings exist with safe defaults.""" from app.models import SystemSetting defaults = [ ('registration_enabled', 'true', 'Allow new users to self-register via /auth/register'), ('survey_enabled', 'true', 'Send a satisfaction survey email when a ticket is resolved'), ('app_timezone', 'America/New_York', 'Display timezone for all dates and times in the UI'), ('sla_critical_hours', '4', 'Hours before a CRITICAL ticket is considered overdue'), ('email_ingestion_enabled', '0', 'Enable automatic ticket creation from inbound email (1=on, 0=off)'), ('email_ingestion_host', '', 'IMAP server hostname (e.g. imap.gmail.com)'), ('email_ingestion_port', '993', 'IMAP SSL port'), ('email_ingestion_user', '', 'Mailbox username / email address'), ('email_ingestion_password', '', 'Mailbox password (stored in plaintext — use a dedicated app password)'), ('email_ingestion_folder', 'INBOX', 'IMAP folder to watch for new mail'), ('email_ingestion_move_to', 'Processed', 'IMAP folder to move processed mail into'), ('email_ingestion_interval', '5', 'Poll interval in minutes'), ('sla_high_hours', '8', 'Hours before a HIGH ticket is considered overdue'), ('sla_medium_hours', '48', 'Hours before a MEDIUM ticket is considered overdue'), ('sla_low_hours', '120', 'Hours before a LOW ticket is considered overdue'), ('app_name', 'TechDesk', 'Application name shown in the sidebar and page titles'), ('app_subtitle', 'IT Helpdesk System', 'Subtitle shown below the app name in the sidebar'), ('company_name', '', 'Company name shown on the login and register pages'), ('logo_stored_name', '', 'Stored filename of the uploaded company logo image'), ('logo_initials', 'TD', 'Two-letter initials shown when no logo is uploaded'), ('primary_color', '#2563eb', 'Primary accent colour (hex) used across the interface'), ] for key, value, description in defaults: if SystemSetting.get(key) is None: SystemSetting.set(key, value, description) db.session.commit() def _seed_admin(app): """Create the default admin account if none exists. Username collision guard ----------------------- The seed username is hardcoded to 'admin'. If a user registered with that username before the first admin seed runs (possible when registration_enabled=True on a fresh install), the INSERT would raise an IntegrityError and break application startup. We guard against this by checking for username conflicts independently of the role check, and falling back to a derived username when 'admin' is already taken. """ from app.models import User, UserRole if User.query.filter_by(role=UserRole.ADMIN).first(): return # Resolve a safe username — 'admin' is preferred but may already be taken. seed_username = 'admin' if User.query.filter_by(username=seed_username).first(): # Derive a unique fallback so startup never fails on a collision. import uuid seed_username = f'admin_{uuid.uuid4().hex[:6]}' app.logger.warning( f'[SEED] Username "admin" is already taken — ' f'seeding admin account with username "{seed_username}". ' f'Rename via Admin → Users after first login.' ) admin = User( email = app.config['ADMIN_EMAIL'], username = seed_username, full_name = 'System Administrator', role = UserRole.ADMIN, department= 'IT', is_active = True, ) admin.set_password(app.config['ADMIN_PASSWORD']) db.session.add(admin) db.session.commit() app.logger.info(f'[SEED] Default admin account created: {admin.email} (username={seed_username})')