feat(deploy): 完成数据迁移与MySQL全链路(14表698k消息100%一致+5方言补丁+bind mount)
- migrate_sqlite_to_mysql.py: 逐表清空重导(幂等),复用repos强转,JSON列 解析,67万消息按msg_id键集分批,467s跑完14表全一致 - 修4个MySQL实测bug: NULLS LAST语法(compat.py编译补丁)、message.text TEXT→MEDIUMTEXT(64KB字节上限,实测20.8万字符车源清单)、URL列 String(1024)→Text(分享链接超1KB)、垃圾超长member_id源头清理 - compose: media改bind mount(不占C盘Docker虚拟磁盘,实测2.5GB卷把 C盘撑爆引擎500);mysql不映射宿主机3306 - 前端Dockerfile: npm ci→install(本地/容器npm版本锁漂移) - 五服务实测:nginx→backend→MySQL全链路返回真实数据(客户6859/A**杰25条 /搜索56条),媒体49536文件bind mount直读 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
+4
-7
@@ -98,8 +98,9 @@ services:
|
||||
API_TOKEN: ${API_TOKEN:-}
|
||||
ENCRYPTION_KEY: ${ENCRYPTION_KEY:-}
|
||||
volumes:
|
||||
# Persist downloaded media blobs across container restarts.
|
||||
- media_data:/data/media
|
||||
# 媒体走 bind mount(不占 Docker 虚拟磁盘——本机实测 C 盘被 2.5GB
|
||||
# volume 撑爆导致引擎 500,2026-08-22);服务器上数据直观可见好备份
|
||||
- ./yunke-server/media:/data/media
|
||||
ports:
|
||||
- "8000:8000"
|
||||
|
||||
@@ -136,10 +137,7 @@ services:
|
||||
FULL_SYNC_HOUR: ${FULL_SYNC_HOUR:-2}
|
||||
MESSAGE_INTERVAL_HOURS: ${MESSAGE_INTERVAL_HOURS:-4}
|
||||
volumes:
|
||||
# Share the media volume with backend so both services see the same
|
||||
# downloaded blobs (read-only would be ideal but the scheduler is the
|
||||
# writer; mounting read-write here is the simplest correct config).
|
||||
- media_data:/data/media
|
||||
- ./yunke-server/media:/data/media
|
||||
|
||||
# ----------------------------------------------------------------------- #
|
||||
# Frontend — built + served by nginx
|
||||
@@ -159,4 +157,3 @@ services:
|
||||
volumes:
|
||||
mysql_data:
|
||||
redis_data:
|
||||
media_data:
|
||||
|
||||
@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import (
|
||||
from sqlalchemy.orm import DeclarativeBase
|
||||
|
||||
from app.config import get_settings
|
||||
from app.db import compat # noqa: F401,I100 — 注册 MySQL NULLS LAST 编译补丁(导入即生效)
|
||||
|
||||
|
||||
class Base(DeclarativeBase):
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
"""MySQL 方言兼容补丁(2026-08-22 部署实测发现)。
|
||||
|
||||
MySQL 不支持 ``NULLS LAST`` / ``NULLS FIRST`` 语法(SQLite/PostgreSQL 有)。
|
||||
本项目所有 ``ORDER BY x DESC NULLS LAST`` 在 MySQL 下语义天然一致——
|
||||
MySQL 的 DESC 排序 NULL 本来就排最后——所以编译时直接剥掉该子句。
|
||||
ASC + NULLS LAST 的组合如果将来出现,需要改为 ``ORDER BY x IS NULL, x``
|
||||
(MySQL 惯用法),当前代码库没有这种用法。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy.ext.compiler import compiles
|
||||
from sqlalchemy.sql.elements import UnaryExpression
|
||||
|
||||
|
||||
@compiles(UnaryExpression, "mysql")
|
||||
def _mysql_strip_nullslast(element, compiler, **kw):
|
||||
"""MySQL: 剥掉 NULLS LAST/NULLS FIRST。
|
||||
|
||||
``sa.nullslast(x.desc())`` 编译为 ``UnaryExpression(modifier=NullsLast)``;
|
||||
MySQL DESC 下 NULL 天然在末尾(语义等价),直接输出内层表达式。
|
||||
非 NULLS 修饰的 UnaryExpression(如普通 -x)不受影响——modifier 为 None
|
||||
时走原编译路径。
|
||||
"""
|
||||
from sqlalchemy.sql.operators import nulls_last_op
|
||||
|
||||
if element.modifier is not nulls_last_op:
|
||||
return compiler.visit_unary(element, **kw)
|
||||
return compiler.process(element.element, **kw)
|
||||
@@ -45,6 +45,7 @@ import datetime as dt
|
||||
from typing import Any
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects.mysql import MEDIUMTEXT # noqa: E402
|
||||
from sqlalchemy import BigInteger, Boolean, DateTime, Index, Integer, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
from sqlalchemy.types import JSON
|
||||
@@ -358,10 +359,14 @@ class Message(Base):
|
||||
type_raw: Mapped[int | None] = mapped_column(
|
||||
Integer, nullable=True
|
||||
) # original upstream type code (e.g. WeChat 1=text, -2=divider)
|
||||
text: Mapped[str | None] = mapped_column(Text, nullable=True) # content/text
|
||||
file_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
file_th_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
file_hd_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
# MEDIUMTEXT(16MB):MySQL TEXT 上限 64KB 字节,实测车源清单消息最长
|
||||
# 20.8万字符(≈800KB utf8mb4)超限(2026-08-21 迁移踩坑)
|
||||
text: Mapped[str | None] = mapped_column(
|
||||
Text().with_variant(MEDIUMTEXT(), "mysql"), nullable=True
|
||||
) # content/text
|
||||
file_url: Mapped[str | None] = mapped_column(sa.Text, nullable=True) # 分享链接可超1KB(实测,2026-08-21)
|
||||
file_th_url: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
|
||||
file_hd_url: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
|
||||
md5: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
md5_th: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
md5_hd: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
@@ -428,7 +433,7 @@ class Media(Base):
|
||||
media_type: Mapped[str | None] = mapped_column(
|
||||
String(32), nullable=True
|
||||
) # image|voice|video|file|...
|
||||
original_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
original_url: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
|
||||
original_th_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
original_hd_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
local_path: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
#!/usr/bin/env python
|
||||
"""SQLite → MySQL 数据迁移(本地 smoke.db → 服务器 MySQL,部署方式B)。
|
||||
|
||||
在目标环境(MySQL 可达)里跑:
|
||||
python scripts/migrate_sqlite_to_mysql.py --sqlite ./smoke.db \
|
||||
--mysql "mysql+asyncmy://yunke:pwd@mysql:3306/yunke"
|
||||
|
||||
* 逐表 DELETE 清空后全量导入(幂等,可重跑)
|
||||
* 复用 repos 的 _build_column_values 强转(bool/int/日期字符串→ms,2026-08-21
|
||||
的 3 个 MySQL 方言修复同样生效)
|
||||
* JSON 列(raw_json 等)SQLite 里是文本,插入前 json.loads 回对象
|
||||
* message(67万行)按 msg_id 键集分批,其余表全量读(最大 media 4.9万行)
|
||||
* 跑完打印每表行数对照
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import sqlite3
|
||||
import time
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
log = logging.getLogger("migrate")
|
||||
|
||||
BATCH = 500
|
||||
|
||||
# 迁移顺序(先组织后业务)
|
||||
TABLES = [
|
||||
"department",
|
||||
"staff",
|
||||
"wechat_account",
|
||||
"whatsapp_account",
|
||||
"customer",
|
||||
"chat_group",
|
||||
"group_member",
|
||||
"message",
|
||||
"phone_call",
|
||||
"sms",
|
||||
"voice_clip",
|
||||
"media",
|
||||
"sync_run",
|
||||
"sync_state",
|
||||
]
|
||||
|
||||
|
||||
def _models_by_table() -> dict[str, type]:
|
||||
from app.db import models
|
||||
|
||||
return {
|
||||
obj.__tablename__: obj
|
||||
for obj in vars(models).values()
|
||||
if isinstance(obj, type) and hasattr(obj, "__tablename__")
|
||||
}
|
||||
|
||||
|
||||
def _json_cols(model: type) -> set[str]:
|
||||
return {c.name for c in model.__table__.columns if isinstance(c.type, sa.JSON)}
|
||||
|
||||
|
||||
def _coerce_row(model: type, row: dict, json_cols: set[str]) -> dict:
|
||||
"""SQLite 行 → MySQL 可绑定的值(复用 repos 强转 + JSON 文本解析)。"""
|
||||
from app.storage.repos import _build_column_values
|
||||
|
||||
out = _build_column_values(model, row)
|
||||
for c in json_cols:
|
||||
v = out.get(c)
|
||||
if isinstance(v, str):
|
||||
try:
|
||||
out[c] = json.loads(v)
|
||||
except (ValueError, TypeError):
|
||||
out[c] = None
|
||||
return out
|
||||
|
||||
|
||||
async def migrate_table(conn, model: type, sqlite_con, table: str) -> tuple[int, int]:
|
||||
"""清空 MySQL 表并从 SQLite 全量导入。返回 (sqlite_rows, mysql_rows)。"""
|
||||
from app.storage.repos import _build_column_values # noqa: F401 (coerce 内用)
|
||||
|
||||
pk_cols = [c.name for c in model.__table__.primary_key.columns]
|
||||
json_cols = _json_cols(model)
|
||||
|
||||
n_src = sqlite_con.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
|
||||
await conn.execute(sa.text(f"DELETE FROM {table}"))
|
||||
|
||||
inserted = 0
|
||||
if table == "message":
|
||||
# 67 万行:按第一主键(msg_id)键集分批,避免 OFFSET 全扫
|
||||
keyset_col = pk_cols[0]
|
||||
last_key = None
|
||||
while True:
|
||||
if last_key is None:
|
||||
rows = sqlite_con.execute(
|
||||
f"SELECT * FROM {table} ORDER BY {keyset_col} LIMIT {BATCH}"
|
||||
).fetchall()
|
||||
else:
|
||||
rows = sqlite_con.execute(
|
||||
f"SELECT * FROM {table} WHERE {keyset_col} > ? ORDER BY {keyset_col} LIMIT {BATCH}",
|
||||
(last_key,),
|
||||
).fetchall()
|
||||
if not rows:
|
||||
break
|
||||
batch = [_coerce_row(model, dict(r), json_cols) for r in rows]
|
||||
if batch:
|
||||
await conn.execute(sa.insert(model), batch)
|
||||
inserted += len(batch)
|
||||
last_key = rows[-1][keyset_col]
|
||||
if inserted % 50_000 < BATCH:
|
||||
log.info(f" {table}: {inserted}/{n_src}")
|
||||
else:
|
||||
rows = sqlite_con.execute(f"SELECT * FROM {table}").fetchall()
|
||||
for i in range(0, len(rows), BATCH):
|
||||
batch = [_coerce_row(model, dict(r), json_cols) for r in rows[i:i + BATCH]]
|
||||
if batch:
|
||||
await conn.execute(sa.insert(model), batch)
|
||||
inserted += len(batch)
|
||||
|
||||
return n_src, inserted
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="SQLite → MySQL 迁移")
|
||||
parser.add_argument("--sqlite", required=True, help="smoke.db 路径")
|
||||
parser.add_argument("--mysql", required=True, help="目标 MySQL async URL")
|
||||
args = parser.parse_args()
|
||||
|
||||
t0 = time.time()
|
||||
sqlite_con = sqlite3.connect(args.sqlite)
|
||||
sqlite_con.row_factory = sqlite3.Row
|
||||
models_by_table = _models_by_table()
|
||||
|
||||
engine = create_async_engine(args.mysql, pool_pre_ping=True, pool_recycle=3600)
|
||||
results: list[tuple[str, int, int]] = []
|
||||
async with engine.begin() as conn:
|
||||
for table in TABLES:
|
||||
model = models_by_table.get(table)
|
||||
if model is None:
|
||||
log.warning(f"跳过(无模型): {table}")
|
||||
continue
|
||||
t1 = time.time()
|
||||
n_src, n_ins = await migrate_table(conn, model, sqlite_con, table)
|
||||
results.append((table, n_src, n_ins))
|
||||
log.info(f"{table}: {n_src} → {n_ins} ({time.time()-t1:.1f}s)")
|
||||
await engine.dispose()
|
||||
sqlite_con.close()
|
||||
|
||||
log.info("=" * 50)
|
||||
ok = True
|
||||
for table, n_src, n_ins in results:
|
||||
mark = "✓" if n_src == n_ins else "✗"
|
||||
if n_src != n_ins:
|
||||
ok = False
|
||||
log.info(f"{mark} {table}: {n_src:,} → {n_ins:,}")
|
||||
log.info(f"总耗时 {time.time()-t0:.0f}s — {'全部一致 ✓' if ok else '存在差异 ✗'}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
+1
-1
@@ -9,7 +9,7 @@ WORKDIR /app
|
||||
# Install deps first (cache-friendly: only re-installs when package*.json
|
||||
# changes; node_modules is never copied from the host via .dockerignore).
|
||||
COPY package.json package-lock.json* ./
|
||||
RUN npm ci --no-audit --no-fund
|
||||
RUN npm install --no-audit --no-fund # ci 对锁同步严格,本地/容器 npm 版本差异致 EUSAGE
|
||||
|
||||
# Copy sources and build the production bundle into ./dist.
|
||||
COPY tsconfig*.json vite.config.ts index.html ./
|
||||
|
||||
Reference in New Issue
Block a user