Clock_out_Time_Calculator/core/database.py

"""
데이터베이스 관리 모듈
SQLite를 사용하여 근무 기록, 연장근무, 휴가 등을 관리
"""
import sqlite3
from contextlib import contextmanager
from datetime import datetime, date
from typing import Optional, List, Dict, Tuple
import os

from core.settings_keys import (
    WORK_HOURS, WORK_MINUTES, ANNUAL_LEAVE_TOTAL, ANNUAL_LEAVE_DAYS,
    INITIAL_OVERTIME_MINUTES, INITIAL_LEAVE_USED_HOURS,
)


class Database:
    def __init__(self, db_path: str = "database.db"):
        self.db_path = db_path
        self.init_database()
        self._enable_concurrency()

    def get_connection(self) -> sqlite3.Connection:
        """데이터베이스 연결 반환.

        timeout=5초: 다른 PC/프로세스가 쓰는 동안 락 충돌 시 대기.
        클라우드 동기화(OneDrive 등)로 같은 DB를 두 PC에서 쓸 때 안전.

        주의: 호출자는 반드시 try/finally로 close()를 보장해야 함.
        가능하면 `_conn()` 컨텍스트 매니저 사용 권장.
        """
        conn = sqlite3.connect(self.db_path, timeout=5.0)
        conn.row_factory = sqlite3.Row
        return conn

    @contextmanager
    def _conn(self):
        """try/finally close()를 자동 처리하는 컨텍스트 매니저.

        예:
            with self._conn() as conn:
                cursor = conn.cursor()
                cursor.execute(...)
                conn.commit()

        예외 발생 시: SQLite의 묵시적 트랜잭션이 commit되지 않은 채 conn이 닫히므로
        해당 트랜잭션 자동 rollback. 단, FK 제약 등으로 부분적 변경이 보일 수 있는
        멀티-statement 케이스는 명시적 try/except + conn.rollback() 필요.
        """
        conn = self.get_connection()
        try:
            yield conn
        finally:
            try:
                conn.close()
            except Exception:
                pass

    def _enable_concurrency(self):
        """WAL 모드 활성화 — 동시 읽기 + 쓰기 가능, 클라우드 동기화 친화."""
        try:
            conn = sqlite3.connect(self.db_path, timeout=5.0)
            conn.execute("PRAGMA journal_mode = WAL")
            conn.execute("PRAGMA synchronous = NORMAL")
            conn.commit()
            conn.close()
        except sqlite3.OperationalError:
            # WAL 미지원 환경(읽기전용 폴더 등) — 기본 모드로 fallback
            pass

    def init_database(self):
        """데이터베이스 초기화 및 테이블 생성"""
        with self._conn() as conn:
            self._create_tables(conn)
            conn.commit()

        # 데이터베이스 마이그레이션 실행
        self.migrate_break_records_cascade()
        self.migrate_lunch_duration_to_minutes()
        self.migrate_leave_records_hours_to_days()
        self.migrate_add_dinner_break()
        self.migrate_cleanup_balance_adjustments()
        self.migrate_work_hours_to_minutes()
        self.migrate_annual_leave_keys()
        self.migrate_v23_break_type()
        self.migrate_v23_notification_log()
        self.migrate_v23_onboarding_for_existing()
        self.migrate_v271_break_indexes()
        self.migrate_v271_work_records_indexes()
        self.migrate_v280_achievements_columns()
        self.migrate_v280_hire_date()
        self.migrate_v290_holidays_auto_sync()

        # 기본 설정 초기화
        self.init_default_settings()

    def migrate_v290_holidays_auto_sync(self) -> None:
        """일 1회 한국 공휴일 자동 동기화 (백그라운드).

        Sentinel: settings['holidays_synced_date'] = 'YYYY-MM-DD' (오늘 날짜).
        값이 오늘과 같으면 스킵 — 즉 같은 날 여러 번 켜도 호출 1회.

        매일 호출하므로 정부가 임시공휴일 발표하면 다음 날 자동 반영.
        일일 한도 10000회, 사용자 50명 × 1회 = 0.5% 소비.

        실제 동기화는 백그라운드 스레드에서 — 부트스트랩이 네트워크에 묶이지 않음.
        실패는 silent, 다음 실행 시 재시도.

        테스트 환경에서는 CLOCKOUT_DISABLE_HOLIDAY_SYNC=1 로 비활성화.
        """
        import os
        if os.environ.get('CLOCKOUT_DISABLE_HOLIDAY_SYNC'):
            return
        from datetime import datetime as _dt
        import threading
        try:
            today = _dt.now().date().isoformat()
            sentinel = self.get_setting('holidays_synced_date', '')
            if sentinel == today:
                return
        except Exception:
            return

        cur_year = _dt.now().year

        def _worker():
            try:
                # 새 연결로 작업 (sqlite3 connection은 thread-affine)
                from core.database import Database
                db = Database(self.db_path)
                added = db.add_korean_holidays_auto(cur_year, include_next_year=True)
                if added >= 0:
                    db.set_setting('holidays_synced_date', today)
            except Exception:
                pass

        t = threading.Thread(target=_worker, daemon=True, name='holiday-sync')
        t.start()

    def _create_tables(self, conn) -> None:
        """init_database()에서 호출되는 CREATE TABLE 묶음. conn은 호출자가 관리."""
        cursor = conn.cursor()

        # 일일 근무 기록 테이블
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS work_records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                date DATE NOT NULL UNIQUE,
                clock_in TIME NOT NULL,
                clock_out TIME,
                lunch_break BOOLEAN DEFAULT 0,
                total_hours REAL,
                overtime_minutes INTEGER DEFAULT 0,
                overtime_earned INTEGER DEFAULT 0,
                overtime_used INTEGER DEFAULT 0,
                work_type TEXT DEFAULT 'normal',
                memo TEXT,
                is_manual BOOLEAN DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        # 연장근무 적립 내역
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS overtime_bank (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                work_record_id INTEGER,
                earned_minutes INTEGER NOT NULL,
                date DATE NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (work_record_id) REFERENCES work_records(id)
            )
        ''')

        # 연장근무 사용 내역
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS overtime_usage (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                work_record_id INTEGER,
                used_minutes INTEGER NOT NULL,
                date DATE NOT NULL,
                reason TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (work_record_id) REFERENCES work_records(id)
            )
        ''')

        # 휴가 기록
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS leave_records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                date DATE NOT NULL,
                leave_type TEXT NOT NULL,
                days REAL,
                memo TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        # 설정
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS settings (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        # 도전과제 (achievements)
        # v2.8.0: code 컬럼 추가 — 평가자가 식별자로 사용
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS achievements (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                code TEXT UNIQUE,
                name TEXT NOT NULL,
                description TEXT,
                category TEXT,
                tier TEXT,
                is_secret BOOLEAN DEFAULT 0,
                progress INTEGER DEFAULT 0,
                target INTEGER DEFAULT 1,
                earned_date DATE,
                badge_icon TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        # 외출 기록
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS break_records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                work_record_id INTEGER,
                date DATE NOT NULL,
                break_out TIME NOT NULL,
                break_in TIME,
                total_minutes INTEGER,
                reason TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (work_record_id) REFERENCES work_records(id) ON DELETE CASCADE
            )
        ''')

        # 공휴일 테이블
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS holidays (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                date DATE NOT NULL UNIQUE,
                name TEXT NOT NULL,
                is_recurring BOOLEAN DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        # 반복 연차 패턴 테이블 (P2)
        # pattern 형식:
        #   'weekly:friday' / 'weekly:mon,wed' (요일 영문 소문자, 콤마 구분)
        #   'biweekly:friday' (격주)
        #   'monthly:15' (매월 N일)
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS recurring_leaves (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                pattern TEXT NOT NULL,
                leave_type TEXT NOT NULL,
                days REAL NOT NULL,
                start_date DATE NOT NULL,
                end_date DATE,
                memo TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

    def migrate_break_records_cascade(self):
        """break_records 테이블에 CASCADE 제약조건 추가 (마이그레이션).

        스키마 introspection이 묵시적 sentinel 역할 — CASCADE가 이미 있으면 no-op.
        DROP/CREATE/INSERT는 단일 트랜잭션 내에서 실행되어 실패 시 자동 rollback.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='break_records'")
            result = cursor.fetchone()

            if not result or 'ON DELETE CASCADE' in result[0]:
                return  # 이미 CASCADE 있음 또는 테이블 없음

            try:
                cursor.execute('SELECT * FROM break_records')
                backup_data = cursor.fetchall()

                cursor.execute('DROP TABLE IF EXISTS break_records')
                cursor.execute('''
                    CREATE TABLE break_records (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        work_record_id INTEGER,
                        date DATE NOT NULL,
                        break_out TIME NOT NULL,
                        break_in TIME,
                        total_minutes INTEGER,
                        reason TEXT,
                        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                        FOREIGN KEY (work_record_id) REFERENCES work_records(id) ON DELETE CASCADE
                    )
                ''')
                for row in backup_data:
                    cursor.execute('''
                        INSERT INTO break_records
                        (id, work_record_id, date, break_out, break_in, total_minutes, reason, created_at, updated_at)
                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                    ''', tuple(row))
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"break_records CASCADE 마이그레이션 실패 (rollback됨): {e}", file=sys.stderr)

    def migrate_lunch_duration_to_minutes(self):
        """lunch_duration을 시간 단위에서 분 단위로 마이그레이션"""
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("SELECT value FROM settings WHERE key = 'lunch_duration'")
                result = cursor.fetchone()
                if result:
                    lunch_hours = float(result['value'])
                    lunch_minutes = int(lunch_hours * 60)
                    cursor.execute('''
                        INSERT OR REPLACE INTO settings (key, value, updated_at)
                        VALUES ('lunch_duration_minutes', ?, CURRENT_TIMESTAMP)
                    ''', (str(lunch_minutes),))
                    # 기존 lunch_duration은 삭제하지 않음 (호환성 유지)

                cursor.execute("SELECT value FROM settings WHERE key = 'lunch_duration_minutes'")
                if not cursor.fetchone():
                    cursor.execute('''
                        INSERT OR IGNORE INTO settings (key, value)
                        VALUES ('lunch_duration_minutes', '60')
                    ''')
                conn.commit()
            except Exception as e:
                import sys
                if ("no such column" not in str(e).lower()
                        and "already exists" not in str(e).lower()):
                    print(f"lunch_duration 마이그레이션 경고: {e}", file=sys.stderr)

    def migrate_leave_records_hours_to_days(self):
        """leave_records.hours 컬럼을 days로 변경 (마이그레이션).

        스키마 introspection이 묵시적 sentinel — 'hours REAL' 사라지면 no-op.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='leave_records'")
                result = cursor.fetchone()
                if not (result and 'hours REAL' in result[0]):
                    return  # 이미 마이그레이션됨 또는 테이블 없음

                cursor.execute('SELECT * FROM leave_records')
                backup_data = cursor.fetchall()

                cursor.execute('DROP TABLE IF EXISTS leave_records')
                cursor.execute('''
                    CREATE TABLE leave_records (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        date DATE NOT NULL,
                        leave_type TEXT NOT NULL,
                        days REAL,
                        memo TEXT,
                        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                    )
                ''')
                for row in backup_data:
                    cursor.execute('''
                        INSERT INTO leave_records
                        (id, date, leave_type, days, memo, created_at)
                        VALUES (?, ?, ?, ?, ?, ?)
                    ''', tuple(row))
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"leave_records 마이그레이션 실패 (rollback됨): {e}", file=sys.stderr)

    def migrate_add_dinner_break(self):
        """work_records 테이블에 dinner_break 컬럼 추가 (마이그레이션)"""
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("PRAGMA table_info(work_records)")
                columns = [row[1] for row in cursor.fetchall()]
                if 'dinner_break' not in columns:
                    cursor.execute('''
                        ALTER TABLE work_records
                        ADD COLUMN dinner_break BOOLEAN DEFAULT 0
                    ''')
                    conn.commit()
                    print("work_records 테이블에 dinner_break 컬럼 추가 완료")
            except Exception as e:
                import sys
                if "duplicate column name" not in str(e).lower():
                    print(f"dinner_break 컬럼 추가 경고: {e}", file=sys.stderr)

    def migrate_cleanup_balance_adjustments(self):
        """기존 잘못된 조정 데이터 정리 마이그레이션

        이전 버전에서 '덮어쓰기' 방식으로 생성된 조정 레코드들을 정리:
        - overtime_bank: work_record_id가 NULL인 레코드는 삭제 (초기값은 settings로 이동)
        - leave_records: 'manual' 타입이고 '이전 사용분 일괄 추가' 메모가 있는 레코드 삭제 (초기값은 settings로 이동)
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("SELECT value FROM settings WHERE key = 'balance_adjustment_migrated_v2'")
                if cursor.fetchone():
                    return  # 이미 완료

                cursor.execute('''
                    DELETE FROM overtime_bank
                    WHERE work_record_id IS NULL
                ''')
                deleted_overtime = cursor.rowcount

                cursor.execute('''
                    DELETE FROM leave_records
                    WHERE leave_type = 'manual' AND memo LIKE '%이전 사용분 일괄 추가%'
                ''')
                deleted_leave = cursor.rowcount

                cursor.execute('''
                    INSERT OR REPLACE INTO settings (key, value, updated_at)
                    VALUES ('balance_adjustment_migrated_v2', 'true', CURRENT_TIMESTAMP)
                ''')
                conn.commit()

                if deleted_overtime > 0 or deleted_leave > 0:
                    print(f"잔액 조정 마이그레이션 v2 완료: 연장근무 {deleted_overtime}건, 연차 {deleted_leave}건 삭제")
            except Exception as e:
                conn.rollback()
                import sys
                print(f"잔액 조정 마이그레이션 오류: {e}", file=sys.stderr)

    def migrate_work_hours_to_minutes(self):
        """work_hours(시간 단위, 정수)를 work_minutes(분 단위)로 마이그레이션."""
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("SELECT value FROM settings WHERE key = 'work_minutes'")
                if cursor.fetchone():
                    return  # 이미 완료

                cursor.execute("SELECT value FROM settings WHERE key = 'work_hours'")
                row = cursor.fetchone()
                if row:
                    try:
                        work_hours_val = float(row[0])
                        work_minutes_val = int(round(work_hours_val * 60))
                    except (ValueError, TypeError):
                        work_minutes_val = 480
                else:
                    work_minutes_val = 480

                cursor.execute('''
                    INSERT OR REPLACE INTO settings (key, value, updated_at)
                    VALUES ('work_minutes', ?, CURRENT_TIMESTAMP)
                ''', (str(work_minutes_val),))
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"work_minutes 마이그레이션 경고: {e}", file=sys.stderr)

    def migrate_annual_leave_keys(self):
        """annual_leave_total(레거시) ↔ annual_leave_days(UI) 동기화.

        UI는 annual_leave_days를 사용하지만 일부 메서드는 annual_leave_total을 읽음.
        둘 중 하나만 있으면 다른 쪽에 복사. sentinel로 1회만 실행.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("SELECT value FROM settings WHERE key = 'annual_leave_keys_migrated'")
                if cursor.fetchone():
                    return  # 이미 완료

                cursor.execute("SELECT value FROM settings WHERE key = 'annual_leave_days'")
                days_row = cursor.fetchone()
                cursor.execute("SELECT value FROM settings WHERE key = 'annual_leave_total'")
                total_row = cursor.fetchone()

                if days_row and not total_row:
                    cursor.execute('''
                        INSERT OR REPLACE INTO settings (key, value, updated_at)
                        VALUES ('annual_leave_total', ?, CURRENT_TIMESTAMP)
                    ''', (days_row[0],))
                elif total_row and not days_row:
                    cursor.execute('''
                        INSERT OR REPLACE INTO settings (key, value, updated_at)
                        VALUES ('annual_leave_days', ?, CURRENT_TIMESTAMP)
                    ''', (total_row[0],))

                cursor.execute('''
                    INSERT OR REPLACE INTO settings (key, value, updated_at)
                    VALUES ('annual_leave_keys_migrated', 'true', CURRENT_TIMESTAMP)
                ''')
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"annual_leave 키 동기화 경고: {e}", file=sys.stderr)

    def migrate_v23_break_type(self):
        """break_records에 break_type 컬럼 추가 (v2.3.0).
        값: 'break'(기본 외출) / 'lunch' / 'dinner'.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("PRAGMA table_info(break_records)")
                cols = [row[1] for row in cursor.fetchall()]
                if 'break_type' not in cols:
                    cursor.execute("ALTER TABLE break_records ADD COLUMN break_type TEXT DEFAULT 'break'")
                    conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"break_type 컬럼 추가 경고: {e}", file=sys.stderr)

    def migrate_v23_notification_log(self):
        """알림 발송 이력 테이블 (v2.3.0). 중복 발송 방지 + 통계."""
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute('''
                    CREATE TABLE IF NOT EXISTS notification_log (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        channel TEXT NOT NULL,
                        event_type TEXT NOT NULL,
                        payload TEXT,
                        sent_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                        success BOOLEAN DEFAULT 1
                    )
                ''')
                cursor.execute('''
                    CREATE INDEX IF NOT EXISTS idx_notif_log_event
                    ON notification_log(event_type, sent_at)
                ''')
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"notification_log 생성 경고: {e}", file=sys.stderr)

    def log_notification(self, channel: str, event_type: str,
                          payload: str = None, success: bool = True) -> None:
        """알림 발송 이력 기록 (중복 방지 가드용)."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "INSERT INTO notification_log (channel, event_type, payload, success) VALUES (?, ?, ?, ?)",
                (channel, event_type, payload, success)
            )
            conn.commit()

    def has_notification_today(self, channel: str, event_type: str) -> bool:
        """오늘 같은 (channel, event_type) 발송 이력 존재 여부.

        주의: SQLite의 CURRENT_TIMESTAMP는 UTC를 반환하므로 sent_at 비교 시에도
        'localtime'을 적용해야 사용자의 로컬 자정 경계와 일치. 적용하지 않으면
        UTC와 로컬 날짜가 다른 시간대(예: KST 00:00~09:00)에 mismatch 발생.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT COUNT(*) FROM notification_log "
                "WHERE channel = ? AND event_type = ? "
                "AND DATE(sent_at, 'localtime') = DATE('now', 'localtime')",
                (channel, event_type)
            )
            return cursor.fetchone()[0] > 0

    def migrate_v271_break_indexes(self):
        """break_records 조회 패턴 인덱스 (v2.7.1).

        가장 자주 쓰이는 쿼리:
          - get_today_break_records / get_break_records_by_date — `WHERE date = ?`
          - 일일 보고서 — `WHERE date = ? AND break_type = ?`
          - get_meal_minutes_today — `WHERE date = ? AND break_type = ?`
          - get_active_break_record — `WHERE date = ? AND break_in IS NULL`

        date 단일 인덱스 + (date, break_type) 복합 인덱스. CREATE INDEX IF NOT EXISTS
        라 idempotent — sentinel 불필요.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute(
                    "CREATE INDEX IF NOT EXISTS idx_break_records_date "
                    "ON break_records(date)"
                )
                cursor.execute(
                    "CREATE INDEX IF NOT EXISTS idx_break_records_date_type "
                    "ON break_records(date, break_type)"
                )
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"break_records 인덱스 생성 경고: {e}", file=sys.stderr)

    def migrate_v271_work_records_indexes(self):
        """work_records / overtime 조회 인덱스 (v2.7.1).

        date는 work_records에서 UNIQUE 제약으로 이미 인덱스가 자동 생성되지만,
        overtime_bank/overtime_usage/leave_records의 date 컬럼은 인덱스가 없어
        get_monthly_stats / get_consecutive_overtime_days 등이 풀스캔.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                for tbl in ('overtime_bank', 'overtime_usage', 'leave_records'):
                    cursor.execute(
                        f"CREATE INDEX IF NOT EXISTS idx_{tbl}_date ON {tbl}(date)"
                    )
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"date 인덱스 생성 경고: {e}", file=sys.stderr)

    def migrate_v280_achievements_columns(self):
        """기존 achievements 테이블에 v2.8.0 컬럼들 추가 (도전과제 시스템).

        code, category, tier, is_secret, progress, target, created_at.
        ALTER TABLE 멱등 — 이미 있는 컬럼은 스킵.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("PRAGMA table_info(achievements)")
                cols = {row[1] for row in cursor.fetchall()}

                additions = [
                    ('code', "ALTER TABLE achievements ADD COLUMN code TEXT"),
                    ('category', "ALTER TABLE achievements ADD COLUMN category TEXT"),
                    ('tier', "ALTER TABLE achievements ADD COLUMN tier TEXT"),
                    ('is_secret', "ALTER TABLE achievements ADD COLUMN is_secret BOOLEAN DEFAULT 0"),
                    ('progress', "ALTER TABLE achievements ADD COLUMN progress INTEGER DEFAULT 0"),
                    ('target', "ALTER TABLE achievements ADD COLUMN target INTEGER DEFAULT 1"),
                    ('created_at', "ALTER TABLE achievements ADD COLUMN created_at TIMESTAMP"),
                ]
                for col_name, sql in additions:
                    if col_name not in cols:
                        cursor.execute(sql)

                # code UNIQUE 인덱스 (UNIQUE 제약은 ALTER로 못 추가하므로 인덱스로)
                cursor.execute(
                    "CREATE UNIQUE INDEX IF NOT EXISTS idx_achievements_code "
                    "ON achievements(code) WHERE code IS NOT NULL"
                )
                conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"achievements 컬럼 마이그레이션 경고: {e}", file=sys.stderr)

    def migrate_v280_hire_date(self):
        """첫 work_records 자동으로 hire_date 설정에 기록 (없으면).

        도전과제(1주년, 365일 후 출근 등)에서 사용. 사용자 입력 없이
        자동 추출 — 첫 출근일 = 가장 오래된 work_records.date.
        """
        if self.get_setting('hire_date', None):
            return  # 이미 있음
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("SELECT MIN(date) FROM work_records")
                row = cursor.fetchone()
                if row and row[0]:
                    cursor.execute('''
                        INSERT OR REPLACE INTO settings (key, value, updated_at)
                        VALUES ('hire_date', ?, CURRENT_TIMESTAMP)
                    ''', (row[0],))
                    conn.commit()
            except Exception as e:
                import sys
                print(f"hire_date 마이그레이션 경고: {e}", file=sys.stderr)

    def migrate_v23_onboarding_for_existing(self):
        """기존 사용자(이미 work_records 데이터 있음)는 온보딩 자동 완료 처리.

        v2.3.0 도입 시 한 번만 실행. 신규 DB(데이터 0)는 영향 없음 → 첫 실행 시 위저드.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("SELECT value FROM settings WHERE key = 'onboarding_completed'")
                row = cursor.fetchone()
                if row and row[0] == 'true':
                    return

                cursor.execute("SELECT COUNT(*) FROM work_records")
                count = cursor.fetchone()[0]
                if count > 0:
                    cursor.execute('''
                        INSERT OR REPLACE INTO settings (key, value, updated_at)
                        VALUES ('onboarding_completed', 'true', CURRENT_TIMESTAMP)
                    ''')
                    conn.commit()
            except Exception as e:
                conn.rollback()
                import sys
                print(f"onboarding 마이그레이션 경고: {e}", file=sys.stderr)

    def get_setting_int(self, key: str, default: int = 0) -> int:
        """설정을 int로 안전하게 조회 (변환 실패 시 default)."""
        raw = self.get_setting(key, None)
        if raw is None:
            return default
        try:
            return int(raw)
        except (ValueError, TypeError):
            return default

    def get_setting_float(self, key: str, default: float = 0.0) -> float:
        """설정을 float로 안전하게 조회."""
        raw = self.get_setting(key, None)
        if raw is None:
            return default
        try:
            return float(raw)
        except (ValueError, TypeError):
            return default

    def get_setting_bool(self, key: str, default: bool = False) -> bool:
        """설정을 bool로 안전하게 조회 ('true'/'1'/'yes' = True)."""
        raw = self.get_setting(key, None)
        if raw is None:
            return default
        return str(raw).lower() in ('1', 'true', 'yes')

    def get_work_minutes(self) -> int:
        """기본 근무시간 (분 단위) 조회.

        work_minutes 우선, 없으면 work_hours * 60으로 폴백.
        7시간 30분 같은 단축근무 케이스에서 분 단위 정확도 보장.
        """
        wm = self.get_setting(WORK_MINUTES, None)
        if wm is not None:
            try:
                return int(wm)
            except (ValueError, TypeError):
                pass

        wh = self.get_setting(WORK_HOURS, '8')
        try:
            return int(round(float(wh) * 60))
        except (ValueError, TypeError):
            return 480

    def init_default_settings(self):
        """기본 설정 초기화"""
        default_settings = {
            'work_hours': '8',
            'work_minutes': '480',
            'lunch_duration_minutes': '60',
            'dinner_duration_minutes': '60',
            'auto_lunch': 'false',
            'auto_overtime': 'true',
            'theme': 'dark',
            'notification_before_minutes': '30',
            'notification_clock_out': 'true',
            'notification_lunch': 'true',
            'notification_dinner': 'true',
            'notification_overtime': 'true',
            'notification_health': 'true',
            # 알림 임계값 (v2.7.1: 하드코딩 → 설정)
            'lunch_reminder_hours': '4',
            'dinner_reminder_hours': '8',
            'overtime_threshold_hours': '20',
            'weekly_hours_threshold': '52',
            'health_consecutive_ot_days': '3',
            'annual_leave_total': '15',
            'annual_leave_days': '15',  # annual_leave_total과 자동 동기화
            'workday_boundary_hour': '6',
            'overtime_unit': '30',
            'time_format': '24',
            # v2.3.0
            'onboarding_completed': 'false',
            'salary_enabled': 'false',
            'hourly_wage': '0',
            'overtime_rate': '1.5',
            'health_break_enabled': 'true',
            'health_break_hours': '4',
            'discord_webhook_url': '',
            'discord_notif_clock_in': 'true',
            'discord_notif_clock_out': 'true',
            'discord_notif_health': 'true',
            # v2.4.0
            'goal_overtime_max_monthly': '0',
            'goal_avg_hours_daily': '0',
            # v2.6.0
            'font_scale': '1.0',
            'high_contrast': 'false',
            # v2.8.0 도전과제
            'birthday': '',
            'stat_weekly_view_count': '0',
            'stat_monthly_view_count': '0',
            'stat_pattern_view_count': '0',
            'calendar_view_count': '0',
            'leave_calendar_view_count': '0',
            'daily_report_count': '0',
            'achievements_view_count': '0',
            'chart_hover_discovered': 'false',
            'notification_achievement': 'true',
            'discord_notif_achievement': 'true',
        }

        with self._conn() as conn:
            cursor = conn.cursor()
            for key, value in default_settings.items():
                cursor.execute('''
                    INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)
                ''', (key, value))
            conn.commit()

    # ===== 근무 기록 관련 메서드 =====

    def add_work_record(self, date: str, clock_in: str, lunch_break: bool = False,
                        is_manual: bool = False) -> int:
        """근무 기록 추가.

        첫 출근 시 hire_date 자동 기록 (도전과제 1주년 등에서 사용).
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO work_records (date, clock_in, lunch_break, is_manual)
                VALUES (?, ?, ?, ?)
            ''', (date, clock_in, lunch_break, is_manual))
            record_id = cursor.lastrowid
            # hire_date가 비어있으면 첫 출근일로 자동 설정 — 별도 트랜잭션 회피
            cursor.execute("SELECT value FROM settings WHERE key = 'hire_date'")
            row = cursor.fetchone()
            if not row or not row[0]:
                cursor.execute('''
                    INSERT OR REPLACE INTO settings (key, value, updated_at)
                    VALUES ('hire_date', ?, CURRENT_TIMESTAMP)
                ''', (date,))
            conn.commit()
            return record_id

    def update_clock_out(self, date: str, clock_out: str, total_hours: float,
                         overtime_minutes: int, overtime_earned: int):
        """퇴근 시간 및 연장근무 업데이트"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                UPDATE work_records
                SET clock_out = ?, total_hours = ?, overtime_minutes = ?,
                    overtime_earned = ?, updated_at = CURRENT_TIMESTAMP
                WHERE date = ?
            ''', (clock_out, total_hours, overtime_minutes, overtime_earned, date))
            conn.commit()

    def cancel_clock_out(self, date: str) -> bool:
        """퇴근 취소 (퇴근 시간 및 연장근무 기록 삭제).

        2개 테이블 처리는 단일 트랜잭션 — 실패 시 자동 rollback.

        Returns:
            bool: 성공 여부 (해당 날짜 기록이 없으면 False)
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT id FROM work_records WHERE date = ?', (date,))
            record = cursor.fetchone()
            if not record:
                return False
            work_record_id = record[0]
            try:
                cursor.execute('''
                    DELETE FROM overtime_bank
                    WHERE work_record_id = ? AND date = ?
                ''', (work_record_id, date))
                cursor.execute('''
                    UPDATE work_records
                    SET clock_out = NULL,
                        total_hours = NULL,
                        overtime_minutes = 0,
                        overtime_earned = 0,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE date = ?
                ''', (date,))
                conn.commit()
                return True
            except Exception:
                conn.rollback()
                raise

    def get_today_record(self) -> Optional[Dict]:
        """오늘 근무 기록 조회"""
        today = date.today().isoformat()
        return self.get_work_record(today)

    def get_work_record(self, date: str) -> Optional[Dict]:
        """특정 날짜 근무 기록 조회"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM work_records WHERE date = ?', (date,))
            row = cursor.fetchone()
            return dict(row) if row else None

    def update_lunch_break(self, date: str, lunch_break: bool):
        """점심시간 사용 여부 업데이트"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                UPDATE work_records
                SET lunch_break = ?, updated_at = CURRENT_TIMESTAMP
                WHERE date = ?
            ''', (lunch_break, date))
            conn.commit()

    def update_dinner_break(self, date: str, dinner_break: bool):
        """저녁시간 사용 여부 업데이트"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                UPDATE work_records
                SET dinner_break = ?, updated_at = CURRENT_TIMESTAMP
                WHERE date = ?
            ''', (dinner_break, date))
            conn.commit()

    def delete_work_record(self, date: str):
        """특정 날짜의 근무 기록 삭제. 3개 테이블이 한 트랜잭션에서 함께 처리."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT id FROM work_records WHERE date = ?', (date,))
            record = cursor.fetchone()
            if not record:
                return  # 삭제할 게 없음 — commit 불필요
            record_id = record[0]
            try:
                cursor.execute('DELETE FROM overtime_bank WHERE work_record_id = ?', (record_id,))
                cursor.execute('DELETE FROM overtime_usage WHERE work_record_id = ?', (record_id,))
                cursor.execute('DELETE FROM work_records WHERE id = ?', (record_id,))
                conn.commit()
            except Exception:
                conn.rollback()
                raise

    def get_work_records_by_range(self, start_date: str, end_date: str) -> List[Dict]:
        """기간별 근무 기록 조회"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM work_records
                WHERE date BETWEEN ? AND ?
                ORDER BY date DESC
            ''', (start_date, end_date))
            return [dict(row) for row in cursor.fetchall()]

    # ===== 연장근무 관련 메서드 =====

    def add_overtime_earned(self, work_record_id: int, earned_minutes: int, date: str):
        """연장근무 적립"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO overtime_bank (work_record_id, earned_minutes, date)
                VALUES (?, ?, ?)
            ''', (work_record_id, earned_minutes, date))
            conn.commit()

    def delete_overtime_earned(self, bank_id: int) -> bool:
        """연장근무 적립(은행) 기록 1건 삭제. 삭제분만큼 잔액이 즉시 감소.

        Returns:
            bool: 실제로 삭제된 행이 있으면 True.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM overtime_bank WHERE id = ?', (bank_id,))
            deleted = cursor.rowcount
            conn.commit()
            return deleted > 0

    def add_overtime_usage(self, work_record_id: int, used_minutes: int,
                          date: str, reason: str = None):
        """연장근무 사용 (음수 잔액 허용 - 선사용 후적립 가능)"""
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute('''
                    INSERT INTO overtime_usage (work_record_id, used_minutes, date, reason)
                    VALUES (?, ?, ?, ?)
                ''', (work_record_id, used_minutes, date, reason))
                if work_record_id is not None:
                    cursor.execute('''
                        UPDATE work_records
                        SET overtime_used = overtime_used + ?
                        WHERE id = ?
                    ''', (used_minutes, work_record_id))
                conn.commit()
            except Exception:
                conn.rollback()
                raise

    def get_total_overtime_balance(self) -> int:
        """총 연장근무 잔액 조회 (초기값 + 적립 - 사용)"""
        initial_overtime = int(self.get_setting(INITIAL_OVERTIME_MINUTES, '0'))
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT
                    COALESCE((SELECT SUM(earned_minutes) FROM overtime_bank), 0) -
                    COALESCE((SELECT SUM(used_minutes) FROM overtime_usage), 0) AS balance
            ''')
            balance = cursor.fetchone()[0]
            return initial_overtime + balance

    def get_today_overtime_usage(self) -> int:
        """오늘 사용한 추가근무 시간 조회 (분)"""
        from datetime import date
        today = date.today().isoformat()
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT SUM(used_minutes) FROM overtime_usage WHERE date = ?',
                            (today,))
            return cursor.fetchone()[0] or 0

    def get_today_leave_minutes(self) -> int:
        """오늘 사용한 연차/반차 시간 조회 (분)"""
        from datetime import date
        return self.get_leave_minutes_for(date.today().isoformat())

    def get_leave_minutes_for(self, date_str: str) -> int:
        """특정 날짜에 등록된 연차 합계를 분 단위로 반환.

        구체 leave_records + 매치되는 recurring_leaves 인스턴스 합산.
        예정/사용 구분 없음.
        """
        days = self._effective_leave_days_for(date_str)
        return int(days * self.get_work_minutes())

    def has_full_day_leave(self, date_str: str) -> bool:
        """해당 날짜에 종일(또는 그 이상) 연차가 등록되어 있는지.

        구체 + 반복 패턴 모두 검사.
        """
        return self._effective_leave_days_for(date_str) >= 1.0

    def _effective_leave_days_for(self, date_str: str) -> float:
        """구체 leave_records + 반복 패턴 매치를 합산한 일수."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT SUM(days) FROM leave_records WHERE date = ?',
                            (date_str,))
            concrete_days = float(cursor.fetchone()[0] or 0.0)

        # 반복 패턴 — 매번 호출되니 lazy import + 가벼운 query
        try:
            from core.recurring_leaves import expand_for_date
            from datetime import datetime as _dt
            target = _dt.strptime(date_str, '%Y-%m-%d').date()
            recs = self.get_recurring_leaves(active_on=date_str)
            recurring_days = sum(o.days for o in expand_for_date(recs, target))
        except Exception:
            recurring_days = 0.0

        return concrete_days + recurring_days

    def get_leave_records_by_date(self, date_str: str) -> List[Dict]:
        """해당 날짜에 등록된 leave_records 전체 (디스플레이/편집용)."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM leave_records WHERE date = ? ORDER BY id',
                            (date_str,))
            return [dict(row) for row in cursor.fetchall()]

    def get_leave_records_by_range(self, start_date: str, end_date: str) -> List[Dict]:
        """기간 내 leave_records (스케줄 화면용). start/end inclusive."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM leave_records
                WHERE date BETWEEN ? AND ?
                ORDER BY date ASC, id ASC
            ''', (start_date, end_date))
            return [dict(row) for row in cursor.fetchall()]

    # ===== 반복 연차 (recurring_leaves) — P2 =====

    def add_recurring_leave(self, pattern: str, leave_type: str, days: float,
                             start_date: str, end_date: str = None,
                             memo: str = None) -> int:
        """반복 연차 패턴 등록.

        Args:
            pattern: 'weekly:friday' / 'biweekly:mon' / 'monthly:15' 등
            leave_type: '연차' / '반차' / '시간' 등 (표시용)
            days: 한 회당 차감 일수 (0.5 = 반차)
            start_date: 시작일 'YYYY-MM-DD'
            end_date: 종료일 또는 None(=무기한)
            memo: 옵션
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO recurring_leaves
                    (pattern, leave_type, days, start_date, end_date, memo)
                VALUES (?, ?, ?, ?, ?, ?)
            ''', (pattern, leave_type, days, start_date, end_date, memo))
            conn.commit()
            return cursor.lastrowid

    def get_recurring_leaves(self, active_on: str = None) -> List[Dict]:
        """반복 패턴 목록. active_on 지정 시 그날 유효한 것만."""
        with self._conn() as conn:
            cursor = conn.cursor()
            if active_on:
                cursor.execute('''
                    SELECT * FROM recurring_leaves
                    WHERE start_date <= ?
                      AND (end_date IS NULL OR end_date >= ?)
                    ORDER BY id
                ''', (active_on, active_on))
            else:
                cursor.execute('SELECT * FROM recurring_leaves ORDER BY id')
            return [dict(row) for row in cursor.fetchall()]

    def delete_recurring_leave(self, rec_id: int) -> None:
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM recurring_leaves WHERE id = ?', (rec_id,))
            conn.commit()

    def add_initial_overtime_balance(self, minutes: int):
        """초기 연장근무 잔액 추가"""
        from datetime import datetime
        today = datetime.now().date().isoformat()
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO overtime_bank (work_record_id, earned_minutes, date)
                VALUES (NULL, ?, ?)
            ''', (minutes, today))
            conn.commit()

    def get_overtime_history(self, limit: int = 30) -> List[Dict]:
        """연장근무 내역 조회 (적립 + 사용)"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT 'earned' as type, earned_minutes as minutes, date,
                       wr.clock_in, wr.clock_out
                FROM overtime_bank ob
                LEFT JOIN work_records wr ON ob.work_record_id = wr.id
                UNION ALL
                SELECT 'used' as type, used_minutes as minutes, date,
                       wr.clock_in, wr.clock_out
                FROM overtime_usage ou
                LEFT JOIN work_records wr ON ou.work_record_id = wr.id
                ORDER BY date DESC
                LIMIT ?
            ''', (limit,))
            return [dict(row) for row in cursor.fetchall()]

    # ===== 휴가 관련 메서드 =====

    def add_leave_record(self, date: str, leave_type: str, days: float, memo: str = None):
        """휴가 기록 추가"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO leave_records (date, leave_type, days, memo)
                VALUES (?, ?, ?, ?)
            ''', (date, leave_type, days, memo))
            conn.commit()

    def get_leave_records(self, start_date: str = None, end_date: str = None,
                            exclude_bulk: bool = False) -> List[Dict]:
        """휴가 기록 조회"""
        with self._conn() as conn:
            cursor = conn.cursor()
            if start_date and end_date:
                if exclude_bulk:
                    cursor.execute('''
                        SELECT * FROM leave_records
                        WHERE date BETWEEN ? AND ?
                        AND COALESCE(memo, '') != '이전 사용분 일괄 추가'
                        ORDER BY date DESC
                    ''', (start_date, end_date))
                else:
                    cursor.execute('''
                        SELECT * FROM leave_records
                        WHERE date BETWEEN ? AND ?
                        ORDER BY date DESC
                    ''', (start_date, end_date))
            else:
                if exclude_bulk:
                    cursor.execute('''
                        SELECT * FROM leave_records
                        WHERE COALESCE(memo, '') != '이전 사용분 일괄 추가'
                        ORDER BY date DESC
                    ''')
                else:
                    cursor.execute('SELECT * FROM leave_records ORDER BY date DESC')
            return [dict(row) for row in cursor.fetchall()]

    def get_annual_leave_balance(self) -> Tuple[float, float]:
        """연차 잔여 조회 (총 연차, 사용한 연차)

        Note:
            현재 UI에서는 get_leave_balance()만 사용됨.
            이 메서드는 leave_records 테이블에서 직접 계산하므로
            settings.leave_balance와 불일치할 수 있음.
            향후 연차 관리 기능 개선 시 활용 가능.
        """
        total = float(self.get_setting(ANNUAL_LEAVE_TOTAL, '15'))
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT SUM(days) FROM leave_records
                WHERE leave_type IS NULL OR leave_type NOT IN ('manual', 'bulk')
            ''')
            used = cursor.fetchone()[0] or 0
        return total, used

    # ===== 설정 관련 메서드 =====

    def get_setting(self, key: str, default: str = None) -> str:
        """설정 값 조회"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT value FROM settings WHERE key = ?', (key,))
            row = cursor.fetchone()
            return row[0] if row else default

    def set_setting(self, key: str, value: str):
        """설정 값 저장"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT OR REPLACE INTO settings (key, value, updated_at)
                VALUES (?, ?, CURRENT_TIMESTAMP)
            ''', (key, value))
            conn.commit()

    # ===== 통계 관련 메서드 =====

    def get_weekly_stats(self) -> Dict:
        """주간 통계"""
        from datetime import datetime, timedelta

        today = datetime.now().date()
        week_ago = today - timedelta(days=7)

        records = self.get_work_records_by_range(week_ago.isoformat(), today.isoformat())

        total_hours = sum(r.get('total_hours', 0) or 0 for r in records)
        total_overtime = sum(r.get('overtime_minutes', 0) or 0 for r in records)
        work_days = len([r for r in records if r.get('clock_out')])

        return {
            'total_hours': total_hours,
            'total_overtime_minutes': total_overtime,
            'work_days': work_days,
            'avg_hours_per_day': total_hours / work_days if work_days > 0 else 0
        }

    def get_consecutive_overtime_days(self, threshold_minutes: int = 30) -> int:
        """오늘부터 거꾸로 연속 연장근무한 일수.

        Args:
            threshold_minutes: 연장근무로 카운트할 최소 분 (기본 30분)
        Returns:
            연속 일수 (오늘 미적립이거나 휴무일이면 0)
        """
        from datetime import date, timedelta
        count = 0
        d = date.today()
        for _ in range(60):  # 최대 60일까지만 거슬러 검사
            rec = self.get_work_record(d.isoformat())
            if not rec or not rec.get('clock_out'):
                break
            if (rec.get('overtime_minutes') or 0) < threshold_minutes:
                break
            count += 1
            d -= timedelta(days=1)
        return count

    def get_monthly_stats(self, year: int, month: int) -> Dict:
        """월간 통계"""
        from calendar import monthrange

        start_date = f"{year}-{month:02d}-01"
        last_day = monthrange(year, month)[1]
        end_date = f"{year}-{month:02d}-{last_day}"

        records = self.get_work_records_by_range(start_date, end_date)

        total_hours = sum(r.get('total_hours', 0) or 0 for r in records)
        total_overtime = sum(r.get('overtime_minutes', 0) or 0 for r in records)
        work_days = len([r for r in records if r.get('clock_out')])

        return {
            'year': year,
            'month': month,
            'total_hours': total_hours,
            'total_overtime_minutes': total_overtime,
            'work_days': work_days,
            'records': records
        }

    # ===== 휴가 관련 메서드 (중복 제거됨 - 356줄의 함수 사용) =====

    def get_leave_record(self, date: str) -> Optional[Dict]:
        """특정 날짜의 휴가 기록 조회"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM leave_records WHERE date = ?', (date,))
            row = cursor.fetchone()
            return dict(row) if row else None

    def get_all_leave_records(self, limit: int = 100) -> List[Dict]:
        """모든 휴가 기록 조회 (최신순)"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM leave_records ORDER BY date DESC LIMIT ?', (limit,))
            return [dict(row) for row in cursor.fetchall()]

    def delete_leave_record(self, leave_id: int):
        """휴가 기록 삭제"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM leave_records WHERE id = ?', (leave_id,))
            conn.commit()

    # ===== 설정 관련 메서드 =====

    def get_settings(self) -> Dict:
        """설정 가져오기"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM settings')
            rows = cursor.fetchall()

        # 딕셔너리로 변환
        settings = {}
        for row in rows:
            key = row['key']
            value = row['value']
            # 타입 변환
            if value.lower() in ['true', 'false']:
                settings[key] = value.lower() == 'true'
            else:
                try:
                    settings[key] = int(value)
                except ValueError:
                    try:
                        settings[key] = float(value)
                    except ValueError:
                        settings[key] = value
        return settings

    def save_settings(self, settings: Dict):
        """설정 저장.

        키 동기화 처리:
        - work_minutes 저장 시 work_hours도 갱신 (호환성)
        - work_hours 저장 시 work_minutes도 갱신
        - annual_leave_days ↔ annual_leave_total 양방향 동기화
        """
        # 동기화 키 미리 보강 (호출자가 일부만 줘도 양쪽 다 저장)
        synced = dict(settings)

        if 'work_minutes' in synced and 'work_hours' not in synced:
            try:
                # floor로 통일 (settings_view와 일관성: 450분 → 7시간)
                synced['work_hours'] = int(float(synced['work_minutes'])) // 60
            except (ValueError, TypeError):
                pass
        elif 'work_hours' in synced and 'work_minutes' not in synced:
            try:
                synced['work_minutes'] = int(round(float(synced['work_hours']) * 60))
            except (ValueError, TypeError):
                pass

        if 'annual_leave_days' in synced and 'annual_leave_total' not in synced:
            synced['annual_leave_total'] = synced['annual_leave_days']
        elif 'annual_leave_total' in synced and 'annual_leave_days' not in synced:
            synced['annual_leave_days'] = synced['annual_leave_total']

        with self._conn() as conn:
            cursor = conn.cursor()
            for key, value in synced.items():
                cursor.execute('''
                    INSERT OR REPLACE INTO settings (key, value)
                    VALUES (?, ?)
                ''', (key, str(value)))
            conn.commit()

    # ===== 외출 관련 메서드 =====

    def add_break_record(self, work_record_id: int, date: str, break_out: str,
                          reason: str = None, break_type: str = 'break') -> int:
        """외출 기록 추가. break_type: 'break'(외출) / 'lunch' / 'dinner'."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO break_records (work_record_id, date, break_out, reason, break_type)
                VALUES (?, ?, ?, ?, ?)
            ''', (work_record_id, date, break_out, reason, break_type))
            record_id = cursor.lastrowid
            conn.commit()
            return record_id

    def add_meal_record(self, date: str, start_time: str, end_time: str,
                         meal_type: str = 'lunch') -> int:
        """식사 시간 기록 (시작·종료 둘 다 알 때).

        Args:
            date: 'YYYY-MM-DD'
            start_time, end_time: 'HH:MM:SS'
            meal_type: 'lunch' or 'dinner'
        Returns:
            새 break_record id
        """
        from datetime import datetime as _dt
        # 분 계산
        start_dt = _dt.strptime(start_time, '%H:%M:%S')
        end_dt = _dt.strptime(end_time, '%H:%M:%S')
        if end_dt < start_dt:
            from datetime import timedelta as _td
            end_dt += _td(days=1)
        total_min = int((end_dt - start_dt).total_seconds() / 60)

        rec = self.get_today_record() if date == _dt.now().date().isoformat() else None
        wid = rec['id'] if rec else None

        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO break_records (work_record_id, date, break_out, break_in,
                                            total_minutes, break_type)
                VALUES (?, ?, ?, ?, ?, ?)
            ''', (wid, date, start_time, end_time, total_min, meal_type))
            rid = cursor.lastrowid
            conn.commit()
            return rid

    def get_meal_minutes_today(self, meal_type: str = 'lunch') -> int:
        """오늘의 식사 시간 합계 (분). 수동 입력된 경우만."""
        from datetime import date as _date
        today = _date.today().isoformat()
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT COALESCE(SUM(total_minutes), 0) FROM break_records
                WHERE date = ? AND break_type = ? AND total_minutes IS NOT NULL
            ''', (today, meal_type))
            return int(cursor.fetchone()[0] or 0)

    def update_break_return(self, break_id: int, break_in: str):
        """외출 복귀 시간 업데이트.

        2 step 처리: ① 복귀시각 저장, ② 총 외출시간 계산해서 갱신.
        예외 발생 시 부분 변경 방지를 위해 명시적 try/except + rollback.
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute('''
                    UPDATE break_records
                    SET break_in = ?, updated_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                ''', (break_in, break_id))

                cursor.execute('''
                    SELECT break_out, break_in FROM break_records WHERE id = ?
                ''', (break_id,))
                row = cursor.fetchone()

                if row and row['break_out'] and row['break_in']:
                    from datetime import datetime, timedelta
                    break_out_time = datetime.strptime(row['break_out'], "%H:%M:%S")
                    break_in_time = datetime.strptime(row['break_in'], "%H:%M:%S")

                    # 복귀가 외출보다 이전이면 자정 넘긴 것
                    if break_in_time < break_out_time:
                        break_in_time += timedelta(days=1)

                    total_minutes = int((break_in_time - break_out_time).total_seconds() / 60)
                    if total_minutes < 0:
                        total_minutes = 0

                    cursor.execute('''
                        UPDATE break_records
                        SET total_minutes = ?
                        WHERE id = ?
                    ''', (total_minutes, break_id))

                conn.commit()
            except Exception:
                conn.rollback()
                raise

    def get_today_break_records(self) -> List[Dict]:
        """오늘의 외출 기록 조회"""
        from datetime import date
        today = date.today().isoformat()
        return self.get_break_records_by_date(today)

    def get_break_records_by_date(self, date: str) -> List[Dict]:
        """특정 날짜의 외출 기록 조회"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM break_records
                WHERE date = ?
                ORDER BY break_out ASC
            ''', (date,))
            return [dict(row) for row in cursor.fetchall()]

    def get_active_break_record(self, target_date: str = None) -> Optional[Dict]:
        """현재 진행 중인 외출 기록 조회 (복귀하지 않은 외출)

        Args:
            target_date: 조회할 날짜 (YYYY-MM-DD), None이면 오늘
        """
        from datetime import date
        if target_date is None:
            target_date = date.today().isoformat()
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM break_records
                WHERE date = ? AND break_in IS NULL
                ORDER BY break_out DESC
                LIMIT 1
            ''', (target_date,))
            row = cursor.fetchone()
            return dict(row) if row else None

    def update_break_record(self, break_id: int, break_out: str, break_in: str = None,
                              reason: str = None):
        """외출 기록 수정"""
        with self._conn() as conn:
            cursor = conn.cursor()
            if break_in:
                from datetime import datetime, timedelta
                break_out_time = datetime.strptime(break_out, "%H:%M:%S")
                break_in_time = datetime.strptime(break_in, "%H:%M:%S")
                if break_in_time < break_out_time:
                    break_in_time += timedelta(days=1)
                total_minutes = int((break_in_time - break_out_time).total_seconds() / 60)
                if total_minutes < 0:
                    total_minutes = 0
                cursor.execute('''
                    UPDATE break_records
                    SET break_out = ?, break_in = ?, total_minutes = ?, reason = ?, updated_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                ''', (break_out, break_in, total_minutes, reason, break_id))
            else:
                cursor.execute('''
                    UPDATE break_records
                    SET break_out = ?, reason = ?, updated_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                ''', (break_out, reason, break_id))
            conn.commit()

    def delete_break_record(self, break_id: int):
        """외출 기록 삭제"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM break_records WHERE id = ?', (break_id,))
            conn.commit()

    def get_total_break_minutes_today(self) -> int:
        """오늘의 총 외출 시간 (분), 진행 중인 외출 포함"""
        from datetime import date, datetime
        today = date.today().isoformat()
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT SUM(total_minutes) FROM break_records
                WHERE date = ? AND total_minutes IS NOT NULL
            ''', (today,))
            total = cursor.fetchone()[0] or 0

            cursor.execute('''
                SELECT break_out FROM break_records
                WHERE date = ? AND break_in IS NULL
                ORDER BY break_out DESC
                LIMIT 1
            ''', (today,))
            active_break = cursor.fetchone()

        if active_break:
            break_out_str = active_break[0]
            now = datetime.now()
            break_out_time = datetime.strptime(f"{today} {break_out_str}",
                                                  "%Y-%m-%d %H:%M:%S")
            active_minutes = int((now - break_out_time).total_seconds() / 60)
            total += active_minutes
        return total

    def update_work_memo(self, date: str, memo: str):
        """근무 기록 메모 업데이트"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                UPDATE work_records
                SET memo = ?, updated_at = CURRENT_TIMESTAMP
                WHERE date = ?
            ''', (memo, date))
            conn.commit()

    def get_leave_balance(self) -> float:
        """연차 잔여 개수 조회 (총 연차 - 초기 사용량 - 프로그램 기록 사용량)"""
        from datetime import datetime

        # 총 연차 일수
        total_annual = int(self.get_setting(ANNUAL_LEAVE_DAYS, '15'))

        # 초기 사용 연차 (프로그램 사용 전)
        initial_leave_hours = float(self.get_setting(INITIAL_LEAVE_USED_HOURS, '0'))
        initial_leave_days = initial_leave_hours / 8.0

        # 올해 프로그램에서 기록된 사용량
        current_year = datetime.now().year
        all_leaves = self.get_all_leave_records(limit=365)
        year_leaves = [r for r in all_leaves if r['date'].startswith(str(current_year))]
        used_days = sum(r['days'] for r in year_leaves)

        # 잔여 = 총 - 초기사용 - 프로그램기록
        return total_annual - initial_leave_days - used_days

    def set_leave_balance(self, balance: float):
        """연차 잔여 개수 설정"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT OR REPLACE INTO settings (key, value, updated_at)
                VALUES ('leave_balance', ?, CURRENT_TIMESTAMP)
            ''', (str(balance),))
            conn.commit()

    def use_leave(self, days: float, date: str, leave_type: str = "연차", memo: str = None):
        """연차 사용

        Args:
            days: 사용할 연차 일수 (예: 1.0=하루, 0.5=반차, 0.25=반반차)

        Note:
            leave_records 테이블의 'days' 컬럼에 일수를 저장함
            예: 1.0 = 1일, 0.5 = 반차, 0.125 = 1시간(8분의 1일)
        """
        current_balance = self.get_leave_balance()
        if current_balance < days:
            raise ValueError(f"연차 잔여 개수가 부족합니다. (잔여: {current_balance}일)")

        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO leave_records (date, leave_type, days, memo)
                VALUES (?, ?, ?, ?)
            ''', (date, leave_type, days, memo))
            conn.commit()
        # 잔여 개수 차감 (별도 트랜잭션 — set_leave_balance 내부 commit)
        self.set_leave_balance(current_balance - days)

    # ===== 공휴일 관련 메서드 =====

    def add_holiday(self, date: str, name: str, is_recurring: bool = False) -> int:
        """공휴일 추가

        Args:
            date: 공휴일 날짜 (YYYY-MM-DD)
            name: 공휴일 이름
            is_recurring: 매년 반복 여부 (음력 명절 등은 False)

        Returns:
            int: 추가된 공휴일 ID
        """
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT OR REPLACE INTO holidays (date, name, is_recurring, created_at)
                VALUES (?, ?, ?, CURRENT_TIMESTAMP)
            ''', (date, name, is_recurring))
            holiday_id = cursor.lastrowid
            conn.commit()
            return holiday_id

    def delete_holiday(self, holiday_id: int):
        """공휴일 삭제"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM holidays WHERE id = ?', (holiday_id,))
            conn.commit()

    def delete_holiday_by_date(self, date: str):
        """날짜로 공휴일 삭제"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM holidays WHERE date = ?', (date,))
            conn.commit()

    def is_holiday(self, date: str) -> bool:
        """해당 날짜가 공휴일인지 확인."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT id FROM holidays WHERE date = ?', (date,))
            return cursor.fetchone() is not None

    def get_holiday(self, date: str) -> Optional[Dict]:
        """해당 날짜의 공휴일 정보 조회."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM holidays WHERE date = ?', (date,))
            row = cursor.fetchone()
            return dict(row) if row else None

    def get_holidays_by_year(self, year: int) -> List[Dict]:
        """해당 연도의 공휴일 목록 조회."""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM holidays
                WHERE date >= ? AND date < ?
                ORDER BY date ASC
            ''', (f"{year}-01-01", f"{year + 1}-01-01"))
            return [dict(row) for row in cursor.fetchall()]

    def get_all_holidays(self) -> List[Dict]:
        """모든 공휴일 목록 조회"""
        with self._conn() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM holidays ORDER BY date ASC')
            return [dict(row) for row in cursor.fetchall()]

    def add_korean_holidays(self, year: int):
        """한국 공휴일 일괄 추가 (고정 공휴일만)

        Args:
            year: 추가할 연도

        Note:
            음력 기반 명절(설날, 추석)은 매년 날짜가 변경되므로
            수동으로 추가해야 합니다.
        """
        fixed_holidays = [
            (f"{year}-01-01", "신정"),
            (f"{year}-03-01", "삼일절"),
            (f"{year}-05-05", "어린이날"),
            (f"{year}-06-06", "현충일"),
            (f"{year}-08-15", "광복절"),
            (f"{year}-10-03", "개천절"),
            (f"{year}-10-09", "한글날"),
            (f"{year}-12-25", "크리스마스"),
        ]

        for date, name in fixed_holidays:
            self.add_holiday(date, name, is_recurring=True)

    def add_korean_holidays_from_api(self, year: int) -> int:
        """공공데이터포털 특일정보 API로 한국 공휴일 등록 (정부 공인).

        임시공휴일 + 근로자의 날 등 holidays 패키지가 놓치는 항목까지 포함.
        네트워크 실패 시 -1 반환 → 호출자 fallback.

        Returns:
            추가된 공휴일 개수 (기존 등록과 중복은 제외). 실패 시 -1.
        """
        try:
            from utils.holiday_api import fetch_korean_holidays
        except ImportError:
            return -1
        items = fetch_korean_holidays(year)
        if items is None:
            return -1
        added = 0
        for it in items:
            if not it.get('is_holiday'):
                continue
            date_str = it['date']
            if not self.is_holiday(date_str):
                self.add_holiday(date_str, it['name'], is_recurring=False)
                added += 1
        return added

    def add_korean_holidays_auto(self, year: int, include_next_year: bool = False) -> int:
        """`holidays` 패키지로 음력 명절 포함 한국 공휴일 자동 등록.

        Timezone 주의:
            `holidays` 패키지는 한국 캘린더 날짜를 timezone-naive `date` 객체로 반환.
            timezone 영향 없음 — 양력/음력 모두 절기 기준이라 timezone과 무관하게
            동일한 캘린더 날짜.

            호출자(settings_view 등)는 보통 `datetime.now().year`을 전달하는데,
            이는 시스템 로컬 시각의 연도 — 한국 사용자가 KST로 설정된 경우 정확.
            만약 시스템 시계가 UTC로 설정된 드문 경우, KST 기준 1/1 직후 ~ 9시 전엔
            UTC 연도가 전년도일 수 있음. 그래도 함수 자체가 `is_holiday`로 중복 가드
            하므로 데이터 손상 없이 단지 한 번 더 등록 시도할 뿐.

        연말 경계:
            12월 말에 호출 시 다음 연도 1/1 신정·설날을 미리 등록해두려면
            include_next_year=True 권장.

        Args:
            year: 등록할 연도 (보통 현재 로컬 연도)
            include_next_year: True면 year + (year+1) 둘 다 등록

        Returns:
            추가된 공휴일 개수. 패키지 미설치 시 -1.
        """
        years_to_add = [year]
        if include_next_year:
            years_to_add.append(year + 1)

        added = 0
        for y in years_to_add:
            # 1차: 정부 API (임시공휴일 포함, 가장 정확)
            api_count = self.add_korean_holidays_from_api(y)
            if api_count >= 0:
                added += api_count
                # API가 응답했으면 근로자의 날도 포함되어 있음. 끝.
                continue

            # 2차 fallback: holidays 패키지
            try:
                import holidays as _holidays
                kr = _holidays.country_holidays('KR', years=y)
            except Exception:
                continue  # 둘 다 실패면 해당 연도만 스킵
            for d, name in kr.items():
                date_str = d.isoformat()
                if not self.is_holiday(date_str):
                    self.add_holiday(date_str, name, is_recurring=False)
                    added += 1

            # holidays.KR이 누락하는 근로자의 날 명시적 보강
            extra = [(f"{y}-05-01", "근로자의 날")]
            for date_str, name in extra:
                if not self.is_holiday(date_str):
                    self.add_holiday(date_str, name, is_recurring=True)
                    added += 1
        return added

    def copy_recurring_holidays(self, from_year: int, to_year: int):
        """반복 공휴일을 다음 연도로 복사

        Args:
            from_year: 복사할 원본 연도
            to_year: 복사 대상 연도
        """
        holidays = self.get_holidays_by_year(from_year)

        for holiday in holidays:
            if holiday['is_recurring']:
                new_date = holiday['date'].replace(str(from_year), str(to_year))
                self.add_holiday(new_date, holiday['name'], is_recurring=True)