from __future__ import annotations import os from dataclasses import dataclass @dataclass(frozen=True) class DbConfig: host: str port: int database: str username: str password: str @dataclass(frozen=True) class GoogleConfig: client_id: str client_secret: str refresh_token: str calendar_id: str def get_db_config() -> DbConfig: return DbConfig( host=os.environ.get("EDOXID_DB_HOST", "203.24.50.52"), port=int(os.environ.get("EDOXID_DB_PORT", "3306")), database=os.environ.get("EDOXID_DB_DATABASE", "doxid2022"), username=os.environ.get("EDOXID_DB_USERNAME", "calendar_sync_ro"), password=os.environ["EDOXID_DB_PASSWORD"], ) def get_google_config() -> GoogleConfig: return GoogleConfig( client_id=os.environ["EDOXID_GOOGLE_CLIENT_ID"], client_secret=os.environ["EDOXID_GOOGLE_CLIENT_SECRET"], refresh_token=os.environ["EDOXID_GOOGLE_REFRESH_TOKEN"], # Calendar "Jadwal Sidang" milik tu@informatika.untan.ac.id calendar_id=os.environ.get( "EDOXID_GOOGLE_CALENDAR_ID", "c_1a4ilomn51i1r0mccnjqe9cumg@group.calendar.google.com", ), ) def get_state_sqlite_path() -> str: return os.environ.get("EDOXID_STATE_SQLITE_PATH", "/opt/airflow/scripts/edoxid_calendar_state.sqlite") def get_sync_min_date() -> str: """Batas bawah tanggal sidang yang disinkron, format 'YYYY-MM-DD'. Default: hari ini — supaya run pertama TIDAK langsung mengirim invite Calendar ke semua submission lama (2019-2026, ~1156 baris per cek 2026-07-27) yang sudah punya schedule. Ubah lewat env var kalau memang mau backfill sengaja. """ return os.environ.get("EDOXID_SYNC_MIN_DATE", "today") def get_business_hours() -> tuple[int, int]: """(jam_mulai, jam_selesai) WIB — di luar rentang ini, sync sungguhan cuma jalan tiap `get_off_hours_interval_minutes()`, bukan tiap trigger DAG. Tidak perlu 2 DAG terpisah — DAG tetap 1 jadwal (tiap 30 menit), tapi run_sync() sendiri yang skip kalau lagi di luar jam kerja & belum waktunya.""" start = int(os.environ.get("EDOXID_BUSINESS_HOURS_START", "7")) end = int(os.environ.get("EDOXID_BUSINESS_HOURS_END", "17")) return start, end def get_off_hours_interval_minutes() -> int: return int(os.environ.get("EDOXID_OFF_HOURS_INTERVAL_MINUTES", "120"))