migration: preserve startup accounts and system slice

app/app_config.py

@@ -1,129 +1,3 @@
-from __future__ import annotations
+"""Compatibility imports for code that still uses the original configuration module."""
 
-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"
-PRIVATE_MENTOR_SKILLS_DIR = DATA_DIR / "private-mentor-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)
+from backend.bootstrap.config import *  # noqa: F401,F403

app/backend/bootstrap/__init__.py

@@ -1,9 +1,23 @@
-from .container import ApplicationContainer, build_application_container
-from .settings import RuntimeSettings, load_runtime_settings
-
 __all__ = [
     "ApplicationContainer",
     "RuntimeSettings",
     "build_application_container",
     "load_runtime_settings",
+    "main",
 ]
+
+
+def __getattr__(name: str):
+    if name in {"ApplicationContainer", "build_application_container"}:
+        from . import container
+
+        return getattr(container, name)
+    if name in {"RuntimeSettings", "load_runtime_settings"}:
+        from . import settings
+
+        return getattr(settings, name)
+    if name == "main":
+        from .runtime import main
+
+        return main
+    raise AttributeError(name)

app/backend/bootstrap/config.py

@@ -0,0 +1,129 @@
+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().parents[2]
+STATIC_DIR = APP_DIR / "static"
+DATA_DIR = APP_DIR / "data"
+ENV_FILE = APP_DIR / ".env"
+MENTOR_SKILLS_DIR = APP_DIR / "游资skills"
+PRIVATE_MENTOR_SKILLS_DIR = DATA_DIR / "private-mentor-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)

app/backend/bootstrap/runtime.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import argparse
+from http.server import ThreadingHTTPServer
+from typing import Any
+
+
+def main(handler_class: type[Any] | None = None, service: Any | None = None) -> None:
+    if handler_class is None or service is None:
+        from backend.application import RequestHandler, SERVICE
+
+        handler_class = handler_class or RequestHandler
+        service = service or SERVICE
+    parser = argparse.ArgumentParser(description="Xiaobai stock review web application")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=8765)
+    args = parser.parse_args()
+    server = ThreadingHTTPServer((args.host, args.port), handler_class)
+    print(f"Xiaobai Review Web is running at http://{args.host}:{args.port}")
+    print("Press Ctrl+C to stop.")
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        service._background_stop.set()
+        server.server_close()

app/backend/bootstrap/settings.py

@@ -4,8 +4,8 @@ import os
 from dataclasses import dataclass
 from typing import Mapping
 
-from app_config import load_local_env, save_local_env
-from security import SecretVault
+from backend.bootstrap.config import load_local_env, save_local_env
+from backend.features.accounts.security import SecretVault
 
 
 def environment_credentials(environment: Mapping[str, str]) -> dict[str, str]:

app/backend/data/policy.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
-from app_config import APP_DIR
+from backend.bootstrap.config import APP_DIR
 from backend.data.contracts import DataUsage, DatasetContract, ProviderContract
 
 

app/backend/data/quality.py

@@ -7,7 +7,7 @@ from pathlib import Path
 from typing import Any
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
-from app_config import APP_DIR
+from backend.bootstrap.config import APP_DIR
 from backend.data.contracts import DataUsage
 from backend.data.policy import DataPolicyError, DataSourcePolicy
 

app/backend/features/accounts/__init__.py

@@ -0,0 +1,24 @@
+__all__ = [
+    "AccountHttpMixin",
+    "AccountService",
+    "SecretVault",
+    "hash_password",
+    "token_hash",
+    "verify_password",
+]
+
+
+def __getattr__(name: str):
+    if name == "AccountHttpMixin":
+        from .http import AccountHttpMixin
+
+        return AccountHttpMixin
+    if name == "AccountService":
+        from .service import AccountService
+
+        return AccountService
+    if name in {"SecretVault", "hash_password", "token_hash", "verify_password"}:
+        from . import security
+
+        return getattr(security, name)
+    raise AttributeError(name)

app/backend/features/accounts/http.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import json
+from http import HTTPStatus
+
+
+class AccountHttpMixin:
+    def auth_register(self) -> None:
+        try:
+            body = self.read_json_body()
+            result = self.application_service.register_account(
+                str(body.get("username") or ""),
+                str(body.get("password") or ""),
+            )
+            self.send_json(
+                {
+                    "ok": True,
+                    "authenticated": True,
+                    "user": result["user"],
+                    "csrf_token": result["csrf_token"],
+                },
+                HTTPStatus.CREATED,
+                {"Set-Cookie": self.session_cookie(result["session_token"])},
+            )
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
+
+    def auth_login(self) -> None:
+        try:
+            body = self.read_json_body()
+            result = self.application_service.login_account(
+                str(body.get("username") or ""),
+                str(body.get("password") or ""),
+            )
+            self.send_json(
+                {
+                    "ok": True,
+                    "authenticated": True,
+                    "user": result["user"],
+                    "csrf_token": result["csrf_token"],
+                },
+                headers={"Set-Cookie": self.session_cookie(result["session_token"])},
+            )
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.UNAUTHORIZED)
+
+    def auth_me(self) -> None:
+        service = self.application_service
+        if not self.require_auth(send_error=False):
+            self.send_json(
+                {
+                    "ok": True,
+                    "authenticated": False,
+                    "registration_required": service.database.count_users() == 0,
+                }
+            )
+            return
+        self.send_json(
+            {
+                "ok": True,
+                "authenticated": True,
+                "user": {
+                    "id": int(self.auth_user["id"]),
+                    "username": str(self.auth_user["username"]),
+                    "role": str(self.auth_user.get("role") or "user"),
+                    "membership": service.membership(),
+                },
+                "csrf_token": str(self.auth_user["csrf_token"]),
+            }
+        )
+
+    def auth_logout(self) -> None:
+        raw_token = self.session_token()
+        if raw_token:
+            from backend.features.accounts.security import token_hash
+
+            self.application_service.database.delete_session(token_hash(raw_token))
+        self.send_json(
+            {"ok": True},
+            headers={"Set-Cookie": self.session_cookie("", clear=True)},
+        )
+
+    def save_birth_profile(self) -> None:
+        try:
+            body = self.read_json_body()
+            personal = self.application_service.save_birth_profile(body)
+            self.send_json({"ok": True, "personal": personal})
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
+
+    def change_password(self) -> None:
+        try:
+            body = self.read_json_body()
+            current = str(body.get("current_password") or "")
+            new = str(body.get("new_password") or "")
+            confirmation = str(body.get("confirm_password") or "")
+            if new != confirmation:
+                raise ValueError("两次输入的新密码不一致。")
+            self.application_service.change_password(current, new)
+            self.send_json({"ok": True})
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
+
+    def save_membership(self) -> None:
+        try:
+            service = self.application_service
+            service.update_membership(self.read_json_body())
+            self.send_json({"ok": True, "users": service.admin_users()})
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)

app/backend/features/accounts/repository.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import sqlite3
+from datetime import datetime, timezone
+from typing import Any
+
+
+class AccountRepositoryMixin:
+    """Original SQLite account persistence methods, moved without query changes."""
+
+    def count_users(self) -> int:
+        with self.connect() as connection:
+            row = connection.execute("SELECT COUNT(*) AS total FROM users").fetchone()
+        return int(row["total"] if row else 0)
+
+    def first_user_id(self) -> int:
+        with self.connect() as connection:
+            row = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
+        return int(row["id"] or 0) if row else 0
+
+    def create_user(
+        self,
+        username: str,
+        password_salt: str,
+        password_hash: str,
+    ) -> dict[str, Any]:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        try:
+            with self.connect() as connection:
+                role = "admin" if int(connection.execute("SELECT COUNT(*) FROM users").fetchone()[0]) == 0 else "user"
+                cursor = connection.execute(
+                    """
+                    INSERT INTO users
+                        (username, password_salt, password_hash, role, created_at, updated_at)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                    """,
+                    (username, password_salt, password_hash, role, now, now),
+                )
+                user_id = int(cursor.lastrowid)
+        except sqlite3.IntegrityError as exc:
+            raise ValueError("该账号名已被使用。") from exc
+        return {"id": user_id, "username": username, "role": role, "created_at": now}
+
+    def user_by_username(self, username: str) -> dict[str, Any] | None:
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT id, username, password_salt, password_hash, role, llm_mode,
+                       membership_status, membership_plan, membership_starts_at,
+                       membership_expires_at, created_at
+                FROM users WHERE username = ? COLLATE NOCASE
+                """,
+                (username,),
+            ).fetchone()
+        return dict(row) if row else None
+
+    def user_password(self, user_id: int) -> dict[str, str] | None:
+        with self.connect() as connection:
+            row = connection.execute(
+                "SELECT password_salt, password_hash FROM users WHERE id = ?",
+                (user_id,),
+            ).fetchone()
+        return dict(row) if row else None
+
+    def update_user_password(self, user_id: int, password_salt: str, password_hash: str) -> bool:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            cursor = connection.execute(
+                "UPDATE users SET password_salt = ?, password_hash = ?, updated_at = ? WHERE id = ?",
+                (password_salt, password_hash, now, user_id),
+            )
+        return cursor.rowcount > 0
+
+    def delete_user(self, user_id: int) -> bool:
+        with self.connect() as connection:
+            cursor = connection.execute("DELETE FROM users WHERE id = ?", (user_id,))
+            return cursor.rowcount > 0
+
+    def create_session(
+        self,
+        session_hash: str,
+        user_id: int,
+        csrf_token: str,
+        expires_at: str,
+    ) -> None:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            connection.execute("DELETE FROM user_sessions WHERE expires_at <= ?", (now,))
+            connection.execute(
+                """
+                INSERT INTO user_sessions
+                    (token_hash, user_id, csrf_token, expires_at, created_at, last_seen_at)
+                VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (session_hash, user_id, csrf_token, expires_at, now, now),
+            )
+
+    def session_user(self, session_hash: str) -> dict[str, Any] | None:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT u.id, u.username, u.role, u.llm_mode, u.membership_status,
+                       u.membership_plan, u.membership_starts_at, u.membership_expires_at,
+                       u.created_at, s.csrf_token, s.expires_at
+                FROM user_sessions AS s
+                JOIN users AS u ON u.id = s.user_id
+                WHERE s.token_hash = ? AND s.expires_at > ?
+                """,
+                (session_hash, now),
+            ).fetchone()
+            if row:
+                connection.execute(
+                    "UPDATE user_sessions SET last_seen_at = ? WHERE token_hash = ?",
+                    (now, session_hash),
+                )
+        return dict(row) if row else None
+
+    def delete_session(self, session_hash: str) -> bool:
+        with self.connect() as connection:
+            cursor = connection.execute(
+                "DELETE FROM user_sessions WHERE token_hash = ?",
+                (session_hash,),
+            )
+            return cursor.rowcount > 0
+
+    def get_user_credentials(self, user_id: int) -> str:
+        with self.connect() as connection:
+            row = connection.execute(
+                "SELECT encrypted_payload FROM user_credentials WHERE user_id = ?",
+                (user_id,),
+            ).fetchone()
+        return str(row["encrypted_payload"]) if row else ""
+
+    def save_user_credentials(self, user_id: int, encrypted_payload: str) -> None:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO user_credentials (user_id, encrypted_payload, updated_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(user_id) DO UPDATE SET
+                    encrypted_payload = excluded.encrypted_payload,
+                    updated_at = excluded.updated_at
+                """,
+                (user_id, encrypted_payload, now),
+            )
+
+    def list_user_credentials(self) -> list[dict[str, Any]]:
+        with self.connect() as connection:
+            rows = connection.execute(
+                "SELECT user_id, encrypted_payload FROM user_credentials ORDER BY user_id"
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def user_access(self, user_id: int) -> dict[str, Any] | None:
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                SELECT id, username, role, llm_mode, membership_status, membership_plan,
+                       membership_starts_at, membership_expires_at, created_at
+                FROM users WHERE id = ?
+                """,
+                (user_id,),
+            ).fetchone()
+        return dict(row) if row else None
+
+    def list_users(self) -> list[dict[str, Any]]:
+        with self.connect() as connection:
+            rows = connection.execute(
+                """
+                SELECT id, username, role, llm_mode, membership_status, membership_plan,
+                       membership_starts_at, membership_expires_at, created_at
+                FROM users ORDER BY id
+                """
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def update_user_llm_mode(self, user_id: int, mode: str) -> None:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            connection.execute(
+                "UPDATE users SET llm_mode = ?, updated_at = ? WHERE id = ?",
+                (mode, now, user_id),
+            )
+
+    def update_membership(
+        self,
+        user_id: int,
+        status: str,
+        plan: str,
+        starts_at: str | None,
+        expires_at: str | None,
+    ) -> bool:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            cursor = connection.execute(
+                """
+                UPDATE users
+                SET membership_status = ?, membership_plan = ?,
+                    membership_starts_at = ?, membership_expires_at = ?, updated_at = ?
+                WHERE id = ?
+                """,
+                (status, plan, starts_at, expires_at, now, user_id),
+            )
+        return cursor.rowcount > 0
+
+    def get_user_birth_profile(self, user_id: int) -> str:
+        with self.connect() as connection:
+            row = connection.execute(
+                "SELECT encrypted_payload FROM user_birth_profiles WHERE user_id = ?",
+                (user_id,),
+            ).fetchone()
+        return str(row["encrypted_payload"]) if row else ""
+
+    def save_user_birth_profile(self, user_id: int, encrypted_payload: str) -> None:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO user_birth_profiles (user_id, encrypted_payload, updated_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(user_id) DO UPDATE SET
+                    encrypted_payload = excluded.encrypted_payload,
+                    updated_at = excluded.updated_at
+                """,
+                (user_id, encrypted_payload, now),
+            )
+
+    def delete_user_birth_profile(self, user_id: int) -> bool:
+        with self.connect() as connection:
+            cursor = connection.execute(
+                "DELETE FROM user_birth_profiles WHERE user_id = ?",
+                (user_id,),
+            )
+            return cursor.rowcount > 0

app/backend/features/accounts/security.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+from typing import Any
+
+from cryptography.fernet import Fernet, InvalidToken
+
+
+PASSWORD_SCRYPT_N = 2**14
+PASSWORD_SCRYPT_R = 8
+PASSWORD_SCRYPT_P = 1
+
+
+class SecretVault:
+    def __init__(self, key: str) -> None:
+        try:
+            self._fernet = Fernet(key.encode("ascii"))
+        except (ValueError, TypeError) as exc:
+            raise ValueError("APP_ENCRYPTION_KEY 格式无效。") from exc
+
+    @staticmethod
+    def generate_key() -> str:
+        return Fernet.generate_key().decode("ascii")
+
+    def encrypt_json(self, payload: dict[str, Any]) -> str:
+        raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+        return self._fernet.encrypt(raw).decode("ascii")
+
+    def decrypt_json(self, token: str) -> dict[str, Any]:
+        if not token:
+            return {}
+        try:
+            payload = json.loads(self._fernet.decrypt(token.encode("ascii")).decode("utf-8"))
+        except (InvalidToken, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise ValueError("账号加密数据无法解密，请检查 APP_ENCRYPTION_KEY。") from exc
+        if not isinstance(payload, dict):
+            raise ValueError("账号加密数据格式无效。")
+        return payload
+
+
+def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]:
+    raw_salt = salt or os.urandom(16)
+    digest = hashlib.scrypt(
+        password.encode("utf-8"),
+        salt=raw_salt,
+        n=PASSWORD_SCRYPT_N,
+        r=PASSWORD_SCRYPT_R,
+        p=PASSWORD_SCRYPT_P,
+        dklen=32,
+    )
+    return (
+        base64.urlsafe_b64encode(raw_salt).decode("ascii"),
+        base64.urlsafe_b64encode(digest).decode("ascii"),
+    )
+
+
+def verify_password(password: str, salt_text: str, expected_hash: str) -> bool:
+    try:
+        salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
+        _, actual_hash = hash_password(password, salt)
+    except (ValueError, TypeError):
+        return False
+    return hmac.compare_digest(actual_hash, expected_hash)
+
+
+def token_hash(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()

app/backend/features/accounts/service.py

@@ -0,0 +1,256 @@
+from __future__ import annotations
+
+import secrets
+import threading
+from collections.abc import Callable
+from datetime import date, datetime, timedelta, timezone
+from typing import Any
+
+from backend.bootstrap.config import (
+    SESSION_MAX_AGE,
+    USERNAME_PATTERN,
+    add_months,
+    normalize_date,
+    parse_iso_datetime,
+)
+from backend.features.accounts.security import (
+    SecretVault,
+    hash_password,
+    token_hash,
+    verify_password,
+)
+
+
+class AccountService:
+    """Preserved account, session, membership and birth-profile behavior."""
+
+    def __init__(
+        self,
+        database: Any,
+        vault: SecretVault,
+        current_user_supplier: Callable[[], int],
+        access_supplier: Callable[[], dict[str, Any]],
+        bind_user: Callable[[int], None],
+        personal_field_builder: Callable[..., dict[str, Any]],
+        auth_lock: threading.Lock,
+    ) -> None:
+        self.database = database
+        self.vault = vault
+        self.current_user_supplier = current_user_supplier
+        self.access_supplier = access_supplier
+        self.bind_user = bind_user
+        self.personal_field_builder = personal_field_builder
+        self.auth_lock = auth_lock
+
+    @property
+    def current_user_id(self) -> int:
+        return int(self.current_user_supplier())
+
+    @staticmethod
+    def membership_for_access(access: dict[str, Any]) -> dict[str, Any]:
+        now = datetime.now(timezone.utc)
+        starts = parse_iso_datetime(access.get("membership_starts_at"))
+        expires = parse_iso_datetime(access.get("membership_expires_at"))
+        subscribed = (
+            access.get("membership_status") == "active"
+            and (not starts or starts <= now)
+            and (not expires or expires > now)
+        )
+        is_admin = str(access.get("role")) == "admin"
+        active = is_admin or subscribed
+        remaining_seconds = None
+        if expires:
+            remaining_seconds = max(0, int((expires - now).total_seconds()))
+        return {
+            "active": active,
+            "subscribed": subscribed,
+            "status": "active" if subscribed else str(access.get("membership_status") or "inactive"),
+            "plan": str(access.get("membership_plan") or ""),
+            "starts_at": str(access.get("membership_starts_at") or ""),
+            "expires_at": str(access.get("membership_expires_at") or ""),
+            "is_admin": is_admin,
+            "remaining_seconds": remaining_seconds,
+            "remaining_days": None if remaining_seconds is None else (remaining_seconds + 86399) // 86400,
+        }
+
+    def membership(self) -> dict[str, Any]:
+        access = self.access_supplier() or self.database.user_access(self.current_user_id) or {}
+        return self.membership_for_access(access)
+
+    def register(self, username: str, password: str) -> dict[str, Any]:
+        username = username.strip()
+        self.validate_input(username, password)
+        with self.auth_lock:
+            salt, password_digest = hash_password(password)
+            user = self.database.create_user(username, salt, password_digest)
+        return self.create_session(user)
+
+    def login(self, username: str, password: str) -> dict[str, Any]:
+        username = username.strip()
+        if not username or not password:
+            raise ValueError("账号名和密码不能为空。")
+        user = self.database.user_by_username(username)
+        if not user or not verify_password(
+            password,
+            str(user.get("password_salt") or ""),
+            str(user.get("password_hash") or ""),
+        ):
+            raise ValueError("账号名或密码不正确。")
+        return self.create_session(user)
+
+    def change_password(self, current_password: str, new_password: str) -> None:
+        current_password = str(current_password or "")
+        access = self.database.user_access(self.current_user_id)
+        self.validate_input(str(access["username"]), new_password)
+        credentials = self.database.user_password(self.current_user_id)
+        if not credentials or not verify_password(
+            current_password,
+            str(credentials.get("password_salt") or ""),
+            str(credentials.get("password_hash") or ""),
+        ):
+            raise ValueError("当前密码不正确。")
+        salt, digest = hash_password(new_password)
+        if not self.database.update_user_password(self.current_user_id, salt, digest):
+            raise ValueError("账号不存在。")
+
+    def create_session(self, user: dict[str, Any]) -> dict[str, Any]:
+        session_token = secrets.token_urlsafe(32)
+        csrf_token = secrets.token_urlsafe(24)
+        expires = datetime.now(timezone.utc) + timedelta(seconds=SESSION_MAX_AGE)
+        self.database.create_session(
+            token_hash(session_token),
+            int(user["id"]),
+            csrf_token,
+            expires.isoformat(timespec="seconds"),
+        )
+        self.bind_user(int(user["id"]))
+        access = self.database.user_access(int(user["id"])) or {}
+        return {
+            "user": {
+                "id": int(user["id"]),
+                "username": str(user["username"]),
+                "role": str(access.get("role") or "user"),
+                "membership": self.membership(),
+            },
+            "session_token": session_token,
+            "csrf_token": csrf_token,
+        }
+
+    @staticmethod
+    def validate_input(username: str, password: str) -> None:
+        if not USERNAME_PATTERN.fullmatch(username):
+            raise ValueError("账号名应为 3 至 30 位中文、字母、数字、下划线或连字符。")
+        if len(password) < 8 or len(password) > 128:
+            raise ValueError("密码长度应为 8 至 128 位。")
+        if password.isalpha() or password.isdigit():
+            raise ValueError("密码应同时包含字母、数字或符号中的至少两类。")
+
+    def save_birth_profile(self, payload: dict[str, Any]) -> dict[str, Any]:
+        birth_datetime = str(payload.get("birth_datetime") or "").strip()
+        gender = str(payload.get("gender") or "unspecified").strip()
+        current_date = normalize_date(str(payload.get("trade_date") or date.today().isoformat()))
+        personal = self.personal_field_builder(birth_datetime, gender, current_date)
+        encrypted = self.vault.encrypt_json(
+            {"birth_datetime": birth_datetime, "gender": gender}
+        )
+        self.database.save_user_birth_profile(self.current_user_id, encrypted)
+        return self.public_personal_profile(personal)
+
+    def stored_birth_profile(self) -> dict[str, str] | None:
+        encrypted = self.database.get_user_birth_profile(self.current_user_id)
+        if not encrypted:
+            return None
+        payload = self.vault.decrypt_json(encrypted)
+        birth_datetime = str(payload.get("birth_datetime") or "").strip()
+        if not birth_datetime:
+            return None
+        return {
+            "birth_datetime": birth_datetime,
+            "gender": str(payload.get("gender") or "unspecified"),
+        }
+
+    def personal_field(
+        self,
+        current_date: str,
+        current_field: dict[str, Any],
+        public: bool = False,
+    ) -> dict[str, Any] | None:
+        stored = self.stored_birth_profile()
+        if not stored:
+            return None
+        personal = self.personal_field_builder(
+            stored["birth_datetime"],
+            stored["gender"],
+            current_date,
+            current_field,
+        )
+        if public:
+            return self.public_personal_profile(personal)
+        personal.pop("birth", None)
+        return personal
+
+    @staticmethod
+    def public_personal_profile(personal: dict[str, Any]) -> dict[str, Any]:
+        allowed = {
+            "day_master",
+            "ten_god_tendency",
+            "element_balance",
+            "balance_tendency",
+            "current",
+            "notice",
+        }
+        return {key: value for key, value in personal.items() if key in allowed}
+
+    def update_membership(self, payload: dict[str, Any]) -> None:
+        try:
+            user_id = int(payload.get("user_id"))
+        except (TypeError, ValueError) as exc:
+            raise ValueError("会员账号不正确。") from exc
+        status = str(payload.get("status") or "inactive")
+        if status not in {"active", "inactive", "suspended"}:
+            raise ValueError("会员状态不正确。")
+        access = self.database.user_access(user_id)
+        if not access:
+            raise ValueError("用户不存在。")
+        starts_at = None
+        expires_at = None
+        plan = ""
+        if status == "active":
+            duration = str(payload.get("duration") or "").strip()
+            durations = {
+                "1_month": (1, "1个月"),
+                "3_months": (3, "3个月"),
+                "12_months": (12, "12个月"),
+                "3_years": (36, "3年"),
+                "permanent": (0, "永久"),
+            }
+            if duration not in durations:
+                raise ValueError("请选择会员开通时长。")
+            now = datetime.now(timezone.utc)
+            existing_start = parse_iso_datetime(access.get("membership_starts_at"))
+            existing_expiry = parse_iso_datetime(access.get("membership_expires_at"))
+            starts = existing_start if existing_start and existing_start <= now else now
+            months, plan = durations[duration]
+            starts_at = starts.isoformat(timespec="seconds")
+            if months:
+                renewal_base = existing_expiry if existing_expiry and existing_expiry > now else now
+                expires_at = add_months(renewal_base, months).isoformat(timespec="seconds")
+        if not self.database.update_membership(
+            user_id, status, plan, starts_at, expires_at
+        ):
+            raise ValueError("用户不存在。")
+
+    def admin_users(
+        self, usage_supplier: Callable[[int], int]
+    ) -> list[dict[str, Any]]:
+        rows = []
+        for user in self.database.list_users():
+            membership = self.membership_for_access(user)
+            used = usage_supplier(int(user["id"])) if membership["active"] else 0
+            rows.append({
+                **user,
+                "membership_active": membership["active"],
+                "membership_subscribed": membership["subscribed"],
+                "used_today": used,
+            })
+        return rows

app/backend/features/alerts/service.py

@@ -4,7 +4,7 @@ import secrets
 from datetime import date, datetime
 from typing import Any
 
-from app_config import validate_text
+from backend.bootstrap.config import validate_text
 from backend.database.repositories import AlertRepository
 
 

app/backend/features/review/trade_journal.py

@@ -4,7 +4,7 @@ import json
 from datetime import date
 from typing import Any
 
-from app_config import normalize_date, validate_stock_code, validate_text
+from backend.bootstrap.config import normalize_date, validate_stock_code, validate_text
 from backend.database.repositories import TradeJournalRepository
 
 

app/backend/features/system/__init__.py

@@ -0,0 +1,3 @@
+from .http import SystemHttpMixin
+
+__all__ = ["SystemHttpMixin"]

app/backend/features/system/http.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import json
+from datetime import date
+from http import HTTPStatus
+
+
+class SystemHttpMixin:
+    def save_system_settings(self) -> None:
+        try:
+            result = self.application_service.save_system_settings(self.read_json_body())
+            self.send_json({"ok": True, **result})
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
+
+    def test_system_llm_settings(self) -> None:
+        try:
+            body = self.read_json_body()
+            result = self.application_service.test_system_llm_profile(
+                str(body.get("model_id") or ""), body.get("profile") or {}
+            )
+            self.send_json({"ok": True, "result": result})
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
+
+    def start_background_refresh(self) -> None:
+        try:
+            body = self.read_json_body(allow_empty=True)
+            started = self.application_service.request_background_sync(
+                str(body.get("trade_date") or date.today().isoformat())
+            )
+            self.send_json(
+                {
+                    "ok": True,
+                    "started": started,
+                    "message": "后台刷新已开始" if started else "已有后台刷新任务正在运行",
+                },
+                HTTPStatus.ACCEPTED,
+            )
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)

app/backend/features/system/repository.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+
+class SystemSettingsRepositoryMixin:
+    """Original encrypted system-setting persistence methods."""
+
+    def get_system_setting(self, key: str) -> str:
+        with self.connect() as connection:
+            row = connection.execute(
+                "SELECT encrypted_payload FROM system_settings WHERE setting_key = ?",
+                (key,),
+            ).fetchone()
+        return str(row["encrypted_payload"]) if row else ""
+
+    def save_system_setting(self, key: str, encrypted_payload: str) -> None:
+        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        with self.connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO system_settings (setting_key, encrypted_payload, updated_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(setting_key) DO UPDATE SET
+                    encrypted_payload = excluded.encrypted_payload,
+                    updated_at = excluded.updated_at
+                """,
+                (key, encrypted_payload, now),
+            )

app/backend/http/__init__.py

@@ -1,8 +1,9 @@
 from .context import correlation_id
 from .errors import normalize_error_payload
+from .handler import HttpTransportMixin
 from .router import AccessRole, ApiRoute, ApiRouteRegistry, RouteRegistryError
 
 __all__ = [
     "AccessRole", "ApiRoute", "ApiRouteRegistry", "RouteRegistryError",
-    "correlation_id", "normalize_error_payload",
+    "HttpTransportMixin", "correlation_id", "normalize_error_payload",
 ]

app/backend/http/handler.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+import json
+import mimetypes
+import secrets
+from http import HTTPStatus
+from http.cookies import SimpleCookie
+from typing import Any
+from urllib.parse import unquote
+
+from backend.bootstrap.config import SESSION_COOKIE, SESSION_MAX_AGE, STATIC_DIR
+from backend.features.accounts.security import token_hash
+from backend.http.context import correlation_id
+from backend.http.errors import normalize_error_payload
+
+
+class HttpTransportMixin:
+    """Original HTTP transport, static-file, session and access behavior."""
+
+    application_service: Any
+    route_registry: Any
+
+    def session_token(self) -> str:
+        cookie = SimpleCookie()
+        try:
+            cookie.load(self.headers.get("Cookie", ""))
+        except Exception:
+            return ""
+        morsel = cookie.get(SESSION_COOKIE)
+        return morsel.value if morsel else ""
+
+    def require_auth(self, send_error: bool = True) -> bool:
+        raw_token = self.session_token()
+        service = self.application_service
+        user = service.database.session_user(token_hash(raw_token)) if raw_token else None
+        if not user:
+            if send_error:
+                self.send_json({"error": "请先登录。"}, HTTPStatus.UNAUTHORIZED)
+            return False
+        self.auth_user = user
+        service.bind_user(int(user["id"]))
+        return True
+
+    def require_csrf(self) -> bool:
+        supplied = self.headers.get("X-CSRF-Token", "")
+        expected = str(getattr(self, "auth_user", {}).get("csrf_token") or "")
+        if not supplied or not secrets.compare_digest(supplied, expected):
+            self.send_json({"error": "请求校验失败，请刷新页面后重试。"}, HTTPStatus.FORBIDDEN)
+            return False
+        return True
+
+    def require_admin(self) -> bool:
+        if str(getattr(self, "auth_user", {}).get("role") or "user") != "admin":
+            self.send_json({"error": "需要管理员权限。"}, HTTPStatus.FORBIDDEN)
+            return False
+        return True
+
+    def require_member(self) -> bool:
+        if self.application_service.membership()["active"]:
+            return True
+        self.send_json(
+            {"error": "该功能仅对有效会员开放，请联系管理员开通会员。", "code": "membership_required"},
+            HTTPStatus.FORBIDDEN,
+        )
+        return False
+
+    def require_access(self, method: str, path: str) -> bool:
+        route = self.route_registry.resolve(method, path)
+        if route is None:
+            self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND)
+            return False
+        role = route.access
+        if role == "public":
+            return True
+        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 = (
+            f"{SESSION_COOKIE}={value}; Path=/; HttpOnly; SameSite=Lax; Max-Age={max_age}"
+        )
+        if self.headers.get("X-Forwarded-Proto", "").lower() == "https":
+            cookie += "; Secure"
+        return cookie
+
+    def read_json_body(self, allow_empty: bool = False) -> dict[str, Any]:
+        length = int(self.headers.get("Content-Length", "0"))
+        if length == 0 and allow_empty:
+            return {}
+        if length <= 0 or length > 65536:
+            raise ValueError("请求内容为空或过大。")
+        return json.loads(self.rfile.read(length).decode("utf-8"))
+
+    def serve_static(self, request_path: str) -> None:
+        relative = unquote(request_path).lstrip("/") or "index.html"
+        candidate = (STATIC_DIR / relative).resolve()
+        try:
+            candidate.relative_to(STATIC_DIR.resolve())
+        except ValueError:
+            self.send_error(HTTPStatus.FORBIDDEN)
+            return
+        if not candidate.is_file():
+            candidate = STATIC_DIR / "index.html"
+        try:
+            content = candidate.read_bytes()
+        except OSError:
+            self.send_error(HTTPStatus.NOT_FOUND)
+            return
+        content_type = mimetypes.guess_type(candidate.name)[0] or "application/octet-stream"
+        if content_type.startswith("text/") or content_type in {"application/javascript", "application/json"}:
+            content_type += "; charset=utf-8"
+        self.send_response(HTTPStatus.OK)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(content)))
+        self.send_header("Cache-Control", "no-cache")
+        self.end_headers()
+        self.wfile.write(content)
+
+    def send_json(
+        self,
+        payload: dict[str, Any],
+        status: HTTPStatus = HTTPStatus.OK,
+        headers: dict[str, str] | None = None,
+    ) -> None:
+        request_id = getattr(self, "_correlation_id", "")
+        if not request_id:
+            request_id = correlation_id(self.headers.get("X-Request-ID", ""))
+            self._correlation_id = request_id
+        payload = normalize_error_payload(payload, status, request_id)
+        content = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(content)))
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("X-Request-ID", request_id)
+        for name, value in (headers or {}).items():
+            self.send_header(name, value)
+        self.end_headers()
+        self.wfile.write(content)
+
+    def log_message(self, format_string: str, *args: Any) -> None:
+        print(f"[{self.log_date_time_string()}] {format_string % args}")

app/backend/http/router.py

@@ -6,7 +6,7 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Literal, cast
 
-from app_config import APP_DIR
+from backend.bootstrap.config import APP_DIR
 
 
 AccessRole = Literal["public", "authenticated", "member", "admin"]

app/backend/jobs/registry.py

@@ -4,7 +4,7 @@ import json
 from dataclasses import dataclass
 from pathlib import Path
 
-from app_config import APP_DIR
+from backend.bootstrap.config import APP_DIR
 
 
 @dataclass(frozen=True)

app/database.py

@@ -7,6 +7,8 @@ from pathlib import Path
 from typing import Any
 
 from backend.database import MIGRATIONS, MigrationRunner, SQLiteConnectionFactory
+from backend.features.accounts.repository import AccountRepositoryMixin
+from backend.features.system.repository import SystemSettingsRepositoryMixin
 
 
 def _optional_float(value: Any) -> float | None:
@@ -18,7 +20,7 @@ def _optional_float(value: Any) -> float | None:
         return None
 
 
-class ReviewDatabase:
+class ReviewDatabase(AccountRepositoryMixin, SystemSettingsRepositoryMixin):
     def __init__(self, path: Path) -> None:
         self.path = path
         self.path.parent.mkdir(parents=True, exist_ok=True)
@@ -647,225 +649,6 @@ class ReviewDatabase:
             )
             MigrationRunner().apply(connection, MIGRATIONS)
 
-    def count_users(self) -> int:
-        with self.connect() as connection:
-            row = connection.execute("SELECT COUNT(*) AS total FROM users").fetchone()
-        return int(row["total"] if row else 0)
-
-    def first_user_id(self) -> int:
-        with self.connect() as connection:
-            row = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
-        return int(row["id"] or 0) if row else 0
-
-    def create_user(
-        self,
-        username: str,
-        password_salt: str,
-        password_hash: str,
-    ) -> dict[str, Any]:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        try:
-            with self.connect() as connection:
-                role = "admin" if int(connection.execute("SELECT COUNT(*) FROM users").fetchone()[0]) == 0 else "user"
-                cursor = connection.execute(
-                    """
-                    INSERT INTO users
-                        (username, password_salt, password_hash, role, created_at, updated_at)
-                    VALUES (?, ?, ?, ?, ?, ?)
-                    """,
-                    (username, password_salt, password_hash, role, now, now),
-                )
-                user_id = int(cursor.lastrowid)
-        except sqlite3.IntegrityError as exc:
-            raise ValueError("该账号名已被使用。") from exc
-        return {"id": user_id, "username": username, "role": role, "created_at": now}
-
-    def user_by_username(self, username: str) -> dict[str, Any] | None:
-        with self.connect() as connection:
-            row = connection.execute(
-                """
-                SELECT id, username, password_salt, password_hash, role, llm_mode,
-                       membership_status, membership_plan, membership_starts_at,
-                       membership_expires_at, created_at
-                FROM users WHERE username = ? COLLATE NOCASE
-                """,
-                (username,),
-            ).fetchone()
-        return dict(row) if row else None
-
-    def user_password(self, user_id: int) -> dict[str, str] | None:
-        with self.connect() as connection:
-            row = connection.execute(
-                "SELECT password_salt, password_hash FROM users WHERE id = ?",
-                (user_id,),
-            ).fetchone()
-        return dict(row) if row else None
-
-    def update_user_password(self, user_id: int, password_salt: str, password_hash: str) -> bool:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            cursor = connection.execute(
-                "UPDATE users SET password_salt = ?, password_hash = ?, updated_at = ? WHERE id = ?",
-                (password_salt, password_hash, now, user_id),
-            )
-        return cursor.rowcount > 0
-
-    def delete_user(self, user_id: int) -> bool:
-        with self.connect() as connection:
-            cursor = connection.execute("DELETE FROM users WHERE id = ?", (user_id,))
-            return cursor.rowcount > 0
-
-    def create_session(
-        self,
-        session_hash: str,
-        user_id: int,
-        csrf_token: str,
-        expires_at: str,
-    ) -> None:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            connection.execute("DELETE FROM user_sessions WHERE expires_at <= ?", (now,))
-            connection.execute(
-                """
-                INSERT INTO user_sessions
-                    (token_hash, user_id, csrf_token, expires_at, created_at, last_seen_at)
-                VALUES (?, ?, ?, ?, ?, ?)
-                """,
-                (session_hash, user_id, csrf_token, expires_at, now, now),
-            )
-
-    def session_user(self, session_hash: str) -> dict[str, Any] | None:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            row = connection.execute(
-                """
-                SELECT u.id, u.username, u.role, u.llm_mode, u.membership_status,
-                       u.membership_plan, u.membership_starts_at, u.membership_expires_at,
-                       u.created_at, s.csrf_token, s.expires_at
-                FROM user_sessions AS s
-                JOIN users AS u ON u.id = s.user_id
-                WHERE s.token_hash = ? AND s.expires_at > ?
-                """,
-                (session_hash, now),
-            ).fetchone()
-            if row:
-                connection.execute(
-                    "UPDATE user_sessions SET last_seen_at = ? WHERE token_hash = ?",
-                    (now, session_hash),
-                )
-        return dict(row) if row else None
-
-    def delete_session(self, session_hash: str) -> bool:
-        with self.connect() as connection:
-            cursor = connection.execute(
-                "DELETE FROM user_sessions WHERE token_hash = ?",
-                (session_hash,),
-            )
-            return cursor.rowcount > 0
-
-    def get_user_credentials(self, user_id: int) -> str:
-        with self.connect() as connection:
-            row = connection.execute(
-                "SELECT encrypted_payload FROM user_credentials WHERE user_id = ?",
-                (user_id,),
-            ).fetchone()
-        return str(row["encrypted_payload"]) if row else ""
-
-    def save_user_credentials(self, user_id: int, encrypted_payload: str) -> None:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            connection.execute(
-                """
-                INSERT INTO user_credentials (user_id, encrypted_payload, updated_at)
-                VALUES (?, ?, ?)
-                ON CONFLICT(user_id) DO UPDATE SET
-                    encrypted_payload = excluded.encrypted_payload,
-                    updated_at = excluded.updated_at
-                """,
-                (user_id, encrypted_payload, now),
-            )
-
-    def list_user_credentials(self) -> list[dict[str, Any]]:
-        with self.connect() as connection:
-            rows = connection.execute(
-                "SELECT user_id, encrypted_payload FROM user_credentials ORDER BY user_id"
-            ).fetchall()
-        return [dict(row) for row in rows]
-
-    def get_system_setting(self, key: str) -> str:
-        with self.connect() as connection:
-            row = connection.execute(
-                "SELECT encrypted_payload FROM system_settings WHERE setting_key = ?",
-                (key,),
-            ).fetchone()
-        return str(row["encrypted_payload"]) if row else ""
-
-    def save_system_setting(self, key: str, encrypted_payload: str) -> None:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            connection.execute(
-                """
-                INSERT INTO system_settings (setting_key, encrypted_payload, updated_at)
-                VALUES (?, ?, ?)
-                ON CONFLICT(setting_key) DO UPDATE SET
-                    encrypted_payload = excluded.encrypted_payload,
-                    updated_at = excluded.updated_at
-                """,
-                (key, encrypted_payload, now),
-            )
-
-    def user_access(self, user_id: int) -> dict[str, Any] | None:
-        with self.connect() as connection:
-            row = connection.execute(
-                """
-                SELECT id, username, role, llm_mode, membership_status, membership_plan,
-                       membership_starts_at, membership_expires_at, created_at
-                FROM users WHERE id = ?
-                """,
-                (user_id,),
-            ).fetchone()
-        return dict(row) if row else None
-
-    def list_users(self) -> list[dict[str, Any]]:
-        with self.connect() as connection:
-            rows = connection.execute(
-                """
-                SELECT id, username, role, llm_mode, membership_status, membership_plan,
-                       membership_starts_at, membership_expires_at, created_at
-                FROM users ORDER BY id
-                """
-            ).fetchall()
-        return [dict(row) for row in rows]
-
-    def update_user_llm_mode(self, user_id: int, mode: str) -> None:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            connection.execute(
-                "UPDATE users SET llm_mode = ?, updated_at = ? WHERE id = ?",
-                (mode, now, user_id),
-            )
-
-    def update_membership(
-        self,
-        user_id: int,
-        status: str,
-        plan: str,
-        starts_at: str | None,
-        expires_at: str | None,
-    ) -> bool:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            cursor = connection.execute(
-                """
-                UPDATE users
-                SET membership_status = ?, membership_plan = ?,
-                    membership_starts_at = ?, membership_expires_at = ?, updated_at = ?
-                WHERE id = ?
-                """,
-                (status, plan, starts_at, expires_at, now, user_id),
-            )
-        return cursor.rowcount > 0
-
     def record_llm_usage(
         self,
         user_id: int,
@@ -907,36 +690,6 @@ class ReviewDatabase:
             ).fetchone()
         return int(row["total"] if row else 0)
 
-    def get_user_birth_profile(self, user_id: int) -> str:
-        with self.connect() as connection:
-            row = connection.execute(
-                "SELECT encrypted_payload FROM user_birth_profiles WHERE user_id = ?",
-                (user_id,),
-            ).fetchone()
-        return str(row["encrypted_payload"]) if row else ""
-
-    def save_user_birth_profile(self, user_id: int, encrypted_payload: str) -> None:
-        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
-        with self.connect() as connection:
-            connection.execute(
-                """
-                INSERT INTO user_birth_profiles (user_id, encrypted_payload, updated_at)
-                VALUES (?, ?, ?)
-                ON CONFLICT(user_id) DO UPDATE SET
-                    encrypted_payload = excluded.encrypted_payload,
-                    updated_at = excluded.updated_at
-                """,
-                (user_id, encrypted_payload, now),
-            )
-
-    def delete_user_birth_profile(self, user_id: int) -> bool:
-        with self.connect() as connection:
-            cursor = connection.execute(
-                "DELETE FROM user_birth_profiles WHERE user_id = ?",
-                (user_id,),
-            )
-            return cursor.rowcount > 0
-
     def get_snapshot(self, trade_date: str) -> dict[str, Any] | None:
         with self.connect() as connection:
             row = connection.execute(

app/security.py

@@ -1,71 +1,3 @@
-from __future__ import annotations
+"""Compatibility imports for the preserved account security API."""
 
-import base64
-import hashlib
-import hmac
-import json
-import os
-from typing import Any
-
-from cryptography.fernet import Fernet, InvalidToken
-
-
-PASSWORD_SCRYPT_N = 2**14
-PASSWORD_SCRYPT_R = 8
-PASSWORD_SCRYPT_P = 1
-
-
-class SecretVault:
-    def __init__(self, key: str) -> None:
-        try:
-            self._fernet = Fernet(key.encode("ascii"))
-        except (ValueError, TypeError) as exc:
-            raise ValueError("APP_ENCRYPTION_KEY 格式无效。") from exc
-
-    @staticmethod
-    def generate_key() -> str:
-        return Fernet.generate_key().decode("ascii")
-
-    def encrypt_json(self, payload: dict[str, Any]) -> str:
-        raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
-        return self._fernet.encrypt(raw).decode("ascii")
-
-    def decrypt_json(self, token: str) -> dict[str, Any]:
-        if not token:
-            return {}
-        try:
-            payload = json.loads(self._fernet.decrypt(token.encode("ascii")).decode("utf-8"))
-        except (InvalidToken, UnicodeDecodeError, json.JSONDecodeError) as exc:
-            raise ValueError("账号加密数据无法解密，请检查 APP_ENCRYPTION_KEY。") from exc
-        if not isinstance(payload, dict):
-            raise ValueError("账号加密数据格式无效。")
-        return payload
-
-
-def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]:
-    raw_salt = salt or os.urandom(16)
-    digest = hashlib.scrypt(
-        password.encode("utf-8"),
-        salt=raw_salt,
-        n=PASSWORD_SCRYPT_N,
-        r=PASSWORD_SCRYPT_R,
-        p=PASSWORD_SCRYPT_P,
-        dklen=32,
-    )
-    return (
-        base64.urlsafe_b64encode(raw_salt).decode("ascii"),
-        base64.urlsafe_b64encode(digest).decode("ascii"),
-    )
-
-
-def verify_password(password: str, salt_text: str, expected_hash: str) -> bool:
-    try:
-        salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
-        _, actual_hash = hash_password(password, salt)
-    except (ValueError, TypeError):
-        return False
-    return hmac.compare_digest(actual_hash, expected_hash)
-
-
-def token_hash(token: str) -> str:
-    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+from backend.features.accounts.security import *  # noqa: F401,F403

app/tests/test_account_access.py

@@ -6,7 +6,7 @@ import sqlite3
 from pathlib import Path
 
 from database import ReviewDatabase
-from security import hash_password, verify_password
+from backend.features.accounts.security import hash_password, verify_password
 
 
 class AccountAccessTests(unittest.TestCase):

app/tests/test_data_gateway.py

@@ -47,7 +47,9 @@ class DataGatewayTests(unittest.TestCase):
     def test_server_has_no_direct_runtime_tushare_construction(self) -> None:
         from pathlib import Path
 
-        source = (Path(__file__).resolve().parents[1] / "server.py").read_text(encoding="utf-8")
+        source = (
+            Path(__file__).resolve().parents[1] / "backend" / "application.py"
+        ).read_text(encoding="utf-8")
         self.assertEqual(source.count("TushareClient(self.token)"), 1)
         self.assertIn("return gateway.tushare()", source)
 

app/tests/test_market_mode.py

@@ -9,7 +9,7 @@ from tushare_client import _sector_coverage_issue
 
 
 def load_method(name: str):
-    source = Path("server.py").read_text(encoding="utf-8")
+    source = Path("backend/application.py").read_text(encoding="utf-8")
     tree = ast.parse(source)
     dashboard_service = next(
         node for node in tree.body

app/tests/test_mentor_registry.py

@@ -81,7 +81,7 @@ class MentorSkillRegistryTests(unittest.TestCase):
         self.assertTrue(all(item.quality_total == 6 for item in skills))
 
     def test_server_applies_private_guard_to_every_mentor_entry_point(self):
-        source = (ROOT / "server.py").read_text(encoding="utf-8")
+        source = (ROOT / "backend" / "application.py").read_text(encoding="utf-8")
         mentor_section = source[source.index("    def mentor_setup"):source.index("    def _heaven_manual_schema")]
         self.assertGreaterEqual(
             mentor_section.count('include_private=self.membership()["is_admin"]'),

app/tests/test_preservation_slice_accounts.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+import server
+from backend.application import DashboardService, RequestHandler
+from backend.bootstrap.config import APP_DIR, STATIC_DIR
+from backend.features.accounts.repository import AccountRepositoryMixin
+from backend.features.accounts.security import SecretVault, token_hash
+from backend.features.accounts.service import AccountService
+from backend.http.handler import HttpTransportMixin
+from database import ReviewDatabase
+
+
+class AccountSliceStructureTests(unittest.TestCase):
+    def test_original_entrypoint_exports_canonical_runtime(self) -> None:
+        self.assertIs(server.DashboardService, DashboardService)
+        self.assertIs(server.RequestHandler, RequestHandler)
+        self.assertIs(server.SERVICE, RequestHandler.application_service)
+
+    def test_runtime_paths_still_point_at_app_root(self) -> None:
+        self.assertEqual(APP_DIR, Path(__file__).resolve().parents[1])
+        self.assertEqual(STATIC_DIR, APP_DIR / "static")
+
+    def test_account_persistence_and_http_transport_have_single_owners(self) -> None:
+        for method in (
+            "create_user",
+            "session_user",
+            "update_membership",
+            "save_user_birth_profile",
+        ):
+            self.assertNotIn(method, ReviewDatabase.__dict__)
+            self.assertIn(method, AccountRepositoryMixin.__dict__)
+        for method in (
+            "require_auth",
+            "require_csrf",
+            "require_access",
+            "serve_static",
+            "send_json",
+        ):
+            self.assertNotIn(method, RequestHandler.__dict__)
+            self.assertIn(method, HttpTransportMixin.__dict__)
+
+
+class AccountSliceBehaviorTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        self.database = ReviewDatabase(Path(self.temporary.name) / "review.db")
+        self.vault = SecretVault(SecretVault.generate_key())
+        self.context: dict[str, object] = {"user_id": 0, "access": {}}
+
+        def bind_user(user_id: int) -> None:
+            self.context["user_id"] = user_id
+            self.context["access"] = self.database.user_access(user_id) or {}
+
+        self.service = AccountService(
+            database=self.database,
+            vault=self.vault,
+            current_user_supplier=lambda: int(self.context["user_id"]),
+            access_supplier=lambda: dict(self.context["access"]),
+            bind_user=bind_user,
+            personal_field_builder=lambda *args: {
+                "birth": "private",
+                "day_master": "甲木",
+                "current": {"trade_date": args[2]},
+                "notice": "test",
+            },
+            auth_lock=threading.Lock(),
+        )
+
+    def tearDown(self) -> None:
+        self.temporary.cleanup()
+
+    def test_register_login_session_and_password_contract(self) -> None:
+        registered = self.service.register("owner_01", "Password123")
+        self.assertEqual(registered["user"]["role"], "admin")
+        self.assertTrue(registered["user"]["membership"]["active"])
+        self.assertIsNotNone(
+            self.database.session_user(token_hash(registered["session_token"]))
+        )
+
+        with self.assertRaisesRegex(ValueError, "账号名或密码不正确"):
+            self.service.login("owner_01", "wrong-password")
+
+        self.service.change_password("Password123", "NewPassword456")
+        logged_in = self.service.login("owner_01", "NewPassword456")
+        self.assertEqual(logged_in["user"]["id"], registered["user"]["id"])
+
+    def test_membership_and_birth_profile_remain_account_scoped(self) -> None:
+        owner = self.service.register("owner_02", "Password123")
+        other = self.database.create_user("other_02", "salt", "hash")
+        self.service.update_membership(
+            {"user_id": other["id"], "status": "active", "duration": "3_months"}
+        )
+        other_access = self.database.user_access(other["id"])
+        self.assertEqual(other_access["membership_plan"], "3个月")
+        self.assertTrue(AccountService.membership_for_access(other_access)["subscribed"])
+
+        personal = self.service.save_birth_profile(
+            {
+                "birth_datetime": "1990-01-01 08:30",
+                "gender": "male",
+                "trade_date": "2026-07-30",
+            }
+        )
+        self.assertNotIn("birth", personal)
+        self.assertEqual(personal["day_master"], "甲木")
+        self.assertTrue(self.database.get_user_birth_profile(owner["user"]["id"]))
+        self.assertEqual(self.database.get_user_birth_profile(other["id"]), "")
+
+
+if __name__ == "__main__":
+    unittest.main()

app/tests/test_stock_detail_realtime.py

@@ -90,8 +90,8 @@ class StockDetailRealtimeTests(unittest.TestCase):
             "moneyflow": {},
         }
 
-        with patch("server.datetime", FixedMarketDatetime), patch(
-            "server.TushareClient", RealtimeClientStub
+        with patch("backend.application.datetime", FixedMarketDatetime), patch(
+            "backend.application.TushareClient", RealtimeClientStub
         ):
             result = self.service._prepare_stock_detail(cached, "002141", today)
 
@@ -112,8 +112,8 @@ class StockDetailRealtimeTests(unittest.TestCase):
             "stock": {"code": "002141", "price": 10, "change": 1.2},
             "prices": [{"trade_date": historical, "close": 10, "change": 1.2}],
         }
-        with patch("server.datetime", FixedMarketDatetime), patch(
-            "server.TushareClient", RealtimeClientStub
+        with patch("backend.application.datetime", FixedMarketDatetime), patch(
+            "backend.application.TushareClient", RealtimeClientStub
         ):
             result = self.service._prepare_stock_detail(payload, "002141", historical)
 
@@ -151,8 +151,8 @@ class StockDetailRealtimeTests(unittest.TestCase):
                 },
             ],
         }
-        with patch("server.datetime", FixedPreopenDatetime), patch(
-            "server.TushareClient", RealtimeClientStub
+        with patch("backend.application.datetime", FixedPreopenDatetime), patch(
+            "backend.application.TushareClient", RealtimeClientStub
         ):
             result = self.service._prepare_stock_detail(payload, "002141", today)
 

app/tools/build_api_registry.py

@@ -60,7 +60,7 @@ def _role(method: str, path: str) -> str:
 
 
 def build() -> dict:
-    text = (ROOT / "server.py").read_text(encoding="utf-8")
+    text = (ROOT / "backend" / "application.py").read_text(encoding="utf-8")
     method_matches = list(re.finditer(r"^    def do_(GET|POST|DELETE)\(", text, re.MULTILINE))
     routes = []
     for index, match in enumerate(method_matches):

docs/migration/evidence/slice-01/README.md

@@ -0,0 +1,57 @@
+# 切片 01：启动、HTTP、账号、会员与系统管理
+
+> 基线：`4083dce`（切片 00 原样可运行副本）  
+> 回档标签：`xiaobai-preservation-slice-01-20260731`  
+> 结论：自动差分通过；前端运行资产未改动，仍等待最终全站人工验收
+
+## 1. 本次范围
+
+本切片只整理原版 `app/` 副本中的启动、通用 HTTP、账号、会员和系统管理实现。所有实现均从
+原文件机械移动，原导入面由兼容外壳保持；没有从冻结的 `next/` 复制产品代码，也没有重新设计
+页面、接口或数据结构。
+
+| 原位置 | 新的唯一实现位置 | 兼容方式 |
+|---|---|---|
+| `app/server.py` 的进程组装与完整应用服务 | `app/backend/application.py`、`app/backend/bootstrap/runtime.py` | `app/server.py` 继续导出原符号并保持原启动命令 |
+| `app/server.py` 的 Cookie、鉴权、CSRF、静态文件和 JSON 传输 | `app/backend/http/handler.py` | `RequestHandler` 组合 `HttpTransportMixin` |
+| `app/server.py` 的账号与会员 HTTP 方法 | `app/backend/features/accounts/http.py` | `RequestHandler` 组合 `AccountHttpMixin` |
+| `app/server.py` 的系统管理 HTTP 方法 | `app/backend/features/system/http.py` | `RequestHandler` 组合 `SystemHttpMixin` |
+| `app/server.py` 的账号、会员和出生信息业务方法 | `app/backend/features/accounts/service.py` | `DashboardService` 委托 `AccountService` |
+| `app/database.py` 的账号、会话、会员与用户凭据方法 | `app/backend/features/accounts/repository.py` | `ReviewDatabase` 组合 `AccountRepositoryMixin` |
+| `app/database.py` 的系统设置存取方法 | `app/backend/features/system/repository.py` | `ReviewDatabase` 组合 `SystemSettingsRepositoryMixin` |
+| `app/security.py` | `app/backend/features/accounts/security.py` | 根文件保留原导出 |
+| `app/app_config.py` | `app/backend/bootstrap/config.py` | 根文件保留原导出 |
+
+## 2. 不变量与差分证据
+
+- API：`config/api.config.json` 与 `app/config/api.config.json` 的 SHA-256 相同；路由路径、方法、
+  访问角色、状态码和错误载荷由原注册表与全量测试继续约束。
+- 数据库：原版和迁移副本均为 36 张业务表、62 个 schema 对象，规范化 schema SHA-256 均为
+  `0615a0423856d0eb02bccd81a071840bd50d6563da96d556cec49967ad8a5f4c`。
+- 账号边界：注册、登录、会话、密码、会员和出生信息使用临时数据库执行同一原版契约，5 项专项
+  测试全部通过。
+- 前端：本切片没有修改 `app/static/`；`index.html`、`app.js`、`styles.css` 与根目录原版对应文件
+  哈希相同。`app-light-1920x1080.png` 保存本切片运行截图。
+- 兼容：原 `python server.py` 命令以及测试和外部模块使用的 `server.DashboardService`、
+  `server.RequestHandler`、`server.SERVICE`、`server.automatic_screener_jobs` 均保持可用。
+
+## 3. 验证结果
+
+| 验证 | 结果 |
+|---|---:|
+| `python -m unittest discover -s tests -q` | 236 项通过 |
+| `python -m unittest tests.test_preservation_slice_accounts -v` | 5 项通过 |
+| `npx playwright test --reporter=dot` | 45 项通过 |
+| `python -m compileall -q ...` | 通过 |
+| `git diff --check` | 通过（仅 Git 的 CRLF 提示） |
+
+Playwright 在自行创建 Python 静态服务器时存在 Windows 子进程退出等待问题；验证时预先启动
+`127.0.0.1:8876` 静态服务器并由 Playwright 复用，45 项用例在 143 秒内正常返回退出码 0。
+
+## 4. 保留与待处理
+
+- `app/backend/application.py` 仍包含其余尚未迁移切片的原版实现，这是刻意保留，不是本切片遗漏。
+- `app/server.py`、`app/app_config.py`、`app/security.py` 和 `app/database.py` 的兼容面须等所有消费者
+  完成归位后再评估；本阶段禁止删除。
+- 没有删除任何不确定代码，没有修改正式根目录数据库，也没有切换 Docker/NAS。
+

docs/migration/保真迁移状态.json

@@ -1,6 +1,6 @@
 {
   "schema_version": 1,
-  "updated_at": "2026-07-30T23:55:00+08:00",
+  "updated_at": "2026-07-31T00:37:00+08:00",
   "status": "active",
   "migration_mode": "behavior_preserving_source_migration",
   "source_of_truth": "current_original_webapp_runtime_and_source",
@@ -9,13 +9,14 @@
   "failed_roots": [
     "next"
   ],
-  "current_slice": null,
-  "last_completed_slice": "slice-00-exact-runtime-copy",
-  "last_checkpoint": "xiaobai-preservation-migration-charter-20260730",
-  "next_action": "commit_and_push_slice-00_then_begin_startup_http_account_migration",
+  "current_slice": "slice-02-market-search-charts-data",
+  "last_completed_slice": "slice-01-startup-http-accounts-system",
+  "last_checkpoint": "xiaobai-preservation-slice-01-20260731",
+  "next_action": "capture_slice-02_market_search_chart_data_contracts_then_move_original_implementations",
   "authoritative_documents": [
     "AGENTS.md",
     "docs/migration/原版保真迁移总纲.md",
+    "docs/migration/目标目录与切片顺序.md",
     "docs/migration/保真迁移账本.md",
     "docs/migration/next失败冻结记录.md"
   ],

docs/migration/保真迁移账本.md

@@ -1,6 +1,6 @@
 # 小白复盘保真迁移账本
 
-> 当前状态：正式迁移，切片00“原样可运行副本”已完成
+> 当前状态：正式迁移，切片01“启动、HTTP、账号、会员与系统管理”已完成
 
 本账本是上下文恢复和人工审计的连续记录。任何迁移提交必须在同一提交中更新本文件及
 `保真迁移状态.json`。
@@ -20,6 +20,7 @@
 | 2026-07-30 | `xiaobai-next-rejected-20260730` | 冻结失败的`next/`实现 | 禁止部署或继续开发 |
 | 2026-07-30 | `xiaobai-preservation-migration-charter-20260730` | 建立保真迁移总纲、状态和恢复协议 | 尚未开始新迁移 |
 | 2026-07-30 | `41329943c4878fc09ed82ec376eb93ab151e4092` | 完成只读资产清查并由用户批准`app/`结构 | 开始切片00 |
+| 2026-07-31 | `xiaobai-preservation-slice-01-20260731` | 启动、HTTP、账号、会员与系统管理原实现归位 | 自动差分通过，进入切片02 |
 
 ## 资产处置登记
 
@@ -50,6 +51,16 @@
 - 完整证据：`docs/migration/evidence/slice-00/README.md`。
 - 资产清单：`docs/migration/原版资产清单.json`，388项，0项哈希差异。
 
+已完成切片：`slice-01-startup-http-accounts-system`。
+
+- 原版基线：提交`4083dce`，即切片00原样副本。
+- 迁移范围：进程入口、应用组装、HTTP传输、账号安全、账号/会员业务、账号/系统设置持久化。
+- 兼容边界：根级`server.py`、`app_config.py`、`security.py`继续保留原导入和命令入口。
+- API与数据库：API/功能注册表哈希一致；原版与副本数据库均为36表、62个schema对象且schema哈希一致。
+- 验收：236项Python测试、5项切片专项测试、45项Playwright测试全部通过；前端运行资产未改动。
+- 回档：标签`xiaobai-preservation-slice-01-20260731`。
+- 完整证据：`docs/migration/evidence/slice-01/README.md`。
+
 ## 决策记录
 
 | 日期 | 决策 | 原因 |

docs/migration/目标目录与切片顺序.md

@@ -0,0 +1,71 @@
+# 保真迁移目标目录与切片顺序
+
+> 状态：已由用户确认
+> 生效日期：2026-07-30
+
+本文件把已批准的`app/`目标结构和迁移顺序固化为可恢复约束。目录只表达职责，迁移时从
+原版复制、移动和拆分真实实现；禁止先建立空业务骨架后按规格书重写。
+
+## 目标目录
+
+```text
+app/
+  server.py                    # 最终仅保留进程入口与HTTP服务器组装
+  backend/
+    bootstrap/                 # 路径、环境、配置和依赖容器
+    http/                      # 路由、鉴权、响应、静态资源和通用传输能力
+    features/
+      accounts/                # 登录、账号、会员及用户身份
+      system/                  # 管理员配置和运行状态
+      market/                  # 公共行情、搜索、详情和图表
+      sentiment/               # 情绪周期
+      pools/                   # 涨停、炸板、跌停、昨日涨停和涨停表现
+      ladder/                  # 市场天梯
+      rotation/                # 板块轮动
+      auction/                 # 集合竞价
+      themes/                  # 题材库
+      popularity/              # 人气热榜
+      dragon_tiger/            # 龙虎榜与游资档案
+      screener/                # 阶段、策略、自定义选股及持续跟踪
+      mentor/                  # 问师
+      heaven/                  # 观势、观气、观心
+      review/                  # 自选、复盘笔记、交易日志和复盘助手
+      alerts/                  # 提醒中心
+    data/                      # 统一数据网关、口径、质量和供应商适配
+    database/                  # 连接、迁移和领域Repository
+    jobs/                      # 后台任务定义、状态、调度和重试
+    llm/                       # 唯一模型网关、流式协议和调用审计
+  frontend/
+    index.html                 # 原版DOM骨架；仅在等价验证后拆分
+    shared/                    # API、状态、Shell、弹窗和通用组件
+    pages/<feature>/           # 页面自己的行为与样式
+    styles/                    # 令牌、基础层、Shell和经验证后的公共样式
+    vendor/                    # 浏览器端第三方静态资产
+  config/                      # 页面、功能、API、任务及数据字段注册表
+  data/                        # SQLite及私有运行数据，保持Git忽略
+  tests/                       # 单元、契约、差异和浏览器回归
+  tools/                       # 清查、迁移、差异验证和维护工具
+  vendor/                      # Python离线依赖
+  游资skills/                 # 原版公开Skill；私有Skill仍位于data/
+```
+
+允许迁移期间保留根级兼容外壳；外壳只能转发到唯一实现，并须在账本登记删除条件。
+
+## 固定切片顺序
+
+| 切片 | 完整纵向范围 |
+|---:|---|
+| 00 | 原版可运行副本、资产清单、数据库副本和视觉基线 |
+| 01 | 启动、HTTP通用能力、登录账号、会员和系统管理 |
+| 02 | 公共行情、全局搜索、详情、悬浮图表、数据网关和数据质量 |
+| 03 | 情绪周期、五类股池和涨停表现 |
+| 04 | 市场天梯和板块轮动 |
+| 05 | 集合竞价、题材库、人气热榜和龙虎榜 |
+| 06 | 智能选股、自定义选股和策略持续跟踪 |
+| 07 | 问师、模型Skill和LLM流式链路 |
+| 08 | 问天：观势、观气和观心 |
+| 09 | 我的复盘、自选、笔记、交易日志、提醒和复盘助手 |
+| 10 | 前端Shell、页面文件、共享组件、CSS层级和移动端职责归位 |
+| 11 | 待定代码试删、全量并行验收、维护文档和切换准备 |
+
+每个切片先记录原版证据，再移动实现，再执行同输入差异；未通过时不得进入下一切片。