系统环境与工具准备
你需要一台安装了Python 3.8或更高版本的计算机。我们将使用以下核心工具包，请在命令行中逐一执行以下安装命令：
```bash
pip install pdfplumber==0.10.3 用于提取PDF文本和元数据
pip install python-magic-bin==0.4.14 用于精确识别文件类型
pip install pandas==2.0.3 用于处理结构化数据
pip install watchdog==3.0.0 用于监控文件夹变化
pip install Pillow==10.0.0 用于处理图像文件
pip install openpyxl==3.1.2 用于读取Excel文件
```
创建一个名为smart_archive的项目文件夹，并在其中建立以下目录结构：
```bash
smart_archive/
├── config/
│ └── rules.json 分类规则配置文件
├── scripts/
│ ├── file_processor.py
│ ├── monitor.py
│ └── utils.py
├── input/ 放置待整理的原始文件
├── output/ 整理后的归档目录
│ ├── 合同协议/
│ ├── 财务票据/
│ ├── 人事档案/
│ └── 其他/
└── logs/ 系统运行日志
```
定义核心分类与命名规则
在config/rules.json中，定义你的档案分类逻辑。这是一个完整的配置文件示例：
```json
{
"categories": {
"合同协议": {
"keywords": ["合同", "协议", "签约", "条款", "甲方", "乙方"],
"file_extensions": [".pdf", ".doc", ".docx", ".txt"],
"target_folder": "合同协议",
"naming_rule": "{年份}{月份}{日期}_{对方单位名称}_{合同类型}.{原扩展名}"
},
"财务票据": {
"keywords": ["发票", "收据", "报销单", "账单", "付款凭证"],
"file_extensions": [".pdf", ".jpg", ".png", ".xlsx"],
"target_folder": "财务票据",
"naming_rule": "{年份}{月份}{日期}_{票据类型}_{金额}元.{原扩展名}"
},
"人事档案": {
"keywords": ["简历", "入职表", "离职证明", "考核表", "薪资单"],
"file_extensions": [".pdf", ".docx", ".xlsx"],
"target_folder": "人事档案",
"naming_rule": "{年份}{月份}{日期}_{员工姓名}_{文档类型}.{原扩展名}"
}
},
"default_category": "其他",
"date_format_in_filename": "YYYYMMDD",
"scan_interval_seconds": 300
}
```
这个配置文件定义了三个核心档案类别，系统将根据文件内容中的关键词和文件扩展名进行自动分类。
构建核心文件处理脚本
在scripts/utils.py中，编写文件处理的基础功能函数：
```python
import os
import magic
import pdfplumber
from datetime import datetime
import json
def get_file_type(file_path):
"""精确识别文件真实类型"""
mime = magic.Magic(mime=True)
file_mime_type = mime.from_file(file_path)
return file_mime_type
def extract_text_from_file(file_path):
"""从多种格式文件中提取文本内容"""
file_type = get_file_type(file_path)
text_content = ""
if 'pdf' in file_type:
with pdfplumber.open(file_path) as pdf:
for page in pdf.pages:
text_content += page.extract_text() + "\n"
elif 'msword' in file_type or 'openxmlformats' in file_type:
处理Word文档
import docx
doc = docx.Document(file_path)
text_content = "\n".join([para.text for para in doc.paragraphs])
elif 'sheet' in file_type:
处理Excel文档
import pandas as pd
df = pd.read_excel(file_path)
text_content = df.to_string()
elif 'text/plain' in file_type:
with open(file_path, 'r', encoding='utf-8') as f:
text_content = f.read()
return text_content.lower() 转为小写便于关键词匹配
def load_config():
"""加载配置文件"""
config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'rules.json')
with open(config_path, 'r', encoding='utf-8') as f:
return json.load(f)
```
实现智能分类与重命名逻辑
在scripts/file_processor.py中，实现核心的分类处理器：
```python
import os
import re
import shutil
from datetime import datetime
from .utils import extract_text_from_file, load_config
class FileProcessor:
def __init__(self):
self.config = load_config()
self.categories = self.config['categories']
def classify_file(self, file_path):
"""对文件进行智能分类"""
if not os.path.exists(file_path):
return self.config['default_category']
提取文件内容文本
try:
content = extract_text_from_file(file_path)
except:
content = ""
获取文件名（不含路径）
filename = os.path.basename(file_path)
filename_lower = filename.lower()
优先匹配分类关键词
for category_name, rules in self.categories.items():
检查文件扩展名
file_ext = os.path.splitext(filename)[1].lower()
if file_ext not in rules['file_extensions']:
continue
检查关键词匹配
for keyword in rules['keywords']:
if keyword in content or keyword in filename_lower:
return category_name
return self.config['default_category']
def generate_new_filename(self, file_path, category):
"""根据规则生成新的文件名"""
if category not in self.categories:
return os.path.basename(file_path)
rules = self.categories[category]
original_name = os.path.basename(file_path)
original_ext = os.path.splitext(original_name)[1]
从文件名和内容中提取信息
content = extract_text_from_file(file_path)
提取日期（优先从文件名，其次从文件修改时间）
date_str = self._extract_date_from_filename(original_name)
if not date_str:
mtime = os.path.getmtime(file_path)
date_obj = datetime.fromtimestamp(mtime)
date_str = date_obj.strftime('%Y%m%d')
提取其他信息（这里以合同为例）
company_name = self._extract_company_name(content)
doc_type = self._extract_document_type(content, category)
构建新文件名
new_name = rules['naming_rule']
new_name = new_name.replace('{年份}{月份}{日期}', date_str)
new_name = new_name.replace('{对方单位名称}', company_name if company_name else '未知单位')
new_name = new_name.replace('{合同类型}', doc_type if doc_type else '未知类型')
new_name = new_name.replace('{原扩展名}', original_ext[1:]) 去掉点号
return new_name
def _extract_date_from_filename(self, filename):
"""从文件名中提取日期"""
patterns = [
r'(\d{4})[-_]?(\d{2})[-_]?(\d{2})', YYYY-MM-DD 或 YYYY_MM_DD
r'(\d{2})[-_]?(\d{2})[-_]?(\d{4})', DD-MM-YYYY
]
for pattern in patterns:
match = re.search(pattern, filename)
if match:
if len(match.group(1)) == 4: YYYY-MM-DD
return f"{match.group(1)}{match.group(2)}{match.group(3)}"
else: DD-MM-YYYY
return f"{match.group(3)}{match.group(2)}{match.group(1)}"
return None
def _extract_company_name(self, content):
"""从内容中提取公司名称（简化版）"""
这里可以扩展更复杂的NLP识别
patterns = [
r'甲方[:：]\s([^\n，。]+)',
r'乙方[:：]\s([^\n，。]+)',
r'公司名称[:：]\s([^\n，。]+)',
]
for pattern in patterns:
match = re.search(pattern, content)
if match:
return match.group(1).strip()
return None
def _extract_document_type(self, content, category):
"""根据分类提取文档类型"""
type_mapping = {
'合同协议': ['采购合同', '服务协议', '租赁合同', '合作协议'],
'财务票据': ['增值税发票', '普通发票', '收据', '报销单'],
'人事档案': ['入职申请表', '离职证明', '绩效考核表', '薪资确认单']
}
if category in type_mapping:
for doc_type in type_mapping[category]:
if doc_type in content:
return doc_type
return '其他'
def process_file(self, file_path):
"""处理单个文件：分类、重命名、移动"""
1. 分类
category = self.classify_file(file_path)
2. 生成新文件名
new_filename = self.generate_new_filename(file_path, category)
3. 确定目标路径
if category in self.categories:
target_folder = self.categories[category]['target_folder']
else:
target_folder = self.config['default_category']
output_root = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'output')
target_path = os.path.join(output_root, target_folder)
4. 创建目标文件夹（如果不存在）
os.makedirs(target_path, exist_ok=True)
5. 移动并重命名文件
final_path = os.path.join(target_path, new_filename)
处理文件名冲突
counter = 1
base_name, ext = os.path.splitext(new_filename)
while os.path.exists(final_path):
final_path = os.path.join(target_path, f"{base_name}_{counter}{ext}")
counter += 1
shutil.move(file_path, final_path)
6. 记录处理结果
self._log_processing(file_path, final_path, category)
return final_path, category
def _log_processing(self, original_path, new_path, category):
"""记录处理日志"""
log_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'logs')
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, 'processing.log')
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_entry = f"{timestamp} | 原文件: {original_path} | 新位置: {new_path} | 分类: {category}\n"
with open(log_file, 'a', encoding='utf-8') as f:
f.write(log_entry)
```
设置自动化文件夹监控
在scripts/monitor.py中，创建自动监控脚本：
```python
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from file_processor import FileProcessor
import os
class ArchiveHandler(FileSystemEventHandler):
def __init__(self):
self.processor = FileProcessor()
self.processed_files = set()
def on_created(self, event):
"""当有新文件创建时触发"""
if not event.is_directory:
等待文件完全写入
time.sleep(1)
if event.src_path not in self.processed_files:
try:
print(f"开始处理新文件: {event.src_path}")
new_path, category = self.processor.process_file(event.src_path)
print(f"文件已归档: {new_path} (分类: {category})")
self.processed_files.add(event.src_path)
except Exception as e:
print(f"处理文件失败 {event.src_path}: {str(e)}")
def start_monitoring():
"""启动文件夹监控"""
input_folder = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'input')
if not os.path.exists(input_folder):
os.makedirs(input_folder)
print(f"已创建监控文件夹: {input_folder}")
event_handler = ArchiveHandler()
observer = Observer()
observer.schedule(event_handler, input_folder, recursive=False)
observer.start()
print(f"开始监控文件夹: {input_folder}")
print("系统已启动，将自动处理放入input文件夹的所有文件")
print("按 Ctrl+C 停止监控")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
if __name__ == "__main__":
start_monitoring()
```
批量处理已有档案
创建scripts/batch_process.py用于批量处理已有文件：
```python
import os
from file_processor import FileProcessor
def batch_process_existing_files(source_folder):
"""批量处理已有文件夹中的所有文件"""
processor = FileProcessor()
if not os.path.exists(source_folder):
print(f"源文件夹不存在: {source_folder}")
return
支持的文件扩展名
supported_extensions = set()
for rules in processor.categories.values():
supported_extensions.update(rules['file_extensions'])
遍历所有文件
processed_count = 0
for root, dirs, files in os.walk(source_folder):
for file in files:
file_path = os.path.join(root, file)
file_ext = os.path.splitext(file)[1].lower()
if file_ext in supported_extensions:
try:
new_path, category = processor.process_file(file_path)
print(f"已处理: {file} -> {new_path} ({category})")
processed_count += 1
except Exception as e:
print(f"处理失败 {file}: {str(e)}")
print(f"批量处理完成，共处理 {processed_count} 个文件")
if __name__ == "__main__":
指定要批量处理的文件夹路径
source_dir = r"C:\Users\YourName\Documents\待整理档案"
batch_process_existing_files(source_dir)
```
系统部署与使用
启动自动化监控服务
打开命令行，切换到项目目录，执行：
```bash
cd /path/to/smart_archive/scripts
python monitor.py
```
系统启动后，将所有需要整理的档案文件放入smart_archive/input/文件夹，系统将自动完成分类、重命名和归档。
手动执行批量处理
对于已有的大量文件，使用批量处理脚本：
```bash
cd /path/to/smart_archive/scripts
python batch_process.py
```
在脚本中修改source_dir变量指向你的档案文件夹。
查看处理日志
所有处理记录保存在smart_archive/logs/processing.log中，格式如下：
```bash
2024-01-15 10:30:25 | 原文件: input/采购合同20231215.pdf | 新位置: output/合同协议/20231215_XX公司_采购合同.pdf | 分类: 合同协议
```
自定义扩展与优化
添加新的档案类别
在config/rules.json的categories部分添加新类别：
```json
"项目文档": {
"keywords": ["项目计划", "需求文档", "设计稿", "测试报告", "验收单"],
"file_extensions": [".pdf", ".docx", ".pptx", ".xmind"],
"target_folder": "项目文档",
"naming_rule": "{项目编号}_{文档类型}_{版本号}.{原扩展名}"
}
```
优化关键词匹配
在file_processor.py的_extract_document_type方法中添加新的类型映射：
```python
type_mapping = {
... 原有映射
'项目文档': ['项目计划书', '需求规格说明书', 'UI设计稿', '测试用例', '项目周报']
}
```
添加邮件通知功能
在file_processor.py的_log_processing方法后添加：
```python
def send_notification(self, original_file, new_location, category):
"""发送处理完成通知（可选）"""
import smtplib
from email.mime.text import MIMEText
配置SMTP服务器
smtp_server = "smtp.your-email-provider.com"
smtp_port = 587
sender_email = "your-email@example.com"
receiver_email = "admin@example.com"
password = "your-email-password"
创建邮件内容
subject = f"档案整理完成：{os.path.basename(original_file)}"
body = f"""
文件整理已完成：
原文件：{original_file}
新位置：{new_location}
分类：{category}
处理时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
msg = MIMET
