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", } 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 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)