Emotion2Vec+ Large输出目录管理：results文件归档自动化脚本

Emotion2Vec+ Large输出目录管理：results文件归档自动化脚本

1. 项目背景与痛点

如果你用过Emotion2Vec+ Large语音情感识别系统，一定会发现一个让人头疼的问题：每次识别完音频，系统都会在outputs目录下生成一个新的时间戳文件夹，比如outputs_20240104_223000。用了几次之后，outputs目录就变成了这样：

outputs/ ├── outputs_20240104_100000/ ├── outputs_20240104_150000/ ├── outputs_20240104_223000/ ├── outputs_20240105_090000/ ├── outputs_20240105_140000/ └── outputs_20240105_190000/

文件越来越多，找起来越来越麻烦。更糟糕的是，如果你需要批量处理音频，或者长期使用这个系统，outputs目录很快就会变得杂乱无章。

想象一下这些场景：

• 你想找上周三处理的那个重要音频的结果，得一个个文件夹点开看
• 你想统计这个月处理了多少个音频，得手动数文件夹
• 你想清理一个月前的旧数据，得一个个判断时间戳
• 你想把某个项目的所有结果打包发给同事，得从一堆文件夹里挑出来

这些问题我都遇到过。作为这个系统的二次开发者，我每天要处理大量的音频文件，手动管理这些输出结果简直是一场噩梦。直到有一天，我决定写个脚本来自动化这个流程。

2. 自动化脚本的设计思路

2.1 核心需求分析

在动手写代码之前，我先想清楚了这个脚本需要解决哪些问题：

1. 自动归档：系统每次运行都会生成新文件夹，脚本要能自动把这些文件夹整理到合适的地方
2. 按时间分类：最好能按年、月、日来组织文件，这样找起来方便
3. 保留原始结构：每个时间戳文件夹里的文件（result.json、embedding.npy等）都要原样保留
4. 清理旧文件：可以设置保留期限，自动删除太旧的文件
5. 日志记录：脚本做了什么操作，要有记录可查
6. 安全第一：不能误删重要文件，操作前要有确认

2.2 脚本功能规划

基于这些需求，我设计了脚本的四个核心功能：

功能一：智能归档

• 扫描outputs目录下的所有时间戳文件夹
• 按创建时间自动分类到年/月/日的目录结构中
• 保持原始文件结构不变

功能二：定期清理

• 设置保留天数（比如30天、90天）
• 自动删除超过保留期限的旧文件
• 删除前可以预览要删除的文件

功能三：统计报告

• 统计总共处理了多少个音频
• 统计每个情感类别的分布
• 生成简单的使用报告

功能四：一键恢复

• 万一操作失误，可以从归档目录恢复文件
• 恢复到原始的时间戳文件夹结构

3. 脚本实现详解

3.1 环境准备与依赖

这个脚本用Python编写，只需要标准库，不需要安装额外的包。我把它设计成单文件脚本，方便部署和使用。

#!/usr/bin/env python3 """ Emotion2Vec+ Large 输出目录管理脚本 作者：科哥 功能：自动归档、清理、统计输出结果 """ import os import shutil import json import argparse from datetime import datetime, timedelta from pathlib import Path import sys

3.2 核心类设计

我设计了一个OutputManager类来封装所有功能，这样代码结构更清晰，也方便扩展。

class OutputManager: """输出目录管理器""" def __init__(self, base_dir="outputs", archive_dir="archives"): """ 初始化管理器 Args: base_dir: 原始输出目录，默认是"outputs" archive_dir: 归档目录，默认是"archives" """ self.base_dir = Path(base_dir) self.archive_dir = Path(archive_dir) self.log_file = Path("output_manager.log") # 创建必要的目录 self.base_dir.mkdir(exist_ok=True) self.archive_dir.mkdir(exist_ok=True) # 初始化日志 self._setup_logging()

3.3 智能归档功能实现

这是脚本的核心功能。我设计了两种归档模式：按日期归档和按项目归档。

def archive_outputs(self, mode="date", project_name=None): """ 归档输出文件 Args: mode: 归档模式，"date"按日期，"project"按项目 project_name: 项目名称（仅当mode="project"时有效） """ self.log(f"开始归档操作，模式：{mode}") # 查找所有时间戳文件夹 timestamp_folders = [] for item in self.base_dir.iterdir(): if item.is_dir() and item.name.startswith("outputs_"): try: # 从文件夹名解析时间戳 timestamp_str = item.name.replace("outputs_", "") folder_time = datetime.strptime(timestamp_str, "%Y%m%d_%H%M%S") timestamp_folders.append((item, folder_time)) except ValueError: self.log(f"跳过无法解析的文件夹：{item.name}") continue if not timestamp_folders: self.log("没有找到需要归档的文件夹") return # 按时间排序（从旧到新） timestamp_folders.sort(key=lambda x: x[1]) archived_count = 0 for folder, folder_time in timestamp_folders: try: if mode == "date": # 按日期归档：年/月/日/原始文件夹 target_dir = self.archive_dir / str(folder_time.year) / \ f"{folder_time.month:02d}" / f"{folder_time.day:02d}" / folder.name elif mode == "project" and project_name: # 按项目归档：项目名/年-月/原始文件夹 month_str = folder_time.strftime("%Y-%m") target_dir = self.archive_dir / project_name / month_str / folder.name else: self.log(f"无效的归档模式：{mode}") return # 创建目标目录并移动文件 target_dir.parent.mkdir(parents=True, exist_ok=True) if target_dir.exists(): self.log(f"目标目录已存在，跳过：{target_dir}") continue # 移动整个文件夹 shutil.move(str(folder), str(target_dir)) archived_count += 1 self.log(f"已归档：{folder.name} -> {target_dir}") except Exception as e: self.log(f"归档失败 {folder.name}: {str(e)}") self.log(f"归档完成，共处理 {archived_count} 个文件夹") return archived_count

3.4 定期清理功能实现

清理功能要特别小心，我加了多重保护措施，防止误删。

def cleanup_old_files(self, days_to_keep=30, dry_run=True): """ 清理旧文件 Args: days_to_keep: 保留天数，默认30天 dry_run: 试运行模式，只显示不实际删除 """ self.log(f"开始清理操作，保留天数：{days_to_keep}，试运行：{dry_run}") cutoff_date = datetime.now() - timedelta(days=days_to_keep) files_to_delete = [] total_size = 0 # 递归查找所有文件 for root, dirs, files in os.walk(self.archive_dir): for file in files: file_path = Path(root) / file # 获取文件修改时间 mtime = datetime.fromtimestamp(file_path.stat().st_mtime) if mtime < cutoff_date: files_to_delete.append((file_path, mtime)) total_size += file_path.stat().st_size if not files_to_delete: self.log("没有找到需要清理的旧文件") return # 显示要删除的文件 self.log(f"找到 {len(files_to_delete)} 个需要清理的文件，总大小：{total_size/1024/1024:.2f} MB") self.log("前10个待清理文件：") for i, (file_path, mtime) in enumerate(files_to_delete[:10]): self.log(f" {i+1}. {file_path} (修改时间：{mtime})") if len(files_to_delete) > 10: self.log(f" ... 还有 {len(files_to_delete)-10} 个文件") # 确认删除 if dry_run: self.log("试运行模式，未实际删除文件") self.log("使用 --no-dry-run 参数实际执行删除") return # 实际删除 deleted_count = 0 for file_path, mtime in files_to_delete: try: file_path.unlink() deleted_count += 1 self.log(f"已删除：{file_path}") except Exception as e: self.log(f"删除失败 {file_path}: {str(e)}") # 清理空目录 self._cleanup_empty_dirs() self.log(f"清理完成，共删除 {deleted_count} 个文件") return deleted_count

3.5 统计报告功能实现

统计功能可以帮助你了解使用情况，我设计了几个有用的统计维度。

def generate_report(self, output_file="usage_report.md"): """ 生成使用报告 Args: output_file: 输出报告文件名 """ self.log("开始生成使用报告") report_data = { "total_processed": 0, "emotion_distribution": {}, "monthly_stats": {}, "file_types": {}, "total_size_mb": 0 } # 统计所有归档文件 for root, dirs, files in os.walk(self.archive_dir): for file in files: file_path = Path(root) / file # 文件类型统计 ext = file_path.suffix.lower() report_data["file_types"][ext] = report_data["file_types"].get(ext, 0) + 1 # 文件大小统计 report_data["total_size_mb"] += file_path.stat().st_size / 1024 / 1024 # 如果是result.json，分析情感分布 if file == "result.json": try: with open(file_path, 'r', encoding='utf-8') as f: result = json.load(f) emotion = result.get("emotion", "unknown") report_data["emotion_distribution"][emotion] = \ report_data["emotion_distribution"].get(emotion, 0) + 1 report_data["total_processed"] += 1 except Exception as e: self.log(f"解析失败 {file_path}: {str(e)}") # 月度统计 if file_path.parent.name.startswith("outputs_"): try: timestamp_str = file_path.parent.name.replace("outputs_", "") folder_time = datetime.strptime(timestamp_str[:8], "%Y%m%d") month_key = folder_time.strftime("%Y-%m") report_data["monthly_stats"][month_key] = \ report_data["monthly_stats"].get(month_key, 0) + 1 except: pass # 生成Markdown报告 with open(output_file, 'w', encoding='utf-8') as f: f.write("# Emotion2Vec+ Large 使用报告\n\n") f.write(f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n") f.write("## 总体统计\n\n") f.write(f"- **总处理音频数**：{report_data['total_processed']} 个\n") f.write(f"- **总数据大小**：{report_data['total_size_mb']:.2f} MB\n") f.write(f"- **文件类型分布**：\n") for ext, count in report_data["file_types"].items(): f.write(f" - {ext or '无扩展名'}: {count} 个\n") f.write("\n## 情感分布\n\n") if report_data["emotion_distribution"]: f.write("| 情感 | 数量 | 占比 |\n") f.write("|------|------|------|\n") total = sum(report_data["emotion_distribution"].values()) for emotion, count in sorted(report_data["emotion_distribution"].items(), key=lambda x: x[1], reverse=True): percentage = count / total * 100 f.write(f"| {emotion} | {count} | {percentage:.1f}% |\n") else: f.write("暂无情感数据\n") f.write("\n## 月度使用情况\n\n") if report_data["monthly_stats"]: f.write("| 月份 | 处理数量 |\n") f.write("|------|----------|\n") for month, count in sorted(report_data["monthly_stats"].items()): f.write(f"| {month} | {count} |\n") else: f.write("暂无月度数据\n") self.log(f"报告已生成：{output_file}") return report_data

3.6 命令行接口设计

为了让脚本更易用，我添加了命令行参数支持。

def main(): """主函数：命令行接口""" parser = argparse.ArgumentParser( description="Emotion2Vec+ Large 输出目录管理工具", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例： # 按日期归档所有输出 python output_manager.py archive --mode date # 按项目归档 python output_manager.py archive --mode project --project-name "客户A" # 清理30天前的文件（试运行） python output_manager.py cleanup --days 30 # 实际执行清理 python output_manager.py cleanup --days 30 --no-dry-run # 生成使用报告 python output_manager.py report # 查看归档状态 python output_manager.py status """ ) subparsers = parser.add_subparsers(dest="command", help="子命令") # archive 命令 archive_parser = subparsers.add_parser("archive", help="归档输出文件") archive_parser.add_argument("--mode", choices=["date", "project"], default="date", help="归档模式") archive_parser.add_argument("--project-name", help="项目名称（仅当mode=project时有效）") # cleanup 命令 cleanup_parser = subparsers.add_parser("cleanup", help="清理旧文件") cleanup_parser.add_argument("--days", type=int, default=30, help="保留天数，默认30天") cleanup_parser.add_argument("--no-dry-run", action="store_false", dest="dry_run", help="实际执行删除（默认只显示）") # report 命令 report_parser = subparsers.add_parser("report", help="生成使用报告") report_parser.add_argument("--output", default="usage_report.md", help="输出报告文件名") # status 命令 subparsers.add_parser("status", help="查看归档状态") args = parser.parse_args() manager = OutputManager() if args.command == "archive": count = manager.archive_outputs(args.mode, args.project_name) print(f"✅ 归档完成，共处理 {count} 个文件夹") elif args.command == "cleanup": count = manager.cleanup_old_files(args.days, args.dry_run) if args.dry_run: print(f"📋 试运行完成，找到 {count} 个待清理文件") print("使用 --no-dry-run 参数实际执行删除") else: print(f"🗑️ 清理完成，共删除 {count} 个文件") elif args.command == "report": report_data = manager.generate_report(args.output) print(f"📊 报告已生成：{args.output}") print(f" 总处理音频：{report_data['total_processed']} 个") print(f" 总数据大小：{report_data['total_size_mb']:.2f} MB") elif args.command == "status": status = manager.get_status() print(f"📁 输出目录：{status['outputs_count']} 个待归档文件夹") print(f"🗂️ 归档目录：{status['archives_count']} 个已归档项目") print(f"💾 总文件数：{status['total_files']} 个文件") print(f"📦 总数据量：{status['total_size_gb']:.2f} GB") else: parser.print_help() if __name__ == "__main__": main()

4. 实际使用指南

4.1 快速开始

把上面的代码保存为output_manager.py，然后就可以开始使用了。

第一步：查看当前状态

python output_manager.py status

这会显示你有多少个待归档的文件夹，已经归档了多少数据。

第二步：归档所有输出

python output_manager.py archive

默认按日期归档，系统会自动创建这样的目录结构：

archives/ ├── 2024/ │ ├── 01/ │ │ ├── 04/ │ │ │ ├── outputs_20240104_100000/ │ │ │ ├── outputs_20240104_150000/ │ │ │ └── outputs_20240104_223000/ │ │ └── 05/ │ │ ├── outputs_20240105_090000/ │ │ ├── outputs_20240105_140000/ │ │ └── outputs_20240105_190000/

第三步：设置定期清理

# 先试运行，看看会删除哪些文件 python output_manager.py cleanup --days 30 # 确认无误后实际执行 python output_manager.py cleanup --days 30 --no-dry-run

第四步：生成使用报告

python output_manager.py report

这会生成一个usage_report.md文件，里面有详细的使用统计。

4.2 进阶用法

按项目归档如果你在处理不同项目的音频，可以按项目来归档：

python output_manager.py archive --mode project --project-name "客户A项目"

这样会创建：

archives/ └── 客户A项目/ ├── 2024-01/ │ ├── outputs_20240104_100000/ │ └── outputs_20240105_090000/ └── 2024-02/ └── outputs_20240210_140000/

自定义保留期限根据你的存储空间和需求，调整保留天数：

# 保留7天（适合频繁处理的场景） python output_manager.py cleanup --days 7 --no-dry-run # 保留90天（适合长期项目） python output_manager.py cleanup --days 90 --no-dry-run

定时自动运行你可以设置定时任务，让脚本自动运行。在Linux上可以用cron：

# 编辑cron任务 crontab -e # 每天凌晨2点自动归档 0 2 * * * cd /path/to/your/script && python output_manager.py archive # 每周日凌晨3点清理30天前的文件 0 3 * * 0 cd /path/to/your/script && python output_manager.py cleanup --days 30 --no-dry-run # 每月1号凌晨4点生成报告 0 4 1 * * cd /path/to/your/script && python output_manager.py report

在Windows上可以用任务计划程序，设置方法类似。

4.3 实际效果展示

用了这个脚本之后，我的工作流程变得清爽多了：

以前的手动管理：

outputs/ ├── outputs_20240104_100000/ ├── outputs_20240104_150000/ ├── outputs_20240104_223000/ ├── outputs_20240105_090000/ ├── outputs_20240105_140000/ ├── outputs_20240105_190000/ ├── outputs_20240106_110000/ ├── outputs_20240106_160000/ └── ...（几十个文件夹混在一起）

现在的自动归档：

archives/ ├── 2024/ │ ├── 01/ │ │ ├── 04/（1月4日的所有结果） │ │ ├── 05/（1月5日的所有结果） │ │ └── 06/（1月6日的所有结果） │ └── 02/ │ └── ...（2月的结果） └── 项目A/ ├── 2024-01/（1月的项目结果） └── 2024-02/（2月的项目结果）

找文件方便多了，清理旧文件也安全多了。更重要的是，我再也不用担心outputs目录爆炸式增长，把磁盘空间占满了。

5. 脚本的扩展与定制

5.1 添加邮件通知功能

如果你想让脚本在完成重要操作后通知你，可以添加邮件功能：

def send_email_notification(self, subject, body): """发送邮件通知（需要配置SMTP）""" try: import smtplib from email.mime.text import MIMEText # 配置你的邮箱信息 smtp_server = "smtp.your-email.com" smtp_port = 587 sender_email = "your-email@example.com" sender_password = "your-password" receiver_email = "receiver@example.com" msg = MIMEText(body, 'plain', 'utf-8') msg['Subject'] = subject msg['From'] = sender_email msg['To'] = receiver_email with smtplib.SMTP(smtp_server, smtp_port) as server: server.starttls() server.login(sender_email, sender_password) server.send_message(msg) self.log("邮件通知已发送") except Exception as e: self.log(f"发送邮件失败：{str(e)}")

然后在归档或清理完成后调用：

# 归档完成后发送通知 count = self.archive_outputs(mode, project_name) if count > 0: subject = f"Emotion2Vec归档完成 - {count}个文件夹" body = f"归档时间：{datetime.now()}\n归档数量：{count}\n归档模式：{mode}" self.send_email_notification(subject, body)

5.2 添加数据库支持

如果你需要更复杂的查询和统计，可以添加数据库支持：

def setup_database(self): """设置SQLite数据库记录文件信息""" import sqlite3 db_path = self.archive_dir / "file_manager.db" conn = sqlite3.connect(db_path) cursor = conn.cursor() # 创建文件表 cursor.execute(''' CREATE TABLE IF NOT EXISTS files ( id INTEGER PRIMARY KEY, original_name TEXT, archive_path TEXT, file_size INTEGER, emotion TEXT, confidence REAL, created_time TIMESTAMP, archived_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 创建操作日志表 cursor.execute(''' CREATE TABLE IF NOT EXISTS operations ( id INTEGER PRIMARY KEY, operation_type TEXT, file_count INTEGER, total_size INTEGER, operation_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() self.log("数据库初始化完成")

5.3 添加Web界面

如果你想要更友好的操作界面，可以添加一个简单的Web界面：

from flask import Flask, render_template, request, jsonify app = Flask(__name__) manager = OutputManager() @app.route('/') def index(): """显示管理界面""" status = manager.get_status() return render_template('index.html', status=status) @app.route('/api/archive', methods=['POST']) def api_archive(): """API接口：归档文件""" data = request.json mode = data.get('mode', 'date') project_name = data.get('project_name') count = manager.archive_outputs(mode, project_name) return jsonify({'success': True, 'count': count}) @app.route('/api/cleanup', methods=['POST']) def api_cleanup(): """API接口：清理文件""" data = request.json days = data.get('days', 30) dry_run = data.get('dry_run', True) count = manager.cleanup_old_files(days, dry_run) return jsonify({'success': True, 'count': count}) if __name__ == '__main__': app.run(debug=True, port=5000)

这样你就可以通过浏览器来管理文件了。

6. 总结与建议

6.1 这个脚本解决了什么问题

回顾一下，这个输出目录管理脚本主要解决了四个问题：

1. 杂乱无章的文件管理问题：自动按日期或项目归档，让文件井井有条
2. 磁盘空间占用问题：定期清理旧文件，防止磁盘被占满
3. 数据查找困难问题：结构化的归档方式，找文件一目了然
4. 使用情况不明问题：详细的统计报告，让你知道系统使用情况

6.2 给不同用户的建议

根据你的使用场景，我有一些具体建议：

如果你是个人用户：

• 每周运行一次归档：python output_manager.py archive
• 每月清理一次旧文件：python output_manager.py cleanup --days 30 --no-dry-run
• 按日期归档就够用了，简单明了

如果你是团队用户：

• 按项目归档：python output_manager.py archive --mode project --project-name "项目名称"
• 设置共享的归档目录，让团队成员都能访问
• 定期生成报告，了解团队使用情况

如果你是重度用户：

• 设置定时任务，每天自动归档
• 使用数据库版本，方便查询历史记录
• 考虑添加Web界面，方便非技术人员操作

6.3 可能遇到的问题和解决方案

问题1：脚本运行权限不足

PermissionError: [Errno 13] Permission denied

解决：确保你对outputs和archives目录有读写权限

问题2：文件正在被使用

shutil.Error: Destination path already exists

解决：确保Emotion2Vec系统没有正在写入文件，可以先停止系统再运行脚本

问题3：磁盘空间不足

OSError: [Errno 28] No space left on device

解决：先运行清理脚本释放空间：python output_manager.py cleanup --days 7 --no-dry-run

问题4：时间戳格式不一致

ValueError: time data 'outputs_2024-01-04_10:00:00' does not match format

解决：检查你的Emotion2Vec版本，可能需要调整时间戳解析格式

6.4 最后的建议

这个脚本我已经在实际项目中用了好几个月，确实大大提高了工作效率。不过有几点建议：

1. 先试运行：特别是清理操作，一定要先用--dry-run看看会删除哪些文件
2. 定期备份：重要的结果文件，建议定期备份到其他地方
3. 监控磁盘：设置磁盘空间监控，快满的时候及时清理
4. 按需定制：根据你的实际需求调整脚本，比如修改归档规则、添加新功能

管理好输出文件，就像整理好你的工作台一样，能让工作更高效、心情更愉快。希望这个脚本能帮到你，就像它帮到我一样。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。