本文目录导读:

我来帮你设计一个通用的自动提取和分类脚本,以下是一个基于Python的完整示例,可以根据你的实际需求调整。
基础文件分类脚本
场景:自动提取并分类文件(按扩展名)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
文件自动提取和分类工具
功能:自动提取指定目录下的文件,并按类型分类到不同文件夹
"""
import os
import shutil
import hashlib
import json
from pathlib import Path
from datetime import datetime
import argparse
class FileClassifier:
def __init__(self, source_dir, target_dir=None):
self.source_dir = Path(source_dir)
self.target_dir = Path(target_dir) if target_dir else self.source_dir / "classified_files"
self.file_categories = {
'Documents': ['.pdf', '.doc', '.docx', '.txt', '.xlsx', '.xls', '.ppt', '.pptx', '.odt'],
'Images': ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.svg', '.webp', '.tiff'],
'Audio': ['.mp3', '.wav', '.flac', '.aac', '.ogg', '.wma'],
'Video': ['.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv'],
'Archives': ['.zip', '.rar', '.tar', '.gz', '.7z', '.bz2'],
'Code': ['.py', '.js', '.html', '.css', '.java', '.cpp', '.c', '.h'],
'Data': ['.csv', '.json', '.xml', '.sql', '.db', '.sqlite'],
'Others': [] # 其他未分类的文件
}
self.stats = {'classified': 0, 'failed': 0, 'duplicates': 0}
def classify_file(self, file_path):
"""根据文件扩展名分类"""
ext = file_path.suffix.lower()
for category, extensions in self.file_categories.items():
if ext in extensions:
return category
return 'Others'
def extract_and_classify(self):
"""提取并分类所有文件"""
print(f"开始处理目录: {self.source_dir}")
print(f"目标目录: {self.target_dir}")
# 创建目标目录结构
self.target_dir.mkdir(parents=True, exist_ok=True)
# 遍历源目录
for file_path in self.source_dir.rglob('*'):
if file_path.is_file():
try:
category = self.classify_file(file_path)
dest_dir = self.target_dir / category
dest_dir.mkdir(parents=True, exist_ok=True)
# 处理重名文件
dest_file = dest_dir / file_path.name
if dest_file.exists():
# 添加时间戳避免重名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
new_name = f"{file_path.stem}_{timestamp}{file_path.suffix}"
dest_file = dest_dir / new_name
shutil.copy2(file_path, dest_file)
self.stats['classified'] += 1
except Exception as e:
print(f"处理文件失败 {file_path}: {e}")
self.stats['failed'] += 1
self.print_statistics()
def print_statistics(self):
"""输出处理统计"""
print("\n" + "="*40)
print("处理完成!统计结果:")
print(f"成功分类: {self.stats['classified']} 个文件")
print(f"处理失败: {self.stats['failed']} 个文件")
# 显示各类别文件数
for category in self.file_categories.keys():
cat_dir = self.target_dir / category
if cat_dir.exists():
count = len(list(cat_dir.iterdir()))
if count > 0:
print(f" {category}: {count} 个文件")
print("="*40)
def main():
parser = argparse.ArgumentParser(description='文件自动提取和分类工具')
parser.add_argument('source', help='源目录路径')
parser.add_argument('-d', '--dest', help='目标目录路径(可选)')
parser.add_argument('--copy', action='store_true', help='复制文件(默认移动)')
args = parser.parse_args()
classifier = FileClassifier(args.source, args.dest)
classifier.extract_and_classify()
if __name__ == "__main__":
main()
数据提取与分类脚本
场景:从文本文件中提取特定信息并分类
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据提取与分类脚本
从文本/日志文件中提取特定模式的数据并分类存储
"""
import re
from collections import defaultdict
import pandas as pd
from datetime import datetime
class DataExtractor:
def __init__(self):
self.patterns = {
'email': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
'phone': r'1[3-9]\d{9}|0\d{2,3}-\d{7,8}',
'url': r'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+[^\s]*',
'ip_address': r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
'date': r'\d{4}[-/]\d{1,2}[-/]\d{1,2}',
'chinese': r'[\u4e00-\u9fa5]+'
}
self.extracted_data = defaultdict(list)
def extract_from_text(self, text, patterns=None):
"""从文本中提取指定模式的数据"""
if patterns is None:
patterns = self.patterns.keys()
for pattern_name in patterns:
if pattern_name in self.patterns:
matches = re.findall(self.patterns[pattern_name], text)
self.extracted_data[pattern_name].extend(matches)
def extract_from_file(self, file_path, patterns=None):
"""从文件中提取数据"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
self.extract_from_text(content, patterns)
except UnicodeDecodeError:
# 尝试其他编码
with open(file_path, 'r', encoding='gbk') as f:
content = f.read()
self.extract_from_text(content, patterns)
def save_to_excel(self, output_file="extracted_data.xlsx"):
"""将提取的数据保存到Excel"""
with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
for category, items in self.extracted_data.items():
if items:
df = pd.DataFrame(items, columns=[category])
df.to_excel(writer, sheet_name=category, index=False)
print(f"数据已保存到: {output_file}")
def save_to_json(self, output_file="extracted_data.json"):
"""将提取的数据保存到JSON"""
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(self.extracted_data, f, ensure_ascii=False, indent=2)
print(f"数据已保存到: {output_file}")
# 使用示例
def process_directory(directory_path):
"""处理目录中的所有文件"""
extractor = DataExtractor()
directory = Path(directory_path)
# 支持的文件类型
supported_extensions = ['.txt', '.log', '.csv', '.md', '.html', '.xml']
for file_path in directory.rglob('*'):
if file_path.suffix.lower() in supported_extensions:
print(f"处理文件: {file_path.name}")
extractor.extract_from_file(file_path)
# 保存结果
extractor.save_to_excel()
extractor.save_to_json()
return extractor.extracted_data
智能分类脚本(基于内容)
场景:根据文件内容自动分类
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
智能文件分类器的关键词和模式进行分类
"""
import os
import json
from collections import Counter
import magic # python-magic 库需要安装
class SmartClassifier:
def __init__(self):
# 定义分类规则
self.classification_rules = {
'财务文件': ['发票', '报销', '财务报表', '预算', '税务', '账目'],
'技术文档': ['API', 'documentation', 'specification', '架构', '设计'],
'项目报告': ['项目', '进度', '报告', '#39;, '汇报', '评审'],
'客户资料': ['客户', '合同', '协议', '报价', '订单'],
'会议记录': ['会议', '纪要', '议程', 'minutes', 'action items'],
}
def analyze_content(self, file_path):
"""分析文件内容"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read().lower()
except:
# 尝试其他编码
try:
with open(file_path, 'r', encoding='gbk') as f:
content = f.read().lower()
except:
return '无法读取'
return content
def smart_classify(self, file_path):
"""智能分类"""
content = self.analyze_content(file_path)
if content == '无法读取':
return '其他'
# 计算匹配分数
scores = {}
for category, keywords in self.classification_rules.items():
score = sum(1 for keyword in keywords if keyword in content)
if score > 0:
scores[category] = score
if not scores:
return '其他'
# 返回匹配分数最高的分类
return max(scores, key=scores.get)
def batch_classify(self, directory_path):
"""批量分类文件"""
directory = Path(directory_path)
classification_results = {}
for file_path in directory.rglob('*'):
if file_path.is_file() and file_path.suffix.lower() in ['.txt', '.md', '.pdf', '.docx']:
category = self.smart_classify(file_path)
classification_results[str(file_path)] = category
return classification_results
# 主执行入口
if __name__ == "__main__":
# 使用示例
import sys
if len(sys.argv) < 2:
print("使用方法: python script.py <目录路径>")
sys.exit(1)
source_dir = sys.argv[1]
# 基础文件分类
print("开始文件分类...")
classifier = FileClassifier(source_dir)
classifier.extract_and_classify()
# 数据提取
print("\n开始数据提取...")
data_processor = DataExtractor()
data_processor.extract_from_file(source_dir)
# 智能分类
print("\n开始智能分类...")
smart_classifier = SmartClassifier()
results = smart_classifier.batch_classify(source_dir)
# 输出结果
print("\n分类结果摘要:")
for file_path, category in results.items():
print(f" {Path(file_path).name} -> {category}")
快速使用指南
安装依赖
# 安装必要的库 pip install pandas openpyxl python-magic # 或一键安装 pip install -r requirements.txt
创建 requirements.txt
pandas>=1.3.0
openpyxl>=3.0.0
python-magic>=0.4.24
argparse
运行脚本
# 基本使用 python file_classifier.py /path/to/source/directory # 指定目标目录 python file_classifier.py /path/to/source -d /path/to/destination # 复制而非移动文件 python file_classifier.py /path/to/source --copy
自定义配置
你可以根据需求调整:
- 添加新的分类:修改
FileClassifier的file_categories字典 - 自定义提取规则:修改
DataExtractor的patterns字典 - 调整智能分类规则:修改
SmartClassifier的classification_rules字典
这个脚本框架可以根据你的具体需求进行扩展和修改,需要我针对特定场景(如邮件分类、日志分析等)提供更具体的实现吗?