feat: add DAN report to JSON extraction script (extract_dan_to_json.py)

docs/参考资料/dan_reports/extract_dan_to_json.py

@@ -0,0 +1,302 @@
+# -*- coding: utf-8 -*-
+"""
+DAN测评报告PDF → JSON提取脚本
+
+对标菌群报告的 extract_full_report_v5.py，将DAN测评报告(PDF)提取为结构化JSON。
+8种报告类型: A1, A2, B2, B3, B4, B5, B6, C1
+
+用法:
+  python extract_dan_to_json.py <pdf_path>   # 单文件模式 → <同名>.json
+  python extract_dan_to_json.py               # 批量模式 → 所有PDF输出JSON
+
+依赖: extract_all_types.py, type_detector.py（同目录下）
+"""
+
+import os
+import sys
+import json
+import csv
+from extract_all_types import (
+    extract_all_data, extract_text_from_pdf,
+    extract_name_from_filename, extract_birthday_from_pdf,
+    extract_date_from_filename, extract_score_and_percentile,
+    extract_a1_data, extract_a2_data, extract_b2_data,
+    extract_b3_data, extract_b4_data, extract_b5_data,
+    extract_b6_data, extract_c1_data,
+)
+from type_detector import detect_type
+
+
+# ==========================================
+# 维度定义：每个报告类型的指标分组
+# ==========================================
+TYPE_DIMENSIONS = {
+    'A1': {
+        '报告类型': '儿童核心认知发展',
+        '维度分组': {
+            '核心认知': ['感知觉', '注意力', '记忆力', '推理能力', '空间能力', '加工速度'],
+        }
+    },
+    'A2': {
+        '报告类型': '核心素养',
+        '维度分组': {
+            '认知维度': ['感知觉', '记忆力', '注意力'],
+            '情绪状态': ['自卑-自信', '抑郁-安详', '焦虑-安详', '无力感-掌控感', '情绪总分'],
+            '大五人格': ['开放性', '宜人性', '责任心', '外倾性', '神经质'],
+            '社会关系': ['与母亲信任', '与父亲信任', '与同伴信任',
+                       '与母亲沟通', '与父亲沟通', '与同伴沟通',
+                       '与母亲亲近', '与父亲亲近', '与同伴亲近'],
+            '身体健康': ['BMI', '身高', '体重', '睡眠_小时', '饮食_小时'],
+        }
+    },
+    'B2': {
+        '报告类型': '儿童自我与家庭教养',
+        '维度分组': {
+            '自我概念': ['行为表现', '能力与学校', '躯体外貌', '情绪状态', '合群', '幸福与满足'],
+            '儿童行为': ['品行问题', '情绪问题', '学习问题', '社交问题',
+                       '生活习惯', '多动倾向', '刻板行为', '拖延行为'],
+            '家庭环境': ['亲密', '情感表达', '和谐', '独立性', '成就向导',
+                       '文化氛围', '娱乐活动', '道德观念', '家务安排', '家庭规则'],
+        }
+    },
+    'B3': {
+        '报告类型': '核心学习能力',
+        '维度分组': {
+            '执行功能': ['抑制控制', '工作记忆', '认知灵活性'],
+            '学习动机': ['深层动机', '表面动机', '自我效能感'],
+            '学习策略': ['深层方法与策略', '表面方法与策略', '学习自我调节'],
+        }
+    },
+    'B4': {
+        '报告类型': '核心认知能力+自我概念+自驱力+成长型思维',
+        '维度分组': {
+            '核心认知': ['感知觉', '注意力', '记忆力', '推理能力', '空间能力', '加工速度'],
+            '自我概念': ['行为表现', '能力与学校', '躯体外貌', '情绪状态', '合群', '幸福与满足'],
+            '自驱力': ['自主性', '胜任感', '归属感'],
+            '成长型思维': ['成长型思维'],
+        }
+    },
+    'B5': {
+        '报告类型': '青春期挑战',
+        '维度分组': {
+            '情绪调节': ['认知重评', '表达抑制'],
+            '学业压力': ['学业负担', '家庭期望', '师生关系', '自我期望', '同伴竞争'],
+            '人际关系': ['母亲关系', '父亲关系', '师生关系', '同伴关系'],
+            '社交能力': ['主动交往', '情感支持', '情感表达', '表达影响', '冲突解决'],
+            '睡眠': ['时长', '效率', '主观质量', '日间功能', '催眠药物', '障碍'],
+            '运动': ['久坐', '步行', '中等强度', '高强度'],
+            '网络依赖': ['总分', '强迫使用', '时间管理', '戒断反应'],
+        }
+    },
+    'B6': {
+        '报告类型': '职业发展',
+        '维度分组': {
+            '职业兴趣': ['艺术型', '社会型', '事业型', '常规型', '现实型', '研究型'],
+            '多元智能': ['内省', '空间', '音乐', '人际关系', '自然', '身体运动', '语言', '逻辑数学'],
+            '职业价值观': ['职业价值观'],
+        }
+    },
+    'C1': {
+        '报告类型': '校园版综合',
+        '维度分组': {
+            '核心认知': ['感知觉', '注意力', '记忆力', '推理能力', '空间能力', '加工速度'],
+            '大五人格': ['开放性', '宜人性', '责任心', '外倾性', '神经质'],
+            '自驱力': ['自主性', '胜任感', '归属感'],
+            '自我概念': ['行为表现', '能力与学校', '躯体外貌', '情绪状态', '合群', '幸福与满足'],
+        }
+    },
+}
+
+
+# ==========================================
+# 平铺行数据 → 嵌套JSON
+# ==========================================
+def flat_row_to_nested(row):
+    """将平铺的CSV行数据转换为按类型分组的嵌套JSON"""
+    result = {
+        '文件名': row.get('filename', ''),
+        '姓名': row.get('姓名', ''),
+        '生日': row.get('生日', ''),
+        '报告类型': row.get('报告类型', ''),
+        '测评日期': row.get('测评日期', ''),
+        '总分': row.get('总分', ''),
+        '百分位': row.get('百分位', ''),
+    }
+
+    report_type = row.get('报告类型', '')
+    type_info = TYPE_DIMENSIONS.get(report_type)
+    if not type_info:
+        return result
+
+    result['报告类型名称'] = type_info['报告类型']
+    result['维度数据'] = {}
+
+    for group_name, dims in type_info['维度分组'].items():
+        group_data = {}
+        for dim in dims:
+            # 尝试多种key匹配
+            val = _find_value(row, dim)
+            if val:
+                group_data[dim] = val
+        if group_data:
+            result['维度数据'][group_name] = group_data
+
+    # B4/B5的特殊处理：B4成长型思维是连续分数，B5有人际指南针子维度
+    if report_type == 'B4' and row.get('成长型思维'):
+        result['维度数据'].setdefault('成长型思维', {})['连续分数'] = row['成长型思维']
+
+    if report_type == 'B5':
+        # 收集19个指南针子维度
+        compass_dims = {}
+        for key, val in row.items():
+            if key.startswith('人际_') and '_' in key and key.count('_') >= 2:
+                compass_dims[key] = val
+        if compass_dims:
+            result['维度数据']['人际指南针'] = compass_dims
+
+    return result
+
+
+def _find_value(row, dim):
+    """从平铺行中查找维度值，尝试各种key命名规则"""
+    # 直接匹配
+    if row.get(dim):
+        return row[dim]
+
+    # 后缀匹配 (如 '感知觉_pct', '行为表现_score')
+    for suffix in ['_pct', '_score', '_小时', '_min', '_频率', '_时长', '_效率', '%']:
+        key = f'{dim}{suffix}'
+        if row.get(key):
+            val = row[key]
+            return val
+
+    # B2: '自我概念_行为表现', '行为_品行问题', '家庭_亲密'
+    for prefix in ['自我概念_', '行为_', '家庭_', '学业压力_', '人际_', '社交_',
+                   '睡眠_', '运动_', '网络依赖_', '兴趣_', '能力_',
+                   '与母亲', '与父亲', '与同伴']:
+        for key in row:
+            if key.startswith(prefix) and dim in key:
+                return row[key]
+
+    # 反向匹配：row中的key包含dim（如 key='感知觉_pct', dim='感知觉'）
+    for key in row:
+        if key.startswith(dim) or key.endswith(dim):
+            return row[key]
+
+    return ''
+
+
+# ==========================================
+# 单PDF → JSON
+# ==========================================
+def extract_dan_to_json(pdf_path):
+    """输入单个PDF文件路径，输出JSON格式的结构化数据"""
+    filename = os.path.basename(pdf_path)
+    row = extract_all_data(pdf_path, filename)
+    result = flat_row_to_nested(row)
+
+    # 补充原始文本（调试用）
+    text = extract_text_from_pdf(pdf_path)
+    result['_raw_text_preview'] = text[:500] if text else ''
+
+    return result
+
+
+# ==========================================
+# 批量模式
+# ==========================================
+def batch_process():
+    """处理当前目录下所有PDF，输出JSON + CSV汇总"""
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    pdf_files = sorted([
+        f for f in os.listdir(script_dir)
+        if f.endswith('.pdf')
+    ])
+
+    if not pdf_files:
+        print('未找到PDF文件')
+        return
+
+    all_results = []
+    for pdf_file in pdf_files:
+        pdf_path = os.path.join(script_dir, pdf_file)
+        label = pdf_file.replace('.pdf', '')
+        try:
+            result = extract_dan_to_json(pdf_path)
+            json_path = os.path.join(script_dir, f'{label}.json')
+            with open(json_path, 'w', encoding='utf-8') as f:
+                json.dump(result, f, ensure_ascii=False, indent=2)
+            print(f'  ✓ {pdf_file} → {label}.json')
+            all_results.append(result)
+        except Exception as e:
+            print(f'  ✗ {pdf_file}: ERROR - {e}')
+
+    # CSV汇总
+    if all_results:
+        csv_path = os.path.join(script_dir, 'dan测评汇总.csv')
+        # 收集所有key
+        all_keys = set()
+        flat_rows = []
+        for r in all_results:
+            flat = _flatten_json(r)
+            flat_rows.append(flat)
+            all_keys.update(flat.keys())
+
+        with open(csv_path, 'w', newline='', encoding='utf-8-sig') as f:
+            sorted_keys = sorted(all_keys)
+            w = csv.DictWriter(f, fieldnames=sorted_keys)
+            w.writeheader()
+            w.writerows(flat_rows)
+        print(f'\nCSV汇总: {csv_path} ({len(all_results)} 条)')
+
+
+def _flatten_json(nested):
+    """将嵌套JSON展平为扁平行"""
+    flat = {}
+    flat['文件名'] = nested.get('文件名', '')
+    flat['姓名'] = nested.get('姓名', '')
+    flat['生日'] = nested.get('生日', '')
+    flat['报告类型'] = nested.get('报告类型', '')
+    flat['报告类型名称'] = nested.get('报告类型名称', '')
+    flat['测评日期'] = nested.get('测评日期', '')
+    flat['总分'] = nested.get('总分', '')
+    flat['百分位'] = nested.get('百分位', '')
+
+    dims = nested.get('维度数据', {})
+    for group_name, group_data in dims.items():
+        for dim_name, dim_val in group_data.items():
+            flat[f'{group_name}_{dim_name}'] = dim_val
+
+    # 补充B5指南针
+    if '人际指南针' in dims:
+        for key, val in dims['人际指南针'].items():
+            flat[key] = val
+
+    return flat
+
+
+# ==========================================
+# 主入口
+# ==========================================
+def main():
+    if len(sys.argv) >= 2:
+        pdf_path = sys.argv[1]
+        if not os.path.isfile(pdf_path):
+            print(f'错误：找不到文件 {pdf_path}')
+            sys.exit(1)
+
+        result = extract_dan_to_json(pdf_path)
+        base, _ = os.path.splitext(pdf_path)
+        json_path = base + '.json'
+        with open(json_path, 'w', encoding='utf-8') as f:
+            json.dump(result, f, ensure_ascii=False, indent=2)
+        print(f'已生成: {json_path}')
+        print(f'  姓名: {result.get("姓名", "")}')
+        print(f'  类型: {result.get("报告类型", "")} - {result.get("报告类型名称", "")}')
+        print(f'  维度: {len(result.get("维度数据", {}))} 组')
+    else:
+        batch_process()
+
+
+if __name__ == '__main__':
+    main()