""" Flask 애플리케이션 초기화 """ from flask import Flask, request, send_from_directory, jsonify from flask_login import LoginManager import sqlite3 from pathlib import Path from typing import Optional from sqlalchemy import create_engine, text, inspect from app.database import db, User from app.core.config import Config, get_config from app.core.logger import get_logger logger = get_logger(__name__) login_manager = LoginManager() login_manager.login_view = 'main.login' login_manager.login_message = '로그인이 필요합니다.' login_manager.login_message_category = 'info' @login_manager.unauthorized_handler def unauthorized(): """인증되지 않은 사용자 처리""" from flask import redirect, url_for, request # API 요청인 경우 JSON 응답 if request.path.startswith('/api/'): from flask import jsonify return jsonify({'error': '로그인이 필요합니다.'}), 401 # 일반 요청인 경우 로그인 페이지로 리다이렉트 return redirect(url_for('main.login', next=request.path)) @login_manager.user_loader def load_user(user_id: str) -> Optional[User]: """ 사용자 로드 함수 (Flask-Login용) Args: user_id: 사용자 ID (문자열) Returns: User 객체 또는 None """ try: return User.query.get(int(user_id)) except (ValueError, TypeError): return None def create_app() -> Flask: """ Flask 애플리케이션 팩토리 함수 Returns: 설정된 Flask 애플리케이션 인스턴스 """ config = get_config() # 필수 디렉토리 생성 config.ensure_directories() # 템플릿 폴더 및 static 폴더 경로 설정 template_folder = str(config.TEMPLATES_FOLDER) static_folder = str(config.STATIC_FOLDER) app = Flask(__name__, template_folder=template_folder, static_folder=static_folder) # Flask 설정 적용 app.config['SECRET_KEY'] = config.SECRET_KEY app.config['SQLALCHEMY_DATABASE_URI'] = config.SQLALCHEMY_DATABASE_URI app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = config.SQLALCHEMY_TRACK_MODIFICATIONS app.config['MAX_CONTENT_LENGTH'] = config.MAX_CONTENT_LENGTH # SQLAlchemy 연결 풀 설정 (클라우드 환경에서 유휴 연결 끊김 방지) # pool_pre_ping: 연결 사용 전에 살아있는지 확인 (ping) # pool_recycle: 연결을 재사용하기 전 최대 시간(초) - 300초(5분) app.config['SQLALCHEMY_ENGINE_OPTIONS'] = { 'pool_pre_ping': True, 'pool_recycle': 300 } # 세션 쿠키 설정 (브라우저 호환성 개선) app.config['SESSION_COOKIE_SECURE'] = True # HTTPS에서만 전송 app.config['SESSION_COOKIE_HTTPONLY'] = True # JavaScript 접근 방지 app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' # CSRF 방지 및 브라우저 호환성 app.config['PERMANENT_SESSION_LIFETIME'] = 86400 # 24시간 # 확장 초기화 db.init_app(app) login_manager.init_app(app) # Blueprint 등록 from app.routes import main_bp app.register_blueprint(main_bp) # 등록된 라우트 확인 (디버깅용) with app.app_context(): logger.info(f"등록된 라우트 수: {len([r for r in app.url_map.iter_rules()])}") logger.info(f"등록된 Blueprint: {list(app.blueprints.keys())}") # 주요 라우트 확인 routes = [str(r) for r in app.url_map.iter_rules() if r.endpoint.startswith('main.')] logger.info(f"등록된 main 라우트: {routes[:10]}...") # 처음 10개만 # favicon.ico 핸들러 추가 @app.route('/favicon.ico') def favicon(): """favicon.ico 요청 처리""" try: return send_from_directory( str(config.STATIC_FOLDER), 'logo.webp', mimetype='image/webp' ) except Exception as e: logger.warning(f"favicon.ico 처리 실패: {e}") return '', 204 # No Content # 404 에러 핸들러 (일시적으로 비활성화하여 디버깅) # 라우트가 제대로 등록되었는지 확인 후 다시 활성화 # @app.errorhandler(404) # def not_found(error): # """404 에러 처리""" # logger.warning(f"404 에러: {request.path} - {request.method}") # if request.path.startswith('/api/'): # return jsonify({'error': '리소스를 찾을 수 없습니다.', 'path': request.path}), 404 # return '', 404 # 요청 로깅 미들웨어 추가 @app.before_request def log_request_info(): """각 HTTP 요청 정보를 로깅""" logger.info(f"[요청] {request.method} {request.path} - IP: {request.remote_addr}") if request.args: logger.debug(f"[요청 파라미터] {dict(request.args)}") @app.after_request def log_response_info(response): """각 HTTP 응답 정보를 로깅""" logger.info(f"[응답] {request.method} {request.path} - 상태: {response.status_code}") # Permissions-Policy 헤더 추가 (최신 브라우저에서 지원하는 기능만 사용) # 인식되지 않는 기능들은 제거하여 경고 방지 # 최신 표준에 맞는 기능들만 포함 permissions_policy = ( "camera=(), " "microphone=(), " "geolocation=(), " "payment=(), " "usb=()" ) response.headers['Permissions-Policy'] = permissions_policy # 브라우저 캐시 제어 헤더 추가 (브라우저 호환성 개선) # HTML 페이지는 캐시하지 않도록 설정하여 브라우저별 차이 최소화 if request.path == '/' or request.path.startswith('/login') or request.path.startswith('/admin') or request.path.startswith('/webnovels'): response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate' response.headers['Pragma'] = 'no-cache' response.headers['Expires'] = '0' return response # 데이터베이스 초기화 및 마이그레이션 with app.app_context(): # 데이터베이스 연결 정보 로깅 db_uri = app.config['SQLALCHEMY_DATABASE_URI'] if db_uri.startswith('postgresql://') or db_uri.startswith('postgres://'): # PostgreSQL 연결 정보 (보안을 위해 비밀번호는 마스킹) masked_uri = db_uri.split('@')[0].split('://')[0] + '://***@' + '@'.join(db_uri.split('@')[1:]) if '@' in db_uri else db_uri logger.info(f"[데이터베이스] PostgreSQL 연결 시도: {masked_uri}") # 연결 테스트 try: engine = create_engine( db_uri, pool_pre_ping=True, pool_recycle=300 ) with engine.connect() as conn: result = conn.execute(text("SELECT version()")) version = result.fetchone()[0] logger.info(f"[데이터베이스] PostgreSQL 연결 성공! 버전: {version[:50]}...") except Exception as e: logger.error(f"[데이터베이스] PostgreSQL 연결 실패: {str(e)}") logger.warning("[데이터베이스] SQLite로 폴백합니다.") # SQLite로 폴백 db_uri = f'sqlite:///{config.INSTANCE_FOLDER / "finance_analysis.db"}' app.config['SQLALCHEMY_DATABASE_URI'] = db_uri else: logger.info(f"[데이터베이스] SQLite 사용: {db_uri}") # 데이터베이스 테이블 생성 try: db.create_all() logger.info("[데이터베이스] 테이블 생성 완료") except Exception as e: logger.error(f"[데이터베이스] 테이블 생성 실패: {str(e)}") raise # 마이그레이션 실행 (SQLite만 지원) migrate_database(app) # 관리자 계정 생성 create_admin_user() logger.info("Flask 애플리케이션이 초기화되었습니다.") return app def migrate_database(app: Flask) -> None: """ 데이터베이스 마이그레이션 실행 (SQLAlchemy Inspector 사용) Args: app: Flask 애플리케이션 인스턴스 """ try: logger.info("데이터베이스 마이그레이션 확인 중...") with app.app_context(): inspector = inspect(db.engine) # 1. user 테이블 - nickname, must_change_password 컬럼 if inspector.has_table('user'): columns = [c['name'] for c in inspector.get_columns('user')] with db.engine.begin() as conn: if 'nickname' not in columns: logger.info("user 테이블에 nickname 컬럼 추가 중...") conn.execute(text("ALTER TABLE \"user\" ADD COLUMN nickname VARCHAR(80)")) if 'must_change_password' not in columns: logger.info("user 테이블에 must_change_password 컬럼 추가 중...") # Boolean 타입은 DB에 따라 다를 수 있으므로 주의 (PostgreSQL/SQLite 모두 BOOLEAN 지원) conn.execute(text("ALTER TABLE \"user\" ADD COLUMN must_change_password BOOLEAN DEFAULT FALSE NOT NULL")) # 2. uploaded_file 테이블 - uploaded_by, parent_file_id, tags 컬럼 if inspector.has_table('uploaded_file'): columns = [c['name'] for c in inspector.get_columns('uploaded_file')] with db.engine.begin() as conn: if 'uploaded_by' not in columns: logger.info("uploaded_file 테이블에 uploaded_by 컬럼 추가 중...") conn.execute(text("ALTER TABLE uploaded_file ADD COLUMN uploaded_by INTEGER")) if 'parent_file_id' not in columns: logger.info("uploaded_file 테이블에 parent_file_id 컬럼 추가 중...") conn.execute(text("ALTER TABLE uploaded_file ADD COLUMN parent_file_id INTEGER")) if 'tags' not in columns: logger.info("uploaded_file 테이블에 tags 컬럼 추가 중...") conn.execute(text("ALTER TABLE uploaded_file ADD COLUMN tags TEXT")) # 3. document_chunk 테이블 - chunk_metadata 컬럼 if inspector.has_table('document_chunk'): columns = [c['name'] for c in inspector.get_columns('document_chunk')] if 'chunk_metadata' not in columns: logger.info("document_chunk 테이블에 chunk_metadata 컬럼 추가 중...") with db.engine.begin() as conn: conn.execute(text("ALTER TABLE document_chunk ADD COLUMN chunk_metadata TEXT")) # 4. chat_session 테이블 - analysis_model, answer_model 컬럼 if inspector.has_table('chat_session'): columns = [c['name'] for c in inspector.get_columns('chat_session')] with db.engine.begin() as conn: if 'analysis_model' not in columns: logger.info("chat_session 테이블에 analysis_model 컬럼 추가 중...") conn.execute(text("ALTER TABLE chat_session ADD COLUMN analysis_model VARCHAR(100)")) if 'answer_model' not in columns: logger.info("chat_session 테이블에 answer_model 컬럼 추가 중...") conn.execute(text("ALTER TABLE chat_session ADD COLUMN answer_model VARCHAR(100)")) # 5. chat_message 테이블 - 토큰 정보 및 모델 정보 컬럼 (PostgreSQL 오류 해결용) if inspector.has_table('chat_message'): columns = [c['name'] for c in inspector.get_columns('chat_message')] with db.engine.begin() as conn: # session_id 컬럼의 NOT NULL 제약조건 제거 (PostgreSQL) if db.engine.dialect.name == 'postgresql': logger.info("PostgreSQL: chat_message.session_id의 NOT NULL 제약조건 제거 시도...") conn.execute(text("ALTER TABLE chat_message ALTER COLUMN session_id DROP NOT NULL")) if 'input_tokens' not in columns: logger.info("chat_message 테이블에 input_tokens 컬럼 추가 중...") conn.execute(text("ALTER TABLE chat_message ADD COLUMN input_tokens INTEGER")) if 'output_tokens' not in columns: logger.info("chat_message 테이블에 output_tokens 컬럼 추가 중...") conn.execute(text("ALTER TABLE chat_message ADD COLUMN output_tokens INTEGER")) if 'model_name' not in columns: logger.info("chat_message 테이블에 model_name 컬럼 추가 중...") conn.execute(text("ALTER TABLE chat_message ADD COLUMN model_name VARCHAR(100)")) if 'usage_type' not in columns: logger.info("chat_message 테이블에 usage_type 컬럼 추가 중...") conn.execute(text("ALTER TABLE chat_message ADD COLUMN usage_type VARCHAR(20) DEFAULT 'user'")) logger.info("데이터베이스 마이그레이션 완료") except Exception as e: logger.error(f"데이터베이스 마이그레이션 중 오류 발생: {e}", exc_info=True) def create_admin_user() -> None: """ 초기 관리자 계정 생성 """ admin_username = 'soymedia' admin_password = 's0ymedi@1@34' try: admin = User.query.filter_by(username=admin_username).first() if not admin: admin = User(username=admin_username, is_admin=True, is_active=True) admin.set_password(admin_password) db.session.add(admin) db.session.commit() logger.info(f'관리자 계정이 생성되었습니다: {admin_username}') else: logger.debug(f'관리자 계정이 이미 존재합니다: {admin_username}') except Exception as e: logger.error(f'관리자 계정 생성 중 오류 발생: {e}', exc_info=True) db.session.rollback()