""" Flask 애플리케이션 초기화 """ import os import time from flask import Flask, request, send_from_directory, jsonify, render_template, flash, redirect, url_for, session from flask_login import LoginManager from app.i18n import gettext import sqlite3 from pathlib import Path from typing import Optional from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode from sqlalchemy import create_engine, text, inspect from sqlalchemy.exc import OperationalError as SAOperationalError from app.database import db, User from app.migrations import check_and_migrate_db 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 """ # DB 장애 모드에서는 로그인 세션이 있어도 사용자 로드를 시도하지 않음 # (에러 페이지 렌더링 중 current_user 로딩이 DB를 다시 건드려 500으로 터지는 것을 방지) try: from flask import current_app if current_app and current_app.config.get('DB_UNAVAILABLE'): return None except Exception: # current_app 접근이 불가능한 경우는 그대로 진행 pass try: return User.query.get(int(user_id)) except Exception as e: # DB 연결/쿼리 장애 시에는 사용자 로드를 포기하고 None 처리 # (flask-login context processor가 error page 렌더링을 막지 않게) try: from sqlalchemy.exc import OperationalError as SAOperationalError if isinstance(e, SAOperationalError): try: from flask import current_app if current_app: current_app.config['DB_UNAVAILABLE'] = True current_app.config['DB_UNAVAILABLE_REASON'] = f"load_user: {e}" except Exception: pass return None except Exception: pass raise 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 # DB 장애 모드 플래그 (Railway DB 장애/지연 시 에러 페이지 노출용) app.config.setdefault('DB_UNAVAILABLE', False) app.config.setdefault('DB_UNAVAILABLE_REASON', None) def _mark_db_unavailable(err: Exception, phase: str) -> None: """DB 연결/쿼리 장애를 감지하면 서버는 살려두고 '장애 모드'로 전환""" app.config['DB_UNAVAILABLE'] = True app.config['DB_UNAVAILABLE_REASON'] = f"{phase}: {err}" logger.error(f"[데이터베이스] DB 장애 감지({phase}): {err}") def _db_error_response(): """ DB 장애 시 사용자에게 노출할 표준 응답. - API는 JSON(503) - 페이지는 HTML 템플릿(503) """ lang = session.get('lang', 'vi') message = gettext('db_error', lang=lang) if request.path.startswith('/api/'): resp = jsonify({"error": message}) resp.status_code = 503 resp.headers['Retry-After'] = '30' return resp return render_template('db_connection_error.html', message=message), 503 def _is_postgres_uri(uri: str) -> bool: return uri.startswith('postgresql://') or uri.startswith('postgres://') def _is_sqlite_uri(uri: str) -> bool: # flask-sqlalchemy/sqlalchemy에서 흔히 쓰는 sqlite URI 형식들 허용 return uri.startswith('sqlite:///') or uri.startswith('sqlite://') def _normalize_postgres_uri(uri: str) -> str: """ - postgres:// -> postgresql:// 로 정규화 - DATABASE_SSLMODE 환경변수 또는 URL query의 sslmode를 반영 - (휴리스틱) Railway 등 일부 호스트는 SSL을 요구하므로, sslmode가 없으면 require를 기본 적용 """ if uri.startswith('postgres://'): uri = 'postgresql://' + uri[len('postgres://'):] sslmode_env = (os.environ.get('DATABASE_SSLMODE') or '').strip() parsed = urlparse(uri) q = dict(parse_qsl(parsed.query, keep_blank_values=True)) # 호스트 기반 기본값: 외부 managed DB는 SSL 요구 케이스가 많아 require를 기본으로 두는 것이 안전함 hostname = (parsed.hostname or '').lower() default_sslmode = '' if hostname.endswith('rlwy.net') or 'proxy.rlwy.net' in hostname or hostname.endswith('neon.tech') or hostname.endswith('supabase.co'): default_sslmode = 'require' # sslmode가 이미 있으면 존중, 없으면 환경변수 기반으로만 추가 if 'sslmode' not in q: if sslmode_env: q['sslmode'] = sslmode_env elif default_sslmode: q['sslmode'] = default_sslmode new_query = urlencode(q) if q else '' return urlunparse(parsed._replace(query=new_query)) # NOTE: # - 로컬/개발 환경에서는 Config 기본값(SQLite)을 정상 허용합니다. # - 운영에서 "PostgreSQL만 강제"하고 싶다면 환경변수로 strict 모드를 켭니다. # (예: REQUIRE_POSTGRES=1) def _require_postgres_only() -> bool: v = (os.environ.get('REQUIRE_POSTGRES') or os.environ.get('DB_STRICT_MODE') or '').strip().lower() return v in ('1', 'true', 'yes', 'y', 'on') def _retry_sleep_seconds(attempt_idx: int) -> float: """ DB 연결이 순간적으로 느린 환경(HF↔Railway)에서 바로 죽지 않도록 지수 backoff 적용. attempt_idx: 0-based """ base = float(os.environ.get('DB_CONNECT_RETRY_SLEEP', '2')) max_sleep = float(os.environ.get('DB_CONNECT_RETRY_MAX_SLEEP', '10')) # 2, 4, 8 ... (max cap) return min(max_sleep, base * (2 ** attempt_idx)) # SQLAlchemy 연결 풀/접속 옵션 (클라우드 환경에서 유휴 연결 끊김 방지 + 빠른 실패) engine_options = { 'pool_pre_ping': True, 'pool_recycle': int(os.environ.get('DB_POOL_RECYCLE', '300')), } # HF/Railway 환경에서 초기 핸드셰이크/SSL이 느릴 수 있어 기본값을 조금 넉넉히 둠 connect_timeout = int(os.environ.get('DB_CONNECT_TIMEOUT', '20')) # IMPORTANT: Flask-SQLAlchemy는 init 시점에 엔진을 만들 수 있으므로, # DB URI/옵션을 "확정"한 뒤 db.init_app(app)을 호출해야 합니다. effective_db_uri = app.config['SQLALCHEMY_DATABASE_URI'] if _is_postgres_uri(effective_db_uri): effective_db_uri = _normalize_postgres_uri(effective_db_uri) app.config['SQLALCHEMY_DATABASE_URI'] = effective_db_uri # psycopg2 연결 옵션: URL query sslmode 외에 connect_timeout은 connect_args로 강제 engine_options['connect_args'] = {'connect_timeout': connect_timeout} # 부팅 시점에 연결 테스트(재시도 포함). 실패하면 (설정에 따라) SQLite로 폴백하거나 종료. masked_uri = effective_db_uri.split('@')[0].split('://')[0] + '://***@' + '@'.join(effective_db_uri.split('@')[1:]) if '@' in effective_db_uri else effective_db_uri max_attempts = int(os.environ.get('DB_CONNECT_RETRIES', '3')) last_err: Exception | None = None for i in range(max_attempts): try: logger.info(f"[데이터베이스] PostgreSQL 연결 테스트: {masked_uri} (attempt {i+1}/{max_attempts})") test_engine = create_engine(effective_db_uri, **engine_options) with test_engine.connect() as conn: conn.execute(text("SELECT 1")) logger.info("[데이터베이스] PostgreSQL 연결 테스트 성공") last_err = None break except Exception as e: last_err = e logger.warning(f"[데이터베이스] PostgreSQL 연결 테스트 실패(재시도 예정): {e}") if i < max_attempts - 1: time.sleep(_retry_sleep_seconds(i)) if last_err is not None: logger.error(f"[데이터베이스] PostgreSQL 연결 테스트 최종 실패: {last_err}") # SQLite 폴백은 하지 않고, 서버는 살려두고 에러 페이지로 안내 _mark_db_unavailable(last_err, phase="connect_test") else: # SQLite는 로컬 기본값으로 허용 (strict 모드가 아니면 정상 부팅) if _is_sqlite_uri(effective_db_uri) and not _require_postgres_only(): try: logger.info(f"[데이터베이스] SQLite 사용: {effective_db_uri}") except Exception: # 로깅 실패는 무시 (부팅에 영향 주지 않음) pass else: # 외부 DB(DATABASE_URL)가 없거나 postgres 형식이 아닌 경우: # strict 모드에서는 DB 사용 불가로 간주하여 장애 모드로 전환 _mark_db_unavailable( ValueError("DATABASE_URL이 없거나 PostgreSQL 형식이 아닙니다. (REQUIRE_POSTGRES=1이면 SQLite 사용 금지)"), phase="invalid_db_uri", ) app.config['SQLALCHEMY_ENGINE_OPTIONS'] = engine_options # 세션 쿠키 설정 (브라우저 호환성 개선) 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 URI/옵션 확정 이후) db.init_app(app) login_manager.init_app(app) # DB 장애/지연 감지 시 표준 에러 페이지/응답 @app.errorhandler(SAOperationalError) def handle_db_operational_error(err): _mark_db_unavailable(err, phase="request_operational_error") return _db_error_response() # Blueprint 등록 from app.routes import main_bp app.register_blueprint(main_bp) from app.routers.creation import creation_bp app.register_blueprint(creation_bp) @app.context_processor def inject_i18n(): lang = session.get('lang', 'vi') return dict( gettext=lambda key, **kwargs: gettext(key, lang=lang, **kwargs), current_lang=lang ) # 등록된 라우트 확인 (디버깅용) 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 에러 처리""" # 정적 파일이나 favicon 등은 로그 레벨을 낮춤 if request.path.startswith('/static') or request.path.endswith('favicon.ico'): logger.debug(f"404 에러(정적): {request.path} - {request.method}") else: logger.warning(f"404 에러: {request.path} - {request.method}") # API 요청이거나 JSON을 기대하는 요청인 경우 JSON 응답 반환 # 경로에 '/api/'가 포함되어 있으면 API 요청으로 간주 (예: /api/, /creation/api/) is_api_request = '/api/' in request.path accepts_json = request.accept_mimetypes.best_match(['application/json', 'text/html']) == 'application/json' is_json_request = request.is_json or request.content_type == 'application/json' if is_api_request or accepts_json or is_json_request: return jsonify({'error': gettext('no_resource', lang=session.get('lang', 'vi')), 'path': request.path}), 404 return 'Not Found', 404 @app.errorhandler(413) def request_entity_too_large(error): """413 에러 처리 (파일 용량 초과)""" logger.warning(f"413 에러: 파일 용량 초과 - {request.path}") lang = session.get('lang', 'vi') if request.path.startswith('/api/'): return jsonify({'error': gettext('upload_limit_exceeded', lang=lang)}), 413 flash(gettext('file_too_large', lang=lang), 'danger') # 이전 페이지로 리다이렉트 시도 return redirect(request.referrer or url_for('main.index')) # 요청 로깅 미들웨어 추가 @app.before_request def log_request_info(): """각 HTTP 요청 정보를 로깅""" # DB 장애 모드면 즉시 안내 페이지 노출(정적/헬스체크류는 제외) if app.config.get('DB_UNAVAILABLE'): path = request.path or '' if not (path.startswith('/static') or path == '/favicon.ico'): return _db_error_response() # 민감한 정보 제외하고 로깅 if request.path.endswith('.json') or request.path.endswith('.ico') or request.path.startswith('/static'): return logger.info(f"[요청] {request.method} {request.path} - IP: {request.remote_addr}") if request.args: logger.debug(f"[요청 파라미터] {dict(request.args)}") # POST 요청 바디 로깅 (JSON인 경우만, 너무 길면 자름) if request.method == 'POST' and request.is_json: try: body = request.get_json() body_str = str(body) if len(body_str) > 1000: body_str = body_str[:1000] + "..." # Windows 콘솔 인코딩 에러 방지 (Emoji 등 처리할 수 없는 문자를 안전하게 처리) try: # cp949로 인코딩 불가능한 문자를 '?'로 대체하여 안전하게 처리 safe_body_str = body_str.encode('cp949', errors='replace').decode('cp949', errors='replace') logger.info(f"[요청 본문] {safe_body_str}") except Exception: # 로깅 실패가 서비스 중단으로 이어지지 않도록 예외 처리 pass except: pass @app.errorhandler(Exception) def handle_exception(e): """처리되지 않은 모든 예외 로깅""" import traceback err_trace = traceback.format_exc() error_msg = f"[Unhandled Exception] {str(e)}\n{err_trace}" logger.error(error_msg) # 터미널에 직접 출력 (콘솔 확인용) print("\n" + "="*80) print("ERROR 발생!") print(error_msg) print("="*80 + "\n") # API 요청인 경우 JSON 응답 if request.path.startswith('/api/'): return jsonify({'error': str(e), 'trace': err_trace}), 500 return str(e), 500 @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 # 데이터베이스 초기화 및 마이그레이션 # DB 장애 모드면(또는 DB 미설정) 초기화 자체를 스킵하고, 요청 시 503 페이지로만 안내한다. if not app.config.get('DB_UNAVAILABLE'): with app.app_context(): try: # 데이터베이스 테이블 생성 (PostgreSQL은 초기 연결이 불안정할 수 있어 재시도) create_attempts = int(os.environ.get('DB_INIT_RETRIES', '3')) last_init_err: Exception | None = None for i in range(create_attempts): try: db.create_all() logger.info("[데이터베이스] 테이블 생성 완료") last_init_err = None break except Exception as e: last_init_err = e logger.warning(f"[데이터베이스] 테이블 생성 실패(재시도 예정): {e}") if i < create_attempts - 1: time.sleep(_retry_sleep_seconds(i)) if last_init_err is not None: logger.error(f"[데이터베이스] 테이블 생성 최종 실패: {last_init_err}") raise last_init_err # 마이그레이션 실행 check_and_migrate_db(app) # 관리자 계정 생성 create_admin_user() # ---- DB 연결/데이터 스냅샷(운영 디버깅용) ---- # "Railway에 연결은 되는데 데이터가 안 보인다" 케이스에서, # 실제로 어떤 DB/스키마/유저에 붙었고 주요 테이블 row가 있는지 로그로 확정한다. try: if db.engine.dialect.name == 'postgresql': try: url_str = db.engine.url.render_as_string(hide_password=True) except Exception: url_str = "" logger.info(f"[데이터베이스] 연결 확인(Engine): {db.engine.dialect.name} {url_str}") with db.engine.connect() as conn: db_name = conn.execute(text("SELECT current_database()")).scalar() db_user = conn.execute(text("SELECT current_user")).scalar() db_schema = conn.execute(text("SELECT current_schema()")).scalar() logger.info(f"[데이터베이스] 현재 DB 스냅샷: db={db_name}, user={db_user}, schema={db_schema}") # 핵심 테이블 카운트(없으면 None) table_names = [ "user", "uploaded_file", "chat_session", "chat_message", "webtoon_project_upload", ] counts = {} for t in table_names: try: counts[t] = conn.execute(text(f'SELECT COUNT(*) FROM "{t}"')).scalar() except Exception: # 테이블이 없거나 권한/스키마 문제면 None 처리 counts[t] = None logger.info(f"[데이터베이스] 주요 테이블 row count: {counts}") except Exception as e: logger.warning(f"[데이터베이스] 스냅샷 로그 실패(무시 가능): {e}") except SAOperationalError as e: # DB 장애여도 앱은 살리고, 요청 시 에러 페이지로 안내 _mark_db_unavailable(e, phase="startup_db_init") except Exception: # DB 장애 이외의 오류는 기존대로 실패시키는 것이 안전 raise else: logger.warning("[데이터베이스] DB 장애 모드로 기동: SQLite 폴백 없이 503 안내 페이지를 노출합니다.") 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, role 컬럼 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")) if 'role' not in columns: logger.info("user 테이블에 role 컬럼 추가 중...") conn.execute(text("ALTER TABLE \"user\" ADD COLUMN role VARCHAR(20) DEFAULT 'user' NOT NULL")) logger.info("기존 관리자 권한 마이그레이션 (is_admin -> role='admin')...") conn.execute(text("UPDATE \"user\" SET role = 'admin' WHERE is_admin = 1")) # 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'")) # 6. webtoon_project_upload 테이블 - analysis_model_name 컬럼 if inspector.has_table('webtoon_project_upload'): columns = [c['name'] for c in inspector.get_columns('webtoon_project_upload')] with db.engine.begin() as conn: if 'analysis_model_name' not in columns: logger.info("webtoon_project_upload 테이블에 analysis_model_name 컬럼 추가 중...") conn.execute(text("ALTER TABLE webtoon_project_upload ADD COLUMN analysis_model_name VARCHAR(200)")) # 7. 웹툰 관련 테이블들 (없는 경우 생성) - 실서버에서 누락되는 케이스 방지 try: from app.database import ( WebtoonProjectUpload, WebtoonProjectTask, WebtoonWBSAnalysis, WebtoonWBSJob, WebtoonProjectDuration, WebtoonEpisodeDuration, WebtoonDurationJob, WebtoonMilestone, ) # NOTE: inspector는 생성 전 상태일 수 있어 checkfirst=True로 안전하게 생성 webtoon_models = [ ("webtoon_project_upload", WebtoonProjectUpload), ("webtoon_project_task", WebtoonProjectTask), ("webtoon_wbs_analysis", WebtoonWBSAnalysis), ("webtoon_wbs_job", WebtoonWBSJob), ("webtoon_project_duration", WebtoonProjectDuration), ("webtoon_episode_duration", WebtoonEpisodeDuration), ("webtoon_duration_job", WebtoonDurationJob), ("webtoon_milestone", WebtoonMilestone), ] missing = [] for table_name, model in webtoon_models: if not inspector.has_table(table_name): missing.append(table_name) logger.info(f"{table_name} 테이블 생성 중...") model.__table__.create(db.engine, checkfirst=True) if missing: logger.warning(f"[웹툰] 누락 테이블 자동 생성 완료: {', '.join(missing)}") # webtoon_episode_duration - active_days 컬럼 if inspector.has_table('webtoon_episode_duration'): columns = [c['name'] for c in inspector.get_columns('webtoon_episode_duration')] with db.engine.begin() as conn: if 'active_days' not in columns: logger.info("webtoon_episode_duration 테이블에 active_days 컬럼 추가 중...") conn.execute(text("ALTER TABLE webtoon_episode_duration ADD COLUMN active_days INTEGER DEFAULT 0 NOT NULL")) except Exception as e: logger.warning(f"웹툰 관련 테이블 생성/보강 중 오류(무시 가능): {e}") 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()