marco
9d93b70b9b
代提工作区既有的他人在制品(非本次 CPS);CPS 迁移链 cps_tables 依赖其 b3f1a2c4d5e6 迁移,需一并提交。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
"""user.username(对外展示账号 ID)+ 默认昵称,并回填存量用户
|
|
|
|
Revision ID: b3f1a2c4d5e6
|
|
Revises: 45047b5a884c
|
|
Create Date: 2026-06-16 12:00:00.000000
|
|
|
|
加 user.username(11 位纯数字、首位非 1 与手机号天然区分、全局唯一)。存量用户:
|
|
生成唯一 username,昵称为空的补 9 位字母数字随机昵称。
|
|
迁移自包含(不 import app 业务代码,逻辑冻结为当时快照)。
|
|
"""
|
|
import secrets
|
|
import string
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision: str = 'b3f1a2c4d5e6'
|
|
down_revision: Union[str, Sequence[str], None] = '45047b5a884c'
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
_USERNAME_FIRST = "23456789" # 首位避前导 0、避 1(手机号都以 1 开头)
|
|
_USERNAME_DIGITS = "0123456789"
|
|
_NICKNAME_ALPHABET = string.ascii_letters + string.digits
|
|
|
|
|
|
def _gen_username() -> str:
|
|
return secrets.choice(_USERNAME_FIRST) + "".join(
|
|
secrets.choice(_USERNAME_DIGITS) for _ in range(10)
|
|
)
|
|
|
|
|
|
def _gen_nickname() -> str:
|
|
return "".join(secrets.choice(_NICKNAME_ALPHABET) for _ in range(9))
|
|
|
|
|
|
def upgrade() -> None:
|
|
# 1. 先加可空列(存量行此刻还没值)
|
|
with op.batch_alter_table('user', schema=None) as batch_op:
|
|
batch_op.add_column(sa.Column('username', sa.String(length=11), nullable=True))
|
|
|
|
# 2. 回填存量:每人一个唯一 username;昵称为空的补随机昵称。
|
|
# 表里 username 原本全空,本批用 used 防互撞即可(无既有非空值需规避)。
|
|
# "user" 是 PostgreSQL 保留字,必须加双引号(SQLite 也接受双引号标识符)。
|
|
bind = op.get_bind()
|
|
rows = bind.execute(sa.text('SELECT id, nickname FROM "user"')).fetchall()
|
|
used = set()
|
|
for row in rows:
|
|
uid, nick = row[0], row[1]
|
|
while True:
|
|
uname = _gen_username()
|
|
if uname not in used:
|
|
used.add(uname)
|
|
break
|
|
if nick is None or not str(nick).strip():
|
|
bind.execute(
|
|
sa.text('UPDATE "user" SET username = :u, nickname = :n WHERE id = :i'),
|
|
{"u": uname, "n": _gen_nickname(), "i": uid},
|
|
)
|
|
else:
|
|
bind.execute(
|
|
sa.text('UPDATE "user" SET username = :u WHERE id = :i'),
|
|
{"u": uname, "i": uid},
|
|
)
|
|
|
|
# 3. 收紧:NOT NULL + 唯一索引(对齐模型 unique=True, index=True)
|
|
with op.batch_alter_table('user', schema=None) as batch_op:
|
|
batch_op.alter_column('username', existing_type=sa.String(length=11), nullable=False)
|
|
batch_op.create_index('ix_user_username', ['username'], unique=True)
|
|
|
|
|
|
def downgrade() -> None:
|
|
with op.batch_alter_table('user', schema=None) as batch_op:
|
|
batch_op.drop_index('ix_user_username')
|
|
batch_op.drop_column('username')
|