from __future__ import annotations import os from functools import lru_cache from pathlib import Path ROOT_DIR = Path(__file__).resolve().parent.parent def _read_env_file(path: Path) -> dict[str, str]: values: dict[str, str] = {} if not path.exists(): return values for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key: values[key] = value return values def _load_env_files() -> None: loaded: dict[str, str] = {} for path in [ ROOT_DIR / ".env", ROOT_DIR / ".env.local", ROOT_DIR / "backend" / ".env", ROOT_DIR / "backend" / ".env.local", ROOT_DIR / "cursor.env", ]: loaded.update(_read_env_file(path)) for key, value in loaded.items(): os.environ[key] = value _load_env_files() def _split_csv(value: str) -> list[str]: return [item.strip() for item in value.split(",") if item.strip()] def _bool_env(name: str, default: str = "false") -> bool: return os.getenv(name, default).lower() in {"1", "true", "yes", "on"} def _is_local_service_host(host: str) -> bool: normalized = host.strip().lower() if not normalized: return True if normalized in {"127.0.0.1", "localhost", "::1"}: return True if normalized.startswith("192.168."): return True if normalized.startswith("10."): return True parts = normalized.split(".") if len(parts) == 4 and parts[0] == "172": try: second = int(parts[1]) if 16 <= second <= 31: return True except ValueError: pass return False def _resolve_minio_connection() -> tuple[str, int, bool, str]: root_domain = ( os.getenv("ROOT_DOMAIN") or os.getenv("NEXT_PUBLIC_ROOT_DOMAIN") or "nomadro.cn" ).strip() bucket = os.getenv("MINIO_BUCKET", "hackrobot") raw_endpoint = (os.getenv("MINIO_ENDPOINT", "127.0.0.1") or "127.0.0.1").strip() host_only = raw_endpoint.replace("https://", "").replace("http://", "").split("/")[0].split(":")[0] remote = not _is_local_service_host(host_only) if os.getenv("MINIO_USE_SSL") is not None: use_ssl = _bool_env("MINIO_USE_SSL", "false") else: use_ssl = remote port_raw = os.getenv("MINIO_PORT", "0") or "0" port = int(port_raw) if port <= 0: port = 443 if remote and use_ssl else 9100 if not remote else 80 public_url = ( os.getenv("MINIO_PUBLIC_URL") or os.getenv("NEXT_PUBLIC_MINIO_PUBLIC_URL") or f"https://minio.{root_domain}/{bucket}" ).rstrip("/") return host_only, port, use_ssl, public_url class Settings: def __init__(self) -> None: self.pb_url = ( os.getenv("PB_INTERNAL_URL") or os.getenv("POCKETBASE_INTERNAL_URL") or os.getenv("PB_URL") or os.getenv("POCKETBASE_URL") or "http://127.0.0.1:8090" ).rstrip("/") self.pb_admin_email = ( os.getenv("PB_ADMIN_EMAIL") or os.getenv("POCKETBASE_EMAIL") or "xiaoshuang.eric@gmail.com" ) self.pb_admin_password = ( os.getenv("PB_ADMIN_PASSWORD") or os.getenv("POCKETBASE_PASSWORD") or "Xiao4669805@" ) self.cookie_name = os.getenv("API_COOKIE_NAME", "pb_session") self.frontend_origins = _split_csv( os.getenv( "FRONTEND_ORIGINS", "http://localhost:3001,http://127.0.0.1:3001", ) ) self.upload_dir = Path(os.getenv("UPLOAD_DIR", "backend/uploads")) self.minio_enabled = _bool_env("MINIO_ENABLED", "true") raw_minio_endpoint = os.getenv("MINIO_ENDPOINT", "127.0.0.1").strip() if "://" in raw_minio_endpoint: raw_minio_endpoint = raw_minio_endpoint.split("://", 1)[1] raw_minio_endpoint = raw_minio_endpoint.split("/", 1)[0] if ":" in raw_minio_endpoint and not raw_minio_endpoint.startswith("["): host_part, port_part = raw_minio_endpoint.rsplit(":", 1) if port_part.isdigit(): raw_minio_endpoint = host_part if not os.getenv("MINIO_PORT"): os.environ.setdefault("MINIO_PORT", port_part) self.minio_endpoint = raw_minio_endpoint or "127.0.0.1" is_local_minio = self.minio_endpoint in {"127.0.0.1", "localhost"} default_minio_port = "9100" if is_local_minio else "443" default_minio_ssl = "false" if is_local_minio else "true" self.minio_port = int(os.getenv("MINIO_PORT", default_minio_port) or default_minio_port) self.minio_use_ssl = _bool_env("MINIO_USE_SSL", default_minio_ssl) self.minio_bucket = os.getenv("MINIO_BUCKET", "hackrobot") self.minio_access_key = os.getenv("MINIO_ACCESS_KEY", "nomadro") self.minio_secret_key = os.getenv("MINIO_SECRET_KEY", "Xiao4669805@") self.minio_region = os.getenv("MINIO_REGION", "china") self.minio_public_url = ( os.getenv("MINIO_PUBLIC_URL") or f"https://minio.nomadro.cn/{self.minio_bucket}" ).rstrip("/") self.minio_upload_prefix = os.getenv("MINIO_UPLOAD_PREFIX", "joins").strip("/").strip() self.app_env = os.getenv("APP_ENV", "development").strip().lower() self.dev_auto_pay = _bool_env("DEV_AUTO_PAY", "false") self.site_id = os.getenv("SITE_ID") or os.getenv("NEXT_PUBLIC_SITE_ID") or "meetup" self.base_url = ( os.getenv("API_PUBLIC_BASE_URL") or os.getenv("BASE_URL") or os.getenv("PAYMENT_API_URL") or os.getenv("NEXT_PUBLIC_API_BASE_URL") or "http://127.0.0.1:8000" ).rstrip("/") if self.base_url == "/api": self.base_url = (os.getenv("NEXT_PUBLIC_SITE_URL") or "http://127.0.0.1:3001").rstrip("/") self.upload_public_base_url = ( os.getenv("UPLOAD_PUBLIC_BASE_URL") or os.getenv("API_PUBLIC_BASE_URL") or os.getenv("BASE_URL") or self.base_url ).rstrip("/") self.email_auth_password = os.getenv("EMAIL_AUTH_PASSWORD", "12345678") self.google_client_id = os.getenv("GOOGLE_CLIENT_ID") or os.getenv("NEXT_PUBLIC_GOOGLE_CLIENT_ID") or "" self.payment_provider = os.getenv("PAYMENT_PROVIDER", "zpay").lower() self.payment_default_amount = int(os.getenv("PAYMENT_DEFAULT_AMOUNT", "100") or "100") self.payment_join_amount = int(os.getenv("PAYMENT_JOIN_AMOUNT", str(self.payment_default_amount)) or str(self.payment_default_amount)) self.payment_join_type = os.getenv("PAYMENT_JOIN_TYPE", "meetup") self.zpay_pid = os.getenv("ZPAY_PID", "2025121809351743") self.zpay_key = os.getenv("ZPAY_KEY", "tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89") self.zpay_submit_url = os.getenv("ZPAY_SUBMIT_URL", "https://zpayz.cn/submit.php") self.zpay_mapi_url = os.getenv("ZPAY_MAPI_URL", "https://zpayz.cn/mapi.php") self.zpay_query_url = os.getenv("ZPAY_QUERY_URL", "https://zpayz.cn/api.php") self.xorpay_aid = os.getenv("XORPAY_AID", "8220") self.xorpay_secret = os.getenv("XORPAY_SECRET", "afcacd99570945f88de62624aaa3578e") self.xorpay_cashier_url = os.getenv("XORPAY_CASHIER_URL", "https://xorpay.com/api/cashier") self.xorpay_query_url = os.getenv("XORPAY_QUERY_URL", "https://xorpay.com/api/query2") self.mirotalk_url = ( os.getenv("MIROTALK_URL") or os.getenv("NEXT_PUBLIC_MIROTALK_URL") or "https://mirotalk.nomadro.cn" ).rstrip("/") self.lounge_url = ( os.getenv("LOUNGE_URL") or os.getenv("NEXT_PUBLIC_LOUNGE_URL") or "https://lounge.nomadro.cn" ).rstrip("/") self.listmonk_url = ( os.getenv("LISTMONK_URL") or os.getenv("NEXT_PUBLIC_LISTMONK_URL") or "https://listmonk.nomadro.cn" ).rstrip("/") self.listmonk_api_url = ( os.getenv("LISTMONK_API_URL") or self.listmonk_url ).rstrip("/") self.listmonk_api_key = os.getenv("LISTMONK_API_KEY", "").strip() list_ids_raw = os.getenv("LISTMONK_DEFAULT_LIST_IDS", "1") self.listmonk_default_list_ids = [ int(x.strip()) for x in list_ids_raw.split(",") if x.strip().isdigit() ] or [1] self.ntfy_url = ( os.getenv("NTFY_URL") or os.getenv("NEXT_PUBLIC_NTFY_URL") or "https://ntfy.nomadro.cn" ).rstrip("/") self.ntfy_auth_token = os.getenv("NTFY_AUTH_TOKEN", "").strip() self.ntfy_enabled = _bool_env("NTFY_ENABLED", "true") self.site_public_url = ( os.getenv("NEXT_PUBLIC_SITE_URL") or os.getenv("SITE_PUBLIC_URL") or "https://nomadro.cn" ).rstrip("/") @lru_cache(maxsize=1) def get_settings() -> Settings: return Settings()