cfc/docs/参考资料/dan_reports/dedup_reports.py

# -*- coding: utf-8 -*-
"""
去重重复报告：找出 姓名+生日+测评日期+报告类型 完全相同的记录，
保留格式规范的最新文件，删除旧格式的重复PDF。

用法: python scripts/reports/dedup_reports.py
"""
import os, csv, re, shutil
from collections import Counter

PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
REPORT_DIR = os.path.join(PROJECT_DIR, '报告')
CSV_PATH = os.path.join(PROJECT_DIR, '素材库', '测评数据_优化提取.csv')
BACKUP_PATH = CSV_PATH.replace('.csv', '_before_dedup.csv')


def is_proper_format(filename):
    """判断是否为规范格式的文件名（保留此类）。
    
    规范格式:
      - 带时间戳: "B4 核心认知能力和成长型思维_姓名_202403231742254603.pdf"
      - 无时间戳: "B6_生涯规划（高中版）_姓名.pdf"
      - 核心特征: 文件名不包含 "DAN素质能力测评报告" 等旧格式标记,
                 且前缀类型与文件名其他内容之间不含旧命名痕迹
    
    旧格式: "A1_DAN素质能力测评报告_姓名*.pdf", 
            "A2_B6生涯规划（高中版）_姓名*.pdf",
            "B6_DAN素质能力测评报告-姓名*.pdf"
    """
    # 旧格式特征：包含 DAN素质能力测评报告 字样
    if 'DAN素质能力测评报告' in filename:
        return False
    
    # 旧格式特征：A2_B6 开头的旧生涯规划文件
    if filename.startswith('A2_B6'):
        return False
    
    # 规范格式特征：
    # 1. {类型}{空格}{中文描述}_{姓名}_{时间戳(14-18位)}.pdf
    if re.match(r'^[A-Z]\d\s+\S+?_\S+?_\d{14,}\.pdf$', filename):
        return True
    # 2. {类型}_{中文描述}_{姓名}.pdf（无时间戳，有中文描述字符≥3）
    if re.match(r'^[A-Z]\d_\S{3,}_.+\.pdf$', filename):
        return True
    
    return False


def main():
    # 1. 读取CSV
    with open(CSV_PATH, 'r', encoding='utf-8-sig') as f:
        reader = csv.DictReader(f)
        rows = list(reader)
        fieldnames = reader.fieldnames
    
    print(f'总记录数: {len(rows)}')
    
    # 2. 按 姓名+生日+测评日期+报告类型 分组
    groups = {}
    for r in rows:
        k = (r.get('姓名', ''), r.get('生日', ''), r.get('测评日期', ''), r.get('报告类型', ''))
        groups.setdefault(k, []).append(r)
    
    # 3. 找出重复组
    dup_groups = {k: v for k, v in groups.items() if len(v) > 1}
    print(f'重复组数: {len(dup_groups)}')
    
    to_delete_files = []
    kept_rows = []
    
    for k, dup_rows in sorted(dup_groups.items(), key=lambda x: -len(x[1])):
        name, birthday, date, rtype = k
        print(f'\n组: {name} | {birthday} | {date} | {rtype} (x{len(dup_rows)})')
        
        # 按文件规范程度排序：规范格式优先保留
        # 如果多个规范格式，保留文件名排序最后的（时间戳最新的）
        proper = [r for r in dup_rows if is_proper_format(r.get('filename', ''))]
        old = [r for r in dup_rows if not is_proper_format(r.get('filename', ''))]
        
        if proper:
            # 保留规范格式中时间戳最新的
            proper.sort(key=lambda r: r.get('filename', ''), reverse=True)
            keep = proper[0]
            delete = proper[1:] + old
        else:
            # 没有规范格式，保留文件名排序最后的
            dup_rows.sort(key=lambda r: r.get('filename', ''), reverse=True)
            keep = dup_rows[0]
            delete = dup_rows[1:]
        
        print(f'  保留: {keep["filename"][:50]}')
        kept_rows.append(keep)
        
        for r in delete:
            fname = r['filename']
            fpath = os.path.join(REPORT_DIR, fname)
            print(f'  删除: {fname[:50]}')
            if os.path.exists(fpath):
                to_delete_files.append(fpath)
    
    # 4. 加入非重复的记录
    non_dup = [rows for k, rows in groups.items() if len(rows) == 1]
    for group in non_dup:
        kept_rows.extend(group)
    
    print(f'\n保留记录: {len(kept_rows)}')
    print(f'待删除文件: {len(to_delete_files)}')
    
    # 5. 备份原CSV
    shutil.copy2(CSV_PATH, BACKUP_PATH)
    print(f'备份CSV: {BACKUP_PATH}')
    
    # 6. 写入新CSV
    with open(CSV_PATH, 'w', encoding='utf-8-sig', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(kept_rows)
    print(f'写入新CSV: {CSV_PATH} ({len(kept_rows)} 记录)')
    
    # 7. 删除旧文件
    if to_delete_files:
        print(f'\n删除 {len(to_delete_files)} 个重复PDF...')
        for fpath in to_delete_files:
            try:
                os.remove(fpath)
                print(f'  DELETED: {os.path.basename(fpath)}')
            except Exception as e:
                print(f'  FAILED: {os.path.basename(fpath)} - {e}')
    
    print(f'\n完成！共去重 {sum(len(v) for v in dup_groups.values()) - len(dup_groups)} 条记录，删除 {len(to_delete_files)} 个文件')


if __name__ == '__main__':
    main()