File size: 6,051 Bytes
e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 a30f196 e28c9e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | 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
@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
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",
),
schedule_timezone=os.getenv("APP_TIMEZONE", SCHEDULE_TIMEZONE),
)
|