from datetime import datetime
|
|
from sqlalchemy import Column, Integer, String, Table, ForeignKey, DateTime
|
from sqlalchemy.orm import relationship, backref
|
|
from app.models.base_model import Base
|
|
user_organization_table = Table('user_organization',Base.metadata
|
, Column('user_id', Integer, ForeignKey('user.id', ondelete='CASCADE'))
|
, Column('organization_id', String(36), ForeignKey('organization.id', ondelete='CASCADE')))
|
|
user_role_table = Table('user_role', Base.metadata
|
, Column('user_id', Integer, ForeignKey('user.id', ondelete='CASCADE'))
|
, Column('role_id', String(36), ForeignKey('role.id', ondelete='CASCADE')))
|
|
user_group_table = Table('user_group', Base.metadata
|
, Column('user_id', Integer, ForeignKey('user.id', ondelete='CASCADE'))
|
, Column('group_id', Integer, ForeignKey('group.id', ondelete='CASCADE')))
|
|
|
|
class UserModel(Base):
|
__tablename__ = "user"
|
id = Column(Integer, primary_key=True, index=True)
|
username = Column(String(255), unique=True, index=True)
|
hashed_password = Column(String(255))
|
compellation = Column(String(255), nullable=False, default="")
|
phone = Column(String(255), nullable=False, default="")
|
email = Column(String(255), nullable=False, default="")
|
description = Column(String(255), nullable=False, default="")
|
ragflow_id = Column(String(32), unique=True, index=True)
|
bisheng_id = Column(Integer, unique=True, index=True)
|
login_name = Column(String(100))
|
status = Column(String(10), nullable=False, default="1")
|
creator = Column(String(36))
|
sex = Column(String(1))
|
permission = Column(String(16), nullable=False, default="1")
|
age = Column(Integer)
|
created_at = Column(DateTime, default=datetime.now())
|
updated_at = Column(DateTime, default=datetime.now(), onupdate=datetime.now())
|
|
|
organizations = relationship('OrganizationModel',
|
secondary=user_organization_table,
|
backref=backref('users', lazy='dynamic'))
|
|
roles = relationship('RoleModel',
|
secondary=user_role_table,
|
backref=backref('users', lazy='dynamic'))
|
|
groups = relationship('GroupModel',
|
secondary=user_group_table,
|
backref=backref('users', lazy='dynamic'))
|
|
|
|
def have_permission(self, url):
|
permissions = []
|
for resource in self.resources:
|
permissions.extend(resource)
|
|
if filter(lambda x: x.URL == url, permissions):
|
return True
|
|
permissions = []
|
for organization in self.organizations:
|
permissions.extend([resource for resource in organization.resources])
|
|
return filter(lambda x: x.NAME == url, permissions)
|
|
def __repr__(self):
|
return '<User %r>\n' % (self.NAME)
|
|
|
def to_dict(self):
|
return {
|
'userId': self.id,
|
'userName': self.username,
|
'loginName': self.login_name if self.login_name else "",
|
'status': self.status,
|
}
|
|
def to_json(self):
|
json = {
|
'userId': self.id,
|
'createTime': self.created_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else "",
|
'updateTime': self.updated_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else "",
|
'userName': self.username,
|
'loginName': self.login_name,
|
'sex': self.sex,
|
'age': self.age,
|
"status": self.status,
|
'phone': self.phone,
|
'email': self.email,
|
# 'phoneNumber': self.phone_number
|
}
|
|
if len(self.organizations) > 0:
|
json['dept'] = [organization.to_json() for organization in self.organizations]
|
|
|
json['groups'] = [group.to_dict() for group in self.groups]
|
|
roles = []
|
# if len(self.roles) > 0:
|
roles = [role.to_json() for role in self.roles]
|
organization_roles = [role.to_json() for organization in self.organizations for role in
|
organization.roles]
|
for role in organization_roles:
|
if role not in roles:
|
roles.append(role)
|
json['roles'] = roles
|
return json
|
|
|
def get_children(self):
|
"""递归获取指定用户的子用户ID列表"""
|
users = UserModel.query.filter_by(CREATOR=self.ID).all()
|
user_id_list = [user.ID for user in users]
|
for user in users:
|
user_id_list.extend(user.get_children())
|
return user_id_list
|
