Spaces:
Paused
Paused
File size: 4,433 Bytes
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 | from __future__ import annotations
import json
import os
import secrets
import shutil
from dataclasses import dataclass
from pathlib import Path
from cryptography.fernet import Fernet
DEFAULT_PARALLEL_LIMIT = 2
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
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
@dataclass(slots=True)
class AppConfig:
root_dir: Path
data_dir: Path
db_path: Path
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
@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)
return cls(
root_dir=root_dir,
data_dir=data_dir,
db_path=data_dir / "course_catcher.db",
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",
),
)
|