diff --git a/docker-compose.yml b/docker-compose.yml index 161343e..0ed72ed 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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: diff --git a/yunke-server/app/db/base.py b/yunke-server/app/db/base.py index 45a15ee..da7a93d 100644 --- a/yunke-server/app/db/base.py +++ b/yunke-server/app/db/base.py @@ -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): diff --git a/yunke-server/app/db/compat.py b/yunke-server/app/db/compat.py new file mode 100644 index 0000000..cf5a188 --- /dev/null +++ b/yunke-server/app/db/compat.py @@ -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) diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index e28001d..0b4bee7 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -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) diff --git a/yunke-server/scripts/migrate_sqlite_to_mysql.py b/yunke-server/scripts/migrate_sqlite_to_mysql.py new file mode 100644 index 0000000..6b30332 --- /dev/null +++ b/yunke-server/scripts/migrate_sqlite_to_mysql.py @@ -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()) diff --git a/yunkes/Dockerfile b/yunkes/Dockerfile index bb08a45..5c5542d 100644 --- a/yunkes/Dockerfile +++ b/yunkes/Dockerfile @@ -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 ./