小鱼开发
5f5dc2242c
用法:
python -m scripts.create_user --mobile 13800138000 --nickname 测试用户
python -m scripts.create_user --mobile 13800138001 --device-id dev-001
自动创建关联的 user_points 记录，可选创建 user_devices 记录。
112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
创建用户 CLI 工具
|
||
================
|
||
|
||
绕过短信验证码，直接操作数据库创建用户。
|
||
|
||
用法：
|
||
cd python-api
|
||
python -m scripts.create_user --mobile 13800138000 --nickname 测试用户
|
||
python -m scripts.create_user --mobile 13800138001 --nickname 用户2 --device-id dev-001
|
||
|
||
环境变量：
|
||
DATABASE_URL 数据库连接（默认从 .env 读取）
|
||
SECRET_KEY JWT 密钥（默认从 .env 读取）
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
import uuid
|
||
|
||
# 将项目根目录加入路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from sqlalchemy import select
|
||
|
||
from app.db.session import AsyncSessionLocal, close_db
|
||
from app.models.user import User
|
||
from app.models.user_device import UserDevice
|
||
from app.models.user_point import UserPoint
|
||
|
||
|
||
async def create_user(
|
||
mobile: str,
|
||
nickname: str | None = None,
|
||
device_id: str | None = None,
|
||
device_name: str | None = None,
|
||
source: str = "cli",
|
||
) -> User:
|
||
"""创建用户并返回 User 实例。"""
|
||
async with AsyncSessionLocal() as session:
|
||
# 检查手机号是否已存在
|
||
stmt = select(User).where(User.mobile == mobile)
|
||
result = await session.execute(stmt)
|
||
existing = result.scalar_one_or_none()
|
||
if existing is not None:
|
||
print(f"⚠️ 手机号 {mobile} 已存在（id={existing.id}），跳过创建")
|
||
return existing
|
||
|
||
# 创建用户
|
||
user = User(
|
||
id=uuid.uuid4(),
|
||
mobile=mobile,
|
||
nickname=nickname,
|
||
source=source,
|
||
)
|
||
session.add(user)
|
||
# 需要先 flush 获取 user.id（虽然 uuid 已预生成，但 flush 确保数据库一致）
|
||
await session.flush()
|
||
|
||
# 创建积分汇总记录
|
||
user_point = UserPoint(user_id=user.id)
|
||
session.add(user_point)
|
||
|
||
# 可选：创建设备记录
|
||
if device_id:
|
||
from datetime import UTC, datetime
|
||
|
||
device = UserDevice(
|
||
user_id=user.id,
|
||
device_id=device_id,
|
||
device_name=device_name,
|
||
last_active_at=datetime.now(UTC),
|
||
)
|
||
session.add(device)
|
||
|
||
await session.commit()
|
||
print(f"✅ 用户创建成功: id={user.id}, mobile={mobile}, nickname={nickname or '(未设置)'}")
|
||
return user
|
||
|
||
|
||
async def run(args: argparse.Namespace) -> None:
|
||
"""主执行逻辑。"""
|
||
try:
|
||
await create_user(
|
||
mobile=args.mobile,
|
||
nickname=args.nickname,
|
||
device_id=args.device_id,
|
||
device_name=args.device_name,
|
||
source=args.source,
|
||
)
|
||
finally:
|
||
await close_db()
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="直接操作数据库创建用户（绕过短信验证）")
|
||
parser.add_argument("--mobile", required=True, help="手机号")
|
||
parser.add_argument("--nickname", default=None, help="用户昵称")
|
||
parser.add_argument("--device-id", default=None, help="设备标识（可选）")
|
||
parser.add_argument("--device-name", default=None, help="设备名称（可选）")
|
||
parser.add_argument("--source", default="cli", help="注册来源（默认: cli）")
|
||
args = parser.parse_args()
|
||
|
||
asyncio.run(run(args))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|