117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
创建测试用户脚本
|
|
在数据库中创建测试用户：admin, market, other
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
|
from sqlalchemy import text
|
|
|
|
# 数据库配置
|
|
DATABASE_URL = 'mysql+aiomysql://root:rootpassword@localhost:3306/project_manager?charset=utf8mb4'
|
|
|
|
# 创建引擎
|
|
engine = create_async_engine(DATABASE_URL, echo=False)
|
|
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""使用bcrypt哈希密码"""
|
|
import bcrypt
|
|
salt = bcrypt.gensalt(rounds=12)
|
|
hashed = bcrypt.hashpw(password.encode("utf-8"), salt)
|
|
return hashed.decode("utf-8")
|
|
|
|
|
|
async def create_test_users():
|
|
"""创建测试用户"""
|
|
async with AsyncSessionLocal() as session:
|
|
# 检查是否已有测试用户
|
|
result = await session.execute(text("SELECT username FROM users WHERE username IN ('market', 'other')"))
|
|
existing = result.fetchall()
|
|
|
|
if len(existing) >= 2:
|
|
print("测试用户已存在，跳过创建")
|
|
return
|
|
|
|
print("开始创建测试用户...")
|
|
|
|
# 检查market用户是否存在
|
|
result = await session.execute(text("SELECT id FROM users WHERE username = 'market'"))
|
|
market_exists = result.fetchone()
|
|
|
|
if not market_exists:
|
|
# 插入市场部用户
|
|
await session.execute(text("""
|
|
INSERT INTO users (username, password_hash, real_name, department, role, phone, is_active)
|
|
VALUES (:username, :password_hash, :real_name, :department, :role, :phone, :is_active)
|
|
"""), {
|
|
'username': 'market',
|
|
'password_hash': hash_password('market123'),
|
|
'real_name': '市场部用户',
|
|
'department': '市场部',
|
|
'role': 'market',
|
|
'phone': '13800000002',
|
|
'is_active': True
|
|
})
|
|
print(" - 市场部用户创建成功")
|
|
else:
|
|
print(" - 市场部用户已存在")
|
|
|
|
# 检查other用户是否存在
|
|
result = await session.execute(text("SELECT id FROM users WHERE username = 'other'"))
|
|
other_exists = result.fetchone()
|
|
|
|
if not other_exists:
|
|
# 插入其他部门用户
|
|
await session.execute(text("""
|
|
INSERT INTO users (username, password_hash, real_name, department, role, phone, is_active)
|
|
VALUES (:username, :password_hash, :real_name, :department, :role, :phone, :is_active)
|
|
"""), {
|
|
'username': 'other',
|
|
'password_hash': hash_password('other123'),
|
|
'real_name': '其他部门用户',
|
|
'department': '其他部门',
|
|
'role': 'other',
|
|
'phone': '13800000003',
|
|
'is_active': True
|
|
})
|
|
print(" - 其他部门用户创建成功")
|
|
else:
|
|
print(" - 其他部门用户已存在")
|
|
|
|
await session.commit()
|
|
print("\n✅ 测试用户创建完成！")
|
|
print("\n用户账号信息：")
|
|
print(" - admin / admin123 (系统管理员)")
|
|
print(" - market / market123 (市场部)")
|
|
print(" - other / other123 (其他部门)")
|
|
|
|
|
|
async def main():
|
|
"""主函数"""
|
|
print("="*60)
|
|
print("创建测试用户")
|
|
print("="*60)
|
|
|
|
try:
|
|
await create_test_users()
|
|
except Exception as e:
|
|
print(f"❌ 错误: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|
|
|