改用mysql

2026-09-02 11:19:49 +08:00
parent 410bd30b2e
commit 73c4d83f42
24 changed files with 831 additions and 258 deletions
@@ -0,0 +1,154 @@
+"""
+管理命令行工具
+
+用法：
+  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()