办公自动化实战合集：Excel、Word、PDF 批量处理、文件自动整理与邮件群发

1
import os
2
import re
3
import glob
4
from pathlib import Path
5
import pandas as pd
6
import numpy as np
7

8
def clean_currency_string(val):
9
    """
10
    清洗货币与数值型脏数据：将 '¥ 1,234.50 元' 或 '(1,234.50)' 转为标准浮点数
11
    """
12
    if pd.isna(val):
13
        return 0.0
14
    if isinstance(val, (int, float)):
15
        return float(val)
16

17
    val_str = str(val).strip()
18
    if not val_str:
19
        return 0.0
20

21
    # 判定会计负数格式：(100.00) -> -100.00
22
    is_negative = False
23
    if val_str.startswith('(') and val_str.endswith(')'):
24
        is_negative = True
25
        val_str = val_str[1:-1]
26

27
    # 剔除货币符号、逗号、中文单位字符
28
    cleaned = re.sub(r'[¥$,\s元]', '', val_str)
29
    try:
30
        num = float(cleaned)
31
        return -num if is_negative else num
32
    except ValueError:
33
        return 0.0
34

35
def detect_header_row(file_path, sheet_name=0, required_keywords=None):
36
    """
37
    智能嗅探表头行号：应对不同人员在表头前随机添加的备注说明行
38
    """
39
    if required_keywords is None:
40
        required_keywords = ["工号", "姓名", "金额"]
41

42
    preview_df = pd.read_excel(file_path, sheet_name=sheet_name, header=None, nrows=10)
43
    for idx, row in preview_df.iterrows():
44
        row_values = [str(x).strip() for x in row.values if pd.notna(x)]
45
        # 如果当前行包含至少两个必需核心字段，则判定为有效表头行
46
        matches = sum(1 for kw in required_keywords if any(kw in cell for cell in row_values))
47
        if matches >= 2:
48
            return idx
49
    return 0
50

51
def merge_enterprise_sales_reports(input_dir, output_filepath):
52
    """
53
    批量合并并深度清洗企业多分支机构销售周报
54
    """
55
    input_path = Path(input_dir)
56
    excel_files = list(input_path.glob("*.xlsx"))
57

58
    if not excel_files:
59
        print(f"[警告] 目录 {input_dir} 下未发现任何 .xlsx 文件")
60
        return False
61

62
    combined_records = []
63

64
    for file_path in excel_files:
65
        # 跳过 Office 打开时产生的临时文件（以 ~$ 开头）
66
        if file_path.name.startswith("~$"):
67
            continue
68

69
        print(f"[正在读取] 正在解析: {file_path.name}")
70
        try:
71
            excel_reader = pd.ExcelFile(file_path)
72
            for sheet in excel_reader.sheet_names:
73
                # 忽略备份或说明 Sheet
74
                if any(ignore_kw in sheet for ignore_kw in ["说明", "参考", "模板", "Sheet"]):
75
                    continue
76

77
                header_idx = detect_header_row(file_path, sheet_name=sheet)
78
                df = pd.read_excel(file_path, sheet_name=sheet, header=header_idx)
79

80
                # 剔除全空白行与全空白列
81
                df.dropna(how='all', inplace=True)
82
                df.dropna(axis=1, how='all', inplace=True)
83

84
                # 剔除员工手动在底部填写的“合计”、“汇总”行
85
                first_col = df.columns[0]
86
                df = df[~df[first_col].astype(str).str.contains(r'合计|汇总|平均', na=False)]
87

88
                # 标准化列名：去除首尾空格、制表符与回车
89
                df.columns = [str(c).strip().replace('\n', '').replace('\r', '') for c in df.columns]
90

91
                # 注入数据审计来源元数据
92
                df['数据源文件'] = file_path.name
93
                df['来源分部'] = sheet
94

95
                combined_records.append(df)
96
        except Exception as e:
97
            print(f"[解析异常] 文件 {file_path.name} 解析失败: {str(e)}")
98

99
    if not combined_records:
100
        print("[错误] 未能成功提取到任何有效数据记录")
101
        return False
102

103
    # 纵向拼接所有 DataFrame
104
    master_df = pd.concat(combined_records, ignore_index=True)
105

106
    # 针对关键业务列进行数据类型强制归一化
107
    if '销售金额' in master_df.columns:
108
        master_df['销售金额'] = master_df['销售金额'].apply(clean_currency_string)
109
    if '订单日期' in master_df.columns:
110
        master_df['订单日期'] = pd.to_datetime(master_df['订单日期'], errors='coerce').dt.strftime('%Y-%m-%d')
111

112
    # 基于主键防重过滤（如：订单号 + 工号唯一）
113
    if '订单编号' in master_df.columns:
114
        initial_len = len(master_df)
115
        master_df.drop_duplicates(subset=['订单编号'], keep='first', inplace=True)
116
        print(f"[去重完成] 成功剔除重复订单 {initial_len - len(master_df)} 条，保留有效数据 {len(master_df)} 条")
117

118
    master_df.to_excel(output_filepath, index=False, engine='openpyxl')
119
    print(f"[SUCCESS] 基础总表已输出至: {output_filepath}")
120
    return True

1
from openpyxl import load_workbook
2
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
3
from openpyxl.utils import get_column_letter
4

5
def beautify_excel_report(excel_path):
6
    """
7
    使用 openpyxl 对生成的汇总报表进行专业视觉排版与公式注入
8
    """
9
    wb = load_workbook(excel_path)
10
    ws = wb.active
11

12
    # 定义企业级专业配色与边框样式
13
    header_font = Font(name="微软雅黑", size=11, bold=True, color="FFFFFF")
14
    header_fill = PatternFill(start_color="1F497D", end_color="1F497D", fill_type="solid")
15
    data_font = Font(name="微软雅黑", size=10)
16
    alert_font = Font(name="微软雅黑", size=10, bold=True, color="9C0006")
17
    alert_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
18

19
    thin_border = Border(
20
        left=Side(style='thin', color='D9D9D9'),
21
        right=Side(style='thin', color='D9D9D9'),
22
        top=Side(style='thin', color='D9D9D9'),
23
        bottom=Side(style='thin', color='D9D9D9')
24
    )
25

26
    max_row = ws.max_row
27
    max_col = ws.max_column
28

29
    # 1. 格式化表头行
30
    for col_idx in range(1, max_col + 1):
31
        cell = ws.cell(row=1, column=col_idx)
32
        cell.font = header_font
33
        cell.fill = header_fill
34
        cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
35
        cell.border = thin_border
36
    ws.row_dimensions[1].height = 28
37

38
    # 2. 格式化数据行与高价值数据条件预警
39
    sales_col_idx = None
40
    for col_idx in range(1, max_col + 1):
41
        if ws.cell(row=1, column=col_idx).value == '销售金额':
42
            sales_col_idx = col_idx
43
            break
44

45
    for row_idx in range(2, max_row + 1):
46
        ws.row_dimensions[row_idx].height = 20
47
        for col_idx in range(1, max_col + 1):
48
            cell = ws.cell(row=row_idx, column=col_idx)
49
            cell.font = data_font
50
            cell.border = thin_border
51
            cell.alignment = Alignment(vertical="center")
52

53
            # 若是金额列，绑定标准财务千分位格式
54
            if col_idx == sales_col_idx:
55
                cell.number_format = '#,##0.00'
56
                cell.alignment = Alignment(horizontal="right", vertical="center")
57
                # 业务规则高亮：单笔业绩大于 50,000 进行红色浅底警示关注
58
                try:
59
                    if float(cell.value or 0) >= 50000:
60
                        cell.fill = alert_fill
61
                        cell.font = alert_font
62
                except (ValueError, TypeError):
63
                    pass
64

65
    # 3. 注入动态求和公式与汇总行
66
    total_row_idx = max_row + 1
67
    ws.row_dimensions[total_row_idx].height = 24
68
    label_cell = ws.cell(row=total_row_idx, column=1, value="系统自动总计")
69
    label_cell.font = Font(name="微软雅黑", size=10, bold=True)
70
    label_cell.alignment = Alignment(horizontal="center", vertical="center")
71

72
    for c in range(1, max_col + 1):
73
        ws.cell(row=total_row_idx, column=c).border = thin_border
74

75
    if sales_col_idx:
76
        col_letter = get_column_letter(sales_col_idx)
77
        sum_formula = f"=SUM({col_letter}2:{col_letter}{max_row})"
78
        sum_cell = ws.cell(row=total_row_idx, column=sales_col_idx, value=sum_formula)
79
        sum_cell.font = Font(name="微软雅黑", size=11, bold=True, color="1F497D")
80
        sum_cell.number_format = '¥#,##0.00'
81
        sum_cell.alignment = Alignment(horizontal="right", vertical="center")
82

83
    # 4. 智能自适应列宽算法（兼容中文双字节宽度）
84
    for col in ws.columns:
85
        max_len = 0
86
        col_letter = get_column_letter(col[0].column)
87
        for cell in col:
88
            val_str = str(cell.value or '')
89
            # 计算字符串显示宽度：中文、日韩字符计 2 个字符宽度，ASCII 计 1 个
90
            current_width = 0
91
            for char in val_str:
92
                current_width += 2 if ord(char) > 127 else 1
93
            if current_width > max_len:
94
                max_len = current_width
95
        # 列宽适当增加安全外边距
96
        ws.column_dimensions[col_letter].width = max(max_len + 4, 12)
97

98
    # 冻结第一行表头，方便滚动查阅
99
    ws.freeze_panes = "A2"
100

101
    wb.save(excel_path)
102
    print(f"[排版完成] 专业样式与自适应列宽已成功应用至: {excel_path}")

1
from openpyxl import load_workbook
2

3
def process_huge_excel_stream(huge_file_path):
4
    """
5
    流式低内存处理几百兆的特大 Excel 文件
6
    """
7
    print(f"[流式启动] 正在以只读优化流打开: {huge_file_path}")
8
    # 启用只读与纯值模式，关闭格式样式解析以最大化提速
9
    wb = load_workbook(filename=huge_file_path, read_only=True, data_only=True)
10
    ws = wb.active
11

12
    valid_count = 0
13
    total_sales = 0.0
14

15
    # iter_rows() 返回的是按需生成的行迭代器，绝不会全量一次性载入内存
16
    for row_idx, row in enumerate(ws.iter_rows(values_only=True), start=1):
17
        if row_idx == 1:
18
            # 读取并记录表头
19
            headers = [str(cell) for cell in row]
20
            continue
21

22
        # 假设第 4 列是金额
23
        sales_val = row[3]
24
        if sales_val is not None:
25
            try:
26
                num = float(sales_val)
27
                total_sales += num
28
                valid_count += 1
29
            except ValueError:
30
                pass
31

32
        if row_idx % 50000 == 0:
33
            print(f"[处理进度] 已稳健解析 {row_idx} 行数据，当前累计金额: {total_sales:,.2f}")
34

35
    # 只读模式下必须显式关闭工作簿连接以释放底层文件句柄
36
    wb.close()
37
    print(f"[统计结束] 最终有效行数: {valid_count}, 总业绩合计: {total_sales:,.2f}")

1
import re
2
from docx import Document
3
from docx.shared import Pt, Inches, RGBColor
4
from docx.oxml.ns import qn
5
from docx.enum.text import WD_ALIGN_PARAGRAPH
6

7
def replace_paragraph_text_safely(paragraph, mapping_dict):
8
    """
9
    攻克 Word Run 切碎断裂难题：跨 Run 安全替换占位符并继承原始排版样式
10
    mapping_dict 示例: {'{{EMPLOYEE_NAME}}': '张三丰', '{{SALARY}}': '28,500.00'}
11
    """
12
    full_text = "".join(run.text for run in paragraph.runs)
13
    has_match = False
14

15
    for placeholder in mapping_dict.keys():
16
        if placeholder in full_text:
17
            has_match = True
18
            break
19

20
    if not has_match or not paragraph.runs:
21
        return
22

23
    # 执行全局段落级文本替换
24
    for placeholder, new_val in mapping_dict.items():
25
        full_text = full_text.replace(placeholder, str(new_val))
26

27
    # 获取第一个 Run 的核心排版元数据以便继承
28
    first_run = paragraph.runs[0]
29
    font_name = first_run.font.name
30
    font_size = first_run.font.size
31
    bold = first_run.bold
32
    italic = first_run.italic
33
    color = first_run.font.color.rgb if first_run.font.color else None
34

35
    # 策略：将替换后的完整文本赋给首个 Run，并彻底清空后续所有 Run 的文本
36
    first_run.text = full_text
37

38
    # 清空后续 Run，避免字符残留重复
39
    for run in paragraph.runs[1:]:
40
        run.text = ""
41

42
def batch_replace_docx_template(template_path, output_path, mapping_dict):
43
    """
44
    对 Word 文档的段落、表格单元格、页眉页脚执行全方位深度替换
45
    """
46
    doc = Document(template_path)
47

48
    # 1. 遍历正文普通段落
49
    for paragraph in doc.paragraphs:
50
        replace_paragraph_text_safely(paragraph, mapping_dict)
51

52
    # 2. 遍历正文所有表格及其嵌套段落
53
    for table in doc.tables:
54
        for row in table.rows:
55
            for cell in row.cells:
56
                for paragraph in cell.paragraphs:
57
                    replace_paragraph_text_safely(paragraph, mapping_dict)
58

59
    # 3. 遍历节（Section）中的页眉与页脚
60
    for section in doc.sections:
61
        for p in section.header.paragraphs:
62
            replace_paragraph_text_safely(p, mapping_dict)
63
        for p in section.footer.paragraphs:
64
            replace_paragraph_text_safely(p, mapping_dict)
65

66
    doc.save(output_path)
67
    print(f"[Word生成成功] 文件已保存至: {output_path}")

1
def render_dynamic_contract_table(doc, items_data, stamp_image_path=None):
2
    """
3
    在 Word 表格中动态追加商品条目，并插入电子公章
4
    """
5
    # 假设文档中的第一个表格是商品结算明细表
6
    if not doc.tables:
7
        return
8
    table = doc.tables[0]
9

10
    # 找到数据模板行（假设第 2 行为示例行，第 1 行为表头）
11
    # 遍历真实采购列表追加行
12
    total_amount = 0.0
13

14
    for item in items_data:
15
        # 添加新行
16
        row_cells = table.add_row().cells
17
        name = item.get("name", "")
18
        qty = item.get("quantity", 0)
19
        price = item.get("unit_price", 0.0)
20
        subtotal = qty * price
21
        total_amount += subtotal
22

23
        row_cells[0].text = str(name)
24
        row_cells[1].text = str(qty)
25
        row_cells[2].text = f"¥{price:,.2f}"
26
        row_cells[3].text = f"¥{subtotal:,.2f}"
27

28
        # 统一设置单元格字体与居中
29
        for c_idx, cell in enumerate(row_cells):
30
            cell.paragraphs[0].alignment = WD_ALIGN_PARAGRAPH.RIGHT if c_idx >= 1 else WD_ALIGN_PARAGRAPH.LEFT
31
            for r in cell.paragraphs[0].runs:
32
                r.font.name = "微软雅黑"
33
                r.font.size = Pt(10)
34
                # 设置中文字体 EastAsia 兼容
35
                r._element.rPr.rFonts.set(qn('w:eastAsia'), '微软雅黑')
36

37
    # 追加总计行
38
    total_cells = table.add_row().cells
39
    total_cells[0].text = "合计总金额"
40
    total_cells[3].text = f"¥{total_amount:,.2f}"
41

42
    # 落款处自动盖章
43
    if stamp_image_path and Path(stamp_image_path).exists():
44
        # 寻找落款段落
45
        for p in doc.paragraphs:
46
            if "乙方（盖章）" in p.text:
47
                run = p.add_run()
48
                # 插入宽度为 1.5 英寸的高清透明电子印章 PNG
49
                run.add_picture(stamp_image_path, width=Inches(1.5))
50
                break

1
import sys
2
import subprocess
3
from pathlib import Path
4

5
def convert_word_to_pdf_cross_platform(docx_path, pdf_path):
6
    """
7
    跨平台 Word 转 PDF：Windows 下调用 Word COM，Linux/Mac 下调用 LibreOffice
8
    """
9
    docx_file = Path(docx_path).resolve()
10
    pdf_file = Path(pdf_path).resolve()
11

12
    if sys.platform == "win32":
13
        try:
14
            import win32com.client
15
            import pythoncom
16
            # 初始化 COM 线程上下文，避免在多线程调度中报错
17
            pythoncom.CoInitialize()
18
            word = win32com.client.DispatchEx("Word.Application")
19
            word.Visible = False
20
            word.DisplayAlerts = False
21

22
            doc = word.Documents.Open(str(docx_file))
23
            # 17 代表 wdFormatPDF 格式常量代码
24
            doc.SaveAs(str(pdf_file), FileFormat=17)
25
            doc.Close()
26
            word.Quit()
27
            pythoncom.CoUninitialize()
28
            print(f"[Win COM 转 PDF 成功] {pdf_file.name}")
29
            return True
30
        except Exception as e:
31
            print(f"[Win COM 转换异常] 回退尝试命令行: {e}")
32

33
    # Linux / macOS 或 Windows 回退方案：调用 libreoffice CLI
34
    cmd = [
35
        "libreoffice",
36
        "--headless",
37
        "--convert-to", "pdf",
38
        "--outdir", str(pdf_file.parent),
39
        str(docx_file)
40
    ]
41
    try:
42
        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
43
        print(f"[LibreOffice 转 PDF 成功] {pdf_file.name}")
44
        return True
45
    except FileNotFoundError:
46
        print("[错误] 未检测到 LibreOffice 环境，请在服务器执行: apt install libreoffice")
47
        return False

1
import fitz  # PyMuPDF 核心模块
2
from pathlib import Path
3

4
def split_pdf_by_bookmark(input_pdf_path, output_dir):
5
    """
6
    根据 PDF 内部的一级书签（目录大纲）自动将大型 PDF 切割为独立章节文件
7
    """
8
    doc = fitz.open(input_pdf_path)
9
    toc = doc.get_toc()  # 返回格式: [[lvl, title, page_num], ...]
10
    out_dir = Path(output_dir)
11
    out_dir.mkdir(parents=True, exist_ok=True)
12

13
    if not toc:
14
        print("[提示] 该 PDF 未包含任何目录书签，回退为固定分页拆分")
15
        return
16

17
    # 筛选出所有 1 级主目录书签
18
    level_1_bookmarks = [b for b in toc if b[0] == 1]
19
    total_pages = len(doc)
20

21
    for i, bm in enumerate(level_1_bookmarks):
22
        title = bm[1].replace('/', '_').replace('\\', '_').strip()
23
        start_page = bm[2] - 1  # fitz 页码从 0 索引
24

25
        # 计算当前章节的结束页码
26
        if i + 1 < len(level_1_bookmarks):
27
            end_page = level_1_bookmarks[i + 1][2] - 2
28
        else:
29
            end_page = total_pages - 1
30

31
        # 实例化新的独立 PDF 容器
32
        sub_doc = fitz.open()
33
        # insert_pdf 是底层页面内存指针浅拷贝，速度极快
34
        sub_doc.insert_pdf(doc, from_page=start_page, to_page=end_page)
35

36
        target_file = out_dir / f"{i+1:02d}_{title}.pdf"
37
        sub_doc.save(str(target_file))
38
        sub_doc.close()
39
        print(f"[拆分完成] 已提取章节: {target_file.name} (页码: {start_page+1} 至 {end_page+1})")
40

41
    doc.close()

1
import pdfplumber
2
import pandas as pd
3

4
def extract_tables_from_pdf(pdf_path):
5
    """
6
    高保真提取 PDF 中的多页复杂表格并拼装为 DataFrame
7
    """
8
    all_rows = []
9

10
    with pdfplumber.open(pdf_path) as pdf:
11
        for page_idx, page in enumerate(pdf.pages):
12
            # extract_table 可以传入自定义的 table_settings，如 explicitly_vertical_lines
13
            table = page.extract_table({
14
                "vertical_strategy": "lines",
15
                "horizontal_strategy": "lines",
16
                "snap_tolerance": 3,
17
                "join_tolerance": 3
18
            })
19

20
            # 如果默认线检测失败，回退到基于文本对齐策略
21
            if not table:
22
                table = page.extract_table({
23
                    "vertical_strategy": "text",
24
                    "horizontal_strategy": "text"
25
                })
26

27
            if table:
28
                # 第一页的第 0 行通常是表头，后续页的第 0 行若重复出现需去重
29
                if page_idx == 0:
30
                    headers = [str(c).strip().replace('\n', '') for c in table[0]]
31
                    data_rows = table[1:]
32
                else:
33
                    data_rows = table
34
                    # 如果后续页重复打印了表头，进行比对剔除
35
                    if [str(c).strip().replace('\n', '') for c in data_rows[0]] == headers:
36
                        data_rows = data_rows[1:]
37

38
                for r in data_rows:
39
                    cleaned_row = [str(c).strip().replace('\n', '') if c is not None else '' for c in r]
40
                    all_rows.append(cleaned_row)
41

42
    if all_rows and headers:
43
        df = pd.DataFrame(all_rows, columns=headers)
44
        print(f"[PDF 表格提取成功] 共提取有效表格记录 {len(df)} 行！")
45
        return df
46
    return None

1
import fitz
2

3
def add_security_watermark_and_encrypt(input_pdf, output_pdf, watermark_text, password=None):
4
    """
5
    为 PDF 页面注入半透明 45 度倾斜矢量水印，并执行安全加密
6
    """
7
    doc = fitz.open(input_pdf)
8

9
    # 逐页压入防伪水印
10
    for page in doc:
11
        rect = page.rect
12
        # 水印起始位置放置在页面中心
13
        p = fitz.Point(rect.width / 4, rect.height / 2)
14

15
        # insert_text 拥有极高的渲染质量与透明度支持
16
        page.insert_text(
17
            p,
18
            watermark_text,
19
            fontsize=36,
20
            fontname="china-s",  # 使用内置简体中文字体
21
            color=(0.7, 0.7, 0.7),  # 柔和浅灰色
22
            fill_opacity=0.35,      # 35% 半透明度，绝不遮挡底层正文
23
            rotate=45               # 45 度对角线倾斜
24
        )
25

26
    # 保存参数配置：如果指定密码，启用 256-bit AES 加密并禁用修改权限
27
    save_args = {
28
        "garbage": 4,      # 深度垃圾回收优化，缩减体积
29
        "deflate": True    # 启用 zlib 流压缩
30
    }
31

32
    if password:
33
        save_args["encryption"] = fitz.PDF_ENCRYPT_AES_256
34
        save_args["user_pw"] = password      # 打开阅读密码
35
        save_args["owner_pw"] = password + "_admin"  # 管理员权限密码
36
        # 仅保留打印权限，禁止复制正文与表单修改
37
        save_args["permissions"] = fitz.PDF_PERM_PRINT
38

39
    doc.save(output_pdf, **save_args)
40
    doc.close()
41
    print(f"[安全加固完成] 水印与加密已成功输出: {output_pdf}")

1
from pathlib import Path
2
import datetime
3

4
# 优雅的跨平台路径拼接与探索
5
base_dir = Path.home() / "Documents" / "FinancialReports"
6
target_year = 2026
7

8
# rglob 实现全盘深层递归遍历
9
for file_path in base_dir.rglob("*.xlsx"):
10
    # 获取无后缀主干名、扩展名、父级目录
11
    stem = file_path.stem
12
    suffix = file_path.suffix
13
    parent = file_path.parent
14

15
    # 获取最后修改时间
16
    mod_time = datetime.datetime.fromtimestamp(file_path.stat().st_mtime)
17
    print(f"发现报表: {stem} [后缀: {suffix}] [修改于: {mod_time:%Y-%m-%d %H:%M}]")

1
import hashlib
2
from pathlib import Path
3
from collections import defaultdict
4

5
def calculate_file_sha256(file_path, chunk_size=65536):
6
    """
7
    分块流式读取文件，低内存计算精准的 SHA-256 哈希值
8
    """
9
    sha256 = hashlib.sha256()
10
    with open(file_path, 'rb') as f:
11
        while True:
12
            chunk = f.read(chunk_size)
13
            if not chunk:
14
                break
15
            sha256.update(chunk)
16
    return sha256.hexdigest()
17

18
def deduplicate_directory(target_folder, quarantine_folder):
19
    """
20
    安全去重引擎：不直接硬删除，而是将重复文件移动至隔离区，并生成硬链接或审计清单
21
    """
22
    target_path = Path(target_folder)
23
    quarantine_path = Path(quarantine_folder)
24
    quarantine_path.mkdir(parents=True, exist_ok=True)
25

26
    # 第一阶段：先基于文件大小进行初筛分组（优化性能：大小不同的文件绝不可能是重复文件）
27
    size_map = defaultdict(list)
28
    for p in target_path.rglob("*"):
29
        if p.is_file() and not p.name.startswith("~$"):
30
            try:
31
                size_map[p.stat().st_size].append(p)
32
            except OSError:
33
                continue
34

35
    # 第二阶段：对大小相同的“嫌疑文件”深度比对 SHA-256 指纹
36
    hash_seen = {}
37
    duplicates_count = 0
38
    reclaimed_bytes = 0
39

40
    for size, files in size_map.items():
41
        if len(files) < 2 or size == 0:
42
            continue
43

44
        for file_p in files:
45
            file_hash = calculate_file_sha256(file_p)
46
            if file_hash in hash_seen:
47
                original_p = hash_seen[file_hash]
48
                duplicates_count += 1
49
                reclaimed_bytes += size
50
                print(f"[发现重复] '{file_p.name}' 完全等同于 '{original_p.name}'")
51

52
                # 安全策略：平移至隔离归档目录，保留源目录干净
53
                dest_p = quarantine_path / file_p.name
54
                # 处理隔离区同名冲突
55
                counter = 1
56
                while dest_p.exists():
57
                    dest_p = quarantine_path / f"{file_p.stem}_dup{counter}{file_p.suffix}"
58
                    counter += 1
59
                file_p.rename(dest_p)
60
            else:
61
                hash_seen[file_hash] = file_p
62

63
    print(f"[清理完毕] 共归档重复文件 {duplicates_count} 个，释放磁盘空间: {reclaimed_bytes / (1024*1024):.2f} MB")

1
import shutil
2
from pathlib import Path
3
import datetime
4

5
def smart_organize_workspace(inbox_dir, archive_root):
6
    """
7
    智能文件分流器：按文件类型与生成时间自动建立多级树状结构
8
    """
9
    inbox = Path(inbox_dir)
10
    archive = Path(archive_root)
11

12
    # 类别映射规则字典
13
    CATEGORY_MAPPING = {
14
        "表格报表": [".xlsx", ".xls", ".csv", ".tsv"],
15
        "正式文书": [".docx", ".doc", ".pdf", ".rtf"],
16
        "演示简报": [".pptx", ".ppt"],
17
        "设计图像": [".png", ".jpg", ".jpeg", ".webp", ".psd"],
18
        "代码脚本": [".py", ".sh", ".bat", ".sql", ".json"],
19
        "归档压缩": [".zip", ".rar", ".7z", ".tar", ".gz"]
20
    }
21

22
    for item in inbox.iterdir():
23
        if item.is_dir() or item.name.startswith("."):
24
            continue
25

26
        ext = item.suffix.lower()
27
        # 寻找匹配分类，未匹配的归入“其他资产”
28
        category = "其他资产"
29
        for cat, ext_list in CATEGORY_MAPPING.items():
30
            if ext in ext_list:
31
                category = cat
32
                break
33

34
        # 提取文件创建/最后修改年月
35
        mtime = datetime.datetime.fromtimestamp(item.stat().st_mtime)
36
        year_str = mtime.strftime("%Y年")
37
        month_str = mtime.strftime("%m月")
38

39
        # 动态构建目的目录：归档根目录 / 资产大类 / 2026年 / 03月
40
        target_folder = archive / category / year_str / month_str
41
        target_folder.mkdir(parents=True, exist_ok=True)
42

43
        target_file = target_folder / item.name
44
        # 目标已存在同名文件时的避让算法
45
        if target_file.exists():
46
            target_file = target_folder / f"{item.stem}_{mtime.strftime('%Y%m%d%H%M%S')}{item.suffix}"
47

48
        shutil.move(str(item), str(target_file))
49
        print(f"[自动归档] {item.name} -> {target_folder.relative_to(archive)}")