sten.wtf / Unstable Instance

config.py

import asyncio
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping, Optional

ROOT = Path(__file__).resolve().parent
DEFAULT_META_HOST = "127.0.0.1"
DEFAULT_META_PORT = 8787
DEFAULT_DB_PATH = str(ROOT / "db" / "open_autosecure.db")
DEFAULT_TELEMETRY_URL = "https://sten.wtf/api/oas/telemetry"


def prepare_runtime() -> None:
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())


def resolve_db_path(raw: str | None) -> str:
    if not raw:
        return DEFAULT_DB_PATH
    p = Path(raw)
    if p.is_absolute():
        return raw
    return str(ROOT / p)


prepare_runtime()


@dataclass
class InstanceConfig:
    instance_id: Optional[str]
    security_email: Optional[str]
    imap_host: Optional[str]
    imap_port: Optional[int]
    imap_user: Optional[str]
    imap_password: Optional[str]
    imap_auth: Optional[str]
    imap_provider: Optional[str]
    imap_refresh_token: Optional[str]
    telemetry_enabled: bool
    telemetry_url: Optional[str]
    proxy_url: Optional[str]
    build_version: Optional[str]
    meta_host: str
    meta_port: int
    db_path: str


def resolve(db_row: dict | None, env: Mapping[str, str]) -> InstanceConfig:
    row = db_row or {}

    meta_host = env.get("OAS_META_HOST") or DEFAULT_META_HOST
    meta_port = env.get("OAS_META_PORT") or DEFAULT_META_PORT
    db_path = resolve_db_path(env.get("OAS_DB_PATH"))

    imap_port = row.get("imap_port")
    configured = bool(row.get("instance_id") or row.get("security_email"))

    return InstanceConfig(
        instance_id=row.get("instance_id"),
        security_email=row.get("security_email"),
        imap_host=row.get("imap_host"),
        imap_port=int(imap_port) if imap_port is not None else None,
        imap_user=row.get("imap_user"),
        imap_password=row.get("imap_password"),
        imap_auth=row.get("imap_auth") or ("xoauth2" if row.get("imap_refresh_token") else "password"),
        imap_provider=row.get("imap_provider"),
        imap_refresh_token=row.get("imap_refresh_token"),
        telemetry_enabled=configured,
        telemetry_url=DEFAULT_TELEMETRY_URL if configured else None,
        proxy_url=row.get("proxy_url"),
        build_version=row.get("build_version"),
        meta_host=str(meta_host),
        meta_port=int(meta_port),
        db_path=str(db_path),
    )