From 9cf1c9ab11a5a3be09bbcc0dfe1a5c83a98af4fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Sat, 1 Aug 2026 11:15:34 +0800 Subject: [PATCH] feat: core ORM models + initial migration Implements the 14 core ORM models for the yunke data-mirror system using SQLAlchemy 2.0 Mapped/mapped_column style, driven by the field lists in docs/superpowers/specs/2026-08-01-yunke-data-dictionary.md: Department, Staff, WechatAccount, WhatsappAccount, Customer, ChatGroup, GroupMember, Message, Media, PhoneCall, Sms, SyncState, YkAccount, SyncRun. Conventions: - Natural source PKs where they exist (friend_id, wx_id, msg_id, room_id, media_hash, call_id, sms_id, dept_id, user_id); surrogate Integer PKs otherwise (SyncState, YkAccount, SyncRun). GroupMember uses a composite (channel, room_id, member_id) PK. - Common audit columns version/raw_json/fetched_at/updated_at on every source-mirrored table. - channel only where meaningful: Customer/Message/Media/ChatGroup/ GroupMember/Sms use messaging channels; PhoneCall.channel is call-type; SyncState.channel scopes the cursor. Omitted on Staff/Department/ WechatAccount/WhatsappAccount/YkAccount/SyncRun. - Message composite index on (channel, owner_account_id, talker_id, timestamp) for the hot per-talker time-order query path. - message_table_name(channel, timestamp) helper documents the future message__ sharding convention; the Message model still maps to a single physical table for now (sharding is a later task). All types are portable generic SQLAlchemy types so tests run on in-memory SQLite while prod targets MySQL. Tests verify Customer round-trip, raw_json JSON round-trip, Message natural PK, GroupMember composite PK, the Message composite index, and full table creation on SQLite. Alembic wired via migrations/env.py to Base.metadata + get_settings().db_url with an async->sync driver swap (mysql+asyncmy -> mysql+pymysql, sqlite+aiosqlite -> sqlite). Initial migration autogenerated against an empty SQLite DB; server defaults normalized to sa.func.now() so the file is MySQL-portable. upgrade head is intentionally not run (no MySQL yet). Co-Authored-By: Claude --- .gitignore | 5 + yunke-server/alembic.ini | 149 +++++ yunke-server/app/db/models.py | 613 ++++++++++++++++++ yunke-server/migrations/README | 1 + yunke-server/migrations/env.py | 91 +++ yunke-server/migrations/script.py.mako | 28 + .../versions/d7a9bc760c6b_initial_schema.py | 331 ++++++++++ yunke-server/pyproject.toml | 4 + yunke-server/tests/test_models.py | 268 ++++++++ 9 files changed, 1490 insertions(+) create mode 100644 yunke-server/alembic.ini create mode 100644 yunke-server/app/db/models.py create mode 100644 yunke-server/migrations/README create mode 100644 yunke-server/migrations/env.py create mode 100644 yunke-server/migrations/script.py.mako create mode 100644 yunke-server/migrations/versions/d7a9bc760c6b_initial_schema.py create mode 100644 yunke-server/tests/test_models.py diff --git a/.gitignore b/.gitignore index 40ffa68..3fe08d0 100644 --- a/.gitignore +++ b/.gitignore @@ -19,3 +19,8 @@ dist/ # Local media storage media/ + +# Local scratch DBs used for Alembic autogenerate (see migrations/env.py) +*.db +*.sqlite +*.sqlite3 diff --git a/yunke-server/alembic.ini b/yunke-server/alembic.ini new file mode 100644 index 0000000..08f1d9f --- /dev/null +++ b/yunke-server/alembic.ini @@ -0,0 +1,149 @@ +# A generic, single database configuration. + +[alembic] +# path to migration scripts. +# this is typically a path given in POSIX (e.g. forward slashes) +# format, relative to the token %(here)s which refers to the location of this +# ini file +script_location = %(here)s/migrations + +# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s +# Uncomment the line below if you want the files to be prepended with date and time +# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file +# for all available tokens +# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s +# Or organize into date-based subdirectories (requires recursive_version_locations = true) +# file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s + +# sys.path path, will be prepended to sys.path if present. +# defaults to the current working directory. for multiple paths, the path separator +# is defined by "path_separator" below. +prepend_sys_path = . + + +# timezone to use when rendering the date within the migration file +# as well as the filename. +# If specified, requires the tzdata library which can be installed by adding +# `alembic[tz]` to the pip requirements. +# string value is passed to ZoneInfo() +# leave blank for localtime +# timezone = + +# max length of characters to apply to the "slug" field +# truncate_slug_length = 40 + +# set to 'true' to run the environment during +# the 'revision' command, regardless of autogenerate +# revision_environment = false + +# set to 'true' to allow .pyc and .pyo files without +# a source .py file to be detected as revisions in the +# versions/ directory +# sourceless = false + +# version location specification; This defaults +# to /versions. When using multiple version +# directories, initial revisions must be specified with --version-path. +# The path separator used here should be the separator specified by "path_separator" +# below. +# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions + +# path_separator; This indicates what character is used to split lists of file +# paths, including version_locations and prepend_sys_path within configparser +# files such as alembic.ini. +# The default rendered in new alembic.ini files is "os", which uses os.pathsep +# to provide os-dependent path splitting. +# +# Note that in order to support legacy alembic.ini files, this default does NOT +# take place if path_separator is not present in alembic.ini. If this +# option is omitted entirely, fallback logic is as follows: +# +# 1. Parsing of the version_locations option falls back to using the legacy +# "version_path_separator" key, which if absent then falls back to the legacy +# behavior of splitting on spaces and/or commas. +# 2. Parsing of the prepend_sys_path option falls back to the legacy +# behavior of splitting on spaces, commas, or colons. +# +# Valid values for path_separator are: +# +# path_separator = : +# path_separator = ; +# path_separator = space +# path_separator = newline +# +# Use os.pathsep. Default configuration used for new projects. +path_separator = os + +# set to 'true' to search source files recursively +# in each "version_locations" directory +# new in Alembic version 1.10 +# recursive_version_locations = false + +# the output encoding used when revision files +# are written from script.py.mako +# output_encoding = utf-8 + +# database URL. This is consumed by the user-maintained env.py script only. +# other means of configuring database URLs may be customized within the env.py +# file. +sqlalchemy.url = driver://user:pass@localhost/dbname + + +[post_write_hooks] +# post_write_hooks defines scripts or Python functions that are run +# on newly generated revision scripts. See the documentation for further +# detail and examples + +# format using "black" - use the console_scripts runner, against the "black" entrypoint +# hooks = black +# black.type = console_scripts +# black.entrypoint = black +# black.options = -l 79 REVISION_SCRIPT_FILENAME + +# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module +# hooks = ruff +# ruff.type = module +# ruff.module = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Alternatively, use the exec runner to execute a binary found on your PATH +# hooks = ruff +# ruff.type = exec +# ruff.executable = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Logging configuration. This is also consumed by the user-maintained +# env.py script only. +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py new file mode 100644 index 0000000..9906312 --- /dev/null +++ b/yunke-server/app/db/models.py @@ -0,0 +1,613 @@ +"""Core ORM models for the yunke data-mirror system. + +Every model maps an entity observed in the yunke CRM API (see +``docs/superpowers/specs/2026-08-01-yunke-data-dictionary.md``) to a SQL table. +Models target **MySQL in production** but keep column types portable so the +suite can exercise them on **in-memory SQLite** (see ``tests/test_models.py``). + +Conventions +----------- +* SQLAlchemy 2.0 ``Mapped``/``mapped_column`` style; all classes extend + ``app.db.base.Base``. +* API field names are mapped to ``snake_case`` columns; the original payload is + preserved verbatim in ``raw_json`` (``sa.JSON``). +* Common audit columns on every source-mirrored table: + - ``version`` — source update stamp (string when the API gives a version + token, BigInteger when it is a ms timestamp). + - ``raw_json`` — the original API record (``sa.JSON``). + - ``fetched_at`` — when we pulled the row (server default ``now()``). + - ``updated_at`` — when the row was last written (server default + onupdate). +* ``channel`` is only added where it is *meaningful*: + - Communication tables (``Customer``, ``Message``, ``Media``, ``ChatGroup``, + ``GroupMember``, ``Sms``) use ``channel`` in {``wechat``, ``whatsapp``, + ``sms``}. + - ``PhoneCall`` uses ``channel`` to mean call-type in {``phone_call``, + ``voice_call``} (calls are not tied to a messaging channel). + - ``SyncState`` uses ``channel`` to scope a sync cursor (together with + ``entity_type``). + - ``Staff``, ``Department``, ``WechatAccount``, ``WhatsappAccount``, + ``YkAccount``, ``SyncRun`` carry no ``channel`` column. +* Primary keys use the natural source id where one exists; otherwise an + integer surrogate PK is used (``SyncState``, ``YkAccount``, ``SyncRun``). +* ``GroupMember`` uses a composite PK ``(channel, room_id, member_id)``. + +The ``Message`` model currently maps to a single physical ``message`` table. +Real-world volume is expected at millions of rows/month, so the long-term plan +is time-sharded physical tables named via :func:`message_table_name`. That +sharding is a LATER task — for now every message row lives in ``message`` and +the helper below is provided so call-sites can already agree on the naming +convention. +""" + +from __future__ import annotations + +import datetime as dt +from typing import Any + +import sqlalchemy as sa +from sqlalchemy import BigInteger, Boolean, DateTime, Index, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.types import JSON + +from app.db.base import Base + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def message_table_name(channel: str, timestamp: int | dt.datetime) -> str: + """Return the conventional sharded table name for a message. + + Format: ``message__`` where ``YYYYMM`` is derived from the + message timestamp. ``timestamp`` may be either: + + * a Unix epoch in **milliseconds** (``int``) — the form used by the + ``Message.timestamp`` column and the yunke API, or + * a ``datetime.datetime`` instance (any tz). + + Examples + -------- + >>> import datetime as dt + >>> message_table_name("wechat", 1_700_000_000_000) + 'message_wechat_202311' + >>> message_table_name("whatsapp", dt.datetime(2024, 1, 5)) + 'message_whatsapp_202401' + """ + if isinstance(timestamp, dt.datetime): + ts = timestamp + else: + # Treat int as ms-since-epoch (UTC). ``float`` also tolerated. + ts = dt.datetime.fromtimestamp(int(timestamp) / 1000, tz=dt.UTC) + return f"message_{channel}_{ts.strftime('%Y%m')}" + + +# --------------------------------------------------------------------------- +# Shared column defaults +# --------------------------------------------------------------------------- + + +def _fetched_at() -> Mapped[dt.datetime]: + return mapped_column( + DateTime, server_default=sa.func.now(), nullable=False + ) + + +def _updated_at() -> Mapped[dt.datetime]: + return mapped_column( + DateTime, + server_default=sa.func.now(), + onupdate=sa.func.now(), + nullable=False, + ) + + +# =========================================================================== +# 1. Organization & people +# =========================================================================== + + +class Department(Base): + """云客 部门/分组 — source: ``POST /usercenter/dept/tree``.""" + + __tablename__ = "department" + + dept_id: Mapped[str] = mapped_column(String(64), primary_key=True) + parent_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + name: Mapped[str | None] = mapped_column(String(255), nullable=True) + path: Mapped[str | None] = mapped_column(String(512), nullable=True) + + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + +class Staff(Base): + """员工 — embedded in account lists (``userId`` + ``userName``).""" + + __tablename__ = "staff" + + user_id: Mapped[str] = mapped_column(String(64), primary_key=True) + name: Mapped[str | None] = mapped_column(String(255), nullable=True) + phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + sale_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + dept_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + is_on_job: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + +# =========================================================================== +# 2. Staff messaging accounts (channel identity, so no `channel` column) +# =========================================================================== + + +class WechatAccount(Base): + """员工微信号 — source: ``POST /wechatweb/wxUser/list`` (wechatId PK).""" + + __tablename__ = "wechat_account" + + wx_id: Mapped[str] = mapped_column(String(128), primary_key=True) + nick_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + alias: Mapped[str | None] = mapped_column(String(255), nullable=True) + phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + user_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + user_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + contacts_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + group_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + send_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + receive_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + is_on_job: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + db_size: Mapped[int | None] = mapped_column(Integer, nullable=True) + online_status: Mapped[str | None] = mapped_column(String(32), nullable=True) + last_update: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + + version: Mapped[str | None] = mapped_column(String(64), nullable=True) + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + +class WhatsappAccount(Base): + """员工 WhatsApp 账号 — source: ``POST /qqweb/WhatsAppUser/businessSalesList``. + + PK is ``uid`` of the form ``@s.whatsapp.net``. + """ + + __tablename__ = "whatsapp_account" + + uid: Mapped[str] = mapped_column(String(128), primary_key=True) + user_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + user_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + nick_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + sale_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + sales_bind_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + contacts_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + group_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + talker_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + friend_add_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + is_on_job: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + last_update: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + + version: Mapped[str | None] = mapped_column(String(64), nullable=True) + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + +# =========================================================================== +# 3. Communication entities (carry a meaningful `channel` column) +# =========================================================================== + + +class Customer(Base): + """客户/好友 (WeChat friend OR WhatsApp contact OR SMS peer). + + Source: ``POST /wechatweb/wxFriends/friend_list/es`` (WeChat) or + ``POST /qqweb/WhatsAppUser/businessFriendList`` (WhatsApp). PK is the + source friend id: ``friendWxId`` for WeChat, ``@s.whatsapp.net`` + for WhatsApp. + """ + + __tablename__ = "customer" + + friend_id: Mapped[str] = mapped_column(String(128), primary_key=True) + channel: Mapped[str] = mapped_column(String(16), nullable=False) + owner_account_id: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) # wxId (WeChat) / uid (WhatsApp) — owning staff account + friend_nickname: Mapped[str | None] = mapped_column(String(255), nullable=True) + friend_remark: Mapped[str | None] = mapped_column(String(255), nullable=True) + real_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + wechat_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + gender: Mapped[int | None] = mapped_column(Integer, nullable=True) + region: Mapped[str | None] = mapped_column(String(255), nullable=True) + description: Mapped[str | None] = mapped_column(Text, nullable=True) + sale_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + user_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + dept_route: Mapped[str | None] = mapped_column(String(255), nullable=True) + dept_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + head_image_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) + last_chat_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + send_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + receive_num: Mapped[int | None] = mapped_column(Integer, nullable=True) + message_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + delete_flag: Mapped[int | None] = mapped_column(Integer, nullable=True) + is_block: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + friend_state: Mapped[int | None] = mapped_column(Integer, nullable=True) + + version: Mapped[str | None] = mapped_column(String(64), nullable=True) + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = ( + Index( + "ix_customer_owner_friend", + "channel", + "owner_account_id", + "friend_id", + ), + ) + + +class ChatGroup(Base): + """聊天群 — source: ``POST /wechatweb/chatroom/getChatroomList``. + + PK ``room_id`` is the source ``chatRoomId`` (ends with ``@chatroom``). + The raw ``memberIds`` (semicolon-joined wxids) is kept verbatim in + ``member_ids``; the parsed members are mirrored into :class:`GroupMember`. + """ + + __tablename__ = "chat_group" + + room_id: Mapped[str] = mapped_column(String(128), primary_key=True) + channel: Mapped[str] = mapped_column(String(16), nullable=False) + room_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + owner_account_id: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) # wechatId / uid of owning staff account + owner_account_nickname: Mapped[str | None] = mapped_column( + String(255), nullable=True + ) + member_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + member_ids: Mapped[str | None] = mapped_column(Text, nullable=True) + head_image_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) + owner_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + owner_user_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + message_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + own_message_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + loss_members: Mapped[int | None] = mapped_column(Integer, nullable=True) + add_members: Mapped[int | None] = mapped_column(Integer, nullable=True) + last_chat_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + update_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + + version: Mapped[str | None] = mapped_column(String(64), nullable=True) + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = ( + Index( + "ix_chat_group_owner", + "channel", + "owner_account_id", + "room_id", + ), + ) + + +class GroupMember(Base): + """群成员 — composite PK (channel, room_id, member_id). + + Source: the ``memberIds`` field of a chatroom payload, parsed into one row + per member. Per-member nickname/avatar may require a separate API call + (see data-dictionary §9.6) and are therefore nullable. + """ + + __tablename__ = "group_member" + + channel: Mapped[str] = mapped_column(String(16), primary_key=True) + room_id: Mapped[str] = mapped_column(String(128), primary_key=True) + member_id: Mapped[str] = mapped_column(String(128), primary_key=True) + nickname: Mapped[str | None] = mapped_column(String(255), nullable=True) + avatar: Mapped[str | None] = mapped_column(String(1024), nullable=True) + display_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + +class Message(Base): + """聊天消息 — WeChat ``msgSvrId`` / WhatsApp ``keyId`` as natural PK. + + Sources: + * WeChat: ``POST /wechatweb/msg/chats`` + * WhatsApp: ``POST /qqweb/WhatsAppUser/chatRecords`` + + The model currently maps to a **single** ``message`` table. The real + sharding across ``message__`` physical tables (see + :func:`message_table_name`) is a LATER task; this single-table mapping is + sufficient for ingestion v1. + """ + + __tablename__ = "message" + + msg_id: Mapped[str] = mapped_column(String(64), primary_key=True) + channel: Mapped[str] = mapped_column(String(16), nullable=False) + owner_account_id: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) # wechatId / uid of sender staff account + talker_id: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) # talker / friendUid — the other party + ori_talker: Mapped[str | None] = mapped_column(String(128), nullable=True) + room_id: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) # @chatroom id; NULL/empty = direct message + is_send: Mapped[bool | None] = mapped_column(Boolean, nullable=True) # mine/isSend + type: Mapped[int | None] = mapped_column(Integer, nullable=True) + 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) + 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) + duration: Mapped[int | None] = mapped_column(Integer, nullable=True) + transcription_text: Mapped[str | None] = mapped_column(Text, nullable=True) + img_path: Mapped[str | None] = mapped_column(String(1024), nullable=True) + original_image: Mapped[str | None] = mapped_column(String(1024), nullable=True) + pic_type: Mapped[int | None] = mapped_column(Integer, nullable=True) + pay_msg_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + refer_msg: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + sender_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + sender_head_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) + timestamp: Mapped[int | None] = mapped_column(BigInteger, nullable=True) # ms epoch + is_del: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + + version: Mapped[str | None] = mapped_column(String(64), nullable=True) + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = ( + # Hot query path: per-account, per-talker, time-ordered pagination. + Index( + "ix_message_channel_owner_talker_time", + "channel", + "owner_account_id", + "talker_id", + "timestamp", + ), + ) + + +class Media(Base): + """Locally cached media (image / voice / video / file) — keyed by hash. + + Deduplicates downloads across messages that reference the same file (via + ``md5``). ``local_path`` is set once the blob has been fetched into the + local media store. + """ + + __tablename__ = "media" + + media_hash: Mapped[str] = mapped_column(String(64), primary_key=True) + channel: Mapped[str] = mapped_column(String(16), nullable=False) + 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_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) + size_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + mime_type: Mapped[str | None] = mapped_column(String(128), nullable=True) + duration: Mapped[int | None] = mapped_column(Integer, nullable=True) + + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = ( + Index("ix_media_type_channel", "media_type", "channel"), + ) + + +# =========================================================================== +# 4. Phone calls & SMS (prefix /pc/) +# =========================================================================== + + +class PhoneCall(Base): + """电话/语音通话记录 — source: ``POST /pc/callLog/list`` (phone calls) and + ``POST /wechatweb/callRecord/queryWeChatVoiceList`` (voice calls). + + ``channel`` here means **call-type** in {``phone_call``, ``voice_call``} + (calls are not tied to a messaging channel). PK is the source ``id``. + """ + + __tablename__ = "phone_call" + + call_id: Mapped[str] = mapped_column(String(64), primary_key=True) + channel: Mapped[str] = mapped_column(String(16), nullable=False) + start_call_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + call_duration: Mapped[int | None] = mapped_column(Integer, nullable=True) + ring_time: Mapped[int | None] = mapped_column(Integer, nullable=True) + call_status: Mapped[int | None] = mapped_column(Integer, nullable=True) + incoming_call: Mapped[int | None] = mapped_column(Integer, nullable=True) + user_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + user_id_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + department_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + call_number: Mapped[str | None] = mapped_column(String(64), nullable=True) + sim_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + sim_number: Mapped[str | None] = mapped_column(String(64), nullable=True) + imei: Mapped[str | None] = mapped_column(String(128), nullable=True) + record_file: Mapped[str | None] = mapped_column(String(1024), nullable=True) + is_upload: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + download_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + ai_asr: Mapped[str | None] = mapped_column(Text, nullable=True) + qc_status: Mapped[int | None] = mapped_column(Integer, nullable=True) + plan_customer_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + + version: Mapped[str | None] = mapped_column(String(64), nullable=True) + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = ( + Index("ix_phone_call_user_time", "user_id", "start_call_time"), + ) + + +class Sms(Base): + """短信记录 — source: ``POST /pc/customer/smsLogsList``.""" + + __tablename__ = "sms" + + sms_id: Mapped[str] = mapped_column(String(64), primary_key=True) + channel: Mapped[str] = mapped_column( + String(16), nullable=False, default="sms" + ) # always 'sms' today; kept for symmetry + customer_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + content: Mapped[str | None] = mapped_column(Text, nullable=True) # full sms text + content_preview: Mapped[str | None] = mapped_column(String(255), nullable=True) + sender: Mapped[str | None] = mapped_column(String(128), nullable=True) + direction: Mapped[int | None] = mapped_column(Integer, nullable=True) + user_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + sender_department: Mapped[str | None] = mapped_column(String(255), nullable=True) + sender_department_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + receive_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + create_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + sim_phone: Mapped[str | None] = mapped_column(String(64), nullable=True) + sim_number: Mapped[str | None] = mapped_column(String(64), nullable=True) + source: Mapped[str | None] = mapped_column(String(64), nullable=True) + + version: Mapped[str | None] = mapped_column(String(64), nullable=True) + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = (Index("ix_sms_user_time", "user_id", "create_time"),) + + +# =========================================================================== +# 5. Operational / bookkeeping tables (local; no raw_json/version) +# =========================================================================== + + +class SyncState(Base): + """Resumable sync cursor — scoped by (entity_type, channel, scope_id). + + ``scope_id`` qualifies the cursor for a sub-stream (e.g. the owning staff + account id when syncing per-account message streams). ``last_version`` + holds the opaque source update stamp; ``last_timestamp`` the ms epoch. + """ + + __tablename__ = "sync_state" + + state_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + entity_type: Mapped[str] = mapped_column(String(32), nullable=False) + channel: Mapped[str | None] = mapped_column(String(16), nullable=True) + scope_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + last_version: Mapped[str | None] = mapped_column(String(64), nullable=True) + last_timestamp: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + last_synced_at: Mapped[dt.datetime | None] = mapped_column( + DateTime, nullable=True + ) + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = ( + Index( + "ix_sync_state_scope", + "entity_type", + "channel", + "scope_id", + unique=True, + ), + ) + + +class YkAccount(Base): + """云客 CRM login account — credentials / session for the upstream API. + + The session cookie / password is stored encrypted with + ``Settings.encryption_key`` (Fernet). ``user_id`` + ``company`` together + identify a yunke session (the two ``userid`` / ``company`` request headers). + """ + + __tablename__ = "yk_account" + + account_id: Mapped[int] = mapped_column( + Integer, primary_key=True, autoincrement=True + ) + user_id: Mapped[str] = mapped_column(String(64), nullable=False) + company: Mapped[str] = mapped_column(String(64), nullable=False) + username: Mapped[str | None] = mapped_column(String(255), nullable=True) + session_blob_enc: Mapped[str | None] = mapped_column(Text, nullable=True) + last_login_at: Mapped[dt.datetime | None] = mapped_column( + DateTime, nullable=True + ) + is_active: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=True, server_default=sa.text("1") + ) + + raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + fetched_at: Mapped[dt.datetime] = _fetched_at() + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = ( + Index("ix_yk_account_user_company", "user_id", "company", unique=True), + ) + + +class SyncRun(Base): + """Audit log for one sync execution (start / end / status / counts).""" + + __tablename__ = "sync_run" + + run_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + entity_type: Mapped[str] = mapped_column(String(32), nullable=False) + channel: Mapped[str | None] = mapped_column(String(16), nullable=True) + scope_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + status: Mapped[str] = mapped_column(String(16), nullable=False) # running|success|failed + started_at: Mapped[dt.datetime] = mapped_column( + DateTime, server_default=sa.func.now(), nullable=False + ) + finished_at: Mapped[dt.datetime | None] = mapped_column(DateTime, nullable=True) + records_fetched: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + records_upserted: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + + updated_at: Mapped[dt.datetime] = _updated_at() + + __table_args__ = (Index("ix_sync_run_entity_status", "entity_type", "status"),) + + +__all__ = [ + "ChatGroup", + "Customer", + "Department", + "GroupMember", + "Media", + "Message", + "PhoneCall", + "Sms", + "Staff", + "SyncRun", + "SyncState", + "WechatAccount", + "WhatsappAccount", + "YkAccount", + "message_table_name", +] diff --git a/yunke-server/migrations/README b/yunke-server/migrations/README new file mode 100644 index 0000000..98e4f9c --- /dev/null +++ b/yunke-server/migrations/README @@ -0,0 +1 @@ +Generic single-database configuration. \ No newline at end of file diff --git a/yunke-server/migrations/env.py b/yunke-server/migrations/env.py new file mode 100644 index 0000000..64cba06 --- /dev/null +++ b/yunke-server/migrations/env.py @@ -0,0 +1,91 @@ +"""Alembic migration environment. + +Wired to: +* ``app.db.base.Base.metadata`` (the single source of truth for our schema), and +* ``app.config.get_settings().db_url`` (the same URL the application uses). + +Production runs against MySQL via ``mysql+asyncmy://``. Alembic drives a +**synchronous** engine, so :func:`_sync_url` swaps the async driver for its +sync counterpart (``mysql+asyncmy`` -> ``mysql+pymysql`` and +``sqlite+aiosqlite`` -> ``sqlite``). For local autogenerate without a running +MySQL server, point ``DB_URL`` at an empty SQLite file, e.g.:: + + DB_URL="sqlite+aiosqlite:///./_autogen.db" alembic revision --autogenerate -m "..." +""" + +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +from app.config import get_settings +from app.db import models # noqa: F401 (registers all tables on Base.metadata) +from app.db.base import Base + +# this is the Alembic Config object, which provides +# access to the values within the .ini file in use. +config = context.config + +# Interpret the config file for Python logging. +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +# Single source of truth for schema; Alembic compares against this. +target_metadata = Base.metadata + + +def _sync_url(url: str) -> str: + """Convert an async DB URL to its synchronous equivalent. + + Alembic's default :func:`engine_from_config` builds a sync engine, so we + swap async drivers out. Pass-through if the URL is already sync. + """ + if url.startswith("mysql+asyncmy://"): + return url.replace("mysql+asyncmy://", "mysql+pymysql://", 1) + if url.startswith("sqlite+aiosqlite://"): + return url.replace("sqlite+aiosqlite://", "sqlite://", 1) + return url + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode (emit SQL without a live connection).""" + url = _sync_url(get_settings().db_url) + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + compare_type=True, + compare_server_default=True, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + """Run migrations in 'online' mode (open a connection, run, close).""" + config.set_main_option("sqlalchemy.url", _sync_url(get_settings().db_url)) + + connectable = engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + with connectable.connect() as connection: + context.configure( + connection=connection, + target_metadata=target_metadata, + compare_type=True, + compare_server_default=True, + ) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/yunke-server/migrations/script.py.mako b/yunke-server/migrations/script.py.mako new file mode 100644 index 0000000..4b41d0a --- /dev/null +++ b/yunke-server/migrations/script.py.mako @@ -0,0 +1,28 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: str | Sequence[str] | None = ${repr(down_revision)} +branch_labels: str | Sequence[str] | None = ${repr(branch_labels)} +depends_on: str | Sequence[str] | None = ${repr(depends_on)} + + +def upgrade() -> None: + """Upgrade schema.""" + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + """Downgrade schema.""" + ${downgrades if downgrades else "pass"} diff --git a/yunke-server/migrations/versions/d7a9bc760c6b_initial_schema.py b/yunke-server/migrations/versions/d7a9bc760c6b_initial_schema.py new file mode 100644 index 0000000..dca7caf --- /dev/null +++ b/yunke-server/migrations/versions/d7a9bc760c6b_initial_schema.py @@ -0,0 +1,331 @@ +"""initial schema + +Revision ID: d7a9bc760c6b +Revises: +Create Date: 2026-08-01 11:11:18.566666 + +Autogenerated against an empty SQLite database (no MySQL server was available +during development). All column types are generic (``sa.String``, ``sa.Text``, +``sa.JSON``, ``sa.BigInteger``, ``sa.Integer``, ``sa.Boolean``, ``sa.DateTime``) +and the ``server_default=sa.func.now()`` timestamps compile per-dialect, so the +migration is portable to MySQL. When a MySQL instance is provisioned, ``alembic +upgrade head`` should be run there directly (no regeneration needed). +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = 'd7a9bc760c6b' +down_revision: str | Sequence[str] | None = None +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('chat_group', + sa.Column('room_id', sa.String(length=128), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=False), + sa.Column('room_name', sa.String(length=255), nullable=True), + sa.Column('owner_account_id', sa.String(length=128), nullable=True), + sa.Column('owner_account_nickname', sa.String(length=255), nullable=True), + sa.Column('member_count', sa.Integer(), nullable=True), + sa.Column('member_ids', sa.Text(), nullable=True), + sa.Column('head_image_url', sa.String(length=1024), nullable=True), + sa.Column('owner_user_id', sa.String(length=64), nullable=True), + sa.Column('owner_user_phone', sa.String(length=64), nullable=True), + sa.Column('message_count', sa.Integer(), nullable=True), + sa.Column('own_message_count', sa.Integer(), nullable=True), + sa.Column('loss_members', sa.Integer(), nullable=True), + sa.Column('add_members', sa.Integer(), nullable=True), + sa.Column('last_chat_time', sa.BigInteger(), nullable=True), + sa.Column('update_time', sa.BigInteger(), nullable=True), + sa.Column('version', sa.String(length=64), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('room_id') + ) + op.create_index('ix_chat_group_owner', 'chat_group', ['channel', 'owner_account_id', 'room_id'], unique=False) + op.create_table('customer', + sa.Column('friend_id', sa.String(length=128), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=False), + sa.Column('owner_account_id', sa.String(length=128), nullable=True), + sa.Column('friend_nickname', sa.String(length=255), nullable=True), + sa.Column('friend_remark', sa.String(length=255), nullable=True), + sa.Column('real_name', sa.String(length=255), nullable=True), + sa.Column('phone', sa.String(length=64), nullable=True), + sa.Column('wechat_phone', sa.String(length=64), nullable=True), + sa.Column('gender', sa.Integer(), nullable=True), + sa.Column('region', sa.String(length=255), nullable=True), + sa.Column('description', sa.Text(), nullable=True), + sa.Column('sale_name', sa.String(length=255), nullable=True), + sa.Column('user_id', sa.String(length=64), nullable=True), + sa.Column('dept_route', sa.String(length=255), nullable=True), + sa.Column('dept_name', sa.String(length=255), nullable=True), + sa.Column('head_image_url', sa.String(length=1024), nullable=True), + sa.Column('last_chat_time', sa.BigInteger(), nullable=True), + sa.Column('send_num', sa.Integer(), nullable=True), + sa.Column('receive_num', sa.Integer(), nullable=True), + sa.Column('message_count', sa.Integer(), nullable=True), + sa.Column('delete_flag', sa.Integer(), nullable=True), + sa.Column('is_block', sa.Boolean(), nullable=True), + sa.Column('friend_state', sa.Integer(), nullable=True), + sa.Column('version', sa.String(length=64), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('friend_id') + ) + op.create_index('ix_customer_owner_friend', 'customer', ['channel', 'owner_account_id', 'friend_id'], unique=False) + op.create_table('department', + sa.Column('dept_id', sa.String(length=64), nullable=False), + sa.Column('parent_id', sa.String(length=64), nullable=True), + sa.Column('name', sa.String(length=255), nullable=True), + sa.Column('path', sa.String(length=512), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('dept_id') + ) + op.create_table('group_member', + sa.Column('channel', sa.String(length=16), nullable=False), + sa.Column('room_id', sa.String(length=128), nullable=False), + sa.Column('member_id', sa.String(length=128), nullable=False), + sa.Column('nickname', sa.String(length=255), nullable=True), + sa.Column('avatar', sa.String(length=1024), nullable=True), + sa.Column('display_name', sa.String(length=255), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('channel', 'room_id', 'member_id') + ) + op.create_table('media', + sa.Column('media_hash', sa.String(length=64), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=False), + sa.Column('media_type', sa.String(length=32), nullable=True), + sa.Column('original_url', sa.String(length=1024), nullable=True), + sa.Column('original_th_url', sa.String(length=1024), nullable=True), + sa.Column('original_hd_url', sa.String(length=1024), nullable=True), + sa.Column('local_path', sa.String(length=1024), nullable=True), + sa.Column('size_bytes', sa.BigInteger(), nullable=True), + sa.Column('mime_type', sa.String(length=128), nullable=True), + sa.Column('duration', sa.Integer(), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('media_hash') + ) + op.create_index('ix_media_type_channel', 'media', ['media_type', 'channel'], unique=False) + op.create_table('message', + sa.Column('msg_id', sa.String(length=64), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=False), + sa.Column('owner_account_id', sa.String(length=128), nullable=True), + sa.Column('talker_id', sa.String(length=128), nullable=True), + sa.Column('ori_talker', sa.String(length=128), nullable=True), + sa.Column('room_id', sa.String(length=128), nullable=True), + sa.Column('is_send', sa.Boolean(), nullable=True), + sa.Column('type', sa.Integer(), nullable=True), + sa.Column('text', sa.Text(), nullable=True), + sa.Column('file_url', sa.String(length=1024), nullable=True), + sa.Column('file_th_url', sa.String(length=1024), nullable=True), + sa.Column('file_hd_url', sa.String(length=1024), nullable=True), + sa.Column('md5', sa.String(length=64), nullable=True), + sa.Column('md5_th', sa.String(length=64), nullable=True), + sa.Column('md5_hd', sa.String(length=64), nullable=True), + sa.Column('duration', sa.Integer(), nullable=True), + sa.Column('transcription_text', sa.Text(), nullable=True), + sa.Column('img_path', sa.String(length=1024), nullable=True), + sa.Column('original_image', sa.String(length=1024), nullable=True), + sa.Column('pic_type', sa.Integer(), nullable=True), + sa.Column('pay_msg_id', sa.String(length=64), nullable=True), + sa.Column('refer_msg', sa.JSON(), nullable=True), + sa.Column('sender_name', sa.String(length=255), nullable=True), + sa.Column('sender_head_url', sa.String(length=1024), nullable=True), + sa.Column('timestamp', sa.BigInteger(), nullable=True), + sa.Column('is_del', sa.Boolean(), nullable=True), + sa.Column('version', sa.String(length=64), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('msg_id') + ) + op.create_index('ix_message_channel_owner_talker_time', 'message', ['channel', 'owner_account_id', 'talker_id', 'timestamp'], unique=False) + op.create_table('phone_call', + sa.Column('call_id', sa.String(length=64), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=False), + sa.Column('start_call_time', sa.BigInteger(), nullable=True), + sa.Column('call_duration', sa.Integer(), nullable=True), + sa.Column('ring_time', sa.Integer(), nullable=True), + sa.Column('call_status', sa.Integer(), nullable=True), + sa.Column('incoming_call', sa.Integer(), nullable=True), + sa.Column('user_id', sa.String(length=64), nullable=True), + sa.Column('user_id_name', sa.String(length=255), nullable=True), + sa.Column('department_id', sa.String(length=64), nullable=True), + sa.Column('call_number', sa.String(length=64), nullable=True), + sa.Column('sim_phone', sa.String(length=64), nullable=True), + sa.Column('sim_number', sa.String(length=64), nullable=True), + sa.Column('imei', sa.String(length=128), nullable=True), + sa.Column('record_file', sa.String(length=1024), nullable=True), + sa.Column('is_upload', sa.Boolean(), nullable=True), + sa.Column('download_count', sa.Integer(), nullable=True), + sa.Column('ai_asr', sa.Text(), nullable=True), + sa.Column('qc_status', sa.Integer(), nullable=True), + sa.Column('plan_customer_name', sa.String(length=255), nullable=True), + sa.Column('version', sa.String(length=64), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('call_id') + ) + op.create_index('ix_phone_call_user_time', 'phone_call', ['user_id', 'start_call_time'], unique=False) + op.create_table('sms', + sa.Column('sms_id', sa.String(length=64), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=False), + sa.Column('customer_phone', sa.String(length=64), nullable=True), + sa.Column('content', sa.Text(), nullable=True), + sa.Column('content_preview', sa.String(length=255), nullable=True), + sa.Column('sender', sa.String(length=128), nullable=True), + sa.Column('direction', sa.Integer(), nullable=True), + sa.Column('user_id', sa.String(length=64), nullable=True), + sa.Column('sender_department', sa.String(length=255), nullable=True), + sa.Column('sender_department_id', sa.String(length=64), nullable=True), + sa.Column('receive_name', sa.String(length=255), nullable=True), + sa.Column('create_time', sa.BigInteger(), nullable=True), + sa.Column('sim_phone', sa.String(length=64), nullable=True), + sa.Column('sim_number', sa.String(length=64), nullable=True), + sa.Column('source', sa.String(length=64), nullable=True), + sa.Column('version', sa.String(length=64), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('sms_id') + ) + op.create_index('ix_sms_user_time', 'sms', ['user_id', 'create_time'], unique=False) + op.create_table('staff', + sa.Column('user_id', sa.String(length=64), nullable=False), + sa.Column('name', sa.String(length=255), nullable=True), + sa.Column('phone', sa.String(length=64), nullable=True), + sa.Column('sale_phone', sa.String(length=64), nullable=True), + sa.Column('dept_id', sa.String(length=64), nullable=True), + sa.Column('is_on_job', sa.Boolean(), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('user_id') + ) + op.create_table('sync_run', + sa.Column('run_id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('entity_type', sa.String(length=32), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=True), + sa.Column('scope_id', sa.String(length=128), nullable=True), + sa.Column('status', sa.String(length=16), nullable=False), + sa.Column('started_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('finished_at', sa.DateTime(), nullable=True), + sa.Column('records_fetched', sa.Integer(), server_default=sa.text('0'), nullable=False), + sa.Column('records_upserted', sa.Integer(), server_default=sa.text('0'), nullable=False), + sa.Column('error', sa.Text(), nullable=True), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('run_id') + ) + op.create_index('ix_sync_run_entity_status', 'sync_run', ['entity_type', 'status'], unique=False) + op.create_table('sync_state', + sa.Column('state_id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('entity_type', sa.String(length=32), nullable=False), + sa.Column('channel', sa.String(length=16), nullable=True), + sa.Column('scope_id', sa.String(length=128), nullable=True), + sa.Column('last_version', sa.String(length=64), nullable=True), + sa.Column('last_timestamp', sa.BigInteger(), nullable=True), + sa.Column('last_synced_at', sa.DateTime(), nullable=True), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('state_id') + ) + op.create_index('ix_sync_state_scope', 'sync_state', ['entity_type', 'channel', 'scope_id'], unique=True) + op.create_table('wechat_account', + sa.Column('wx_id', sa.String(length=128), nullable=False), + sa.Column('nick_name', sa.String(length=255), nullable=True), + sa.Column('alias', sa.String(length=255), nullable=True), + sa.Column('phone', sa.String(length=64), nullable=True), + sa.Column('user_id', sa.String(length=64), nullable=True), + sa.Column('user_name', sa.String(length=255), nullable=True), + sa.Column('contacts_num', sa.Integer(), nullable=True), + sa.Column('group_num', sa.Integer(), nullable=True), + sa.Column('send_num', sa.Integer(), nullable=True), + sa.Column('receive_num', sa.Integer(), nullable=True), + sa.Column('is_on_job', sa.Boolean(), nullable=True), + sa.Column('db_size', sa.Integer(), nullable=True), + sa.Column('online_status', sa.String(length=32), nullable=True), + sa.Column('last_update', sa.BigInteger(), nullable=True), + sa.Column('version', sa.String(length=64), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('wx_id') + ) + op.create_table('whatsapp_account', + sa.Column('uid', sa.String(length=128), nullable=False), + sa.Column('user_id', sa.String(length=64), nullable=True), + sa.Column('user_name', sa.String(length=255), nullable=True), + sa.Column('nick_name', sa.String(length=255), nullable=True), + sa.Column('phone', sa.String(length=64), nullable=True), + sa.Column('sale_phone', sa.String(length=64), nullable=True), + sa.Column('sales_bind_phone', sa.String(length=64), nullable=True), + sa.Column('contacts_num', sa.Integer(), nullable=True), + sa.Column('group_num', sa.Integer(), nullable=True), + sa.Column('talker_count', sa.Integer(), nullable=True), + sa.Column('friend_add_num', sa.Integer(), nullable=True), + sa.Column('is_on_job', sa.Boolean(), nullable=True), + sa.Column('last_update', sa.BigInteger(), nullable=True), + sa.Column('version', sa.String(length=64), nullable=True), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('uid') + ) + op.create_table('yk_account', + sa.Column('account_id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('user_id', sa.String(length=64), nullable=False), + sa.Column('company', sa.String(length=64), nullable=False), + sa.Column('username', sa.String(length=255), nullable=True), + sa.Column('session_blob_enc', sa.Text(), nullable=True), + sa.Column('last_login_at', sa.DateTime(), nullable=True), + sa.Column('is_active', sa.Boolean(), server_default=sa.text('1'), nullable=False), + sa.Column('raw_json', sa.JSON(), nullable=True), + sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint('account_id') + ) + op.create_index('ix_yk_account_user_company', 'yk_account', ['user_id', 'company'], unique=True) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index('ix_yk_account_user_company', table_name='yk_account') + op.drop_table('yk_account') + op.drop_table('whatsapp_account') + op.drop_table('wechat_account') + op.drop_index('ix_sync_state_scope', table_name='sync_state') + op.drop_table('sync_state') + op.drop_index('ix_sync_run_entity_status', table_name='sync_run') + op.drop_table('sync_run') + op.drop_table('staff') + op.drop_index('ix_sms_user_time', table_name='sms') + op.drop_table('sms') + op.drop_index('ix_phone_call_user_time', table_name='phone_call') + op.drop_table('phone_call') + op.drop_index('ix_message_channel_owner_talker_time', table_name='message') + op.drop_table('message') + op.drop_index('ix_media_type_channel', table_name='media') + op.drop_table('media') + op.drop_table('group_member') + op.drop_table('department') + op.drop_index('ix_customer_owner_friend', table_name='customer') + op.drop_table('customer') + op.drop_index('ix_chat_group_owner', table_name='chat_group') + op.drop_table('chat_group') + # ### end Alembic commands ### diff --git a/yunke-server/pyproject.toml b/yunke-server/pyproject.toml index 13b20f2..e43d166 100644 --- a/yunke-server/pyproject.toml +++ b/yunke-server/pyproject.toml @@ -8,6 +8,10 @@ dependencies = [ "uvicorn[standard]", "sqlalchemy[asyncio]", "asyncmy", + # Pure-Python sync MySQL driver used by Alembic migrations (env.py swaps + # the production ``mysql+asyncmy`` URL to ``mysql+pymysql`` for the sync + # migration engine). + "pymysql", "alembic", "redis", "httpx", diff --git a/yunke-server/tests/test_models.py b/yunke-server/tests/test_models.py new file mode 100644 index 0000000..2b4f8a9 --- /dev/null +++ b/yunke-server/tests/test_models.py @@ -0,0 +1,268 @@ +"""Tests for app.db.models — ORM model definitions round-trip through SQLite. + +These tests build an isolated in-memory SQLite async engine (same pattern as +tests/test_db_base.py) and exercise the real ORM models defined in +app.db.models. They never touch the production MySQL engine. +""" + +import datetime as dt + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db.base import Base +from app.db.models import ( + ChatGroup, + Customer, + Department, + GroupMember, + Media, + Message, + PhoneCall, + Sms, + Staff, + SyncRun, + SyncState, + WechatAccount, + WhatsappAccount, + YkAccount, + message_table_name, +) + +ALL_MODELS = ( + Department, + Staff, + WechatAccount, + WhatsappAccount, + Customer, + ChatGroup, + GroupMember, + Message, + Media, + PhoneCall, + Sms, + SyncState, + YkAccount, + SyncRun, +) + +EXPECTED_TABLES = ( + "department", + "staff", + "wechat_account", + "whatsapp_account", + "customer", + "chat_group", + "group_member", + "message", + "media", + "phone_call", + "sms", + "sync_state", + "yk_account", + "sync_run", +) + + +async def _make_engine(): + """Create an in-memory async SQLite engine with all tables created.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return engine + + +def test_all_models_extend_base(): + """Every model must subclass Base and declare a __tablename__.""" + for model in ALL_MODELS: + assert issubclass(model, Base), f"{model.__name__} must extend Base" + assert isinstance(model.__tablename__, str) and model.__tablename__, ( + f"{model.__name__} must declare __tablename__" + ) + + +def test_table_names_match_expected(): + """The 14 tables must use the documented snake_case names.""" + for model, expected in zip(ALL_MODELS, EXPECTED_TABLES): + assert model.__tablename__ == expected, ( + f"{model.__name__}.__tablename__ = {model.__tablename__!r}, " + f"expected {expected!r}" + ) + + +async def test_all_tables_create_on_sqlite(): + """All 14 tables must be createable on SQLite (types stay portable).""" + engine = await _make_engine() + try: + async with engine.begin() as conn: + tables = await conn.run_sync( + lambda sync_conn: sa.inspect(sync_conn).get_table_names() + ) + for name in EXPECTED_TABLES: + assert name in tables, f"missing table: {name}" + finally: + await engine.dispose() + + +async def test_customer_round_trip(): + """A Customer row round-trips its key fields through SQLite.""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + cust = Customer( + friend_id="wxid_abc123", + channel="wechat", + owner_account_id="wxid_staff", + friend_nickname="Alice", + phone="13800138000", + message_count=42, + ) + session.add(cust) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Customer).where(Customer.friend_id == "wxid_abc123") + ) + ).scalar_one() + assert fetched.friend_id == "wxid_abc123" + assert fetched.channel == "wechat" + assert fetched.owner_account_id == "wxid_staff" + assert fetched.friend_nickname == "Alice" + assert fetched.phone == "13800138000" + assert fetched.message_count == 42 + assert fetched.fetched_at is not None + finally: + await engine.dispose() + + +async def test_customer_raw_json_round_trip(): + """raw_json must round-trip an arbitrary dict (stored as JSON).""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + payload = { + "friendWxId": "wxid_abc", + "list": [1, 2, 3], + "nested": {"k": "v", "flag": True}, + } + cust = Customer( + friend_id="wxid_raw", + channel="wechat", + owner_account_id="wxid_staff", + raw_json=payload, + ) + session.add(cust) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Customer).where(Customer.friend_id == "wxid_raw") + ) + ).scalar_one() + assert fetched.raw_json == payload + finally: + await engine.dispose() + + +async def test_message_natural_pk(): + """Message PK is msg_id (the source msgSvrId / keyId).""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + msg = Message( + msg_id="1000069900000000001", + channel="wechat", + owner_account_id="wxid_staff", + talker_id="wxid_abc", + is_send=False, + type=1, + text="hello", + timestamp=1700000000000, + ) + session.add(msg) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Message).where(Message.msg_id == "1000069900000000001") + ) + ).scalar_one() + assert fetched.msg_id == "1000069900000000001" + assert fetched.text == "hello" + assert fetched.timestamp == 1700000000000 + assert fetched.is_send is False + finally: + await engine.dispose() + + +def test_message_table_name_helper_with_ms_timestamp(): + """message_table_name(channel, ms_epoch) -> message__.""" + ts_ms = int( + dt.datetime(2023, 11, 15, 12, 0, 0, tzinfo=dt.UTC).timestamp() * 1000 + ) + assert message_table_name("wechat", ts_ms) == "message_wechat_202311" + + +def test_message_table_name_helper_with_datetime(): + """message_table_name also accepts a datetime instance.""" + assert ( + message_table_name("whatsapp", dt.datetime(2024, 1, 5, tzinfo=dt.UTC)) + == "message_whatsapp_202401" + ) + + +async def test_message_composite_index_present(): + """The (channel, owner_account_id, talker_id, timestamp) index must exist.""" + engine = await _make_engine() + try: + async with engine.begin() as conn: + indexes = await conn.run_sync( + lambda sync_conn: sa.inspect(sync_conn).get_indexes("message") + ) + names = {idx["name"] for idx in indexes} + # The composite index name is implementation detail; assert one index + # covers exactly those four columns. + cols_to_index = { + "channel", + "owner_account_id", + "talker_id", + "timestamp", + } + assert any( + set(idx["column_names"]) == cols_to_index for idx in indexes + ), f"no composite index on {cols_to_index}; got: {names}" + finally: + await engine.dispose() + + +async def test_group_member_composite_pk(): + """GroupMember uses a composite PK (channel, room_id, member_id).""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + gm = GroupMember( + channel="wechat", + room_id="12345@chatroom", + member_id="wxid_member1", + nickname="Bob", + ) + session.add(gm) + await session.commit() + + fetched = ( + await session.execute( + sa.select(GroupMember).where( + GroupMember.room_id == "12345@chatroom", + GroupMember.member_id == "wxid_member1", + GroupMember.channel == "wechat", + ) + ) + ).scalar_one() + assert fetched.nickname == "Bob" + finally: + await engine.dispose()