档案管理系统数据权限控制：可直接落地的实操解决方案

档案管理系统数据权限控制实操指南

一、前置环境搭建

需准备Python 3.8+、MySQL 5.7+，执行以下命令安装依赖包：

pip install flask flask-sqlalchemy flask-login mysql-connector-python

创建项目目录与核心文件：

mkdir archive_system && cd archive_system && touch app.py schema.sql

二、数据库表结构设计

在schema.sql中写入以下完整建表语句，包含用户、角色、用户角色关联、档案四个核心表：

CREATE DATABASE IF NOT EXISTS archive_db;
USE archive_db;
-- 用户表
CREATE TABLE user (
id INT PRIMARY KEY AUTO_INCREMENT,
username VARCHAR(50) UNIQUE NOT NULL,
password VARCHAR(100) NOT NULL
);
-- 角色表（关联部门）
CREATE TABLE role (
id INT PRIMARY KEY AUTO_INCREMENT,
role_name VARCHAR(50) NOT NULL,
department VARCHAR(50) NOT NULL COMMENT '角色对应的部门，超级管理员设为'
);
-- 用户-角色关联表（多对多）
CREATE TABLE user_role (
user_id INT NOT NULL,
role_id INT NOT NULL,
PRIMARY KEY(user_id, role_id),
FOREIGN KEY(user_id) REFERENCES user(id),
FOREIGN KEY(role_id) REFERENCES role(id)
);
-- 档案表（含所属部门）
CREATE TABLE archive (
id INT PRIMARY KEY AUTO_INCREMENT,
title VARCHAR(100) NOT NULL,
content TEXT,
department VARCHAR(50) NOT NULL COMMENT '档案所属部门'
);

执行建库建表：mysql -u root -p < schema.sql，输入MySQL密码即可完成。

三、核心逻辑代码实现（app.py）

写入以下完整代码，实现权限控制的核心逻辑：

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, login_user, login_required, current_user, UserMixin
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key_123'   替换为自定义密钥
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqlconnector://root:你的MySQL密码@localhost/archive_db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
login_manager = LoginManager()
login_manager.init_app(app)
模型定义
class User(UserMixin, db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
password = db.Column(db.String(100), nullable=False)
class Role(db.Model):
id = db.Column(db.Integer, primary_key=True)
role_name = db.Column(db.String(50), nullable=False)
department = db.Column(db.String(50), nullable=False)
class UserRole(db.Model):
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), primary_key=True)
role_id = db.Column(db.Integer, db.ForeignKey('role.id'), primary_key=True)
class Archive(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
content = db.Column(db.Text)
department = db.Column(db.String(50), nullable=False)
加载用户
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
获取当前用户可访问的部门
def get_allowed_departments():
user_roles = UserRole.query.filter_by(user_id=current_user.id).all()
departments = []
for ur in user_roles:
role = Role.query.get(ur.role_id)
if role.department == '':   超级管理员
return ['']
departments.append(role.department)
return departments
登录接口
@app.route('/api/login', methods=['POST'])
def login():
data = request.get_json()
user = User.query.filter_by(username=data.get('username')).first()
if user and user.password == data.get('password'):
login_user(user)
return jsonify({'code':200, 'msg':'登录成功'})
return jsonify({'code':401, 'msg':'账号或密码错误'})
获取档案接口（核心权限控制）
@app.route('/api/archives', methods=['GET'])
@login_required
def get_archives():
allowed_depts = get_allowed_departments()
重点：仅返回用户权限内的档案，禁止漏写过滤逻辑
if '' in allowed_depts:
archives = Archive.query.all()
else:
archives = Archive.query.filter(Archive.department.in_(allowed_depts)).all()
return jsonify({'code':200, 'data':[{'id':a.id, 'title':a.title, 'dept':a.department} for a in archives]})
初始化测试数据（首次运行执行一次）
@app.route('/api/init', methods=['POST'])
def init_data():
if not Role.query.first():
role_super = Role(role_name='超级管理员', department='')
role_hr = Role(role_name='人事部管理员', department='人事部')
role_finance = Role(role_name='财务部管理员', department='财务部')
db.session.add_all([role_super, role_hr, role_finance])
db.session.commit()
if not User.query.filter_by(username='admin').first():
admin = User(username='admin', password='123456')
user_hr = User(username='hr_user', password='123456')
user_fin = User(username='fin_user', password='123456')
db.session.add_all([admin, user_hr, user_fin])
db.session.commit()
db.session.add_all([
UserRole(user_id=admin.id, role_id=1),
UserRole(user_id=user_hr.id, role_id=2),
UserRole(user_id=user_fin.id, role_id=3)
])
db.session.add_all([
Archive(title='人事部2024招聘计划', content='含校招/社招名额', department='人事部'),
Archive(title='财务部Q2预算表', content='部门经费预算明细', department='财务部'),
Archive(title='全员考勤通知', content='月度考勤规则调整', department='人事部')
])
db.session.commit()
return jsonify({'code':200, 'msg':'初始化完成'})
if __name__ == '__main__':
with app.app_context():
db.create_all()
app.run(debug=True)

替换代码中你的MySQL密码为实际数据库密码，SECRET_KEY可自定义修改。

四、权限测试验证

1. 启动服务：执行python app.py，服务默认运行在http://127.0.0.1:5000

2. 初始化测试数据：用Postman或curl调用POST接口：curl -X POST http://127.0.0.1:5000/api/init

3. 权限验证步骤：

• 登录hr_user（密码123456），调用/archives接口，仅返回人事部的2条档案；
• 登录fin_user，调用/archives接口，仅返回财务部的1条档案；
• 登录admin，调用/archives接口，返回全部3条档案，验证超级管理员权限。

五、生产环境适配

1. 将代码中debug=True改为False，禁止生产环境开启调试模式；

2. 用户密码需加密，示例用明文，生产环境需替换为bcrypt或werkzeug的加密工具；

3. 新增/修改档案接口需加入权限校验，确保仅对应角色可操作本部门档案。