155 lines
4.6 KiB
Python
155 lines
4.6 KiB
Python
"""
|
|
管理命令行工具
|
|
|
|
用法：
|
|
python manage.py create-internal-user # 交互式创建内部员工
|
|
python manage.py create-internal-user --username admin --email admin@company.com --password 12345678
|
|
python manage.py list-users # 列出所有用户
|
|
python manage.py list-users --role internal # 只列出内部员工
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 确保能导入 app 模块
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.db import get_session_factory, get_engine
|
|
from app.core.security import hash_password
|
|
from app.models import Base, User, Plan
|
|
from sqlalchemy import select
|
|
|
|
|
|
def ensure_tables():
|
|
"""确保表存在。"""
|
|
engine = get_engine()
|
|
Base.metadata.create_all(engine)
|
|
|
|
|
|
def get_or_create_free_plan(session) -> Plan:
|
|
"""获取或创建 free plan。"""
|
|
plan = session.scalars(select(Plan).where(Plan.code == "free")).first()
|
|
if not plan:
|
|
plan = Plan(
|
|
code="free",
|
|
name="免费版",
|
|
storage_quota=104_857_600,
|
|
max_file_size=20_971_520,
|
|
is_active=True,
|
|
)
|
|
session.add(plan)
|
|
session.flush()
|
|
return plan
|
|
|
|
|
|
def cmd_create_internal_user(args):
|
|
"""创建内部员工账号。"""
|
|
ensure_tables()
|
|
factory = get_session_factory()
|
|
|
|
username = args.username
|
|
email = args.email
|
|
password = args.password
|
|
|
|
# 交互式输入
|
|
if not username:
|
|
username = input("用户名: ").strip()
|
|
if not email:
|
|
email = input("邮箱: ").strip()
|
|
if not password:
|
|
import getpass
|
|
password = getpass.getpass("密码: ").strip()
|
|
|
|
if not username or not email or not password:
|
|
print("错误：用户名、邮箱、密码不能为空。")
|
|
sys.exit(1)
|
|
|
|
if len(password) < 8:
|
|
print("错误：密码长度不能少于 8 位。")
|
|
sys.exit(1)
|
|
|
|
with factory() as session:
|
|
# 检查用户名/邮箱是否已存在
|
|
existing = session.scalars(
|
|
select(User).where((User.username == username) | (User.email == email.lower()))
|
|
).first()
|
|
if existing:
|
|
if existing.username == username:
|
|
print(f"错误：用户名 '{username}' 已被占用。")
|
|
else:
|
|
print(f"错误：邮箱 '{email}' 已被注册。")
|
|
sys.exit(1)
|
|
|
|
plan = get_or_create_free_plan(session)
|
|
user = User(
|
|
username=username,
|
|
email=email.lower(),
|
|
password_hash=hash_password(password),
|
|
status="active",
|
|
role="internal",
|
|
plan_id=plan.id,
|
|
storage_used=0,
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
|
|
print(f"\n✅ 内部员工账号创建成功！")
|
|
print(f" 用户名：{username}")
|
|
print(f" 邮箱：{email}")
|
|
print(f" 角色：internal")
|
|
print(f" 登录地址：/internal-login")
|
|
|
|
|
|
def cmd_list_users(args):
|
|
"""列出所有用户。"""
|
|
ensure_tables()
|
|
factory = get_session_factory()
|
|
|
|
with factory() as session:
|
|
stmt = select(User).order_by(User.created_at.desc())
|
|
if args.role:
|
|
stmt = stmt.where(User.role == args.role)
|
|
|
|
users = list(session.scalars(stmt).all())
|
|
|
|
if not users:
|
|
print("暂无用户。")
|
|
return
|
|
|
|
print(f"\n{'用户名':<15} {'邮箱':<25} {'角色':<10} {'状态':<8} {'创建时间'}")
|
|
print("-" * 80)
|
|
for u in users:
|
|
print(f"{u.username:<15} {u.email:<25} {u.role:<10} {u.status:<8} {u.created_at[:19]}")
|
|
|
|
print(f"\n共 {len(users)} 个用户")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="AI Knowledge Link 管理工具")
|
|
subparsers = parser.add_subparsers(dest="command", help="可用命令")
|
|
|
|
# create-internal-user
|
|
p_create = subparsers.add_parser("create-internal-user", help="创建内部员工账号")
|
|
p_create.add_argument("--username", "-u", help="用户名")
|
|
p_create.add_argument("--email", "-e", help="邮箱")
|
|
p_create.add_argument("--password", "-p", help="密码")
|
|
|
|
# list-users
|
|
p_list = subparsers.add_parser("list-users", help="列出所有用户")
|
|
p_list.add_argument("--role", "-r", choices=["internal", "customer"], help="按角色过滤")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "create-internal-user":
|
|
cmd_create_internal_user(args)
|
|
elif args.command == "list-users":
|
|
cmd_list_users(args)
|
|
else:
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|