如何基于Python登录系统实现用户大厅/群组的创建、加入及专属功能调用？

基于你已有的用户登录系统，我们可以通过扩展数据库结构、实现核心群组功能、添加权限校验这三步来完成需求，以下是具体实现细节：

1. 数据库表结构扩展 #

你需要在现有用户表基础上新增三张表，用来存储大厅信息、用户与大厅的关联、以及邀请记录：

• halls（大厅表）：

  • id：主键，唯一标识大厅
  • name：大厅名称（必填）
  • owner_id：关联用户表的创建者ID（外键）
  • created_at：大厅创建时间（默认当前时间）
  • description：可选，大厅描述字段

• user_hall_relations（用户-大厅关联表）：

  • user_id：关联用户ID（外键，联合主键）
  • hall_id：关联大厅ID（外键，联合主键）
  • role：用户在大厅的角色（比如owner/admin/member，默认member）
  • joined_at：用户加入时间（默认当前时间）

• hall_invites（邀请记录表）：

  • id：主键
  • hall_id：关联大厅ID（外键）
  • inviter_id：邀请者ID（外键）
  • invitee_email：被邀请者邮箱（匹配用户表的邮箱字段）
  • code：唯一邀请码（用于验证）
  • status：邀请状态（pending/accepted/rejected，默认pending）
  • expired_at：邀请过期时间（默认7天后）

2. 核心功能代码实现 #

下面用SQLAlchemy ORM示例（如果你用原生SQL，逻辑完全一致，只是语法不同）：

2.1 创建大厅 #

登录用户创建大厅后，自动成为大厅所有者并加入：

import datetime
import uuid
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()

# 假设你已有的User模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True)
    email = Column(String(100), unique=True)

class Hall(Base):
    __tablename__ = 'halls'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    owner_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    owner = relationship('User')

class UserHallRelation(Base):
    __tablename__ = 'user_hall_relations'
    user_id = Column(Integer, ForeignKey('users.id'), primary_key=True)
    hall_id = Column(Integer, ForeignKey('halls.id'), primary_key=True)
    role = Column(String(20), default='member')
    joined_at = Column(DateTime, default=datetime.datetime.utcnow)

# 初始化数据库会话
engine = create_engine('sqlite:///your_database.db')
Session = sessionmaker(bind=engine)
db_session = Session()

def create_hall(user_id, hall_name, description=None):
    # 校验用户存在
    user = db_session.query(User).filter_by(id=user_id).first()
    if not user:
        raise ValueError("用户不存在")
    # 创建大厅
    new_hall = Hall(name=hall_name, owner_id=user_id)
    db_session.add(new_hall)
    db_session.commit()
    # 绑定创建者到大厅
    owner_relation = UserHallRelation(user_id=user_id, hall_id=new_hall.id, role='owner')
    db_session.add(owner_relation)
    db_session.commit()
    return new_hall.id

2.2 邀请与接受邀请 #

邀请者必须是大厅成员，被邀请者通过邮箱匹配验证：

class HallInvite(Base):
    __tablename__ = 'hall_invites'
    id = Column(Integer, primary_key=True)
    hall_id = Column(Integer, ForeignKey('halls.id'))
    inviter_id = Column(Integer, ForeignKey('users.id'))
    invitee_email = Column(String(100))
    code = Column(String(64), unique=True)
    status = Column(String(20), default='pending')
    expired_at = Column(DateTime, default=lambda: datetime.datetime.utcnow() + datetime.timedelta(days=7))

def send_hall_invite(inviter_id, hall_id, invitee_email):
    # 校验邀请者是否在大厅内
    inviter_in_hall = db_session.query(UserHallRelation).filter_by(user_id=inviter_id, hall_id=hall_id).first()
    if not inviter_in_hall:
        raise ValueError("你不是该大厅成员，无法发送邀请")
    # 生成唯一邀请码
    invite_code = str(uuid.uuid4())
    new_invite = HallInvite(
        hall_id=hall_id,
        inviter_id=inviter_id,
        invitee_email=invitee_email,
        code=invite_code
    )
    db_session.add(new_invite)
    db_session.commit()
    # 这里可以加发送邮件的逻辑，把invite_code发给被邀请者邮箱
    return invite_code

def accept_hall_invite(user_id, invite_code):
    # 校验邀请有效性
    invite = db_session.query(HallInvite).filter_by(code=invite_code, status='pending').first()
    if not invite or datetime.datetime.utcnow() > invite.expired_at:
        raise ValueError("邀请无效或已过期")
    # 校验被邀请者是当前用户
    user = db_session.query(User).filter_by(id=user_id, email=invite.invitee_email).first()
    if not user:
        raise ValueError("该邀请不属于你")
    # 校验用户未加入该大厅
    already_joined = db_session.query(UserHallRelation).filter_by(user_id=user_id, hall_id=invite.hall_id).first()
    if already_joined:
        raise ValueError("你已加入该大厅")
    # 添加用户到大厅
    new_relation = UserHallRelation(user_id=user_id, hall_id=invite.hall_id)
    db_session.add(new_relation)
    # 更新邀请状态
    invite.status = 'accepted'
    db_session.commit()

2.3 直接加入大厅（公开场景） #

如果是公开大厅，允许用户直接申请加入：

def join_hall_directly(user_id, hall_id):
    # 校验大厅存在
    hall = db_session.query(Hall).filter_by(id=hall_id).first()
    if not hall:
        raise ValueError("大厅不存在")
    # 校验用户未加入
    already_joined = db_session.query(UserHallRelation).filter_by(user_id=user_id, hall_id=hall_id).first()
    if already_joined:
        raise ValueError("你已加入该大厅")
    # 添加关联
    new_relation = UserHallRelation(user_id=user_id, hall_id=hall_id)
    db_session.add(new_relation)
    db_session.commit()

3. 权限控制与大厅专属函数 #

要确保只有大厅内用户能调用专属函数，并且函数仅影响大厅成员：

3.1 权限校验工具 #

写一个通用的校验函数，或者用装饰器简化重复逻辑：

def is_user_in_hall(user_id, hall_id):
    return bool(db_session.query(UserHallRelation).filter_by(user_id=user_id, hall_id=hall_id).first())

# 装饰器：要求用户必须是大厅成员
def require_hall_member(func):
    def wrapper(user_id, hall_id, *args, **kwargs):
        if not is_user_in_hall(user_id, hall_id):
            raise PermissionError("你不属于该大厅，无法执行此操作")
        return func(user_id, hall_id, *args, **kwargs)
    return wrapper

# 装饰器：要求用户是大厅所有者
def require_hall_owner(func):
    def wrapper(user_id, hall_id, *args, **kwargs):
        relation = db_session.query(UserHallRelation).filter_by(user_id=user_id, hall_id=hall_id, role='owner').first()
        if not relation:
            raise PermissionError("仅大厅所有者可执行此操作")
        return func(user_id, hall_id, *args, **kwargs)
    return wrapper

3.2 实现大厅专属函数 #

比如发送大厅通知、统计成员、修改大厅名称等：

@require_hall_member
def send_hall_announcement(user_id, hall_id, content):
    # 获取所有大厅成员ID
    member_ids = [rel.user_id for rel in db_session.query(UserHallRelation).filter_by(hall_id=hall_id).all()]
    # 这里实现通知逻辑，比如插入通知表、推送消息等，仅针对这些member_ids
    print(f"向大厅{hall_id}的成员{member_ids}发送公告：{content}")

@require_hall_member
def get_hall_member_list(user_id, hall_id):
    members = db_session.query(User).join(UserHallRelation).filter(UserHallRelation.hall_id == hall_id).all()
    return [{"id": u.id, "username": u.username} for u in members]

@require_hall_owner
def rename_hall(user_id, hall_id, new_name):
    hall = db_session.query(Hall).filter_by(id=hall_id).first()
    hall.name = new_name
    db_session.commit()

4. Web场景下的会话处理（可选） #

如果是Web应用（比如Flask/Django），需要从请求会话中获取当前登录用户ID，示例用Flask：

from flask import Flask, session, request, jsonify

app = Flask(__name__)
app.secret_key = 'your_secret_key_here'

# 创建大厅接口
@app.route('/api/halls', methods=['POST'])
def api_create_hall():
    if 'user_id' not in session:
        return jsonify({"error": "请先登录"}), 401
    data = request.get_json()
    hall_name = data.get('name')
    try:
        hall_id = create_hall(session['user_id'], hall_name)
        return jsonify({"hall_id": hall_id, "message": "大厅创建成功"}), 200
    except ValueError as e:
        return jsonify({"error": str(e)}), 400

# 接受邀请接口
@app.route('/api/halls/invite/<invite_code>', methods=['POST'])
def api_accept_invite(invite_code):
    if 'user_id' not in session:
        return jsonify({"error": "请先登录"}), 401
    try:
        accept_hall_invite(session['user_id'], invite_code)
        return jsonify({"message": "成功加入大厅"}), 200
    except ValueError as e:
        return jsonify({"error": str(e)}), 400

内容的提问来源于stack exchange，提问作者jeff425