80 lines
3.4 KiB
Python
80 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import date
|
|
from http import HTTPStatus
|
|
from urllib.parse import parse_qs, unquote
|
|
from backend.bootstrap.config import validate_text
|
|
|
|
|
|
class HeavenRoutesMixin:
|
|
def _handle_heaven_get(self, parsed) -> bool:
|
|
if parsed.path == "/api/heaven/readings":
|
|
query = parse_qs(parsed.query)
|
|
try:
|
|
self.send_json(
|
|
self.application_service.heaven_readings(
|
|
query.get("mode", [""])[0],
|
|
query.get("context_date", [""])[0],
|
|
int(query.get("limit", ["100"])[0]),
|
|
)
|
|
)
|
|
except (TypeError, ValueError) as exc:
|
|
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
|
|
return True
|
|
if parsed.path == "/api/heaven/setup":
|
|
query = parse_qs(parsed.query)
|
|
trade_date = query.get("trade_date", [date.today().isoformat()])[0]
|
|
sector_name = query.get("sector", [""])[0]
|
|
stock_code = query.get("stock_code", [""])[0]
|
|
manual_data = None
|
|
manual_text = query.get("manual_data", [""])[0]
|
|
if manual_text:
|
|
try:
|
|
manual_data = json.loads(manual_text)
|
|
except json.JSONDecodeError:
|
|
self.send_json({"error": "六爻补录数据格式不正确。"}, HTTPStatus.BAD_REQUEST)
|
|
return True
|
|
try:
|
|
self.send_json(
|
|
self.application_service.heaven_setup(
|
|
trade_date,
|
|
sector_name,
|
|
stock_code,
|
|
manual_data,
|
|
)
|
|
)
|
|
except ValueError as exc:
|
|
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
|
|
return True
|
|
return False
|
|
|
|
def _handle_heaven_delete(self, parsed) -> bool:
|
|
heaven_reading_match = re.fullmatch(r"/api/heaven/readings/(\d+)", parsed.path)
|
|
if heaven_reading_match:
|
|
deleted = self.application_service.database.delete_heaven_reading(
|
|
self.application_service.current_user_id, int(heaven_reading_match.group(1))
|
|
)
|
|
self.send_json({"ok": True, "deleted": deleted})
|
|
return True
|
|
sector_phase_match = re.fullmatch(r"/api/heaven/sector-phases/(.+)", parsed.path)
|
|
if sector_phase_match:
|
|
name = unquote(sector_phase_match.group(1)).strip()
|
|
deleted = self.application_service.database.delete_sector_phase_override(name)
|
|
self.send_json({"ok": True, "deleted": deleted})
|
|
return True
|
|
return False
|
|
|
|
def save_sector_phase_override(self) -> None:
|
|
try:
|
|
body = self.read_json_body()
|
|
name = validate_text(body.get("name"), "行业或题材名称", 50, required=True)
|
|
element = str(body.get("element") or "").strip()
|
|
if element not in {"木", "火", "土", "金", "水"}:
|
|
raise ValueError("五行归类必须是木、火、土、金或水。")
|
|
self.application_service.database.save_sector_phase_override(name, element)
|
|
self.send_json({"ok": True})
|
|
except (ValueError, json.JSONDecodeError) as exc:
|
|
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
|