影刀RPA文件自动化大全：批量重命名、格式转换与智能归档

影刀RPA文件自动化大全：批量重命名、格式转换与智能归档

作者：林焱 | 影刀RPA技术专栏

标签： 影刀RPA / 文件自动化 / 批量重命名 / 文件归档 / 格式转换 / 文件监控

适合读者： 需要处理大量文件、实现文件自动整理的RPA开发者

前言：文件自动化的巨大价值

你每天在文件管理上浪费多少时间？

• 从各渠道收到的报表，需要按部门/日期/类型手动归档
• 合同文件需要按"供应商-金额-日期"格式重命名
• 图片素材需要压缩、转格式、加水印
• 下载的发票PDF需要识别内容并自动归类

这些工作都可以让影刀RPA自动完成。

本文将带你全面掌握文件自动化的各种技巧，从基础的文件读写、批量重命名，到高级的文件监控触发、跨目录智能归档。

第一章：文件与目录基础操作

拼多多店群自动化报活动上架！

1.1 路径操作

import os
import pathlib

# ===== 路径操作 =====
# 使用 pathlib（推荐，面向对象）
p = pathlib.Path("C:/RPA/data/reports")
p = pathlib.Path(r"C:\\RPA\\data\\reports") # Windows路径的两种写法

# 路径拼接（跨平台安全）
file_path = p / "2024" / "01" / "report.xlsx"
# 结果：C:\\RPA\\data\\reports\\2024\\01\\report.xlsx

# 路径属性
print(file_path.parent) # C:\\RPA\\data\\reports\\2024\\01
print(file_path.name) # report.xlsx
print(file_path.stem) # report（不含扩展名）
print(file_path.suffix) # .xlsx
print(file_path.suffixes) # ['.xlsx']（多重扩展名，如.tar.gz）

# 路径检查
file_path.exists() # 是否存在
file_path.is_file() # 是否为文件
file_path.is_dir() # 是否为目录

# os.path方式（传统）
import os
dir_path = os.path.join("C:", "RPA", "data", "reports")
file_name = os.path.basename("C:/RPA/data/report.xlsx") # "report.xlsx"
dir_name = os.path.dirname("C:/RPA/data/report.xlsx") # "C:/RPA/data"
stem, ext = os.path.splitext("report.xlsx") # ("report", ".xlsx")

1.2 目录操作

import os
import shutil
import pathlib

# ===== 创建目录 =====
os.makedirs("C:/RPA/output/2024/01", exist_ok=True) # 递归创建，已存在不报错
pathlib.Path("C:/RPA/output/2024/02").mkdir(parents=True, exist_ok=True) # 等价

# ===== 列出目录内容 =====
# 列出当前目录所有文件和子目录
for item in os.listdir("C:/RPA/data"):
print(item)

# 只列出文件
files = [f for f in os.listdir("C:/RPA/data")
if os.path.isfile(os.path.join("C:/RPA/data", f))]

# 按扩展名过滤
xlsx_files = [f for f in os.listdir("C:/RPA/data") if f.endswith(".xlsx")]

# 使用pathlib（更优雅）
data_dir = pathlib.Path("C:/RPA/data")
all_files = list(data_dir.iterdir()) # 当前目录

all_xlsx = list(data_dir.glob("*.xlsx")) # 当前目录的xlsx文件
all_xlsx_recursive = list(data_dir.rglob("*.xlsx")) # 递归所有子目录的xlsx文件

# ===== 文件信息 =====
stat = os.stat("C:/RPA/data/report.xlsx")
file_size_bytes = stat.st_size
file_size_mb = stat.st_size / 1024 / 1024
modified_time = os.path.getmtime("C:/RPA/data/report.xlsx") # Unix时间戳

import datetime
modified_datetime = datetime.datetime.fromtimestamp(modified_time)
print(f"文件大小：{file_size_mb:.2f}MB，修改时间：{modified_datetime}")

# ===== 复制/移动/删除 =====
shutil.copy("source.xlsx", "destination.xlsx") # 复制文件
shutil.copy2("source.xlsx", "destination/") # 复制并保留元数据
shutil.move("source.xlsx", "archive/source.xlsx") # 移动文件（也可重命名）
shutil.copytree("source_dir", "dest_dir") # 复制整个目录树
os.rename("old_name.xlsx", "new_name.xlsx") # 重命名
os.remove("file_to_delete.xlsx") # 删除文件（不经回收站！）
shutil.rmtree("dir_to_delete", ignore_errors=True) # 删除目录树

第二章：批量重命名——最常用的文件自动化

2.1 按规则批量重命名

import os
import re
import pathlib
import datetime

def batch_rename(source_dir: str, rename_rule: callable,
file_pattern: str = "*",
dry_run: bool = True) –> list:
"""
批量重命名文件

参数：
– source_dir: 目标目录
– rename_rule: 重命名规则函数 (old_name: str) -> new_name: str
– file_pattern: 文件匹配模式（如 "*.xlsx", "report_*.pdf"）
– dry_run: True=只预览不执行，False=真实执行

返回：[(旧名, 新名)] 列表
"""
dir_path = pathlib.Path(source_dir)
files = sorted(dir_path.glob(file_pattern))

rename_plan = []

for file_path in files:
if not file_path.is_file():
continue

old_name = file_path.name
try:
new_name = rename_rule(old_name)
except Exception as e:
print(f"⚠️ 规则处理失败 {old_name}：{e}")
continue

if old_name != new_name:
rename_plan.append((old_name, new_name, file_path))

print(f"\\n重命名计划（{'预览' if dry_run else '执行'}）：")
for old_name, new_name, _ in rename_plan:
print(f" {old_name:40s} → {new_name}")

if not rename_plan:
print(" 无需重命名的文件")
return []

if not dry_run:
success_count = 0
for old_name, new_name, file_path in rename_plan:
new_path = file_path.parent / new_name

# 避免覆盖已有文件
if new_path.exists():
base = new_path.stem
ext = new_path.suffix
counter = 1
while new_path.exists():
new_path = file_path.parent / f"{base}_{counter}{ext}"
counter += 1
print(f" ⚠️ 文件已存在，重命名为：{new_path.name}")

file_path.rename(new_path)
success_count += 1

print(f"\\n✅ 成功重命名 {success_count}/{len(rename_plan)} 个文件")

return [(old, new) for old, new, _ in rename_plan]

# ===== 常用重命名规则 =====

# 规则1：添加日期前缀
def add_date_prefix(filename: str) –> str:
today = datetime.date.today().strftime("%Y%m%d")
return f"{today}_{filename}"

# 规则2：删除特定前缀
def remove_prefix(prefix: str):
def rule(filename: str) –> str:
if filename.startswith(prefix):
return filename[len(prefix):]
return filename
return rule

# 规则3：从文件名提取信息并重组
def normalize_report_filename(filename: str) –> str:
"""
将各种格式的报表文件名统一为：YYYY-MM-DD_类型_来源.xlsx

例如：
– "销售报表20240115.xlsx" → "2024-01-15_销售报表_手动.xlsx"
– "report_sales_2024_01_15.xlsx" → "2024-01-15_销售报表_系统.xlsx"
"""
stem, ext = os.path.splitext(filename)

# 提取日期（多种格式）
date_patterns = [
(r'(\\d{4})(\\d{2})(\\d{2})', "{}-{}-{}"), # 20240115
(r'(\\d{4})[_\\-](\\d{2})[_\\-](\\d{2})', "{}-{}-{}"), # 2024-01-15 或 2024_01_15
(r'(\\d{4})年(\\d{1,2})月(\\d{1,2})日', "{}-{:02d}-{:02d}") # 2024年1月15日
]

date_str = None
for pattern, fmt in date_patterns:
match = re.search(pattern, stem)
if match:
year, month, day = match.groups()
date_str = fmt.format(year, int(month), int(day))
break

if not date_str:
date_str = datetime.date.today().strftime("%Y-%m-%d")

# 提取类型关键词
type_keywords = {
"销售": "销售报表",
"sale": "销售报表",
"采购": "采购报表",
"库存": "库存报表",
"财务": "财务报表",
"考勤": "考勤报表"
}

report_type = "未分类报表"
for kw, type_name in type_keywords.items():
if kw.lower() in stem.lower():
report_type = type_name
break

return f"{date_str}_{report_type}{ext}"

# 实际使用
# 预览重命名结果
batch_rename(
"C:/RPA/downloads",
normalize_report_filename,
file_pattern="*.xlsx",
dry_run=True # 先预览
)

# 确认后执行
# batch_rename("C:/RPA/downloads", normalize_report_filename, dry_run=False)

2.2 按序号批量重命名

def rename_with_sequence(source_dir: str, prefix: str = "file",
start_num: int = 1, padding: int = 4,
file_pattern: str = "*",
sort_by: str = "name") –> None:
"""
按序号批量重命名文件

示例：
– img_20240115_001.jpg → product_0001.jpg
– img_20240115_002.jpg → product_0002.jpg
"""
dir_path = pathlib.Path(source_dir)
files = [f for f in dir_path.glob(file_pattern) if f.is_file()]

# 排序方式
if sort_by == "name":
files.sort(key=lambda f: f.name)
elif sort_by == "modified":
files.sort(key=lambda f: f.stat().st_mtime)
elif sort_by == "created":
files.sort(key=lambda f: f.stat().st_ctime)

for i, file_path in enumerate(files):
num = start_num + i
num_str = str(num).zfill(padding) # 补零
new_name = f"{prefix}_{num_str}{file_path.suffix}"
new_path = file_path.parent / new_name

print(f"{file_path.name} → {new_name}")
file_path.rename(new_path)

print(f"\\n✅ 完成，共重命名 {len(files)} 个文件")

# 示例：将产品图片按序号重命名
rename_with_sequence(
source_dir="C:/RPA/product_images",
prefix="product",
start_num=1,
padding=4,
file_pattern="*.jpg"
)

第三章：智能文件归档

3.1 按规则自动归档

import shutil
import pathlib
import datetime
import re

def smart_archive(source_dir: str, archive_root: str,
archive_rules: list, dry_run: bool = True) –> dict:
"""
智能文件归档系统

archive_rules格式：[{
"name": "规则名称",
"pattern": "文件名匹配模式（glob）",
"condition": lambda filename: bool, # 额外条件（可选）
"target_path": "目标路径（支持变量）", # {year}/{month}/{day}
"action": "copy" 或 "move"
}]
"""
source = pathlib.Path(source_dir)
archive = pathlib.Path(archive_root)
today = datetime.date.today()

stats = {
"processed": 0,
"archived": 0,
"skipped": 0,
"errors": 0,
"details": []
}

# 时间变量替换
time_vars = {
"{year}": today.strftime("%Y"),
"{month}": today.strftime("%m"),
"{day}": today.strftime("%d"),
"{week}": f"W{today.isocalendar()[1]:02d}",
"{quarter}": f"Q{(today.month–1)//3+1}"
}

for file_path in source.iterdir():
if not file_path.is_file():
continue

stats["processed"] += 1
matched_rule = None

# 匹配归档规则
for rule in archive_rules:
# 检查文件名模式
import fnmatch
if fnmatch.fnmatch(file_path.name, rule["pattern"]):
# 检查额外条件
condition = rule.get("condition")
if condition is None or condition(file_path.name):
matched_rule = rule
break

if not matched_rule:
stats["skipped"] += 1
continue

# 解析目标路径
target_subpath = matched_rule["target_path"]
for var, value in time_vars.items():
target_subpath = target_subpath.replace(var, value)

# 替换文件名变量
target_subpath = target_subpath.replace("{filename}", file_path.stem)
target_subpath = target_subpath.replace("{ext}", file_path.suffix.lstrip("."))

target_dir = archive / target_subpath
target_file = target_dir / file_path.name

action = matched_rule.get("action", "move")

detail = {
"source": str(file_path),
"target": str(target_file),
"rule": matched_rule["name"],
"action": action
}

if dry_run:
print(f"[{action.upper()}] {file_path.name}")
print(f" 规则：{matched_rule['name']}")
print(f" 目标：{target_file}")
stats["details"].append(detail)
stats["archived"] += 1
else:
try:
target_dir.mkdir(parents=True, exist_ok=True)

# 处理同名文件
if target_file.exists():
stem = target_file.stem
suffix = target_file.suffix
counter = 1
while target_file.exists():
target_file = target_dir / f"{stem}_{counter}{suffix}"
counter += 1

if action == "move":
shutil.move(str(file_path), str(target_file))
elif action == "copy":
shutil.copy2(str(file_path), str(target_file))

print(f"✅ [{action.upper()}] {file_path.name} → {target_file.relative_to(archive)}")
stats["archived"] += 1
stats["details"].append(detail)

except Exception as e:
print(f"❌ 归档失败 {file_path.name}：{e}")
stats["errors"] += 1

print(f"\\n归档{'预览' if dry_run else ''}完成：处理{stats['processed']}个，归档{stats['archived']}个，跳过{stats['skipped']}个，失败{stats['errors']}个")
return stats

# ===== 使用示例 =====

# 定义归档规则
archive_rules = [
{
"name": "月度销售报表",
"pattern": "*销售报表*",
"condition": lambda name: name.endswith(".xlsx"),
"target_path": "报表/{year}/{month}/销售",
"action": "move"
},
{
"name": "采购合同PDF",
"pattern": "*.pdf",
"condition": lambda name: "合同" in name or "contract" in name.lower(),
"target_path": "合同/{year}/采购合同",
"action": "move"
},
{
"name": "系统截图",
"pattern": "Screenshot_*",
"target_path": "截图/{year}/{month}",
"action": "move"
},
{
"name": "所有Excel（兜底规则）",
"pattern": "*.xlsx",
"target_path": "Excel/{year}/{month}",
"action": "copy" # 复制而不是移动
}
]

# 先预览
smart_archive(
source_dir="C:/Users/Admin/Downloads",
archive_root="D:/Archives",
archive_rules=archive_rules,
dry_run=True
)

# 确认无误后执行
# smart_archive(…, dry_run=False)

第四章：文件监控——实时触发自动化

4.1 轮询监控（简单可靠）

import time
import os
import pathlib
import hashlib

class FileWatcher:
"""文件目录监控（轮询方式）"""

def __init__(self, watch_dir: str, interval_seconds: int = 10,
file_pattern: str = "*"):
self.watch_dir = pathlib.Path(watch_dir)
self.interval = interval_seconds
self.file_pattern = file_pattern
self.known_files = {} # {文件路径: 最后修改时间}
self._running = False

def _scan(self) –> dict:
"""扫描目录获取当前文件状态"""
current_files = {}
for f in self.watch_dir.glob(self.file_pattern):
if f.is_file():
current_files[str(f)] = f.stat().st_mtime
return current_files

def _detect_changes(self, current: dict) –> dict:
"""检测文件变化"""
changes = {"new": [], "modified": [], "deleted": []}

# 新增或修改的文件
for path, mtime in current.items():
if path not in self.known_files:
changes["new"].append(path)
elif self.known_files[path] != mtime:
changes["modified"].append(path)

# 删除的文件
for path in self.known_files:
if path not in current:
changes["deleted"].append(path)

return changes

def watch(self,
on_new=None, on_modified=None, on_deleted=None,
max_duration_seconds: int = None):
"""
开始监控

回调函数签名：
– on_new(file_path: str)
– on_modified(file_path: str)
– on_deleted(file_path: str)
"""
print(f"开始监控目录：{self.watch_dir}")
print(f"文件模式：{self.file_pattern}，检查间隔：{self.interval}s")

# 初始化已知文件
self.known_files = self._scan()
print(f"初始文件数：{len(self.known_files)}")

self._running = True
start_time = time.time()

while self._running:
time.sleep(self.interval)

# 检查是否超时
if max_duration_seconds and (time.time() – start_time) > max_duration_seconds:
print("达到最大监控时长，停止监控")
break

current = self._scan()
changes = self._detect_changes(current)

# 触发回调
for file_path in changes["new"]:
print(f"📁 新文件：{os.path.basename(file_path)}")
if on_new:
try:
on_new(file_path)
except Exception as e:
print(f"处理新文件出错：{e}")

for file_path in changes["modified"]:
print(f"✏️ 文件变更：{os.path.basename(file_path)}")
if on_modified:
try:
on_modified(file_path)
except Exception as e:
print(f"处理文件变更出错：{e}")

for file_path in changes["deleted"]:
print(f"🗑️ 文件删除：{os.path.basename(file_path)}")
if on_deleted:
try:
on_deleted(file_path)
except Exception as e:
print(f"处理文件删除出错：{e}")

# 更新已知文件状态
self.known_files = current

def stop(self):
self._running = False

# ===== 使用示例：监控下载目录并自动归档 =====

def on_new_file(file_path: str):
"""收到新文件时自动归档"""
path = pathlib.Path(file_path)

# 等待文件写入完成（避免文件被写到一半就处理）
time.sleep(2)

# 根据扩展名决定归档位置
ext = path.suffix.lower()
archive_map = {
".xlsx": "C:/Archive/Excel",
".pdf": "C:/Archive/PDF",
".jpg": "C:/Archive/Images",
".png": "C:/Archive/Images",
".docx": "C:/Archive/Word"
}

target_dir = archive_map.get(ext, "C:/Archive/Others")
os.makedirs(target_dir, exist_ok=True)

target_path = os.path.join(target_dir, path.name)
shutil.copy2(file_path, target_path)
print(f"✅ 已归档：{path.name} → {target_dir}")

# 发送通知
send_notification(f"新文件已归档：{path.name}")

# 启动监控
watcher = FileWatcher(
watch_dir="C:/Users/Admin/Downloads",
interval_seconds=30,
file_pattern="*"
)

watcher.watch(on_new=on_new_file)

TEMU店群矩阵自动化运营核价报活动

第五章：文件格式处理

5.1 CSV与Excel互转

import csv
import openpyxl

def csv_to_excel(csv_path: str, excel_path: str,
encoding: str = "utf-8-sig",
sheet_name: str = "Sheet1") –> str:
"""CSV转Excel（自动识别编码，保留格式）"""

# 读取CSV
with open(csv_path, "r", encoding=encoding, newline="") as f:
reader = csv.DictReader(f)
headers = reader.fieldnames
rows = list(reader)

# 写入Excel
wb = openpyxl.Workbook()
ws = wb.active
ws.title = sheet_name

# 写表头（加粗）
ws.append(headers)
header_row = ws[1]
for cell in header_row:
cell.font = openpyxl.styles.Font(bold=True)
cell.fill = openpyxl.styles.PatternFill(
fill_type="solid",
fgColor="4472C4" # 蓝色背景
)
cell.font = openpyxl.styles.Font(bold=True, color="FFFFFF")

# 写数据
for row in rows:
ws.append([row.get(h, "") for h in headers])

# 自动列宽
for column_cells in ws.columns:
max_len = max(len(str(cell.value or "")) for cell in column_cells)
ws.column_dimensions[column_cells[0].column_letter].width = min(max_len + 2, 50)

wb.save(excel_path)
print(f"✅ CSV转Excel完成：{len(rows)}行 → {excel_path}")
return excel_path

def excel_to_csv(excel_path: str, csv_path: str,
sheet_name: str = None,
encoding: str = "utf-8-sig") –> str:
"""Excel转CSV"""

wb = openpyxl.load_workbook(excel_path, read_only=True, data_only=True)
ws = wb[sheet_name] if sheet_name else wb.active

with open(csv_path, "w", encoding=encoding, newline="") as f:
writer = csv.writer(f)
for row in ws.iter_rows(values_only=True):
writer.writerow([str(v) if v is not None else "" for v in row])

wb.close()
print(f"✅ Excel转CSV完成：{csv_path}")
return csv_path

def batch_csv_to_excel(source_dir: str, output_dir: str = None) –> list:
"""批量将目录下所有CSV转为Excel"""
source = pathlib.Path(source_dir)
output = pathlib.Path(output_dir or source_dir)
output.mkdir(exist_ok=True)

csv_files = list(source.glob("*.csv"))
results = []

for csv_file in csv_files:
excel_path = output / (csv_file.stem + ".xlsx")

# 尝试自动检测编码
encodings = ["utf-8-sig", "gbk", "utf-8", "gb18030"]
converted = False

for enc in encodings:
try:
csv_to_excel(str(csv_file), str(excel_path), encoding=enc)
results.append({"source": csv_file.name, "target": excel_path.name, "success": True})
converted = True
break
except UnicodeDecodeError:
continue

if not converted:
print(f"⚠️ {csv_file.name} 编码识别失败")
results.append({"source": csv_file.name, "target": "", "success": False})

print(f"\\n批量转换完成：{len([r for r in results if r['success']]}/{len(csv_files)}成功")
return results

第六章：文件去重与清理

6.1 基于MD5的文件去重

import hashlib
import os
import pathlib
from collections import defaultdict

def find_duplicate_files(search_dir: str, file_pattern: str = "*") –> dict:
"""
查找重复文件（基于MD5哈希）

返回：{md5哈希: [文件路径列表]} （只包含重复的）
"""
hash_map = defaultdict(list)

for file_path in pathlib.Path(search_dir).rglob(file_pattern):
if not file_path.is_file():
continue

# 计算文件MD5（大文件只读部分）
try:
file_size = file_path.stat().st_size

with open(file_path, "rb") as f:
if file_size > 10 * 1024 * 1024: # 大于10MB
# 只读取头部、中部、尾部各4KB（快速判断）
content = f.read(4096)
f.seek(file_size // 2)
content += f.read(4096)
f.seek(–4096, 2)
content += f.read(4096)
else:
content = f.read()

file_hash = hashlib.md5(content).hexdigest()
hash_map[file_hash].append(str(file_path))

except Exception as e:
print(f"⚠️ 读取失败 {file_path}：{e}")

# 只返回有重复的
duplicates = {h: paths for h, paths in hash_map.items() if len(paths) > 1}

if duplicates:
total_dup_files = sum(len(paths) – 1 for paths in duplicates.values())
print(f"找到 {len(duplicates)} 组重复文件，共 {total_dup_files} 个多余副本")

for hash_val, paths in list(duplicates.items())[:5]: # 显示前5组
print(f"\\n哈希：{hash_val[:8]}…")
for i, p in enumerate(paths):
mark = "✅ 保留" if i == 0 else "🗑️ 重复"
size = os.path.getsize(p) / 1024
print(f" {mark} {os.path.basename(p)} ({size:.0f}KB)")
else:
print("未发现重复文件")

return duplicates

def remove_duplicates(duplicates: dict, keep_strategy: str = "first",
dry_run: bool = True) –> int:
"""
删除重复文件

keep_strategy：
– "first": 保留列表中第一个（通常是最早发现的）
– "oldest": 保留修改时间最早的（原始文件）
– "newest": 保留修改时间最新的
"""
removed_count = 0

for hash_val, paths in duplicates.items():
if keep_strategy == "oldest":
keep_path = min(paths, key=lambda p: os.path.getmtime(p))
elif keep_strategy == "newest":
keep_path = max(paths, key=lambda p: os.path.getmtime(p))
else:
keep_path = paths[0]

to_remove = [p for p in paths if p != keep_path]

for path in to_remove:
if dry_run:
print(f"[预览] 将删除：{os.path.basename(path)}")
else:
try:
os.remove(path)
removed_count += 1
print(f"✅ 已删除：{os.path.basename(path)}")
except Exception as e:
print(f"❌ 删除失败：{path}：{e}")

if not dry_run:
print(f"\\n共删除 {removed_count} 个重复文件")

return removed_count

总结

本文覆盖了影刀RPA文件自动化的核心技能：

文件自动化是RPA中非常高频的场景，掌握这些技巧后，大量日常文件整理工作都能实现全自动处理。

作者：林焱 | 影刀RPA技术专栏 | 欢迎点赞收藏！