diff --git a/docs/migration/重建迁移章程.md b/docs/migration/重建迁移章程.md index ba92e3f..5f7d46c 100644 --- a/docs/migration/重建迁移章程.md +++ b/docs/migration/重建迁移章程.md @@ -124,7 +124,7 @@ HTTP/后台任务 -> 业务服务 -> Repository/DataGateway/LLMGateway -> 基础 | 11 | 问天观势、观气、观心 | 公式、安全门、动画和历史验收 | 已完成 | | 12 | 我的复盘、提醒和复盘助手 | 私有数据、汇总和弹窗验收 | 已完成 | | 13 | 全站移动端重组和无障碍 | 320/390/430/768/横屏验收 | 已完成 | -| 14 | 数据迁移、Docker、备份恢复和性能安全 | 旧库副本迁移、回滚演练 | 进行中 | +| 14 | 数据迁移、Docker、备份恢复和性能安全 | 旧库副本迁移、回滚演练 | 已完成 | | 15 | 全量验收、减法审计和切换准备 | 规格覆盖矩阵、最终报告 | 未开始 | 阶段编号不会因为上下文压缩重新规划。只有发现产品规格自身矛盾时，才记录决策并调整阶段内容；不得通过新增阶段掩盖未完成工作。 @@ -255,4 +255,9 @@ HTTP/后台任务 -> 业务服务 -> Repository/DataGateway/LLMGateway -> 基础 - 可访问性补齐跳到主内容、排序`aria-sort`、弹窗焦点回收、44px触控目标和有效的1ms减少动态效果令牌。移动规则集中在537行的唯一`mobile.css`，未复制页面、API或业务状态。 - 阶段13视觉与验收证据位于`next/docs/evidence/stage-13/`。 - 新系统进入阶段14：交付可重复的旧库迁移、Docker运行、备份恢复、升级回退及性能安全检查，所有演练仅作用于副本和新系统容器资产。 +- 阶段14已完成：一次性迁移工具以只读 URI 打开旧库，真实副本迁移保留3个账号、会员与模型配置、全部已归属私有记录、196次历史选股、16条跟踪及受控展示归档；完整性检查通过且外键违规为0。 +- 备份使用SQLite在线备份API并将数据库、环境密钥材料和私有Skill归入同一带SHA-256清单的归档；恢复默认拒绝覆盖并在落盘前校验安全路径、文件摘要和数据库完整性。真实恢复演练通过。 +- 生产前端改为FastAPI同端口提供，SPA深链接与API 404边界通过测试；新容器定义采用多阶段构建、非root、只读根文件系统、持久卷和健康检查。开发机没有Docker，实际镜像构建和容器重启恢复列为NAS切换前阻断验收项。 +- 阶段14门禁：Ruff、96项pytest、Vue类型检查、7项Vitest、Vite生产构建及17项Playwright全部通过；真实迁移库副本行情摘要读取P95为23.55ms。证据位于`next/docs/evidence/stage-14/`。 +- 新系统进入阶段15：执行规格覆盖矩阵、减法审计、性能安全总验收、人工维护演练和最终切换准备；阶段15不自动切换NAS生产容器。 - 生产切换明确保留为最终人工确认项。 diff --git a/next/.dockerignore b/next/.dockerignore new file mode 100644 index 0000000..212dea7 --- /dev/null +++ b/next/.dockerignore @@ -0,0 +1,11 @@ +.git +.venv +.pytest_cache +.ruff_cache +data +docs/evidence +frontend/node_modules +frontend/dist +tests +*.log +.env diff --git a/next/.env.example b/next/.env.example new file mode 100644 index 0000000..9f8aea8 --- /dev/null +++ b/next/.env.example @@ -0,0 +1,3 @@ +APP_ENCRYPTION_KEY=replace-with-a-persistent-fernet-key +APP_LOG_LEVEL=INFO +APP_TIMEZONE=Asia/Shanghai diff --git a/next/Dockerfile b/next/Dockerfile new file mode 100644 index 0000000..f23449b --- /dev/null +++ b/next/Dockerfile @@ -0,0 +1,29 @@ +FROM node:24-alpine AS frontend-build +WORKDIR /build/frontend +COPY frontend/package*.json ./ +RUN npm ci +COPY frontend/ ./ +RUN npm run build + +FROM python:3.14-slim AS runtime +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + APP_ENV=production \ + APP_HOST=0.0.0.0 \ + APP_PORT=8765 \ + APP_DATA_DIR=/app/data \ + APP_FRONTEND_DIST=/app/frontend/dist +WORKDIR /app +RUN addgroup --system xiaobai && adduser --system --ingroup xiaobai xiaobai +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt +COPY backend/ ./backend/ +COPY config/ ./config/ +COPY tools/ ./tools/ +COPY --from=frontend-build /build/frontend/dist ./frontend/dist +RUN mkdir -p /app/data && chown -R xiaobai:xiaobai /app +USER xiaobai +EXPOSE 8765 +HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \ + CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8765/api/health', timeout=3)"] +CMD ["python", "-m", "uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "8765", "--proxy-headers", "--no-server-header"] diff --git a/next/README.md b/next/README.md index ac8cd87..853b73a 100644 --- a/next/README.md +++ b/next/README.md @@ -46,11 +46,24 @@ npm.cmd run dev 回退只允许显式执行；代码回退不会自动回退数据库。 +旧库迁移、生产部署与恢复不进入正常运行时代码。迁移只读打开旧库并写入独立目标： + +```powershell +$env:APP_ENCRYPTION_KEY='<现有固定密钥>' +.\.venv\Scripts\python.exe -m tools.legacy_migration ` + --source ..\data\review.db ` + --target data\xiaobai.db ` + --report data\migration-report.json +``` + +一致性备份、校验恢复和 Docker 部署见 [docs/deployment.md](docs/deployment.md)。正式 NAS +容器不会被脚本自动替换，切换必须在最终验收后单独人工确认。 + ## 最小质量门禁 ```powershell cd next -.\.venv\Scripts\python.exe -m ruff check backend tests +.\.venv\Scripts\python.exe -m ruff check backend tools tests .\.venv\Scripts\python.exe -m pytest cd frontend npm.cmd run check diff --git a/next/backend/bootstrap/application.py b/next/backend/bootstrap/application.py index 7d36f29..60c4a40 100644 --- a/next/backend/bootstrap/application.py +++ b/next/backend/bootstrap/application.py @@ -2,7 +2,8 @@ import asyncio import logging from contextlib import asynccontextmanager -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException +from fastapi.responses import FileResponse from backend.bootstrap.container import build_container from backend.bootstrap.logging import configure_logging @@ -59,4 +60,16 @@ def create_application(settings: Settings | None = None) -> FastAPI: install_request_context(application) install_error_handlers(application) application.include_router(api_router, prefix="/api") + if runtime.frontend_dist_directory.joinpath("index.html").is_file(): + frontend_root = runtime.frontend_dist_directory.resolve() + + @application.get("/{requested_path:path}", include_in_schema=False) + def frontend(requested_path: str) -> FileResponse: + if requested_path == "api" or requested_path.startswith("api/"): + raise HTTPException(status_code=404) + candidate = frontend_root.joinpath(requested_path).resolve() + if candidate.is_relative_to(frontend_root) and candidate.is_file(): + return FileResponse(candidate) + return FileResponse(frontend_root / "index.html") + return application diff --git a/next/backend/bootstrap/settings.py b/next/backend/bootstrap/settings.py index 3a1caa8..920f3b8 100644 --- a/next/backend/bootstrap/settings.py +++ b/next/backend/bootstrap/settings.py @@ -41,6 +41,7 @@ class Settings: host: str port: int encryption_key: str | None + frontend_dist_directory: Path = PROJECT_ROOT / "frontend" / "dist" timezone: str = "Asia/Shanghai" @classmethod @@ -63,6 +64,9 @@ class Settings: os.getenv("APP_PRIVATE_MENTOR_SKILLS_DIR", ""), data_directory / "private-mentor-skills", ) + frontend_dist_directory = _resolve_path( + os.getenv("APP_FRONTEND_DIST", ""), PROJECT_ROOT / "frontend" / "dist" + ) log_file = _resolve_path( os.getenv("APP_LOG_FILE", ""), data_directory / "logs" / "application.log", @@ -80,6 +84,7 @@ class Settings: database_path=database_path, mentor_skills_directory=mentor_skills_directory, private_mentor_skills_directory=private_mentor_skills_directory, + frontend_dist_directory=frontend_dist_directory, log_file=log_file, log_level=log_level, host=os.getenv("APP_HOST", "127.0.0.1").strip() or "127.0.0.1", @@ -98,6 +103,7 @@ class Settings: database_path=root / "xiaobai-test.db", mentor_skills_directory=PROJECT_ROOT / "config" / "mentor-skills", private_mentor_skills_directory=root / "private-mentor-skills", + frontend_dist_directory=root / "frontend-dist", log_file=None, log_level="CRITICAL", host="127.0.0.1", diff --git a/next/compose.yaml b/next/compose.yaml new file mode 100644 index 0000000..0b34ae6 --- /dev/null +++ b/next/compose.yaml @@ -0,0 +1,26 @@ +services: + web: + build: + context: . + image: xiaobai-review-next:local + restart: unless-stopped + ports: + - "8765:8765" + env_file: + - .env + environment: + APP_ENV: production + APP_HOST: 0.0.0.0 + APP_PORT: 8765 + APP_DATA_DIR: /app/data + APP_DATABASE_PATH: /app/data/xiaobai.db + APP_PRIVATE_MENTOR_SKILLS_DIR: /app/data/private-mentor-skills + volumes: + - ./data:/app/data + read_only: true + tmpfs: + - /tmp:size=64m,mode=1777 + security_opt: + - no-new-privileges:true + cap_drop: + - ALL diff --git a/next/docs/deployment.md b/next/docs/deployment.md new file mode 100644 index 0000000..58c35fc --- /dev/null +++ b/next/docs/deployment.md @@ -0,0 +1,65 @@ +# 新系统部署与回退 + +本说明只适用于 `next/` 重建版。旧容器在最终人工确认前继续运行，不得原地覆盖。 + +## 首次准备 + +1. 将 `.env.example` 复制为 `.env`，写入固定的 Fernet `APP_ENCRYPTION_KEY`。该密钥必须与迁移旧数据时使用的密钥一致。 +2. 创建 `data/private-mentor-skills/`，把不公开的 Skill 放在该持久化目录；镜像内只含公开 Skill。 +3. 先在独立目录运行迁移，并查看报告中的迁移数、主动跳过项、完整性和校验和。 +4. 使用尚未占用的端口启动新容器并完成验收。不要停止旧容器。 + +```powershell +$env:APP_ENCRYPTION_KEY='<现有固定密钥>' +python -m tools.legacy_migration ` + --source ..\data\review.db ` + --target .\data\xiaobai.db ` + --report .\data\migration-report.json +docker compose config +docker compose build +docker compose up -d +``` + +`legacy_migration` 以只读 URI 打开旧库，可重复执行；会保留用户编号和密码散列，不迁移会话。旧的每用户 LLM 配置和可由正式数据源重建的原始因子表会明确记录为主动跳过，不会进入新运行时。 + +## 备份 + +SQLite 必须通过在线备份 API 取得一致快照，不能在运行中直接复制数据库文件。生产备份应把数据库、环境/密钥文件与私有 Skill 放入同一归档，并把归档存放在容器数据卷之外。 + +```powershell +python -m tools.backup ` + --database .\data\xiaobai.db ` + --output D:\backups\xiaobai-2026-07-30.zip ` + --private-skills .\data\private-mentor-skills ` + --environment-file .\.env ` + --apply-retention +``` + +保留策略至少为最近 7 份日备份与 4 个不同周的周备份。密钥随归档备份意味着归档自身必须保存到受控、加密的 NAS 位置。 + +## 恢复演练 + +恢复默认拒绝覆盖，且在写入前校验清单、逐文件 SHA-256、路径安全、SQLite 完整性。 + +```powershell +python -m tools.restore ` + --archive D:\backups\xiaobai-2026-07-30.zip ` + --database .\data\restore-drill.db ` + --private-skills .\data\restore-private-skills ` + --environment-file .\data\restore.env ` + --confirm-restore +``` + +恢复后必须抽验：原账号登录、共享行情快照、自选与复盘、会员状态、模型配置、问天历史、选股历史和跟踪记录。 + +## 升级与回退 + +1. 升级前生成一致备份，记录当前镜像标签与数据库 migration 版本。 +2. 用新镜像启动独立验收容器；健康检查通过后再切换反向代理或宿主端口。 +3. 代码回退只切换至旧镜像，不自动降级数据库。 +4. 若新 migration 与旧镜像不兼容，使用升级前的整套备份恢复到独立目录，再启动旧镜像。 +5. NAS 正式切换属于单独人工确认项，本仓库脚本不会自动执行切换。 + +## 本机验证边界 + +开发机未安装 Docker 时仍可执行全部 Python、前端构建、迁移和备份恢复测试，但不能声称镜像已经实际构建或容器健康检查已经运行。最终切换前应在有 Docker 的 NAS 或预发布机补做 `docker compose build`、只读文件系统、数据卷写入和重启恢复验证。 diff --git a/next/docs/evidence/stage-14/acceptance.md b/next/docs/evidence/stage-14/acceptance.md new file mode 100644 index 0000000..21d5dbd --- /dev/null +++ b/next/docs/evidence/stage-14/acceptance.md @@ -0,0 +1,51 @@ +# 阶段 14 验收 + +## 交付边界 + +- 旧系统只作为只读来源；真实演练源为 `data/review.db`，目标为新建的 `next/data/stage14-migrated.db`。 +- 迁移逻辑仅存在于一次性工具 `tools/legacy_migration.py`，正常应用不导入旧系统代码或旧数据库。 +- 新 Docker、备份和恢复资产仅位于 `next/`，没有停止、修改或替换 NAS 正式容器。 + +## 真实迁移结果 + +- 源库 SHA-256：`b7c444359c3b9b201a47ce6b23e55ad7ca4eba3d9dd67dfc8544b575d43f4924`。 +- SQLite `integrity_check=ok`，外键违规 0。 +- 3 个账号与会员状态、1 份出生资料、2 项系统数据凭据、3 个模型配置均已转换。 +- 6 条自选、3 条复盘、2 条提醒、28 条问师消息、45 条问师偏好、31 条问天历史均已迁移并保留账号归属。 +- 196 次历史选股与 16 条手动策略跟踪已迁移；每个历史交易日使用明确标注的归档因子快照，不伪造旧因子覆盖率。 +- 5,534 个股票目录、261 个交易日、24 份市场摘要、5,567 份最近 90 根日 K 展示归档和 24 份兼容市场洞察已迁移。 +- 真实账号密码验证、管理员与永久会员状态、模型选择、问天历史和选股历史抽验通过。 +- 逐项机器报告见 `real-migration-report.json`，报告不含明文令牌或模型密钥。 + +## 主动删除或重建 + +- 会话不迁移，切换后所有账号重新登录。 +- 已取消的用户自主 LLM 配置不迁移。 +- 旧原始因子表不复制成兼容表；由新系统受治理的数据同步和因子任务重建。 +- 旧指数行仅有收盘价，缺少新图表契约要求的 OHLC，未伪造成指数 K 线。 +- 日 K 展示归档每个标的保留最近 90 根；旧全量数据库和一致性备份继续作为审计资产保留。 + +## 备份恢复演练 + +- 使用 SQLite 在线备份 API 创建一致快照，归档同时包含环境密钥材料与 9 个私有 Skill 文件，共 11 个文件。 +- 恢复前验证安全路径、逐文件 SHA-256 和 SQLite 完整性；默认拒绝覆盖已有数据库、环境、密钥和私有 Skill 目录。 +- 恢复副本 `integrity_check=ok`，账号数 3、历史选股数 196，与迁移目标一致。 +- 单元测试确认篡改数据库内容会因校验和不一致被拒绝。 + +## 生产与安全抽验 + +- Vite 生产资源由 FastAPI 同端口提供；SPA 深链接可返回入口文件，未知 `/api/*` 保持统一 JSON 404。 +- 生产登录 Cookie 含 `HttpOnly`、`Secure`、`SameSite=Lax`；生产环境缺少加密密钥会拒绝启动。 +- 真实迁移库副本的行情摘要读取 20 次：中位数 20.90ms，P95 23.55ms；日 K 返回 90 个点。 +- 容器使用多阶段构建、非 root 用户、只读根文件系统、持久化数据卷、健康检查、移除 Linux capabilities 和 `no-new-privileges`。 +- 本机未安装 Docker，因此未声称镜像已实构建；NAS 切换前必须在有 Docker 的预发布环境补做构建、健康检查、只读根文件系统和重启恢复验证。 + +## 质量门 + +- Ruff：通过。 +- pytest：96 项通过。 +- Vue 类型检查：通过。 +- Vitest：3 个文件、7 项通过。 +- Vite 生产构建：通过，CSS 89.25KB，JS 279.23KB，均为构建前原始体积。 +- Playwright：17 项通过，单 worker，耗时 53.6 秒。 +- `git diff --check` 与已知令牌/密码扫描：通过。 diff --git a/next/docs/evidence/stage-14/real-migration-report.json b/next/docs/evidence/stage-14/real-migration-report.json new file mode 100644 index 0000000..cc08e3b --- /dev/null +++ b/next/docs/evidence/stage-14/real-migration-report.json @@ -0,0 +1,51 @@ +{ + "source": "C:\\Users\\MoBai\\Documents\\gupiaofupan\\webapp\\data\\review.db", + "target": "C:\\Users\\MoBai\\Documents\\gupiaofupan\\webapp\\next\\data\\stage14-migrated.db", + "source_sha256": "b7c444359c3b9b201a47ce6b23e55ad7ca4eba3d9dd67dfc8544b575d43f4924", + "target_sha256": "9b0ca7faed7a15b4099bd09ae7a0c94b52a0c8a7dbfedb9d2b26bdf47e34d5d2", + "integrity": "ok", + "foreign_key_violations": 0, + "migrated": { + "alerts": 2, + "birth_profiles": 1, + "chart_series": 5567, + "heaven_readings": 31, + "llm_models": 3, + "llm_usage_daily": 9, + "market_entities": 5534, + "market_insight_snapshots": 24, + "market_summaries": 24, + "memberships": 3, + "mentor_messages": 28, + "mentor_preferences": 45, + "review_notes": 3, + "screener_runs": 196, + "strategy_tracks": 16, + "system_credentials": 2, + "trading_days": 261, + "users": 3, + "watchlist_entries": 6 + }, + "target_counts": { + "users": 3, + "memberships": 3, + "market_entities": 5537, + "market_summaries": 24, + "chart_series": 5567, + "watchlist_entries": 6, + "review_notes": 3, + "trade_entries": 0, + "alerts": 2, + "mentor_messages": 28, + "heaven_readings": 31, + "screener_runs": 196, + "custom_screener_strategies": 0, + "strategy_tracks": 16 + }, + "intentionally_skipped": { + "sessions": "sessions are intentionally invalidated during cutover", + "user_credentials": "per-user LLM configuration was removed from the product", + "raw_factor_tables": "reproducible provider inputs are rebuilt by governed sync jobs", + "benchmark_bars": "legacy rows lack OHLC values required by the chart contract" + } +} diff --git a/next/tests/test_backup_restore.py b/next/tests/test_backup_restore.py new file mode 100644 index 0000000..43f45db --- /dev/null +++ b/next/tests/test_backup_restore.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import sqlite3 +import zipfile + +import pytest + +from tools.backup import create_backup +from tools.restore import restore_backup + + +def test_backup_restore_round_trip_and_rejects_tampering(tmp_path) -> None: + database = tmp_path / "source.db" + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE sample (value TEXT)") + connection.execute("INSERT INTO sample VALUES ('preserved')") + skills = tmp_path / "skills" + skills.mkdir() + skills.joinpath("private.md").write_text("private", encoding="utf-8") + environment = tmp_path / "source.env" + environment.write_text("APP_ENCRYPTION_KEY=hidden\n", encoding="utf-8") + archive = tmp_path / "xiaobai-2026-07-30.zip" + create_backup(database, archive, private_skills=skills, environment_file=environment) + + restored = tmp_path / "restored.db" + restored_skills = tmp_path / "restored-skills" + restore_backup(archive, restored, private_skills=restored_skills) + with sqlite3.connect(restored) as connection: + assert connection.execute("SELECT value FROM sample").fetchone()[0] == "preserved" + assert restored_skills.joinpath("private.md").read_text(encoding="utf-8") == "private" + + tampered = tmp_path / "tampered.zip" + with zipfile.ZipFile(archive) as source, zipfile.ZipFile(tampered, "w") as destination: + for item in source.infolist(): + content = source.read(item.filename) + replacement = b"broken" if item.filename == "database.sqlite3" else content + destination.writestr(item, replacement) + with pytest.raises(ValueError, match="checksum mismatch"): + restore_backup(tampered, tmp_path / "rejected.db") + + +def test_restore_refuses_implicit_overwrite(tmp_path) -> None: + database = tmp_path / "source.db" + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE sample (value TEXT)") + archive = tmp_path / "backup.zip" + create_backup(database, archive) + destination = tmp_path / "existing.db" + destination.write_bytes(b"keep") + with pytest.raises(FileExistsError): + restore_backup(archive, destination) + assert destination.read_bytes() == b"keep" diff --git a/next/tests/test_health.py b/next/tests/test_health.py index f09e088..1e2e06a 100644 --- a/next/tests/test_health.py +++ b/next/tests/test_health.py @@ -73,3 +73,21 @@ def test_validation_error_does_not_expose_framework_details(tmp_path) -> None: assert response.status_code == 422 assert response.json()["error"]["code"] == "invalid_request" assert "integer" not in response.text.lower() + + +def test_production_frontend_serves_spa_without_capturing_api_404(tmp_path) -> None: + settings = Settings.for_test(tmp_path) + settings.frontend_dist_directory.mkdir(parents=True) + settings.frontend_dist_directory.joinpath("index.html").write_text( + "
application
", encoding="utf-8" + ) + settings.frontend_dist_directory.joinpath("asset.js").write_text( + "window.ready=true", encoding="utf-8" + ) + application = create_application(settings) + + assert request(application, "/review/history").text == "
application
" + assert request(application, "/asset.js").text == "window.ready=true" + api_response = request(application, "/api/unknown") + assert api_response.status_code == 404 + assert api_response.json()["error"]["code"] == "not_found" diff --git a/next/tests/test_legacy_migration.py b/next/tests/test_legacy_migration.py new file mode 100644 index 0000000..e0189d0 --- /dev/null +++ b/next/tests/test_legacy_migration.py @@ -0,0 +1,170 @@ +# ruff: noqa: E501 - compact SQL fixtures intentionally mirror the legacy schema. + +from __future__ import annotations + +import base64 +import hashlib +import json +import sqlite3 + +from cryptography.fernet import Fernet + +from backend.security.passwords import PasswordHasher +from tools.legacy_migration import LegacyMigrator + + +def _legacy_database(path, key: str) -> None: + salt = b"0123456789abcdef" + digest = hashlib.scrypt( + b"Password123", salt=salt, n=2**14, r=8, p=1, dklen=32 + ) + encoded_salt = base64.urlsafe_b64encode(salt).decode() + encoded_hash = base64.urlsafe_b64encode(digest).decode() + fernet = Fernet(key.encode()) + credentials = { + "tushare_token": "secret-tushare", + "ifind_refresh_token": "secret-refresh", + "member_daily_limit": 61, + "llm_models": [{ + "id": "main", "name": "Primary", "base_url": "https://model.invalid/v1", + "model": "model-a", "api_key": "secret-model", + }], + "primary_model_id": "main", + } + with sqlite3.connect(path) as connection: + connection.executescript( + """ + CREATE TABLE users (id INTEGER PRIMARY KEY,username TEXT,password_salt TEXT, + password_hash TEXT,created_at TEXT,updated_at TEXT,role TEXT,llm_mode TEXT, + membership_status TEXT,membership_plan TEXT,membership_starts_at TEXT, + membership_expires_at TEXT); + CREATE TABLE user_birth_profiles + (user_id INTEGER PRIMARY KEY,encrypted_payload TEXT,updated_at TEXT); + CREATE TABLE llm_usage (id INTEGER PRIMARY KEY,user_id INTEGER,feature TEXT, + source TEXT,model TEXT,status TEXT,latency_ms INTEGER,created_at TEXT, + role TEXT,prompt_version TEXT,error_code TEXT,input_tokens INTEGER,output_tokens INTEGER); + CREATE TABLE system_settings + (setting_key TEXT PRIMARY KEY,encrypted_payload TEXT,updated_at TEXT); + CREATE TABLE stock_master (ts_code TEXT PRIMARY KEY,code TEXT,name TEXT, + industry TEXT,market TEXT,list_date TEXT,updated_at TEXT); + CREATE TABLE daily_bars (trade_date TEXT,ts_code TEXT,open REAL,high REAL, + low REAL,close REAL,pct_chg REAL,vol REAL,amount REAL); + CREATE TABLE dashboard_snapshots + (trade_date TEXT PRIMARY KEY,source TEXT,payload TEXT,record_count INTEGER,updated_at TEXT); + CREATE TABLE watchlist (user_id INTEGER,code TEXT,name TEXT,sector TEXT,color TEXT, + created_at TEXT,updated_at TEXT,remark TEXT); + CREATE TABLE review_notes (id INTEGER PRIMARY KEY,code TEXT,stock_name TEXT, + trade_date TEXT,content TEXT,plan TEXT,created_at TEXT,updated_at TEXT, + user_id INTEGER,summary TEXT); + CREATE TABLE trade_entries (id INTEGER PRIMARY KEY,user_id INTEGER,trade_date TEXT, + code TEXT,name TEXT,action TEXT,price REAL,quantity INTEGER,position_pct REAL, + pnl_amount REAL,pnl_pct REAL,thesis TEXT,execution TEXT,emotion TEXT,tags TEXT, + created_at TEXT,updated_at TEXT); + CREATE TABLE alerts (id INTEGER PRIMARY KEY,user_id INTEGER,kind TEXT,title TEXT, + content TEXT,available_date TEXT,code TEXT,dedupe_key TEXT,is_read INTEGER, + created_at TEXT,updated_at TEXT,read_at TEXT); + CREATE TABLE assistant_messages (id INTEGER PRIMARY KEY,user_id INTEGER,role TEXT, + content TEXT,context_date TEXT,created_at TEXT); + CREATE TABLE mentor_preferences (user_id INTEGER,mentor_id TEXT,pinned INTEGER, + sort_order INTEGER,updated_at TEXT); + CREATE TABLE mentor_messages (id INTEGER PRIMARY KEY,user_id INTEGER,mentor_id TEXT, + trade_date TEXT,role TEXT,content TEXT,meta TEXT,created_at TEXT); + CREATE TABLE heaven_readings (id INTEGER PRIMARY KEY,user_id INTEGER,mode TEXT, + context_date TEXT,subject TEXT,subject_detail TEXT,answer TEXT, + context_snapshot TEXT,dedupe_key TEXT,created_at TEXT); + CREATE TABLE screener_runs (id INTEGER PRIMARY KEY,trade_date TEXT,regime TEXT, + strategy_name TEXT,formula TEXT,result TEXT,created_at TEXT,user_id INTEGER,mode TEXT); + CREATE TABLE screener_strategies (id INTEGER PRIMARY KEY,name TEXT,description TEXT, + regimes TEXT,formula TEXT,builtin INTEGER,created_at TEXT,updated_at TEXT,user_id INTEGER); + CREATE TABLE strategy_tracks (id INTEGER PRIMARY KEY,user_id INTEGER,run_id INTEGER, + selection_date TEXT,strategy_name TEXT,ts_code TEXT,code TEXT,name TEXT,sector TEXT, + entry_price REAL,created_at TEXT,updated_at TEXT); + CREATE TABLE data_snapshots (kind TEXT,cache_key TEXT,source TEXT,payload TEXT, + updated_at TEXT,PRIMARY KEY(kind,cache_key)); + """ + ) + now = "2026-07-29T16:00:00+08:00" + connection.execute( + "INSERT INTO users VALUES (8,'leefer',?,?,?,?,'admin','auto','active','永久',?,NULL)", + (encoded_salt, encoded_hash, now, now, now), + ) + profile = fernet.encrypt(b'{"gender":"male"}').decode() + connection.execute("INSERT INTO user_birth_profiles VALUES (8,?,?)", (profile, now)) + connection.execute( + "INSERT INTO llm_usage VALUES (1,8,'mentor','model','','success',3,?,'','','',0,0)", + (now,), + ) + encrypted = fernet.encrypt(json.dumps(credentials).encode()).decode() + connection.execute("INSERT INTO system_settings VALUES ('credentials',?,?)", (encrypted, now)) + connection.execute( + "INSERT INTO stock_master VALUES ('000001.SZ','000001','平安银行','银行','主板','19910403',?)", + (now,), + ) + connection.execute( + "INSERT INTO daily_bars VALUES ('20260729','000001.SZ',10,11,9,10.5,5,100,1000)" + ) + connection.execute( + "INSERT INTO dashboard_snapshots VALUES ('20260729','tushare','{}',1,?)", (now,) + ) + connection.execute( + "INSERT INTO watchlist VALUES (8,'000001','平安银行','银行','red',?,?, '长期')", + (now, now), + ) + connection.execute( + "INSERT INTO review_notes VALUES (1,'000001','平安银行','20260729','复盘','计划',?,?,8,'总结')", + (now, now), + ) + connection.execute( + "INSERT INTO trade_entries VALUES (1,8,'20260729','000001','平安银行','buy',10,100,20,NULL,NULL,'逻辑','执行','calm','[]',?,?)", + (now, now), + ) + connection.execute( + "INSERT INTO alerts VALUES (1,8,'manual','提醒','','20260730','','a',0,?,?,NULL)", + (now, now), + ) + connection.execute("INSERT INTO assistant_messages VALUES (1,8,'user','问题','20260729',?)", (now,)) + connection.execute("INSERT INTO mentor_preferences VALUES (8,'mentor',1,1,?)", (now,)) + connection.execute( + "INSERT INTO mentor_messages VALUES (1,8,'mentor','20260729','user','问题','',?)", (now,) + ) + connection.execute( + "INSERT INTO heaven_readings VALUES (1,8,'fortune','20260729','','','结果','{}','d',?)", (now,) + ) + result = json.dumps({"candidates": [{"identifier": "000001.SZ", "close": 10.5}]}) + connection.execute( + "INSERT INTO screener_runs VALUES (1,'20260729','','策略','{}',?, ?,8,'curated')", + (result, now), + ) + connection.execute( + "INSERT INTO screener_strategies VALUES (1,'自定义','','[]','{}',0,?,?,8)", + (now, now), + ) + connection.execute( + "INSERT INTO strategy_tracks VALUES (1,8,1,'20260729','策略','000001.SZ','000001','平安银行','银行',10.5,?,?)", + (now, now), + ) + connection.execute( + "INSERT INTO data_snapshots VALUES ('popularity_v1','20260729','legacy','{}',?)", (now,) + ) + + +def test_legacy_migration_is_idempotent_and_preserves_login(tmp_path) -> None: + source = tmp_path / "legacy.db" + target = tmp_path / "next.db" + key = Fernet.generate_key().decode() + _legacy_database(source, key) + + first = LegacyMigrator(source, target, key).run() + second = LegacyMigrator(source, target, key).run() + + assert first["integrity"] == second["integrity"] == "ok" + rendered = json.dumps(first) + assert "secret-tushare" not in rendered + assert "secret-model" not in rendered + with sqlite3.connect(target) as connection: + password = connection.execute("SELECT password_hash FROM users WHERE id=8").fetchone()[0] + assert PasswordHasher().verify("Password123", password) + assert connection.execute("SELECT count(*) FROM watchlist_entries").fetchone()[0] == 1 + assert connection.execute("SELECT count(*) FROM screener_runs").fetchone()[0] == 1 + assert connection.execute("SELECT count(*) FROM strategy_tracks").fetchone()[0] == 1 + assert connection.execute("SELECT daily_llm_limit FROM memberships").fetchone()[0] == 61 diff --git a/next/tools/backup.py b/next/tools/backup.py new file mode 100644 index 0000000..67e92ce --- /dev/null +++ b/next/tools/backup.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import argparse +import hashlib +import json +import sqlite3 +import tempfile +import zipfile +from collections.abc import Sequence +from contextlib import closing +from datetime import datetime +from pathlib import Path + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def snapshot_database(source: Path, destination: Path) -> None: + source_uri = f"file:{source.resolve().as_posix()}?mode=ro" + with closing(sqlite3.connect(source_uri, uri=True)) as origin: + with closing(sqlite3.connect(destination)) as snapshot: + origin.backup(snapshot) + if snapshot.execute("PRAGMA integrity_check").fetchone()[0] != "ok": + raise RuntimeError("database snapshot failed integrity check") + + +def create_backup( + database: Path, + output: Path, + *, + private_skills: Path | None = None, + environment_file: Path | None = None, + encryption_key_file: Path | None = None, +) -> dict: + database = database.resolve() + if not database.is_file(): + raise FileNotFoundError(database) + output = output.resolve() + output.parent.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="xiaobai-backup-") as raw: + staging = Path(raw) + database_copy = staging / "database.sqlite3" + snapshot_database(database, database_copy) + files: list[tuple[Path, str]] = [(database_copy, "database.sqlite3")] + for source, archive_name in ( + (environment_file, "secrets/environment"), + (encryption_key_file, "secrets/app-encryption.key"), + ): + if source: + resolved = source.resolve() + if not resolved.is_file(): + raise FileNotFoundError(resolved) + files.append((resolved, archive_name)) + if private_skills and private_skills.exists(): + files.extend( + (item, f"private-mentor-skills/{item.relative_to(private_skills).as_posix()}") + for item in private_skills.rglob("*") + if item.is_file() + ) + manifest = { + "format": 1, + "created_at": datetime.now().astimezone().isoformat(timespec="seconds"), + "files": { + name: {"sha256": sha256(path), "size": path.stat().st_size} for path, name in files + }, + } + manifest_path = staging / "manifest.json" + manifest_path.write_text( + json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + temporary = output.with_suffix(output.suffix + ".tmp") + with zipfile.ZipFile(temporary, "w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.write(manifest_path, "manifest.json") + for path, name in files: + archive.write(path, name) + temporary.replace(output) + return manifest + + +def apply_retention(directory: Path, *, keep_daily: int = 7, keep_weekly: int = 4) -> list[Path]: + backups = sorted( + directory.glob("xiaobai-*.zip"), key=lambda item: item.stat().st_mtime, reverse=True + ) + keep = set(backups[:keep_daily]) + weeks: set[str] = set() + for backup in backups: + week = datetime.fromtimestamp(backup.stat().st_mtime).strftime("%G-%V") + if week not in weeks and len(weeks) < keep_weekly: + weeks.add(week) + keep.add(backup) + removed = [] + for backup in backups: + if backup not in keep: + backup.unlink() + removed.append(backup) + return removed + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Create a consistent Xiaobai backup archive") + parser.add_argument("--database", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--private-skills", type=Path) + parser.add_argument("--environment-file", type=Path) + parser.add_argument("--encryption-key-file", type=Path) + parser.add_argument("--apply-retention", action="store_true") + return parser + + +def main(arguments: Sequence[str] | None = None) -> int: + parsed = build_parser().parse_args(arguments) + manifest = create_backup( + parsed.database, + parsed.output, + private_skills=parsed.private_skills, + environment_file=parsed.environment_file, + encryption_key_file=parsed.encryption_key_file, + ) + removed = apply_retention(parsed.output.parent) if parsed.apply_retention else [] + print( + json.dumps( + { + "backup": str(parsed.output.resolve()), + "files": len(manifest["files"]), + "removed_by_retention": len(removed), + }, + ensure_ascii=False, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/next/tools/legacy_migration.py b/next/tools/legacy_migration.py new file mode 100644 index 0000000..4222fb3 --- /dev/null +++ b/next/tools/legacy_migration.py @@ -0,0 +1,629 @@ +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sqlite3 +from collections import defaultdict +from collections.abc import Sequence +from datetime import datetime +from pathlib import Path +from typing import Any + +from cryptography.fernet import Fernet + +from backend.database import MIGRATIONS, Database, MigrationRunner + +ARCHIVE_VERSION = "legacy-archive-v1" + + +def _iso_date(value: Any) -> str: + text = str(value or "").strip() + compact = text[:10].replace("-", "") + return f"{compact[:4]}-{compact[4:6]}-{compact[6:8]}" if len(compact) >= 8 else "" + + +def _json(value: Any, fallback: Any) -> Any: + if isinstance(value, (dict, list)): + return value + try: + return json.loads(str(value)) + except (TypeError, ValueError, json.JSONDecodeError): + return fallback + + +def _dump(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + +def _hash_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _table_exists(connection: sqlite3.Connection, table: str) -> bool: + return ( + connection.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() + is not None + ) + + +class LegacyMigrator: + def __init__(self, source: Path, target: Path, encryption_key: str | None) -> None: + self.source_path = source.resolve() + self.target_path = target.resolve() + self.key = encryption_key + self.counts: dict[str, int] = defaultdict(int) + self.skipped: dict[str, str] = { + "sessions": "sessions are intentionally invalidated during cutover", + "user_credentials": "per-user LLM configuration was removed from the product", + "raw_factor_tables": "reproducible provider inputs are rebuilt by governed sync jobs", + "benchmark_bars": "legacy rows lack OHLC values required by the chart contract", + } + self.user_ids: set[int] = set() + self.run_ids: set[int] = set() + + def run(self) -> dict[str, Any]: + if self.source_path == self.target_path: + raise ValueError("source and target database paths must differ") + if not self.source_path.is_file(): + raise FileNotFoundError(self.source_path) + database = Database(self.target_path) + MigrationRunner(database).upgrade(MIGRATIONS) + source = sqlite3.connect(f"file:{self.source_path.as_posix()}?mode=ro", uri=True) + source.row_factory = sqlite3.Row + try: + with database.transaction() as target: + self._accounts(source, target) + self._system_settings(source, target) + self._market(source, target) + self._private_data(source, target) + self._screener(source, target) + self._insights(source, target) + with database.read() as target: + integrity = str(target.execute("PRAGMA integrity_check").fetchone()[0]) + foreign_keys = list(target.execute("PRAGMA foreign_key_check")) + target_counts = { + table: int(target.execute(f"SELECT count(*) FROM {table}").fetchone()[0]) + for table in ( + "users", "memberships", "market_entities", "market_summaries", + "chart_series", "watchlist_entries", "review_notes", "trade_entries", + "alerts", "mentor_messages", "heaven_readings", "screener_runs", + "custom_screener_strategies", "strategy_tracks", + ) + } + finally: + source.close() + if integrity != "ok" or foreign_keys: + raise RuntimeError("migrated database failed integrity validation") + return { + "source": str(self.source_path), + "target": str(self.target_path), + "source_sha256": _hash_file(self.source_path), + "target_sha256": _hash_file(self.target_path), + "integrity": integrity, + "foreign_key_violations": 0, + "migrated": dict(sorted(self.counts.items())), + "target_counts": target_counts, + "intentionally_skipped": self.skipped, + } + + def _accounts(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: + users = source.execute("SELECT * FROM users ORDER BY id").fetchall() + for row in users: + user_id = int(row["id"]) + self.user_ids.add(user_id) + password = f"scrypt$16384$8$1${row['password_salt']}${row['password_hash']}" + target.execute( + """INSERT INTO users + (id,username,username_key,password_hash,is_admin,status,created_at,updated_at) + VALUES (?,?,?,?,?,'active',?,?) ON CONFLICT(id) DO UPDATE SET + username=excluded.username, username_key=excluded.username_key, + password_hash=excluded.password_hash, is_admin=excluded.is_admin, + status=excluded.status, updated_at=excluded.updated_at""", + ( + user_id, + row["username"], + str(row["username"]).casefold(), + password, + int(str(row["role"]) == "admin"), + row["created_at"], + row["updated_at"], + ), + ) + permanent = str(row["membership_plan"]) == "永久" + state = "active" if row["membership_status"] == "active" else "inactive" + target.execute( + """INSERT INTO memberships + (user_id,state,expires_at,is_permanent,daily_llm_limit,updated_at,updated_by) + VALUES (?,?,?,?,50,?,NULL) ON CONFLICT(user_id) DO UPDATE SET + state=excluded.state,expires_at=excluded.expires_at, + is_permanent=excluded.is_permanent,updated_at=excluded.updated_at""", + (user_id, state, row["membership_expires_at"], int(permanent), row["updated_at"]), + ) + self.counts["users"] = len(users) + self.counts["memberships"] = len(users) + if _table_exists(source, "user_birth_profiles"): + for row in source.execute("SELECT * FROM user_birth_profiles"): + target.execute( + """INSERT INTO birth_profiles + (user_id,encrypted_payload,created_at,updated_at) VALUES (?,?,?,?) + ON CONFLICT(user_id) DO UPDATE SET + encrypted_payload=excluded.encrypted_payload,updated_at=excluded.updated_at""", + ( + row["user_id"], + row["encrypted_payload"], + row["updated_at"], + row["updated_at"], + ), + ) + self.counts["birth_profiles"] += 1 + if _table_exists(source, "llm_usage"): + for row in source.execute( + """SELECT user_id,substr(created_at,1,10) usage_date,count(*) calls, + max(created_at) updated_at FROM llm_usage WHERE status='success' + GROUP BY user_id,substr(created_at,1,10)""" + ): + target.execute( + """INSERT INTO llm_usage_daily VALUES (?,?,?,?) + ON CONFLICT(user_id,usage_date) DO UPDATE SET + successful_calls=excluded.successful_calls,updated_at=excluded.updated_at""", + (row["user_id"], row["usage_date"], row["calls"], row["updated_at"]), + ) + self.counts["llm_usage_daily"] += 1 + + def _system_settings(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: + row = source.execute( + """SELECT encrypted_payload,updated_at FROM system_settings + WHERE setting_key='credentials'""" + ).fetchone() + if row is None: + return + if not self.key: + raise ValueError("APP_ENCRYPTION_KEY is required to migrate encrypted settings") + fernet = Fernet(self.key.encode("ascii")) + credentials = json.loads(fernet.decrypt(str(row["encrypted_payload"]).encode("ascii"))) + admin_id = min(self.user_ids) + limit = max(1, min(int(credentials.get("member_daily_limit") or 50), 1000)) + target.execute("UPDATE memberships SET daily_llm_limit=?", (limit,)) + for name in ("tushare_token", "ifind_refresh_token", "ifind_access_token"): + value = str(credentials.get(name) or "").strip() + if value: + target.execute( + """INSERT INTO system_credentials VALUES (?,?,?,?) + ON CONFLICT(name) DO UPDATE SET encrypted_value=excluded.encrypted_value, + updated_at=excluded.updated_at,updated_by=excluded.updated_by""", + (name, fernet.encrypt(value.encode()).decode(), row["updated_at"], admin_id), + ) + self.counts["system_credentials"] += 1 + model_map: dict[str, int] = {} + for model in credentials.get("llm_models") or []: + key = str(model.get("id") or model.get("name") or "") + display = str(model.get("name") or model.get("model") or "model").strip() + api_key = str(model.get("api_key") or "") + target.execute( + """INSERT INTO llm_models + (display_name,display_name_key,base_url,model_identifier,encrypted_api_key, + created_at,updated_at,updated_by) VALUES (?,?,?,?,?,?,?,?) + ON CONFLICT(display_name_key) DO UPDATE SET base_url=excluded.base_url, + model_identifier=excluded.model_identifier,encrypted_api_key=excluded.encrypted_api_key, + updated_at=excluded.updated_at,updated_by=excluded.updated_by""", + ( + display, + display.casefold(), + str(model.get("base_url") or ""), + str(model.get("model") or ""), + fernet.encrypt(api_key.encode()).decode(), + row["updated_at"], + row["updated_at"], + admin_id, + ), + ) + model_id = int( + target.execute( + "SELECT id FROM llm_models WHERE display_name_key=?", (display.casefold(),) + ).fetchone()[0] + ) + model_map[key] = model_id + self.counts["llm_models"] += 1 + primary = model_map.get(str(credentials.get("primary_model_id") or "")) + fallback = model_map.get(str(credentials.get("fallback_model_id") or "")) + if fallback == primary: + fallback = None + target.execute( + """UPDATE llm_configuration SET primary_model_id=?,fallback_model_id=?, + updated_at=?,updated_by=? WHERE id=1""", + (primary, fallback, row["updated_at"], admin_id), + ) + + def _market(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: + observed = datetime.now().astimezone().isoformat(timespec="seconds") + stocks = source.execute("SELECT * FROM stock_master ORDER BY ts_code").fetchall() + for row in stocks: + target.execute( + """INSERT INTO market_entities VALUES ('stock',?,?,?,?,?,1,'legacy',?) + ON CONFLICT(entity_type,identifier) DO UPDATE SET code=excluded.code, + name=excluded.name,search_key=excluded.search_key,sector=excluded.sector, + active=1,source='legacy',observed_at=excluded.observed_at""", + ( + row["ts_code"], + row["code"], + row["name"], + f"{row['code']} {row['ts_code']} {row['name']} {row['industry']}".casefold(), + row["industry"] or None, + row["updated_at"] or observed, + ), + ) + self.counts["market_entities"] = len(stocks) + dates = [ + row[0] + for row in source.execute( + "SELECT DISTINCT trade_date FROM daily_bars ORDER BY trade_date" + ) + ] + previous = None + for raw_date in dates: + trade_date = _iso_date(raw_date) + target.execute( + """INSERT INTO trading_days VALUES (?,1,?,'legacy',?) + ON CONFLICT(trade_date) DO UPDATE SET is_open=1, + previous_open_date=excluded.previous_open_date""", + (trade_date, previous, observed), + ) + previous = trade_date + self.counts["trading_days"] = len(dates) + for row in source.execute("SELECT * FROM dashboard_snapshots"): + target.execute( + """INSERT INTO market_summaries VALUES (?,?,'archive','legacy',1,?,?) + ON CONFLICT(trade_date) DO UPDATE SET observed_at=excluded.observed_at, + state='archive',source='legacy',coverage=1,payload_json=excluded.payload_json, + created_at=excluded.created_at""", + ( + _iso_date(row["trade_date"]), + row["updated_at"], + row["payload"], + row["updated_at"], + ), + ) + self.counts["market_summaries"] += 1 + query = """WITH ranked AS ( + SELECT *,row_number() OVER (PARTITION BY ts_code ORDER BY trade_date DESC) rank + FROM daily_bars) SELECT * FROM ranked WHERE rank<=90 ORDER BY ts_code,trade_date""" + current = "" + points: list[dict[str, Any]] = [] + for row in source.execute(query): + code = str(row["ts_code"]) + if current and code != current: + self._save_chart(target, current, points, observed) + points = [] + current = code + points.append( + { + "time": _iso_date(row["trade_date"]), + "open": row["open"], + "high": row["high"], + "low": row["low"], + "close": row["close"], + "volume": float(row["vol"] or 0) * 100, + "amount": float(row["amount"] or 0) * 1000, + "average": None, + } + ) + if current: + self._save_chart(target, current, points, observed) + + def _save_chart( + self, + target: sqlite3.Connection, + identifier: str, + points: list[dict[str, Any]], + observed: str, + ) -> None: + previous = points[-2]["close"] if len(points) > 1 else None + target.execute( + """INSERT INTO chart_series VALUES ('stock',?,'day',?,?,'tushare','display', + 'none',1,?,?) ON CONFLICT(entity_type,identifier,interval,trade_date) + DO UPDATE SET payload_json=excluded.payload_json,observed_at=excluded.observed_at""", + ( + identifier, + points[-1]["time"], + observed, + _dump({"previous_close": previous, "points": points}), + observed, + ), + ) + self.counts["chart_series"] += 1 + + def _private_data(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: + stock_ids = { + row["code"]: row["ts_code"] + for row in source.execute("SELECT code,ts_code FROM stock_master") + } + for row in source.execute("SELECT * FROM watchlist"): + if int(row["user_id"]) not in self.user_ids: + continue + target.execute( + """INSERT INTO watchlist_entries VALUES (?,?,?,?,?,?) + ON CONFLICT(user_id,identifier) DO UPDATE SET name=excluded.name, + sector=excluded.sector,remark=excluded.remark""", + ( + row["user_id"], + stock_ids.get(row["code"], row["code"]), + row["name"], + row["sector"] or None, + row["created_at"], + row["remark"], + ), + ) + self.counts["watchlist_entries"] += 1 + self._copy_review_rows(source, target) + for row in source.execute("SELECT * FROM mentor_preferences"): + target.execute( + "INSERT OR REPLACE INTO mentor_preferences VALUES (?,?,?,?,?)", + ( + row["user_id"], + row["mentor_id"], + row["pinned"], + row["sort_order"], + row["updated_at"], + ), + ) + self.counts["mentor_preferences"] += 1 + for row in source.execute("SELECT * FROM mentor_messages"): + target.execute( + """INSERT OR REPLACE INTO mentor_messages + (id,user_id,mentor_id,trade_date,role,content,request_id,status,created_at) + VALUES (?,?,?,?,?,?,NULL,'complete',?)""", + ( + row["id"], + row["user_id"], + row["mentor_id"], + _iso_date(row["trade_date"]), + row["role"], + row["content"], + row["created_at"], + ), + ) + self.counts["mentor_messages"] += 1 + for row in source.execute("SELECT * FROM heaven_readings"): + snapshot = _json(row["context_snapshot"], {}) + if row["subject_detail"]: + snapshot.setdefault("legacy_subject_detail", row["subject_detail"]) + target.execute( + """INSERT OR REPLACE INTO heaven_readings + (id,user_id,mode,reading_date,subject_key,result_json,interpretation, + interpretation_status,request_id,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,'complete',NULL,?,?)""", + ( + row["id"], + row["user_id"], + row["mode"], + _iso_date(row["context_date"]), + row["subject"], + _dump(snapshot), + row["answer"], + row["created_at"], + row["created_at"], + ), + ) + self.counts["heaven_readings"] += 1 + + def _copy_review_rows(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: + for row in source.execute("SELECT * FROM review_notes WHERE user_id IS NOT NULL"): + target.execute( + """INSERT OR REPLACE INTO review_notes VALUES (?,?,?,?,?,?,?,?,?,?)""", + ( + row["id"], + row["user_id"], + row["code"], + row["stock_name"], + _iso_date(row["trade_date"]), + row["summary"], + row["content"], + row["plan"], + row["created_at"], + row["updated_at"], + ), + ) + self.counts["review_notes"] += 1 + actions = {"buy", "sell", "add", "trim", "watch"} + emotions = {"calm", "confident", "hesitant", "anxious", "impulsive"} + for row in source.execute("SELECT * FROM trade_entries"): + target.execute( + """INSERT OR REPLACE INTO trade_entries + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + row["id"], + row["user_id"], + _iso_date(row["trade_date"]), + row["code"], + row["name"], + row["action"] if row["action"] in actions else "watch", + row["price"], + row["quantity"], + row["position_pct"], + row["pnl_amount"], + row["pnl_pct"], + row["emotion"] if row["emotion"] in emotions else "calm", + row["tags"], + row["thesis"], + row["execution"], + row["created_at"], + row["updated_at"], + ), + ) + self.counts["trade_entries"] += 1 + for row in source.execute("SELECT * FROM alerts"): + kind = ( + row["kind"] if row["kind"] in {"manual", "strategy_t1", "strategy_t5"} else "manual" + ) + target.execute( + "INSERT OR REPLACE INTO alerts VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", + ( + row["id"], + row["user_id"], + kind, + row["title"], + row["content"], + _iso_date(row["available_date"]), + row["code"], + row["dedupe_key"], + row["is_read"], + row["read_at"], + row["created_at"], + row["updated_at"], + ), + ) + self.counts["alerts"] += 1 + for row in source.execute("SELECT * FROM assistant_messages"): + target.execute( + """INSERT OR REPLACE INTO review_assistant_messages + VALUES (?,?,?,?,?,NULL,'complete',?)""", + ( + row["id"], + row["user_id"], + row["role"], + row["content"], + _iso_date(row["context_date"]), + row["created_at"], + ), + ) + self.counts["review_assistant_messages"] += 1 + + def _screener(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: + snapshots: dict[str, int] = {} + for row in source.execute("SELECT * FROM screener_runs ORDER BY id"): + trade_date = _iso_date(row["trade_date"]) + if trade_date not in snapshots: + target.execute( + """INSERT OR IGNORE INTO screener_factor_snapshots + (trade_date,version,observed_at,state,source_set_json,coverage_json,created_at) + VALUES (?,? ,?,'archive','[\"legacy\"]','{}',?)""", + (trade_date, ARCHIVE_VERSION, row["created_at"], row["created_at"]), + ) + snapshots[trade_date] = int( + target.execute( + "SELECT id FROM screener_factor_snapshots WHERE trade_date=? AND version=?", + (trade_date, ARCHIVE_VERSION), + ).fetchone()[0] + ) + result = _json(row["result"], {}) + candidates = result.get("candidates") if isinstance(result, dict) else [] + candidates = candidates if isinstance(candidates, list) else [] + for candidate in candidates: + if isinstance(candidate, dict) and not candidate.get("identifier"): + candidate["identifier"] = ( + candidate.get("ts_code") or candidate.get("code") or "" + ) + mode = {"smart": "stage", "curated": "curated", "quant": "custom"}.get( + str(row["mode"]), "custom" + ) + formula = _json(row["formula"], {}) + strategy_id = str(formula.get("id") or f"legacy-{row['id']}") + target.execute( + """INSERT OR REPLACE INTO screener_runs + (id,owner_user_id,mode,strategy_id,strategy_name,strategy_version, + selection_date,factor_snapshot_id,status,started_at,completed_at,coverage, + missing_fields_json,result_json,error_message) + VALUES (?,?,?,?,?,1,?,?,?, ?,?,0,'[]',?,'')""", + ( + row["id"], + row["user_id"], + mode, + strategy_id, + row["strategy_name"], + trade_date, + snapshots[trade_date], + "completed" if candidates else "no_signal", + row["created_at"], + row["created_at"], + _dump(candidates), + ), + ) + self.run_ids.add(int(row["id"])) + self.counts["screener_runs"] += 1 + for row in source.execute("SELECT * FROM screener_strategies WHERE user_id IS NOT NULL"): + target.execute( + """INSERT OR REPLACE INTO custom_screener_strategies + (id,user_id,name,version,formula_json,created_at,updated_at) + VALUES (?,?,?,1,?,?,?)""", + ( + row["id"], + row["user_id"], + row["name"], + row["formula"], + row["created_at"], + row["updated_at"], + ), + ) + self.counts["custom_screener_strategies"] += 1 + for row in source.execute("SELECT * FROM strategy_tracks"): + if int(row["run_id"]) not in self.run_ids: + continue + target.execute( + """INSERT OR REPLACE INTO strategy_tracks + (id,user_id,run_id,identifier,code,name,sector,selection_date, + strategy_name,entry_price,added_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + ( + row["id"], + row["user_id"], + row["run_id"], + row["ts_code"], + row["code"], + row["name"], + row["sector"], + _iso_date(row["selection_date"]), + row["strategy_name"], + row["entry_price"], + row["created_at"], + ), + ) + self.counts["strategy_tracks"] += 1 + + def _insights(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: + mappings = { + "auction_center_v6": "auction", + "theme_library_v1": "themes", + "popularity_v1": "popularity", + "dragon_tiger": "dragon-list", + } + for old_kind, new_kind in mappings.items(): + for row in source.execute("SELECT * FROM data_snapshots WHERE kind=?", (old_kind,)): + trade_date = _iso_date(str(row["cache_key"]).split(":", 1)[0]) + if not trade_date: + continue + target.execute( + """INSERT OR REPLACE INTO market_insight_snapshots + VALUES (?,?,'',?,'archive','legacy',1,?)""", + (new_kind, trade_date, row["updated_at"], row["payload"]), + ) + self.counts["market_insight_snapshots"] += 1 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Migrate a read-only legacy database copy") + parser.add_argument("--source", type=Path, required=True) + parser.add_argument("--target", type=Path, required=True) + parser.add_argument("--report", type=Path) + return parser + + +def main(arguments: Sequence[str] | None = None) -> int: + parsed = build_parser().parse_args(arguments) + report = LegacyMigrator(parsed.source, parsed.target, os.getenv("APP_ENCRYPTION_KEY")).run() + rendered = json.dumps(report, ensure_ascii=False, indent=2) + if parsed.report: + parsed.report.parent.mkdir(parents=True, exist_ok=True) + parsed.report.write_text(rendered + "\n", encoding="utf-8") + print(rendered) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/next/tools/restore.py b/next/tools/restore.py new file mode 100644 index 0000000..bd71200 --- /dev/null +++ b/next/tools/restore.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import argparse +import json +import shutil +import sqlite3 +import tempfile +import zipfile +from collections.abc import Sequence +from contextlib import closing +from pathlib import Path + +from tools.backup import sha256 + + +def _safe_members(archive: zipfile.ZipFile) -> list[zipfile.ZipInfo]: + members = archive.infolist() + for member in members: + path = Path(member.filename) + if path.is_absolute() or ".." in path.parts: + raise ValueError("backup contains an unsafe path") + return members + + +def restore_backup( + archive_path: Path, + database: Path, + *, + private_skills: Path | None = None, + environment_file: Path | None = None, + encryption_key_file: Path | None = None, + overwrite: bool = False, +) -> dict: + archive_path = archive_path.resolve() + database = database.resolve() + destinations = [database] + destinations.extend(path.resolve() for path in (environment_file, encryption_key_file) if path) + if private_skills: + destinations.append(private_skills.resolve()) + if not overwrite and any(path.exists() for path in destinations): + raise FileExistsError( + "restore destination exists; explicit overwrite confirmation is required" + ) + with tempfile.TemporaryDirectory(prefix="xiaobai-restore-") as raw: + staging = Path(raw) + with zipfile.ZipFile(archive_path) as archive: + _safe_members(archive) + archive.extractall(staging) + manifest = json.loads((staging / "manifest.json").read_text(encoding="utf-8")) + if manifest.get("format") != 1 or not isinstance(manifest.get("files"), dict): + raise ValueError("backup manifest is invalid") + for name, metadata in manifest["files"].items(): + path = staging / name + if not path.is_file() or sha256(path) != metadata.get("sha256"): + raise ValueError(f"backup checksum mismatch: {name}") + restored_db = staging / "database.sqlite3" + with closing( + sqlite3.connect(f"file:{restored_db.as_posix()}?mode=ro", uri=True) + ) as connection: + if connection.execute("PRAGMA integrity_check").fetchone()[0] != "ok": + raise ValueError("restored database failed integrity check") + database.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(restored_db, database) + for destination, name in ( + (environment_file, "secrets/environment"), + (encryption_key_file, "secrets/app-encryption.key"), + ): + source = staging / name + if destination and source.is_file(): + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + skills_source = staging / "private-mentor-skills" + if private_skills and skills_source.is_dir(): + private_skills.mkdir(parents=True, exist_ok=True) + shutil.copytree(skills_source, private_skills, dirs_exist_ok=True) + return manifest + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Validate and restore a Xiaobai backup") + parser.add_argument("--archive", type=Path, required=True) + parser.add_argument("--database", type=Path, required=True) + parser.add_argument("--private-skills", type=Path) + parser.add_argument("--environment-file", type=Path) + parser.add_argument("--encryption-key-file", type=Path) + parser.add_argument("--confirm-restore", action="store_true") + parser.add_argument("--confirm-overwrite", action="store_true") + return parser + + +def main(arguments: Sequence[str] | None = None) -> int: + parsed = build_parser().parse_args(arguments) + if not parsed.confirm_restore: + raise SystemExit("restore requires --confirm-restore") + manifest = restore_backup( + parsed.archive, + parsed.database, + private_skills=parsed.private_skills, + environment_file=parsed.environment_file, + encryption_key_file=parsed.encryption_key_file, + overwrite=parsed.confirm_overwrite, + ) + print( + json.dumps( + {"restored": str(parsed.database.resolve()), "files": len(manifest["files"])}, + ensure_ascii=False, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())