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_login import current_user from app.models import SystemSetting unread = 0 if 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) # ── DB initialisation (first run) ───────────────────────────────────────── with app.app_context(): db.create_all() _seed_admin(app) _seed_settings() _seed_ticket_templates() # ── Email ingestion background scheduler ──────────────────────────────────── _start_email_ingestion_scheduler(app) # ── SLA background scheduler ────────────────────────────────────────────── # APScheduler runs inside the gunicorn worker process (single-worker # eventlet setup), so no cross-process coordination is needed. # The scheduler is only started in the main process — not during Flask's # reloader child process — to prevent duplicate job execution. _start_sla_scheduler(app) return app def _start_sla_scheduler(app): """Start the APScheduler background job that checks for SLA breaches. Safe to call on every app startup: the scheduler is idempotent and jobstore deduplication prevents double-registration on hot-reloads. Under gunicorn with preload_app=False each worker calls create_app() once, so there is exactly one scheduler per worker. """ # Skip inside Flask's reloader subprocess (identified by the env var it sets). import os as _os if _os.environ.get('WERKZEUG_RUN_MAIN') == 'true': # Reloader is active — the child process will start its own scheduler. return try: from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from app.services.sla_service import check_sla_breaches scheduler = BackgroundScheduler(daemon=True) scheduler.add_job( func = check_sla_breaches, trigger = IntervalTrigger(minutes=30), id = 'sla_check', name = 'SLA Breach Check', replace_existing = True, args = [app], ) scheduler.start() app.logger.info('[SLA] APScheduler started — SLA breach check every 30 minutes') except Exception as exc: app.logger.error(f'[SLA] Failed to start APScheduler: {exc}') def _start_email_ingestion_scheduler(app): """Start the APScheduler job that polls the inbound mailbox for new emails. The interval is read from SystemSetting at job creation time (default 5 min). The job is a no-op when email_ingestion_enabled = '0', so it is safe to always register it — no credentials are required until the admin enables it. """ 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 app.services.email_ingestion_service import check_inbound_email from app.models import SystemSetting with app.app_context(): interval = int(SystemSetting.get('email_ingestion_interval', '5') or '5') scheduler = BackgroundScheduler(daemon=True) scheduler.add_job( func = check_inbound_email, trigger = IntervalTrigger(minutes=interval), id = 'email_ingest', name = 'Inbound Email Ingestion', replace_existing = True, args = [app], ) scheduler.start() app.logger.info( f'[EMAIL INGEST] APScheduler started — polling every {interval} minute(s)' ) except Exception as exc: app.logger.error(f'[EMAIL INGEST] Failed to start scheduler: {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})')