62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
"""Create or update a local user account with a hashed password."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.security import hash_password
|
|
from app.models.entities import UserAccount
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
|
|
async def create_or_update_user(
|
|
emp_no: str,
|
|
name: str,
|
|
password: str,
|
|
role: str,
|
|
email: str | None,
|
|
) -> int:
|
|
settings = get_settings()
|
|
engine = create_async_engine(settings.database_url, pool_pre_ping=True)
|
|
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
try:
|
|
async with session_factory() as session:
|
|
user = await session.scalar(select(UserAccount).where(UserAccount.emp_no == emp_no))
|
|
if user is None:
|
|
user = UserAccount(emp_no=emp_no, name=name, role=role, status="启用")
|
|
session.add(user)
|
|
user.name = name
|
|
user.role = role
|
|
user.status = "启用"
|
|
user.password_hash = hash_password(password)
|
|
user.email = email
|
|
await session.commit()
|
|
return user.id
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--emp-no", required=True, help="Employee number used for login")
|
|
parser.add_argument("--name", required=True, help="Display name")
|
|
parser.add_argument("--password", required=True, help="Plain password; only its hash is stored")
|
|
parser.add_argument(
|
|
"--role",
|
|
default="admin",
|
|
choices=("user", "product_manager", "reviewer", "admin"),
|
|
)
|
|
parser.add_argument("--email")
|
|
args = parser.parse_args()
|
|
if len(args.password) < 8:
|
|
parser.error("password must be at least 8 characters")
|
|
user_id = asyncio.run(create_or_update_user(args.emp_no, args.name, args.password, args.role, args.email))
|
|
print(f"user_id={user_id}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|