from __future__ import annotations import json import os import secrets import shutil from dataclasses import dataclass from pathlib import Path from urllib.parse import quote from cryptography.fernet import Fernet DEFAULT_PARALLEL_LIMIT = 4 POLL_INTERVAL_SECONDS = 10 LOGIN_RETRY_LIMIT = 6 TASK_BACKOFF_SECONDS = 6 BROWSER_PAGE_TIMEOUT = 40 LOGS_PAGE_SIZE = 180 SELENIUM_ERROR_LIMIT = 5 SELENIUM_RESTART_LIMIT = 5 SUBMIT_CAPTCHA_RETRY_LIMIT = 5 SCHEDULE_TIMEZONE = "Asia/Shanghai" AIVEN_MYSQL_TEMPLATE = "mysql://avnadmin:{password}@mysql-2bace9cd-cacode.i.aivencloud.com:21260/SACC?ssl-mode=REQUIRED" def _pick_binary(env_name: str, *fallbacks: str) -> str: env_value = os.getenv(env_name) if env_value: return env_value for candidate in fallbacks: if not candidate: continue if Path(candidate).exists(): return candidate resolved = shutil.which(candidate) if resolved: return resolved return fallbacks[0] if fallbacks else "" def _load_or_create_internal_secrets(data_dir: Path) -> tuple[str, str]: secret_file = data_dir / ".app_secrets.json" payload: dict[str, str] = {} if secret_file.exists(): try: raw_payload = json.loads(secret_file.read_text(encoding="utf-8")) if isinstance(raw_payload, dict): payload = { str(key): str(value) for key, value in raw_payload.items() if isinstance(value, str) } except (OSError, json.JSONDecodeError): payload = {} session_secret = payload.get("session_secret", "").strip() encryption_key = payload.get("encryption_key", "").strip() changed = False if not session_secret: session_secret = secrets.token_hex(32) changed = True if not encryption_key: encryption_key = Fernet.generate_key().decode("utf-8") changed = True if changed or not secret_file.exists(): secret_file.write_text( json.dumps( { "session_secret": session_secret, "encryption_key": encryption_key, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) return session_secret, encryption_key def _build_database_target(root_dir: Path, data_dir: Path) -> tuple[str | Path, str, Path | None]: explicit_database_url = os.getenv("DATABASE_URL", "").strip() if explicit_database_url: normalized_url = explicit_database_url else: sql_password = os.getenv("SQL_PASSWORD", "").strip() if sql_password: normalized_url = AIVEN_MYSQL_TEMPLATE.format(password=quote(sql_password, safe="")) else: return (data_dir / "course_catcher.db").resolve(), "sqlite", None if normalized_url.startswith("mysql://"): normalized_url = f"mysql+pymysql://{normalized_url[len('mysql://') :]}" if normalized_url.startswith("mysql+pymysql://"): mysql_ssl_ca_path = Path(os.getenv("MYSQL_SSL_CA", str(root_dir / "ca.pem"))).resolve() return normalized_url, "mysql", mysql_ssl_ca_path if normalized_url.startswith("sqlite:///"): sqlite_path = Path(normalized_url[len("sqlite:///") :]).resolve() return sqlite_path, "sqlite", None return Path(normalized_url).resolve(), "sqlite", None def _build_browser_proxy_url() -> str: explicit_proxy_url = os.getenv("SACC_SELENIUM_PROXY_URL", "").strip() if explicit_proxy_url: return _normalize_browser_proxy_url(explicit_proxy_url) proxy_host = os.getenv("SACC_SELENIUM_PROXY_HOST", "").strip() proxy_port = os.getenv("SACC_SELENIUM_PROXY_PORT", "").strip() if not proxy_host or not proxy_port: return "" proxy_scheme = os.getenv("SACC_SELENIUM_PROXY_SCHEME", "socks5h").strip().lower() or "socks5h" proxy_username = os.getenv("SACC_SELENIUM_PROXY_USERNAME", "").strip() proxy_password = os.getenv("SACC_SELENIUM_PROXY_PASSWORD", "").strip() if proxy_username or proxy_password: credentials = f"{quote(proxy_username, safe='')}:{quote(proxy_password, safe='')}@" else: credentials = "" return _normalize_browser_proxy_url(f"{proxy_scheme}://{credentials}{proxy_host}:{proxy_port}") def _normalize_browser_proxy_url(proxy_url: str) -> str: normalized = str(proxy_url or "").strip() remote_dns_enabled = os.getenv("SACC_SELENIUM_PROXY_REMOTE_DNS", "1").strip().lower() not in {"0", "false", "no", "off"} if remote_dns_enabled and normalized.lower().startswith("socks5://"): return f"socks5h://{normalized[len('socks5://') :]}" return normalized @dataclass(slots=True) class AppConfig: root_dir: Path data_dir: Path db_path: str | Path database_backend: str mysql_ssl_ca_path: Path | None session_secret: str encryption_key: str super_admin_username: str super_admin_password: str default_parallel_limit: int poll_interval_seconds: int login_retry_limit: int task_backoff_seconds: int browser_page_timeout: int logs_page_size: int selenium_error_limit: int selenium_restart_limit: int submit_captcha_retry_limit: int chrome_binary: str chromedriver_path: str browser_proxy_url: str schedule_timezone: str @classmethod def load(cls) -> "AppConfig": root_dir = Path(__file__).resolve().parent.parent data_dir = Path(os.getenv("DATA_DIR", str(root_dir / "data"))).resolve() data_dir.mkdir(parents=True, exist_ok=True) super_admin_username = os.getenv("ADMIN", "superadmin") super_admin_password = os.getenv("PASSWORD", "change-me-in-hf-space") session_secret, encryption_key = _load_or_create_internal_secrets(data_dir) db_path, database_backend, mysql_ssl_ca_path = _build_database_target(root_dir, data_dir) return cls( root_dir=root_dir, data_dir=data_dir, db_path=db_path, database_backend=database_backend, mysql_ssl_ca_path=mysql_ssl_ca_path, session_secret=session_secret, encryption_key=encryption_key, super_admin_username=super_admin_username, super_admin_password=super_admin_password, default_parallel_limit=DEFAULT_PARALLEL_LIMIT, poll_interval_seconds=POLL_INTERVAL_SECONDS, login_retry_limit=LOGIN_RETRY_LIMIT, task_backoff_seconds=TASK_BACKOFF_SECONDS, browser_page_timeout=BROWSER_PAGE_TIMEOUT, logs_page_size=LOGS_PAGE_SIZE, selenium_error_limit=SELENIUM_ERROR_LIMIT, selenium_restart_limit=SELENIUM_RESTART_LIMIT, submit_captcha_retry_limit=SUBMIT_CAPTCHA_RETRY_LIMIT, chrome_binary=_pick_binary( "CHROME_BIN", "/usr/bin/chromium", "/usr/bin/chromium-browser", "chromium", "chromium-browser", ), chromedriver_path=_pick_binary( "CHROMEDRIVER_PATH", "/usr/bin/chromedriver", "chromedriver", ), browser_proxy_url=_build_browser_proxy_url(), schedule_timezone=os.getenv("APP_TIMEZONE", SCHEDULE_TIMEZONE), )