如何写一个脚本自动提取并分类

wen 实用脚本 4

本文目录导读:

如何写一个脚本自动提取并分类

  1. 基础文件分类脚本
  2. 数据提取与分类脚本
  3. 智能分类脚本(基于内容)
  4. 快速使用指南
  5. 自定义配置

我来帮你设计一个通用的自动提取和分类脚本,以下是一个基于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

自定义配置

你可以根据需求调整:

  1. 添加新的分类:修改 FileClassifierfile_categories 字典
  2. 自定义提取规则:修改 DataExtractorpatterns 字典
  3. 调整智能分类规则:修改 SmartClassifierclassification_rules 字典

这个脚本框架可以根据你的具体需求进行扩展和修改,需要我针对特定场景(如邮件分类、日志分析等)提供更具体的实现吗?

抱歉,评论功能暂时关闭!