228 lines
9.0 KiB
Python
228 lines
9.0 KiB
Python
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",
|
|
]:
|
|
loaded.update(_read_env_file(path))
|
|
for key, value in loaded.items():
|
|
os.environ.setdefault(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()
|