116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
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)
|