一、环境准备与依赖安装
在开始构建档案监督系统之前，必须确保运行环境满足以下要求。本系统基于Python 3.8开发，利用其强大的文件处理能力和正则表达式库来实现对档案合规性的自动化检查。
检查Python版本。打开终端或命令提示符，输入以下命令：
```bash
python3 --version
```
如果版本低于3.8，请前往Python官网下载最新版安装包并完成安装。接着，我们需要安装两个核心第三方库：watchdog（用于实时监控文件变动）和pandas（用于生成审计报告）。执行以下安装命令：
```bash
pip install watchdog pandas openpyxl
```
安装完成后，在本地创建一个项目文件夹，例如命名为 archive_audit_system。在该文件夹下新建以下目录结构以支撑系统运行：
- data/：存放待审计的档案文件。
- rules/：存放合规性规则配置文件。
- logs/：存放审计结果日志。
- audit_system.py：核心脚本文件。
二、定义合规性检查规则
档案制度落实不到位的核心原因往往是“无法量化”标准。我们将制度转化为可执行的代码规则。在 rules 目录下创建一个名为 config.json 的配置文件。该文件定义了档案命名规范、必须包含的元数据字段以及审批流程的状态检查。
请复制以下完整的JSON配置内容，直接保存到 config.json 中：
```json
{
"audit_rules": {
"naming_convention": {
"enabled": true,
"pattern": "^(\\d{4})-(\\d{2})-(\\d{2})_(.?)_(\\w+)$",
"description": "文件名必须遵循：日期_部门_项目名.扩展名，例如 2023-10-01_财务部_年度决算.pdf"
},
"required_metadata": {
"enabled": true,
"fields": ["creator", "approval_id", "confidential_level"],
"description": "文件属性或关联数据库中必须包含创建人、审批单号、密级"
},
"file_size_limit": {
"enabled": true,
"max_size_mb": 50,
"description": "单个档案文件大小不得超过50MB"
}
},
"monitor_path": "../data",
"approval_database_mock": {
"valid_approvals": ["APP20231001", "APP20231002"]
}
}
```
上述配置中，naming_convention 使用正则表达式强制规范文件命名，这是解决档案乱放乱存的第一道防线；required_metadata 模拟了对审批流程的检查，确保每一个档案都有对应的审批单号，防止“未审先存”。
三、编写核心监督脚本
接下来是系统的核心部分。我们将编写一个Python脚本，它不仅扫描现有文件，还能实时监控新文件的合规性。打开 audit_system.py，输入以下完整代码：
```python
import os
import json
import re
import time
import logging
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import pandas as pd
加载配置文件
def load_config(config_path):
with open(config_path, 'r', encoding='utf-8') as f:
return json.load(f)
CONFIG = load_config('./rules/config.json')
MONITOR_PATH = os.path.abspath(CONFIG['monitor_path'])
LOG_PATH = './logs'
配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(os.path.join(LOG_PATH, 'audit_error.log'), encoding='utf-8'),
logging.StreamHandler()
]
)
class ArchiveAuditor:
def __init__(self, config):
self.config = config['audit_rules']
self.approvals = config['approval_database_mock']['valid_approvals']
def check_naming(self, filename):
"""检查文件名是否符合规范"""
if not self.config['naming_convention']['enabled']:
return True, "规则未启用"
pattern = self.config['naming_convention']['pattern']
if re.match(pattern, filename):
return True, "命名合规"
else:
return False, f"命名不符合规范: {self.config['naming_convention']['description']}"
def check_metadata(self, file_path):
"""
模拟元数据检查。
实际场景中，这里可以连接数据库或读取文件属性。
这里我们模拟：文件名中必须包含有效的审批单号（简化逻辑）
"""
if not self.config['required_metadata']['enabled']:
return True, "规则未启用"
模拟逻辑：假设文件名中包含审批单号，或者有对应的meta文件
这里为了演示，假设文件名中包含APP开头的ID
filename = os.path.basename(file_path)
found_approval = None
for app_id in self.approvals:
if app_id in filename:
found_approval = app_id
break
if found_approval:
return True, f"包含有效审批单号: {found_approval}"
else:
return False, "未找到关联的有效审批单号，涉嫌违规归档"
def check_file_size(self, file_path):
"""检查文件大小"""
if not self.config['file_size_limit']['enabled']:
return True, "规则未启用"
size_mb = os.path.getsize(file_path) / (1024 1024)
limit = self.config['file_size_limit']['max_size_mb']
if size_mb <= limit:
return True, f"大小合规: {size_mb:.2f}MB"
else:
return False, f"文件超限: {size_mb:.2f}MB (最大允许: {limit}MB)"
def audit_file(self, file_path):
"""执行单文件完整审计"""
filename = os.path.basename(file_path)
results = {
"文件名": filename,
"路径": file_path,
"审计时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"状态": "通过",
"违规详情": []
}
1. 命名检查
pass_naming, msg_naming = self.check_naming(filename)
if not pass_naming:
results["状态"] = "不通过"
results["违规详情"].append(msg_naming)
2. 元数据/审批检查
pass_meta, msg_meta = self.check_metadata(file_path)
if not pass_meta:
results["状态"] = "不通过"
results["违规详情"].append(msg_meta)
3. 大小检查
pass_size, msg_size = self.check_file_size(file_path)
if not pass_size:
results["状态"] = "不通过"
results["违规详情"].append(msg_size)
return results
class AuditEventHandler(FileSystemEventHandler):
def __init__(self, auditor):
self.auditor = auditor
self.report_data = []
def on_created(self, event):
if not event.is_directory:
logging.info(f"检测到新文件: {event.src_path}")
result = self.auditor.audit_file(event.src_path)
self.report_data.append(result)
if result["状态"] == "不通过":
logging.error(f"审计失败: {result['文件名']} - {'; '.join(result['违规详情'])}")
else:
logging.info(f"审计通过: {result['文件名']}")
def generate_report(self):
if not self.report_data:
return
df = pd.DataFrame(self.report_data)
report_file = os.path.join(LOG_PATH, f"audit_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx")
df.to_excel(report_file, index=False)
logging.info(f"审计报告已生成: {report_file}")
def run_initial_scan(auditor, handler):
"""启动时扫描现有文件"""
logging.info("正在执行存量档案初始扫描...")
for root, dirs, files in os.walk(MONITOR_PATH):
for file in files:
full_path = os.path.join(root, file)
result = auditor.audit_file(full_path)
handler.report_data.append(result)
if result["状态"] == "不通过":
logging.warning(f"存量文件违规: {result['文件名']} - {'; '.join(result['违规详情'])}")
handler.generate_report()
if __name__ == "__main__":
确保目录存在
os.makedirs(LOG_PATH, exist_ok=True)
os.makedirs(MONITOR_PATH, exist_ok=True)
auditor = ArchiveAuditor(CONFIG)
event_handler = AuditEventHandler(auditor)
先扫描一次存量
run_initial_scan(auditor, event_handler)
启动实时监控
observer = Observer()
observer.schedule(event_handler, MONITOR_PATH, recursive=True)
observer.start()
logging.info(f"档案监督系统已启动，正在监控目录: {MONITOR_PATH}")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
event_handler.generate_report()
observer.join()
```
四、模拟测试与结果验证
代码编写完成后，我们需要模拟真实的违规场景来验证系统的有效性。请严格按照以下步骤进行操作。
1. 创建合规测试文件
在 data 文件夹下，创建一个符合规则的文本文件。文件名必须严格匹配正则表达式 ^(\\d{4})-(\\d{2})-(\\d{2})_(.?)_(\\w+)$，并且为了通过元数据检查，文件名中需要包含配置文件中定义的有效审批单号（如 APP20231001）。
- 操作：在
data 目录下新建一个文件，命名为 2023-10-27_技术部_研发文档_APP20231001.txt。
- 内容：随意输入一些文本。
2. 创建违规测试文件
为了测试监督系统的拦截能力，我们创建几个典型的违规文件。
- 违规案例1（命名不规范）： 创建文件 随便起个名字.pdf。这应该触发“命名不符合规范”的报警。
- 违规案例2（无审批单号）： 创建文件 2023-10-27_市场部_宣传计划.txt。虽然命名格式正确，但缺少
APP... 审批单号，应触发“未找到关联的有效审批单号”报警。
- 违规案例3（文件过大）： 我们可以通过修改配置文件来快速测试此功能。将
config.json 中的 max_size_mb 改为 0.0001（极小值）。然后创建一个大于该大小的文件，系统应报警“文件超限”。测试完毕后记得改回50MB。
3. 运行系统并观察日志
在项目根目录下，打开终端运行脚本：
```bash
python audit_system.py
```
观察终端输出。你应该能看到系统首先进行了“存量档案初始扫描”，对刚才放入的文件进行了审计。对于合规文件，显示“审计通过”；对于违规文件，显示红色或醒目的“审计失败”及具体原因。
接着，保持脚本运行状态，尝试再次向 data 文件夹拖入一个新的违规文件。你会看到日志实时刷新，捕捉到了这次违规操作。这证明了监督机制是实时生效的。
4. 查看审计报告
按下 Ctrl+C 停止脚本，或者等待一段时间后查看 logs 目录。系统会自动生成一个 Excel 文件，文件名包含时间戳，例如 audit_report_20231027_103000.xlsx。
打开该Excel文件，你将看到一张结构化的表格，列出了所有被审计的文件、审计时间、状态以及详细的违规原因。这份报告可以直接作为档案制度整改的依据，发给相关责任人进行整改。
五、部署与自动化运行
为了确保监督系统长期有效，不能依赖手动开启终端。我们需要将其部署为后台服务。
1. Linux/Mac 环境部署 (使用 Systemd)
创建一个服务文件 /etc/systemd/system/archive_audit.service：
```ini
[Unit]
Description=Archive Audit System
After=network.target
[Service]
User=root
WorkingDirectory=/your/path/to/archive_audit_system
ExecStart=/usr/bin/python3 /your/path/to/archive_audit_system/audit_system.py
Restart=always
[Install]
WantedBy=multi-user.target
```
执行以下命令启动服务并设置开机自启：
```bash
sudo systemctl daemon-reload
sudo systemctl start archive_audit
sudo systemctl enable archive_audit
```
2. Windows 环境部署 (使用任务计划程序)
- 打开“任务计划程序”，点击“创建基本任务”。
- 名称：档案合规监督，触发器：启动时。
- 操作：启动程序。
- 程序/脚本：选择你的
python.exe 路径（例如 C:\Python39\python.exe）。
- 添加参数：输入脚本的完整路径（例如
D:\archive_audit_system\audit_system.py）。
- 起始于：输入脚本所在的目录（例如
D:\archive_audit_system\）。
- 完成设置，并在“属性”中勾选“不管用户是否登录都要运行”。
通过以上步骤，你已经完全拥有了一套自动化的档案制度监督系统。它通过代码将抽象的“制度”固化为不可逾越的“技术关卡”，任何不合规的档案归档行为都会被实时抓取并记录，彻底解决了落实不到位的问题。
微信咨询
电话联系
QQ客服
微信咨询一对一服务