refactor: centralize runtime configuration and API access policy

ARCHITECTURE.md

@@ -0,0 +1,36 @@
+# Architecture
+
+The application intentionally keeps a small deployment footprint: one Python process, one
+SQLite database, and a build-free browser client. The internal boundaries are nevertheless
+explicit so new features do not bypass account isolation or data-quality rules.
+
+## Backend boundaries
+
+- `server.py`: application services and HTTP request/response wiring.
+- `api_access.py`: the single authorization policy for authenticated, member, and admin APIs.
+- `app_config.py`: runtime paths, local environment loading, and shared input validation.
+- `database.py`: SQLite schema, migrations, and persistence operations.
+- `tushare_client.py` and `realtime_aggregator.py`: external market-data adapters.
+- `sentiment_engine.py`, `screener.py`, and `heaven_engine.py`: deterministic domain logic.
+- `mentor_agent.py`, `heaven_agent.py`, and `llm_strategy.py`: bounded LLM adapters.
+
+## Data ownership
+
+Public market snapshots, stock factors, built-in strategies, limit-up reasons, seat aliases,
+and sector-element mappings are shared. Only administrators can modify shared knowledge.
+
+Watchlists, review notes, custom strategies, screener runs, mentor conversations, birth data,
+alerts, trading journals, and assistant conversations are owned by a user ID and must be
+queried with that ID. LLM features additionally require active membership.
+
+## Data integrity
+
+Production reads never synthesize market prices. A failed live request may use the latest real
+snapshot at or before the requested date. When no real snapshot exists, the API reports that
+the data is unavailable. Demo builders remain test fixtures only.
+
+## Change contract
+
+New endpoints must be added to `api_access.required_role` when they need member or admin
+access. New user-owned tables must include `user_id`, an ownership index, and cross-account
+tests. API payload compatibility is protected by the Python and Playwright suites.

api_access.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import re
+from typing import Literal
+
+
+AccessRole = Literal["authenticated", "member", "admin"]
+
+MEMBER_GET_PATHS = frozenset(
+    {
+        "/api/screener/setup",
+        "/api/mentors/setup",
+        "/api/mentors/messages",
+        "/api/heaven/setup",
+    }
+)
+
+MEMBER_POST_PATHS = frozenset(
+    {
+        "/api/screener/sync",
+        "/api/screener/compile",
+        "/api/screener/strategies",
+        "/api/screener/run",
+        "/api/mentors/chat",
+        "/api/heaven/hexagram",
+        "/api/heaven/personal",
+        "/api/heaven/interpret",
+    }
+)
+
+ADMIN_POST_PATHS = frozenset(
+    {
+        "/api/backfill",
+        "/api/reasons",
+        "/api/seat-aliases",
+        "/api/heaven/sector-phases",
+    }
+)
+
+
+def required_role(method: str, path: str) -> AccessRole:
+    method = method.upper()
+    if path.startswith("/api/admin/"):
+        return "admin"
+    if method == "GET" and path in MEMBER_GET_PATHS:
+        return "member"
+    if method == "POST":
+        if path in ADMIN_POST_PATHS:
+            return "admin"
+        if path in MEMBER_POST_PATHS:
+            return "member"
+    if method == "DELETE":
+        if re.fullmatch(r"/api/heaven/sector-phases/.+", path):
+            return "admin"
+        if path == "/api/mentors/messages" or re.fullmatch(
+            r"/api/screener/strategies/\d+", path
+        ):
+            return "member"
+    return "authenticated"

app_config.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import calendar
+import os
+import re
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+
+APP_DIR = Path(__file__).resolve().parent
+STATIC_DIR = APP_DIR / "static"
+DATA_DIR = APP_DIR / "data"
+ENV_FILE = APP_DIR / ".env"
+MENTOR_SKILLS_DIR = APP_DIR / "游资skills"
+TOKEN_PATTERN = re.compile(r"^[A-Za-z0-9_-]{20,128}$")
+USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9_\-\u4e00-\u9fff]{3,30}$")
+SESSION_COOKIE = "xiaobai_session"
+SESSION_MAX_AGE = 30 * 24 * 60 * 60
+
+
+def load_local_env() -> None:
+    if not ENV_FILE.exists():
+        return
+    for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
+
+
+def save_local_env(updates: dict[str, str]) -> None:
+    values: dict[str, str] = {}
+    if ENV_FILE.exists():
+        for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines():
+            if "=" in raw_line and not raw_line.lstrip().startswith("#"):
+                key, value = raw_line.split("=", 1)
+                values[key.strip()] = value.strip().strip('"').strip("'")
+    values.update(updates)
+    ENV_FILE.write_text(
+        "".join(f"{key}={value}\n" for key, value in values.items()),
+        encoding="utf-8",
+    )
+
+
+def remove_local_env(keys: set[str]) -> None:
+    if not ENV_FILE.exists():
+        return
+    kept = []
+    for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines():
+        if "=" in raw_line and not raw_line.lstrip().startswith("#"):
+            key = raw_line.split("=", 1)[0].strip()
+            if key in keys:
+                continue
+        kept.append(raw_line)
+    ENV_FILE.write_text("".join(f"{line}\n" for line in kept), encoding="utf-8")
+    for key in keys:
+        os.environ.pop(key, None)
+
+
+def normalize_date(value: str) -> str:
+    compact = value.replace("-", "").strip()
+    try:
+        parsed = datetime.strptime(compact, "%Y%m%d")
+    except ValueError as exc:
+        raise ValueError("日期格式应为 YYYY-MM-DD。") from exc
+    if parsed.date() > date.today():
+        raise ValueError("不能查询未来日期。")
+    return parsed.strftime("%Y%m%d")
+
+
+def validate_stock_code(value: str) -> str:
+    code = value.strip()
+    if not re.fullmatch(r"\d{6}", code):
+        raise ValueError("股票代码应为 6 位数字。")
+    return code
+
+
+def tushare_code(code: str) -> str:
+    if code.startswith(("4", "8", "9")):
+        suffix = "BJ"
+    elif code.startswith("6"):
+        suffix = "SH"
+    else:
+        suffix = "SZ"
+    return f"{code}.{suffix}"
+
+
+def validate_text(value: Any, label: str, maximum: int, required: bool = False) -> str:
+    text = str(value or "").strip()
+    if required and not text:
+        raise ValueError(f"{label}不能为空。")
+    if len(text) > maximum:
+        raise ValueError(f"{label}不能超过 {maximum} 个字符。")
+    return text
+
+
+def parse_iso_datetime(value: Any) -> datetime | None:
+    text = str(value or "").strip()
+    if not text:
+        return None
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
+
+
+def membership_boundary(value: Any, end: bool) -> str | None:
+    text = str(value or "").strip()
+    if not text:
+        return None
+    try:
+        day = datetime.strptime(text, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+    except ValueError as exc:
+        raise ValueError("会员日期格式应为 YYYY-MM-DD。") from exc
+    if end:
+        day += timedelta(days=1)
+    return day.isoformat(timespec="seconds")
+
+
+def add_months(value: datetime, months: int) -> datetime:
+    month_index = value.year * 12 + value.month - 1 + months
+    year, zero_based_month = divmod(month_index, 12)
+    month = zero_based_month + 1
+    day = min(value.day, calendar.monthrange(year, month)[1])
+    return value.replace(year=year, month=month, day=day)

server.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import argparse
-import calendar
 import copy
 import json
 import mimetypes
@@ -13,10 +12,29 @@ from datetime import date, datetime, timedelta, timezone
 from http import HTTPStatus
 from http.cookies import SimpleCookie
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qs, unquote, urlparse
 
+from api_access import required_role
+from app_config import (
+    DATA_DIR,
+    MENTOR_SKILLS_DIR,
+    SESSION_COOKIE,
+    SESSION_MAX_AGE,
+    STATIC_DIR,
+    TOKEN_PATTERN,
+    USERNAME_PATTERN,
+    add_months as _add_months,
+    load_local_env,
+    membership_boundary as _membership_boundary,
+    normalize_date,
+    parse_iso_datetime as _parse_iso_datetime,
+    remove_local_env,
+    save_local_env,
+    tushare_code,
+    validate_stock_code,
+    validate_text,
+)
 from database import ReviewDatabase
 from heaven_agent import HeavenAgentError, interpret_heaven
 from heaven_engine import (
@@ -47,15 +65,6 @@ from sentiment_engine import (
 from tushare_client import TushareClient, TushareError
 
 
-APP_DIR = Path(__file__).resolve().parent
-STATIC_DIR = APP_DIR / "static"
-DATA_DIR = APP_DIR / "data"
-ENV_FILE = APP_DIR / ".env"
-MENTOR_SKILLS_DIR = APP_DIR / "游资skills"
-TOKEN_PATTERN = re.compile(r"^[A-Za-z0-9_-]{20,128}$")
-USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9_\-\u4e00-\u9fff]{3,30}$")
-SESSION_COOKIE = "xiaobai_session"
-SESSION_MAX_AGE = 30 * 24 * 60 * 60
 LEGACY_SECRET_KEYS = {
     "TUSHARE_TOKEN",
     "LLM_API_KEY",
@@ -87,46 +96,6 @@ THS_SEARCH_TYPES = {
 }
 
 
-def load_local_env() -> None:
-    if not ENV_FILE.exists():
-        return
-    for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines():
-        line = raw_line.strip()
-        if not line or line.startswith("#") or "=" not in line:
-            continue
-        key, value = line.split("=", 1)
-        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
-
-
-def save_local_env(updates: dict[str, str]) -> None:
-    values: dict[str, str] = {}
-    if ENV_FILE.exists():
-        for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines():
-            if "=" in raw_line and not raw_line.lstrip().startswith("#"):
-                key, value = raw_line.split("=", 1)
-                values[key.strip()] = value.strip().strip('"').strip("'")
-    values.update(updates)
-    ENV_FILE.write_text(
-        "".join(f"{key}={value}\n" for key, value in values.items()),
-        encoding="utf-8",
-    )
-
-
-def remove_local_env(keys: set[str]) -> None:
-    if not ENV_FILE.exists():
-        return
-    kept = []
-    for raw_line in ENV_FILE.read_text(encoding="utf-8").splitlines():
-        if "=" in raw_line and not raw_line.lstrip().startswith("#"):
-            key = raw_line.split("=", 1)[0].strip()
-            if key in keys:
-                continue
-        kept.append(raw_line)
-    ENV_FILE.write_text("".join(f"{line}\n" for line in kept), encoding="utf-8")
-    for key in keys:
-        os.environ.pop(key, None)
-
-
 class DashboardService:
     def __init__(self) -> None:
         load_local_env()
@@ -3369,12 +3338,7 @@ class RequestHandler(BaseHTTPRequestHandler):
             return
         if parsed.path.startswith("/api/") and not self.require_auth():
             return
-        if parsed.path.startswith("/api/admin/") and not self.require_admin():
-            return
-        if parsed.path in {
-            "/api/screener/setup", "/api/mentors/setup", "/api/mentors/messages",
-            "/api/heaven/setup",
-        } and not self.require_member():
+        if not self.require_access("GET", parsed.path):
             return
         if parsed.path == "/api/admin/settings":
             self.send_json(
@@ -3570,15 +3534,7 @@ class RequestHandler(BaseHTTPRequestHandler):
             return
         if not self.require_auth() or not self.require_csrf():
             return
-        if parsed.path.startswith("/api/admin/") and not self.require_admin():
-            return
-        if parsed.path == "/api/backfill" and not self.require_admin():
-            return
-        if parsed.path in {
-            "/api/screener/sync", "/api/screener/compile", "/api/screener/strategies",
-            "/api/screener/run", "/api/mentors/chat", "/api/heaven/hexagram",
-            "/api/heaven/personal", "/api/heaven/interpret", "/api/heaven/sector-phases",
-        } and not self.require_member():
+        if not self.require_access("POST", parsed.path):
             return
         if parsed.path == "/api/auth/logout":
             self.auth_logout()
@@ -3608,18 +3564,12 @@ class RequestHandler(BaseHTTPRequestHandler):
             self.save_note()
             return
         if parsed.path == "/api/reasons":
-            if not self.require_admin():
-                return
             self.save_reason()
             return
         if parsed.path == "/api/seat-aliases":
-            if not self.require_admin():
-                return
             self.save_seat_alias()
             return
         if parsed.path == "/api/heaven/sector-phases":
-            if not self.require_admin():
-                return
             self.save_sector_phase_override()
             return
         if parsed.path == "/api/backfill":
@@ -3655,13 +3605,13 @@ class RequestHandler(BaseHTTPRequestHandler):
         parsed = urlparse(self.path)
         if not self.require_auth() or not self.require_csrf():
             return
+        if not self.require_access("DELETE", parsed.path):
+            return
         if parsed.path == "/api/account/birth-profile":
             deleted = SERVICE.database.delete_user_birth_profile(SERVICE.current_user_id)
             self.send_json({"ok": True, "deleted": deleted})
             return
         if parsed.path == "/api/mentors/messages":
-            if not self.require_member():
-                return
             query = parse_qs(parsed.query)
             try:
                 deleted = SERVICE.clear_mentor_messages(
@@ -3674,8 +3624,6 @@ class RequestHandler(BaseHTTPRequestHandler):
             return
         strategy_match = re.fullmatch(r"/api/screener/strategies/(\d+)", parsed.path)
         if strategy_match:
-            if not self.require_member():
-                return
             try:
                 result = SERVICE.delete_screener_strategy(int(strategy_match.group(1)))
                 self.send_json({"ok": True, **result})
@@ -3698,8 +3646,6 @@ class RequestHandler(BaseHTTPRequestHandler):
             return
         sector_phase_match = re.fullmatch(r"/api/heaven/sector-phases/(.+)", parsed.path)
         if sector_phase_match:
-            if not self.require_admin():
-                return
             name = unquote(sector_phase_match.group(1)).strip()
             deleted = SERVICE.database.delete_sector_phase_override(name)
             self.send_json({"ok": True, "deleted": deleted})
@@ -3842,6 +3788,14 @@ class RequestHandler(BaseHTTPRequestHandler):
         )
         return False
 
+    def require_access(self, method: str, path: str) -> bool:
+        role = required_role(method, path)
+        if role == "admin":
+            return self.require_admin()
+        if role == "member":
+            return self.require_member()
+        return True
+
     def session_cookie(self, value: str, clear: bool = False) -> str:
         max_age = 0 if clear else SESSION_MAX_AGE
         cookie = (
@@ -4151,75 +4105,6 @@ class RequestHandler(BaseHTTPRequestHandler):
         print(f"[{self.log_date_time_string()}] {format_string % args}")
 
 
-def normalize_date(value: str) -> str:
-    compact = value.replace("-", "").strip()
-    try:
-        parsed = datetime.strptime(compact, "%Y%m%d")
-    except ValueError as exc:
-        raise ValueError("日期格式应为 YYYY-MM-DD。") from exc
-    if parsed.date() > date.today():
-        raise ValueError("不能查询未来日期。")
-    return parsed.strftime("%Y%m%d")
-
-
-def validate_stock_code(value: str) -> str:
-    code = value.strip()
-    if not re.fullmatch(r"\d{6}", code):
-        raise ValueError("股票代码应为 6 位数字。")
-    return code
-
-
-def tushare_code(code: str) -> str:
-    if code.startswith(("4", "8", "9")):
-        suffix = "BJ"
-    elif code.startswith("6"):
-        suffix = "SH"
-    else:
-        suffix = "SZ"
-    return f"{code}.{suffix}"
-
-
-def validate_text(value: Any, label: str, maximum: int, required: bool = False) -> str:
-    text = str(value or "").strip()
-    if required and not text:
-        raise ValueError(f"{label}不能为空。")
-    if len(text) > maximum:
-        raise ValueError(f"{label}不能超过 {maximum} 个字符。")
-    return text
-
-
-def _parse_iso_datetime(value: Any) -> datetime | None:
-    text = str(value or "").strip()
-    if not text:
-        return None
-    try:
-        parsed = datetime.fromisoformat(text)
-    except ValueError:
-        return None
-    return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
-
-
-def _membership_boundary(value: Any, end: bool) -> str | None:
-    text = str(value or "").strip()
-    if not text:
-        return None
-    try:
-        day = datetime.strptime(text, "%Y-%m-%d").replace(tzinfo=timezone.utc)
-    except ValueError as exc:
-        raise ValueError("会员日期格式应为 YYYY-MM-DD。") from exc
-    if end:
-        day += timedelta(days=1)
-    return day.isoformat(timespec="seconds")
-
-
-def _add_months(value: datetime, months: int) -> datetime:
-    month_index = value.year * 12 + value.month - 1 + months
-    year, zero_based_month = divmod(month_index, 12)
-    month = zero_based_month + 1
-    day = min(value.day, calendar.monthrange(year, month)[1])
-    return value.replace(year=year, month=month, day=day)
-
-
 def main() -> None:
     parser = argparse.ArgumentParser(description="Xiaobai stock review web application")
     parser.add_argument("--host", default="127.0.0.1")

tests/test_api_access.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import unittest
+
+from api_access import required_role
+
+
+class ApiAccessPolicyTests(unittest.TestCase):
+    def test_member_workspaces_are_consistently_protected(self):
+        cases = {
+            ("GET", "/api/screener/setup"): "member",
+            ("GET", "/api/mentors/messages"): "member",
+            ("GET", "/api/heaven/setup"): "member",
+            ("POST", "/api/screener/run"): "member",
+            ("POST", "/api/mentors/chat"): "member",
+            ("POST", "/api/heaven/interpret"): "member",
+            ("DELETE", "/api/screener/strategies/42"): "member",
+            ("DELETE", "/api/mentors/messages"): "member",
+        }
+        for (method, path), role in cases.items():
+            with self.subTest(method=method, path=path):
+                self.assertEqual(required_role(method, path), role)
+
+    def test_shared_knowledge_mutations_require_admin(self):
+        cases = (
+            ("POST", "/api/reasons"),
+            ("POST", "/api/seat-aliases"),
+            ("POST", "/api/heaven/sector-phases"),
+            ("DELETE", "/api/heaven/sector-phases/油气开采"),
+            ("POST", "/api/backfill"),
+            ("GET", "/api/admin/settings"),
+        )
+        for method, path in cases:
+            with self.subTest(method=method, path=path):
+                self.assertEqual(required_role(method, path), "admin")
+
+    def test_personal_market_data_routes_need_login_only(self):
+        cases = (
+            ("GET", "/api/dashboard"),
+            ("GET", "/api/watchlist"),
+            ("POST", "/api/notes"),
+            ("DELETE", "/api/notes/3"),
+        )
+        for method, path in cases:
+            with self.subTest(method=method, path=path):
+                self.assertEqual(required_role(method, path), "authenticated")
+
+
+if __name__ == "__main__":
+    unittest.main()