使用 Python + Flask 构建钉钉部门和人员管理系统
·
使用 Python + Flask 构建钉钉部门和人员管理系统
摘要: 本文详细介绍如何使用 Python 和 Flask 框架开发一个完整的钉钉组织架构管理系统,包括从钉钉API同步数据、构建Web管理界面、实现部门人员查看和编辑等功能。
一、项目背景
在企业日常管理中,钉钉的组织架构数据是非常重要的信息。为了方便管理和查看钉钉的部门和人员信息,我开发了一个基于 Python + Flask 的Web管理系统。该系统可以实现:
- ✅ 自动从钉钉API同步部门和人员数据
- ✅ 可视化展示组织架构
- ✅ 按部门筛选查看人员
- ✅ 搜索和编辑人员信息
- ✅ 实时统计部门人数
二、技术栈
- 后端框架: Flask 3.x
- 前端技术: HTML5 + CSS3 + JavaScript (原生)
- 数据存储: JSON文件
- 钉钉API: 钉钉开放平台API v2
- 开发语言: Python 3.x
三、系统架构
钉钉部门和人员管理系统
── dingtalk_sync.py # 钉钉数据同步脚本
├── app.py # Flask Web应用
├── templates/
│ └── index.html # 前端管理界面
├── departments.json # 部门数据
├── users.json # 人员数据
└── requirements.txt # 依赖包
四、核心功能实现
4.1 钉钉数据同步模块
首先,我们需要从钉钉API获取组织架构数据。钉钉提供了完善的API接口,我们可以通过以下步骤获取数据:
1. 获取 Access Token
def get_access_token(self) -> str:
"""获取访问令牌"""
url = f"{self.base_url}/gettoken"
params = {
"appkey": self.app_key,
"appsecret": self.app_secret
}
response = requests.get(url, params=params)
result = response.json()
if result.get("errcode") == 0:
self.access_token = result.get("access_token")
return self.access_token
关键点:
- Access Token 有效期为 2 小时
- 需要缓存 Token,避免频繁请求
- 使用 AppKey 和 AppSecret 进行认证
2. 获取部门列表
def get_all_departments(self) -> List[Dict]:
"""获取所有部门列表"""
access_token = self.get_access_token()
url = f"{self.base_url}/topapi/v2/department/listsub"
# 获取根部门（dept_id=1）下的所有子部门
data = {"dept_id": 1}
response = requests.post(url, params=params, json=data)
result = response.json()
if result.get("errcode") == 0:
return result.get("result", [])
钉钉API说明:
- 接口地址:
/topapi/v2/department/listsub - 根部门 ID 固定为 1
- 返回所有子部门列表
3. 获取部门用户 (支持分页)
钉钉的用户列表接口支持分页查询,需要处理分页逻辑:
def get_department_users(self, dept_id: int) -> List[Dict]:
"""获取指定部门下的用户列表"""
all_users = []
cursor = 0
size = 100 # 每页最多100条
while True:
data = {
"dept_id": dept_id,
"cursor": cursor,
"size": size
}
response = requests.post(url, params=params, json=data)
result = response.json()
if result.get("errcode") == 0:
result_data = result.get("result", {})
users = result_data.get("list", [])
all_users.extend(users)
# 检查是否有更多数据
if not result_data.get("has_more", False):
break
cursor = result_data.get("next_cursor", 0)
return all_users
分页处理要点:
- 使用
cursor进行分页 - 每页最多获取 100 条数据
- 通过
has_more判断是否还有下一页 - 使用
next_cursor获取下一页的游标
4. 完整同步流程
def get_all_users(self) -> List[Dict]:
"""获取所有用户信息"""
departments = self.get_all_departments()
all_users = []
# 获取根部门用户
root_users = self.get_department_users(1)
all_users.extend(root_users)
# 获取所有子部门的用户
for dept in departments:
dept_id = dept.get("dept_id")
users = self.get_department_users(dept_id)
all_users.extend(users)
return all_users
4.2 Flask Web 后端
使用 Flask 框架构建 RESTful API,提供数据查询和更新接口。
1. 基础配置
from flask import Flask, render_template, jsonify, request
import json
import os
app = Flask(__name__)
# 数据文件路径
DATA_DIR = os.path.dirname(os.path.abspath(__file__))
DEPARTMENTS_FILE = os.path.join(DATA_DIR, 'departments.json')
USERS_FILE = os.path.join(DATA_DIR, 'users.json')
2. 部门列表 API (带人数统计)
@app.route('/api/departments')
def get_departments():
"""API - 获取所有部门"""
departments = load_json(DEPARTMENTS_FILE)
users = load_json(USERS_FILE)
# 计算每个部门的人数
dept_user_count = {}
for user in users:
dept_ids = user.get('dept_id_list', [])
for dept_id in dept_ids:
dept_user_count[dept_id] = dept_user_count.get(dept_id, 0) + 1
# 为每个部门添加人数统计
for dept in departments:
dept_id = dept.get('dept_id')
dept['user_count'] = dept_user_count.get(dept_id, 0)
return jsonify({
'success': True,
'data': departments,
'total': len(departments)
})
设计思路:
- 同时加载部门和用户数据
- 统计每个部门的用户数量
- 将人数信息附加到部门对象中
3. 按部门查询用户
@app.route('/api/users/by-department/<int:dept_id>')
def get_users_by_department(dept_id):
"""API - 获取指定部门的用户"""
users = load_json(USERS_FILE)
dept_users = [u for u in users if dept_id in u.get('dept_id_list', [])]
return jsonify({
'success': True,
'data': dept_users,
'total': len(dept_users)
})
4. 更新用户信息
@app.route('/api/user/<string:user_id>', methods=['PUT'])
def update_user(user_id):
"""API - 更新用户信息"""
users = load_json(USERS_FILE)
for user in users:
if user.get('userid') == user_id:
data = request.json
# 更新允许的字段
if 'name' in data:
user['name'] = data['name']
if 'mobile' in data:
user['mobile'] = data['mobile']
if 'email' in data:
user['email'] = data['email']
if 'job_number' in data:
user['job_number'] = data['job_number']
if 'title' in data:
user['title'] = data['title']
save_json(USERS_FILE, users)
return jsonify({'success': True, 'message': '用户更新成功'})
return jsonify({'success': False, 'message': '用户不存在'}), 404
5. 触发数据同步
@app.route('/api/sync', methods=['POST'])
def sync_data():
"""API - 触发数据同步"""
try:
import subprocess
result = subprocess.run(
['python3', 'dingtalk_sync.py'],
capture_output=True,
text=True,
cwd=DATA_DIR
)
if result.returncode == 0:
return jsonify({
'success': True,
'message': '同步成功',
'output': result.stdout
})
else:
return jsonify({
'success': False,
'message': '同步失败',
'error': result.stderr
}), 500
except Exception as e:
return jsonify({
'success': False,
'message': f'同步异常: {str(e)}'
}), 500
实现原理:
- 通过
subprocess调用同步脚本 - 捕获标准输出和错误输出
- 返回执行结果给前端
4.3 前端界面实现
前端使用原生 HTML + CSS + JavaScript,无需额外的前端框架,轻量且高效。
1. 页面布局
<div class="container">
<header>
<h1>钉钉部门和人员管理系统</h1>
<div class="header-actions">
<button class="btn btn-success" onclick="syncData()">同步数据</button>
<button class="btn btn-primary" onclick="refreshData()">刷新</button>
</div>
</header>
<div class="stats-grid">
<!-- 统计卡片 -->
</div>
<div class="main-content">
<div class="panel">
<!-- 部门列表 -->
</div>
<div class="panel">
<!-- 人员列表 -->
</div>
</div>
</div>
2. 加载统计数据
async function loadStatistics() {
const response = await fetch('/api/statistics');
const result = await response.json();
if (result.success) {
const stats = result.data;
document.getElementById('statsGrid').innerHTML = `
<div class="stat-card">
<h3>部门总数</h3>
<div class="value">${stats.total_departments}</div>
</div>
<div class="stat-card">
<h3>人员总数</h3>
<div class="value">${stats.total_users}</div>
</div>
<div class="stat-card">
<h3>平均部门人数</h3>
<div class="value">${Math.round(stats.total_users / stats.total_departments)}</div>
</div>
`;
}
}
3. 加载所有用户 (默认显示)
async function loadAllUsers() {
const response = await fetch('/api/users');
const result = await response.json();
if (result.success) {
allUsers = result.data;
filteredUsers = [...allUsers];
renderUsers(filteredUsers);
}
}
4. 按部门筛选用户
async function selectDepartment(deptId) {
currentDeptId = deptId;
const response = await fetch(`/api/users/by-department/${deptId}`);
const result = await response.json();
if (result.success) {
allUsers = result.data;
filteredUsers = [...allUsers];
renderUsers(filteredUsers);
}
}
5. 搜索功能
function filterUsers() {
const keyword = document.getElementById('userSearch').value.toLowerCase();
filteredUsers = allUsers.filter(user => {
return (user.name && user.name.toLowerCase().includes(keyword)) ||
(user.job_number && user.job_number.toLowerCase().includes(keyword)) ||
(user.mobile && user.mobile.includes(keyword));
});
renderUsers(filteredUsers);
}
搜索逻辑:
- 支持按姓名搜索
- 支持按工号搜索
- 支持按手机号搜索
- 实时过滤,无需刷新页面
6. 编辑用户
function editUser(userId) {
const user = allUsers.find(u => u.userid === userId);
document.getElementById('editUserId').value = user.userid;
document.getElementById('editName').value = user.name || '';
document.getElementById('editJobNumber').value = user.job_number || '';
document.getElementById('editTitle').value = user.title || '';
document.getElementById('editMobile').value = user.mobile || '';
document.getElementById('editEmail').value = user.email || '';
document.getElementById('editModal').classList.add('show');
}
// 保存用户
document.getElementById('editForm').addEventListener('submit', async function(e) {
e.preventDefault();
const userId = document.getElementById('editUserId').value;
const data = {
name: document.getElementById('editName').value,
job_number: document.getElementById('editJobNumber').value,
title: document.getElementById('editTitle').value,
mobile: document.getElementById('editMobile').value,
email: document.getElementById('editEmail').value
};
const response = await fetch(`/api/user/${userId}`, {
method: 'PUT',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(data)
});
const result = await response.json();
if (result.success) {
showToast('保存成功', 'success');
}
});
五、项目部署
5.1 安装依赖
# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate
# 安装依赖包
pip install requests flask
5.2 配置钉钉应用
- 登录 钉钉开放平台
- 创建应用,获取以下信息:
- CorpId: 企业 ID
- AppKey: 应用唯一标识
- AppSecret: 应用密钥
- 配置应用权限:
- 通讯录只读权限
- 部门管理权限
- 员工信息权限
5.3 修改配置文件
编辑 dingtalk_sync.py,填入你的钉钉应用信息:
CORP_ID = "your_corp_id"
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
5.4 运行项目
# 首次同步数据
python3 dingtalk_sync.py
# 启动 Web 服务
python3 app.py
访问 http://localhost:5001 即可使用管理系统。
六、功能展示
6.1 首页概览
- 顶部统计卡片: 显示部门总数、人员总数、平均部门人数
- 左侧部门列表: 展示所有部门及人数,支持搜索
- 右侧人员列表: 默认显示所有人员,支持筛选和搜索
6.2 部门筛选
点击左侧部门,右侧自动显示该部门的所有人员。
6.3 人员编辑
点击"编辑"按钮,弹出模态框修改人员信息:
- 姓名
- 工号
- 职位
- 手机号
- 邮箱
6.4 数据同步
点击"同步数据"按钮,自动从钉钉API获取最新组织架构数据。
七、技术亮点
7.1 分页处理
钉钉用户列表接口采用游标分页,系统自动处理所有分页数据,确保获取完整信息。
7.2 数据缓存
- Access Token 自动缓存,避免重复请求
- 部门人数统计在服务端计算,减轻前端压力
7.3 响应式设计
使用 CSS Grid 和 Flexbox 布局,适配不同屏幕尺寸。
7.4 实时搜索
前端搜索无需刷新页面,通过 JavaScript 实时过滤数据。
7.5 错误处理
- API 请求失败时显示友好提示
- 数据加载异常时显示加载状态
- 表单提交失败时回滚数据
八、扩展功能建议
基于当前系统,还可以扩展以下功能:
- 定时同步: 使用 cron 或 celery 实现定时自动同步
- 权限管理: 添加用户登录和权限控制
- 数据导出: 支持导出 Excel 或 CSV 格式
- 组织架构树: 以树形结构展示部门层级关系
- 数据统计图表: 使用 ECharts 展示部门人员分布
- 操作日志: 记录所有编辑操作
- 批量导入: 支持批量导入人员信息
- 消息通知: 人员变动时发送通知
九、常见问题
Q1: 获取 Access Token 失败?
A: 检查 AppKey 和 AppSecret 是否正确,确保应用有足够的权限。
Q2: 部门人数显示为 0?
A: 确保已执行同步脚本,生成了 departments.json 和 users.json 文件。
Q3: 端口被占用?
A: 修改 app.py 中的端口号,例如改为 5001:
app.run(host='0.0.0.0', port=5001, debug=True)
Q4: 如何部署到生产环境?
A: 建议使用 Gunicorn + Nginx:
pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:5000 app:app
十、总结
本文详细介绍了一个完整的钉钉部门和人员管理系统的开发过程。系统采用 Python + Flask 技术栈,实现了数据同步、Web管理、人员编辑等核心功能。
项目优势:
- 代码简洁,易于理解和维护
- 功能完整,满足日常管理需求
- 扩展性强,可根据需求添加新功能
- 部署简单,无需复杂配置
适用场景:
- 企业组织架构管理
- 人员信息统计和分析
- 钉钉数据本地化存储
- 自动化办公流程集成
希望本文对你有所帮助!如果觉得不错,欢迎点赞和收藏。有任何问题或建议,欢迎在评论区留言交流。
完整源码: 项目所有代码已开源,可根据实际需求进行修改和扩展。
相关链接:
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
11
0- 0
已为社区贡献1条内容
所有评论(0)