django 利用钉钉扩展用户系统

Posted 2021-01-15

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了django 利用钉钉扩展用户系统相关的知识，希望对你有一定的参考价值。

django 利用钉钉扩展用户系统

用户架构图

技术分享图片

说明

以django 自带 user,group 为基础 ,结合 钉钉用户系统 进行结合，以达到能够满足实际需求。

组关系: 钉钉部门组 通过 多对多 关系到 系统Group。 
然后设置计划任务，如果 有钉钉部门组 关联到了 系统组，就把该用户组下的 用户 自动添加相应的 系统组。

组关系2: 自定义权限组 通过 多对多 关系到 系统Group。 
根据 用户 系统组 的名字 查询到 自定义权限信息，进行判断

版本

Django==2.0.9
jsonfield==2.0.2

建立表结构

钉钉部门表
扩展User表
自定义权限组
钉钉token信息表

from django.db import models
from django.contrib.auth.models import AbstractUser, Group, User
from jsonfield import JSONField

class Dept(models.Model):
    name = models.CharField(max_length=128, verbose_name=‘部门名称‘, )
    group = models.ManyToManyField(to=Group, blank=True)
    parentid = models.CharField(max_length=128, verbose_name=‘部门上一级id‘, blank=True, null=True)
    parentname1 = models.CharField(max_length=128, verbose_name=‘部门上一级名字‘, blank=True, null=True)
    parentname2 = models.CharField(max_length=128, verbose_name=‘部门上二级名字‘, blank=True, null=True)
    parentname3 = models.CharField(max_length=128, verbose_name=‘部门上三级名字‘, blank=True, null=True)
    parentname4 = models.CharField(max_length=128, verbose_name=‘部门上四级名字‘, blank=True, null=True)

    class Meta:
        db_table = "dept"
        verbose_name = "钉钉部门"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.name

class Users(AbstractUser):
    position = models.CharField(max_length=64, verbose_name=‘职位信息‘, blank=True, null=True)
    unionid = models.CharField(max_length=64, verbose_name=‘钉钉unionid‘, blank=True, null=True)
    userid = models.CharField(max_length=64, verbose_name=‘钉钉userid‘, blank=True, null=True)
    dingId = models.CharField(max_length=64, verbose_name=‘钉钉dingId‘, blank=True, null=True)

    avatar = models.CharField(max_length=256, verbose_name=‘头像‘, blank=True, null=True)
    mobile = models.CharField(max_length=11, verbose_name=‘手机‘, blank=True, null=True)

    isBoss = models.BooleanField(verbose_name=‘是否为BOSS‘, default=False, )
    is_ding_admin = models.BooleanField(verbose_name=‘是否为钉钉admin‘, default=False, )
    ding_group = models.ManyToManyField(to=Dept, blank=True, verbose_name=‘钉钉组‘, related_name=‘dept‘)

    class Meta:
        db_table = ‘users‘
        verbose_name = ‘用户信息‘
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.username

 class Request(models.Model):
    request = models.CharField(max_length=16, verbose_name=‘请求类型(大写)‘)

    class Meta:
        db_table = "request"
        verbose_name = "请求类型"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.request

class RolePermission(models.Model):
    role = models.CharField(max_length=32, verbose_name=‘角色组‘)
    group = models.ManyToManyField(Group, verbose_name=‘用户系统组‘, related_name=‘roles‘,blank=True)
    table = models.CharField(max_length=32, verbose_name=‘表名字‘)
    request = models.ManyToManyField(Request, verbose_name=‘请求‘, related_name=‘re‘, )
    permission = JSONField(max_length=1024, verbose_name=‘权限条件‘)
    level = models.IntegerField(verbose_name=‘相同表下,权限优先级‘,default=‘1‘)

    class Meta:
        db_table = "role_permission"
        verbose_name = "自定义权限组"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.role

class DingdingToken(models.Model):
    DINGDING_CHOICES = (
        (‘0‘, ‘登录认证token‘),
        (‘1‘, ‘通讯录token‘),
    )
    appId = models.CharField(max_length=128, verbose_name=‘appId‘)
    appSecret = models.CharField(max_length=128, verbose_name=‘appSecret‘)
    access_token = models.CharField(max_length=128, verbose_name=‘access_token‘, blank=True, null=True)
    type = models.CharField(choices=DINGDING_CHOICES, max_length=16, verbose_name=‘钉钉token类型‘)
    up_time = models.DateTimeField(auto_now=True, verbose_name=‘更新时间‘)

    class Meta:
        db_table = "dingding_token"
        verbose_name = "钉钉token"
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.appId

自定义权限组例子

技术分享图片

以常见的资产 asset 为例

表名字  asset     字段 group     (分组 为 dev,ops)

在Request 表添加

GET （代表只读）
POST （代表更新删除）
在RolePermission 添加

角色资产只读组
系统组资产只读组
表名字 assset
请求 GET
权限条件 {"group":‘dev‘}
level 1 （如系统组有多条相同表的权限，根据级别判断最高权限）

把钉钉部门系统运维部添加系统组资产只读，这样系统运维部下面的用户就自动有了只读资产权限。

获取钉钉信息

首先把钉钉用户、部门，注册到 user 和钉钉部门表内。

定时更新钉钉用户钉钉部门

import time
import requests
from  test.models import DingdingToken, Users, Dept

def ding_book_token():
    """
    通讯录token
    :return:
    """
    ding = DingdingToken.objects.get(type=‘1‘)
    app_id = ding.appId
    app_secret = ding.appSecret
    token = requests.get(f‘https://oapi.dingtalk.com/gettoken?corpid={app_id}&corpsecret={app_secret}‘)
    ding.access_token = token.json()["access_token"]
    ding.save()

def ding_get_user_info():
    """
    更新 钉钉 用户 信息 ,根据部门ID
    :return:
    """
    access_token = DingdingToken.objects.get(type=‘1‘).access_token
    list_ids = Dept.objects.all()
    for u in list_ids:
        if u.id < 1000:
            continue
        user_list_requests = requests.get(
            f‘https://oapi.dingtalk.com/user/simplelist?access_token={access_token}&department_id={u.id}‘)
        user_list = user_list_requests.json()[‘userlist‘]

        for i in user_list:
            user_info_request = requests.get(
                f‘https://oapi.dingtalk.com/user/get?access_token={access_token}&userid={i["userid"]}‘)
            user_info = user_info_request.json()
            try:
                username = i[‘name‘]
                print(username)
                Users.objects.update_or_create(userid=i[‘userid‘],
                                               defaults={‘username‘: username, ‘email‘: user_info[
                                                   ‘email‘] if ‘email‘ in user_info.keys() else ‘‘,
                                                         ‘position‘: user_info[‘position‘],
                                                         ‘unionid‘: user_info[‘unionid‘], ‘dingId‘: user_info[‘dingId‘],
                                                         ‘mobile‘: user_info[‘mobile‘],
                                                         ‘is_active‘: user_info[‘active‘],
                                                         ‘isBoss‘: user_info[‘isBoss‘],
                                                         ‘avatar‘: user_info[‘avatar‘],
                                                         ‘is_ding_admin‘: user_info[‘isAdmin‘]
                                                         })

            except Exception as e:
                Users.objects.update_or_create(userid=i[‘userid‘],
                                               defaults={‘username‘: f"{i[‘name‘]}{(user_info[‘mobile‘])[-4:]}",
                                                         ‘email‘: user_info[
                                                             ‘email‘] if ‘email‘ in user_info.keys() else ‘‘,
                                                         ‘position‘: user_info[‘position‘],
                                                         ‘unionid‘: user_info[‘unionid‘], ‘dingId‘: user_info[‘dingId‘],
                                                         ‘mobile‘: user_info[‘mobile‘],
                                                         ‘is_active‘: user_info[‘active‘],
                                                         ‘isBoss‘: user_info[‘isBoss‘],
                                                         ‘avatar‘: user_info[‘avatar‘],
                                                         ‘is_ding_admin‘: user_info[‘isAdmin‘]
                                                         })

            u = Users.objects.get(userid=i[‘userid‘])
            u.ding_group.set(user_info[‘department‘])
            u.save()
            time.sleep(0.2)

def ding_book_update():
    ding = DingdingToken.objects.filter(type=‘1‘).first()
    print(ding)
    if ding:
        token = ding.access_token
        department = requests.get(f"https://oapi.dingtalk.com/department/list?access_token={token}")

        for i in department.json()[‘department‘]:
            if i[‘id‘] == 1:
                Dept.objects.update_or_create(id=i[‘id‘], defaults={‘name‘: ‘‘, ‘parentid‘: ‘0‘, })
            else:
                parent_name2 = ‘‘
                parent_name3 = ‘‘
                parent_name4 = ‘‘
                try:
                    parent = Dept.objects.get(id=i[‘parentid‘])
                    parent_id1 = parent.parentid
                    parent_name1 = parent.name

                    try:
                        parent2 = Dept.objects.get(id=parent_id1)
                        parent_name2 = parent2.name
                        parent_id2 = parent2.parentid

                        try:
                            parent3 = Dept.objects.get(id=parent_id2)
                            parent_name3 = parent3.name
                            parent_id3 = parent3.parentid
                            try:
                                parent4 = Dept.objects.get(id=parent_id3)
                                parent_name4 = parent4.name
                            except Exception as e:
                                parent_name4 = ‘‘
                        except Exception as e:
                            parent_name3 = ‘‘

                    except Exception as e:
                        parent_name2 = ‘‘

                except Exception as e:
                    parent_name1 = ‘‘

                Dept.objects.update_or_create(id=i[‘id‘],
                                              defaults={‘name‘: i[‘name‘], ‘parentid‘: i[‘parentid‘],
                                                        ‘parentname1‘: parent_name1,
                                                        ‘parentname2‘: parent_name2,
                                                        ‘parentname3‘: parent_name3,
                                                        ‘parentname4‘: parent_name4})

计划任务, 更新钉钉部门组和系统的组的对应关系。

def ding_user_group_update():
    """
    更新 User   钉钉组 和 系统组 的权限关系
    :return:
    """
    for i in Users.objects.all():
        g_list = []
        for d in i.ding_group.all():
            if d.group.all():
                for g in d.group.all():
                    g_list.append(g.id)

        old_list = []
        for o in i.groups.all():
            old_list.append(o.id)
        if old_list != g_list:
            print(i.username, ‘用户组更新了‘)
            i.groups.set(g_list)
            i.save()

钉钉登录

http://blog.51cto.com/hequan/2304690

admin设置

from django.contrib import admin
from test.models import  Request, RolePermission, DingdingToken, Users, Dept
from django.contrib.auth.admin import UserAdmin

class DingdingTokenAdmin(admin.ModelAdmin):
    list_display = (‘appId‘, ‘type‘, ‘access_token‘, ‘up_time‘)

class RolePermissionAdmin(admin.ModelAdmin):
    list_display = (‘role‘, ‘table‘, ‘show_group‘, ‘permission‘, ‘level‘,)

    @classmethod
    def show_group(self, obj):
        return [i.name for i in obj.group.all()]

    filter_horizontal = (‘group‘, )

class DeptAdmin(admin.ModelAdmin):

    @classmethod
    def show_group(self, obj):
        return [i.name for i in obj.group.all()]

    list_display = (‘id‘, ‘name‘, ‘show_group‘, ‘parentname1‘, ‘parentname2‘, ‘parentname3‘, ‘parentname4‘, ‘parentid‘,)
    search_fields = (‘id‘, ‘name‘)
    list_display_links = (‘id‘, ‘name‘)
    filter_horizontal = (‘group‘,)

class UsersAdmin(UserAdmin):
    fieldsets = (
        (None, {‘fields‘: (‘username‘, ‘password‘)}),
        (‘基本信息‘, {‘fields‘: (‘first_name‘, ‘last_name‘, ‘email‘)}),
        (‘权限‘, {‘fields‘: (‘is_active‘, ‘is_staff‘, ‘is_superuser‘, ‘groups‘, ‘user_permissions‘)}),
        (‘登录时间‘, {‘fields‘: (‘last_login‘, ‘date_joined‘)}),
        (‘钉钉信息‘, {‘fields‘: (
            ‘position‘, ‘unionid‘, ‘dingId‘, ‘userid‘, ‘avatar‘, ‘mobile‘, ‘isBoss‘, ‘is_ding_admin‘, ‘ding_group‘)}),
    )

admin.site.register(Users, UsersAdmin)
admin.site.register(Request)
admin.site.register(RolePermission, RolePermissionAdmin)
admin.site.register(DingdingToken, DingdingTokenAdmin)
admin.site.register(Dept, DeptAdmin)
admin.site.site_header = ‘运维管理后台‘
admin.site.site_title = admin.site.site_header

权限判断

import json
from test.models import RolePermission
from functools import wraps
from django.shortcuts import HttpResponse
from django.db.models import Q

def role_permission_get_list(function):
    """
    列表页面
    :param function:
    :return:
    """

    @wraps(function)
    def wrapped(self):
        user = self.request.user
        groups = [x[‘name‘] for x in self.request.user.groups.values()]
        request_type = self.request.method
        model = str(self.model._meta).split(".")[1]

        filter_dict = {}
        not_list = [‘page‘, ‘order_by‘, ‘csrfmiddlewaretoken‘]
        for k, v in dict(self.request.GET).items():
            if [i for i in v if i != ‘‘] and (k not in not_list):
                if ‘__in‘ in k:
                    filter_dict[k] = v
                else:
                    filter_dict[k] = v[0]

        if not user.is_superuser:
            role_groups = RolePermission.objects.filter(
                Q(group__name__in=groups) & Q(request__request=request_type) & Q(table=model)).values_list(‘table‘,
                                                                                                           ‘request__request‘,
                                                                                                           ‘permission‘,
                                                                                                           ‘level‘).order_by(
                ‘-level‘).first()

            permission_dict = json.loads(role_groups[2])
            if permission_dict:
                if filter_dict:
                    for k, v in permission_dict.items():
                        if ‘__in‘ in k:
                            k1 = k.replace(‘__in‘, ‘‘)
                        if ‘__gt‘ in k:
                            k1 = k.replace(‘__gt‘, ‘‘)
                        if ‘__lt‘ in k:
                            k1 = k.replace(‘__lt‘, ‘‘)
                        else:
                            k1 = k
                        if k1 in list(filter_dict.keys()):
                            del filter_dict[k1]

                    if filter_dict:
                        filter_dict.update(**permission_dict)
                    else:
                        print(‘查询条件处理后为空,默认权限‘)
                        filter_dict = permission_dict
                else:
                    print(‘查询条件为空,默认权限‘)
                    filter_dict = permission_dict
            else:
                print(‘没有权限‘)
                filter_dict = {‘id‘: -1}

        self.filter_dict = filter_dict
        filter_dict = self.filter_dict
        self.queryset = self.model.objects.filter(**filter_dict)
        order_by_val = self.request.GET.get(‘order_by‘, ‘‘)
        if order_by_val:
            self.queryset = self.queryset.order_by(order_by_val) if self.queryset else self.queryset
        result = function(self)
        return result

    return wrapped

def role_permission_detail(function):
    """
    详情 页面
    :param function:
    :return:
    """

    @wraps(function)
    def wrapped(self, request, *args, **kwargs):
        user = self.request.user

        if not user.is_superuser:
            groups = [x[‘name‘] for x in self.request.user.groups.values()]
            request_type = self.request.method
            model = str(self.model._meta).split(".")[1]
            pk = self.kwargs.get(self.pk_url_kwarg, None)
            role_groups = RolePermission.objects.filter(
                Q(group__name__in=groups) & Q(request__request=request_type) & Q(table=model)).values_list(‘table‘,
                                                                                                           ‘request__request‘,
                                                                                                           ‘permission‘,
                                                                                                           ‘level‘).order_by(
                ‘-level‘).first()
            permission_dict = json.loads(role_groups[2])
            if pk:
                permission_dict[‘id‘] = pk
            obj = self.model.objects.filter(**permission_dict).count()
            if not obj:
                return HttpResponse(status=403)
        result = function(self, request, *args, **kwargs)
        return result

    return wrapped

def role_permission_update_delete(function):
    """
     更新 删除
    :param function:
    :return:
    """

    @wraps(function)
    def wrapped(self, request):
        user = self.request.user
        if not user.is_superuser:

            groups = [x[‘name‘] for x in self.request.user.groups.values()]
            request_type = self.request.method
            model = str(self.model._meta).split(".")[1]
            pk = self.request.POST.get(‘nid‘, None)
            role_groups = RolePermission.objects.filter(
                Q(group__name__in=groups) & Q(request__request=request_type) & Q(table=model)).values_list(‘table‘,
                                                                                                           ‘request__request‘,
                                                                                                           ‘permission‘,
                                                                                                           ‘level‘).order_by(
                ‘-level‘).first()
            permission_dict = json.loads(role_groups[2])
            if pk:
                permission_dict[‘id‘] = pk
            else:
                instance = (self.get_form_kwargs())[‘instance‘]
                permission_dict[‘id‘] = instance.id
            obj = self.model.objects.filter(**permission_dict).count()
            if not obj:
                ret = {‘status‘: None, ‘error‘: "没有权限,拒绝", ‘msg‘: ‘Without permission, rejected‘}
                return HttpResponse(json.dumps(ret))
        result = function(self, request)
        return result

    return wrapped

例如
@role_permission_get_list
def get_context_data(self, **kwargs):

以上是关于django 利用钉钉扩展用户系统的主要内容，如果未能解决你的问题，请参考以下文章

如何在扩展另一个文件的 django 模板中使用带有动态内容的 html 块片段？

Python3.7配合Django2.0来调用钉钉(dingding)在线api实时监测员工考勤打卡情况