xiaobai-review/backend/http/dispatch.py

from __future__ import annotations

from http import HTTPStatus
from urllib.parse import urlparse


PUBLIC_POST_HANDLERS = {
    "/api/auth/register": "auth_register",
    "/api/auth/login": "auth_login",
    "/api/auth/switch": "auth_switch",
    "/api/auth/forget": "auth_forget",
}

# Service-to-service bridge for the data hub console. These paths are guarded by
# the shared HUB_ADMIN_TOKEN header instead of a browser session, so they stay
# out of the user-facing route registry on purpose.
HUB_SERVICE_HANDLERS = {
    "/api/hub-admin/session": "hub_session_check",
    "/api/hub-admin/session/logout": "hub_session_logout",
    "/api/hub-admin/password/check": "hub_password_check",
    "/api/hub-admin/status": "hub_system_status",
    "/api/hub-admin/settings/save": "hub_save_settings",
    "/api/hub-admin/settings/test": "hub_test_model",
    "/api/hub-admin/models/fetch": "hub_fetch_models",
    "/api/hub-admin/members": "hub_members",
    "/api/hub-admin/membership/save": "hub_save_membership",
    "/api/hub-admin/invites": "hub_invites",
    "/api/hub-admin/invites/create": "hub_create_invites",
    "/api/hub-admin/invites/revoke": "hub_revoke_invite",
}

AUTHENTICATED_POST_HANDLERS = {
    "/api/auth/logout": "auth_logout",
    "/api/account/birth-profile": "save_birth_profile",
    "/api/account/password": "change_password",
    "/api/alerts": "save_alert",
    "/api/trades": "save_trade_entry",
    "/api/assistant/chat": "stream_assistant_chat",
    "/api/admin/settings": "save_system_settings",
    "/api/admin/settings/test": "test_system_llm_settings",
    "/api/admin/membership": "save_membership",
    "/api/admin/refresh": "start_background_refresh",
    "/api/watchlist": "save_watchlist",
    "/api/notes": "save_note",
    "/api/reasons": "save_reason",
    "/api/seat-aliases": "save_seat_alias",
    "/api/heaven/sector-phases": "save_sector_phase_override",
    "/api/backfill": "backfill_data",
    "/api/screener/sync": "sync_screener_data",
    "/api/screener/compile": "compile_screener_strategy",
    "/api/screener/strategies": "save_screener_strategy",
    "/api/screener/run": "run_screener",
    "/api/screener/tracking/refresh": "refresh_screener_tracking",
    "/api/mentors/chat": "stream_mentor_chat",
    "/api/heaven/hexagram": "heaven_hexagram",
    "/api/heaven/personal": "heaven_personal",
    "/api/heaven/interpret": "heaven_interpret",
}


class ApplicationHttpDispatchMixin:
    def _dispatch_named_handler(self, path: str, handlers: dict[str, str]) -> bool:
        handler_name = handlers.get(path)
        if handler_name is None:
            return False
        getattr(self, handler_name)()
        return True

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if self._handle_system_public_get(parsed):
            return
        if self._handle_accounts_public_get(parsed):
            return
        if parsed.path.startswith("/api/"):
            if not self.require_auth():
                return
            if not self.require_access("GET", parsed.path):
                return
        for handler in (
            self._handle_system_get,
            self._handle_accounts_get,
            self._handle_alerts_get,
            self._handle_review_get,
            self._handle_market_get,
            self._handle_auction_get,
            self._handle_themes_get,
            self._handle_popularity_get,
            self._handle_sentiment_get,
            self._handle_rotation_get,
            self._handle_dragon_tiger_get,
            self._handle_screener_get,
            self._handle_mentor_get,
            self._handle_heaven_get,
        ):
            if handler(parsed):
                return
        self.serve_static(parsed.path)

    def do_POST(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path in HUB_SERVICE_HANDLERS:
            if self.require_service_token():
                self._dispatch_named_handler(parsed.path, HUB_SERVICE_HANDLERS)
            return
        if self._dispatch_named_handler(parsed.path, PUBLIC_POST_HANDLERS):
            return
        if not self.require_auth() or not self.require_csrf():
            return
        if not self.require_access("POST", parsed.path):
            return
        if self._dispatch_named_handler(parsed.path, AUTHENTICATED_POST_HANDLERS):
            return
        for handler in (
            self._handle_alerts_post,
            self._handle_screener_post,
            self._handle_mentor_post,
        ):
            if handler(parsed):
                return
        self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND)

    def do_DELETE(self) -> None:
        parsed = urlparse(self.path)
        if not self.require_auth() or not self.require_csrf():
            return
        if not self.require_access("DELETE", parsed.path):
            return
        for handler in (
            self._handle_accounts_delete,
            self._handle_review_delete,
            self._handle_mentor_delete,
            self._handle_screener_delete,
            self._handle_alerts_delete,
            self._handle_heaven_delete,
        ):
            if handler(parsed):
                return
        self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND)