data_processing_common.py

import os
import re
import datetime  # 导入datetime用于日期操作
from rich.progress import Progress, TextColumn, BarColumn, TimeElapsedColumn


def sanitize_filename(name, max_length=50, max_words=5):
    """
    清理文件名，移除不需要的词汇和字符

    参数:
        name: 原始文件名
        max_length: 文件名最大长度，默认50
        max_words: 最大单词数，默认5

    返回:
        清理后的文件名
    """
    # 移除文件扩展名（如果存在）
    name = os.path.splitext(name)[0]

    # 移除不需要的词汇和数据类型词汇
    name = re.sub(
        r'\b(jpg|jpeg|png|gif|bmp|txt|md|pdf|docx|xls|xlsx|csv|ppt|pptx|image|picture|photo|this|that|these|those|here|there|'
        r'please|note|additional|notes|folder|name|sure|heres|a|an|the|and|of|in|'
        r'to|for|on|with|your|answer|should|be|only|summary|summarize|text|category)\b',
        '',
        name,
        flags=re.IGNORECASE
    )

    # 移除非单词字符，保留下划线
    sanitized = re.sub(r'[^\w\s]', '', name).strip()

    # 将多个下划线或空格替换为单个下划线
    sanitized = re.sub(r'[\s_]+', '_', sanitized)

    # 转换为小写
    sanitized = sanitized.lower()

    # 移除首尾的下划线
    sanitized = sanitized.strip('_')

    # 分割为单词并限制单词数量
    words = sanitized.split('_')
    limited_words = [word for word in words if word]  # 移除空字符串
    limited_words = limited_words[:max_words]
    limited_name = '_'.join(limited_words)

    # 限制长度
    return limited_name[:max_length] if limited_name else 'untitled'


def process_files_by_date(file_paths, output_path, dry_run=False, silent=False, log_file=None):
    """
    按日期处理文件，将文件组织到年份/月份文件夹中

    参数:
        file_paths: 文件路径列表
        output_path: 输出路径
        dry_run: 是否为模拟运行
        silent: 是否静默模式
        log_file: 日志文件路径

    返回:
        操作列表
    """
    operations = []
    for file_path in file_paths:
        # 获取文件修改时间
        mod_time = os.path.getmtime(file_path)

        # 转换为datetime对象
        mod_datetime = datetime.datetime.fromtimestamp(mod_time)
        year = mod_datetime.strftime('%Y')  # 年份
        month = mod_datetime.strftime('%B')  # 月份名称（如'January'），使用'%m'可获取月份数字

        # 创建目录路径
        dir_path = os.path.join(output_path, year, month)

        # 准备新文件路径
        new_file_name = os.path.basename(file_path)
        new_file_path = os.path.join(dir_path, new_file_name)

        # 决定使用硬链接还是软链接
        link_type = 'hardlink'  # 暂时默认使用硬链接

        # 记录操作
        operation = {
            'source': file_path,
            'destination': new_file_path,
            'link_type': link_type,
        }
        operations.append(operation)

    return operations


def process_files_by_type(file_paths, output_path, dry_run=False, silent=False, log_file=None):
    """
    按文件类型处理文件，首先分为文本类和图片类文件

    参数:
        file_paths: 文件路径列表
        output_path: 输出路径
        dry_run: 是否为模拟运行
        silent: 是否静默模式
        log_file: 日志文件路径

    返回:
        操作列表
    """
    operations = []

    # 定义文件扩展名
    image_extensions = ('.png', '.jpg', '.jpeg', '.gif', '.bmp', '.tiff')  # 图片扩展名
    text_extensions = ('.txt', '.md', '.docx', '.doc', '.pdf', '.xls', '.xlsx', '.epub', '.mobi', '.azw',
                       '.azw3')  # 文本扩展名

    for file_path in file_paths:
        # 排除隐藏文件（额外安全检查）
        if os.path.basename(file_path).startswith('.'):
            continue

        # 获取文件扩展名
        ext = os.path.splitext(file_path)[1].lower()

        # 检查是否为图片文件
        if ext in image_extensions:
            # 图片类文件
            top_folder = '图片文件'
            # 如需要可以在此添加子分类
            folder_name = top_folder

        elif ext in text_extensions:
            # 文本类文件
            top_folder = '文本文件'
            # 将扩展名映射到子文件夹
            if ext in ('.txt', '.md'):
                sub_folder = '纯文本文件'
            elif ext in ('.doc', '.docx'):
                sub_folder = 'Word文档'
            elif ext == '.pdf':
                sub_folder = 'PDF文件'
            elif ext in ('.xls', '.xlsx'):
                sub_folder = 'Excel文件'
            elif ext in ('.epub', '.mobi', '.azw', '.azw3'):
                sub_folder = '电子书'
            else:
                sub_folder = '其他'
            folder_name = os.path.join(top_folder, sub_folder)

        else:
            # 其他类型
            folder_name = '其他文件'

        # 创建目录路径
        dir_path = os.path.join(output_path, folder_name)

        # 准备新文件路径
        new_file_name = os.path.basename(file_path)
        new_file_path = os.path.join(dir_path, new_file_name)

        # 决定使用硬链接还是软链接
        link_type = 'hardlink'  # 暂时默认使用硬链接

        # 记录操作
        operation = {
            'source': file_path,
            'destination': new_file_path,
            'link_type': link_type,
        }
        operations.append(operation)

    return operations


def compute_operations(data_list, new_path, renamed_files, processed_files):
    """
    基于生成的元数据计算文件操作

    参数:
        data_list: 数据列表，包含文件路径、文件夹名、文件名等信息
        new_path: 新路径
        renamed_files: 已重命名文件集合
        processed_files: 已处理文件集合

    返回:
        操作列表，用于显示或进一步处理
    """
    operations = []
    for data in data_list:
        file_path = data['file_path']

        # 跳过已处理的文件
        if file_path in processed_files:
            continue
        processed_files.add(file_path)

        # 准备文件夹名和文件名
        folder_name = data['foldername']
        new_file_name = data['filename'] + os.path.splitext(file_path)[1]

        # 准备新文件路径
        dir_path = os.path.join(new_path, folder_name)
        new_file_path = os.path.join(dir_path, new_file_name)

        # 处理重复文件
        counter = 1
        original_new_file_name = new_file_name
        while new_file_path in renamed_files:
            new_file_name = f"{data['filename']}_{counter}" + os.path.splitext(file_path)[1]
            new_file_path = os.path.join(dir_path, new_file_name)
            counter += 1

        # 决定使用硬链接还是软链接
        link_type = 'hardlink'  # 暂时默认使用硬链接

        # 记录操作
        operation = {
            'source': file_path,
            'destination': new_file_path,
            'link_type': link_type,
            'folder_name': folder_name,
            'new_file_name': new_file_name
        }
        operations.append(operation)
        renamed_files.add(new_file_path)

    return operations  # 返回操作列表用于显示或进一步处理


def execute_operations(operations, dry_run=False, silent=False, log_file=None):
    """
    执行文件操作

    参数:
        operations: 操作列表
        dry_run: 是否为模拟运行（不实际执行）
        silent: 是否静默模式（不输出到控制台）
        log_file: 日志文件路径
    """
    total_operations = len(operations)

    # 创建进度条
    with Progress(
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TimeElapsedColumn(),
            transient=True
    ) as progress:
        task = progress.add_task("正在整理文件...", total=total_operations)

        for operation in operations:
            source = operation['source']
            destination = operation['destination']
            link_type = operation['link_type']
            dir_path = os.path.dirname(destination)

            if dry_run:
                # 模拟运行模式
                message = f"模拟运行: 将创建{link_type}从 '{source}' 到 '{destination}'"
            else:
                # 在执行操作前确保目录存在
                os.makedirs(dir_path, exist_ok=True)

                try:
                    if link_type == 'hardlink':
                        os.link(source, destination)
                        message = f"成功创建硬链接: '{source}' -> '{destination}'"
                    else:
                        os.symlink(source, destination)
                        message = f"成功创建软链接: '{source}' -> '{destination}'"
                except Exception as e:
                    message = f"错误: 创建{link_type}失败 '{source}' -> '{destination}': {e}"

            # 更新进度
            progress.advance(task)

            # 静默模式处理
            if silent:
                # 如果指定了日志文件，写入日志
                if log_file:
                    with open(log_file, 'a', encoding='utf-8') as f:
                        f.write(message + '\n')
            else:
                # 非静默模式，输出到控制台
                print(message)