#!/usr/bin/env python3 """ 一键运行：实体属性表 + 三元组 的读取、清洗、导出 用法: python scripts/run_all.py """ import json import os import sys _script_dir = os.path.dirname(os.path.abspath(__file__)) _project_root = os.path.dirname(_script_dir) sys.path.insert(0, _project_root) def _json_serializer(obj): if hasattr(obj, 'isoformat'): return obj.isoformat() if hasattr(obj, 'item'): return obj.item() if isinstance(obj, float) and (obj != obj or obj == float('inf') or obj == float('-inf')): return None raise TypeError(f'Object of type {type(obj).__name__} is not JSON serializable') def load_config(project_root: str) -> dict: config_dir = os.path.join(project_root, 'config') json_path = os.path.join(config_dir, 'config.json') yaml_path = os.path.join(config_dir, 'config.yaml') if os.path.exists(json_path): with open(json_path, 'r', encoding='utf-8') as f: return json.load(f) if os.path.exists(yaml_path): try: import yaml with open(yaml_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) except ImportError: pass return {'source': 'json', 'json': {}, 'triplet': {}, 'db': {}} def _resolve_path(path: str, project_root: str) -> str: return path if os.path.isabs(path) else os.path.join(project_root, path) def main(): from src.cleaner import AdvancedDataCleaner from src.data_loader import load_data, load_triplets_from_dm, load_from_json, run_formalization_pipeline from src.triplet_cleaner import clean_triplets config = load_config(_project_root) source = config.get('source', 'json') tc = config.get('triplet', {}) # 三元组：db 模式默认处理；json 模式需 triplet.input_file；triplet.enabled=false 可关闭 triplet_input = tc.get('input_file') run_triplets = tc.get('enabled', source == 'db' or bool(triplet_input)) print("=" * 50) print("作战体系数据清洗 - 一键运行") print("=" * 50) print(f"数据源: {source}") # ---------- 1. 实体属性表 ---------- print("\n【1/2】实体属性表 读取与清洗") try: data, paths = load_data(config, _project_root) except (FileNotFoundError, ValueError) as e: print(f"错误：{e}") sys.exit(1) os.makedirs(os.path.dirname(paths['report']), exist_ok=True) cleaner = AdvancedDataCleaner(output_file=paths['output'], report_file=paths['report'], data=data) cleaner.run() valid_ids = cleaner.get_valid_target_ids() print(f" -> 输出: {paths['output']}") print(f" -> 报告: {paths['report']}") # ---------- 2. 三元组 ---------- if not run_triplets: print("\n【2/2】三元组 已跳过（triplet.enabled=false 或 json 模式未配置 input_file）") print("\n完成。") return print("\n【2/2】三元组 读取与清洗") if source == 'db': triplets = load_triplets_from_dm(config, _project_root) print(f" 从达梦读取 {len(triplets)} 条三元组") elif triplet_input: input_path = _resolve_path(triplet_input, _project_root) if os.path.exists(input_path): raw = load_from_json(input_path, _project_root) triplets = raw.get("triples", raw) if isinstance(raw, dict) else raw triplets = triplets if isinstance(triplets, list) else [] print(f" 从 {triplet_input} 读取 {len(triplets)} 条三元组") else: print(f" 跳过：三元组文件不存在 {input_path}") triplets = [] else: print(" 跳过：json 模式未配置 triplet.input_file") triplets = [] if not triplets: print(" 无三元组数据，跳过导出") print("\n完成。") return # 三元组清洗配置 clean_cfg = tc.get('clean', True) do_clean = {'deduplicate': True, 'filter_orphans': False, 'remove_null_core': True, 'normalize_relation_type': True} if clean_cfg is True else (clean_cfg if isinstance(clean_cfg, dict) else {}) deduplicate = do_clean.get('deduplicate', True) if do_clean else False filter_orphans = do_clean.get('filter_orphans', False) if do_clean else False remove_null_core = do_clean.get('remove_null_core', True) if do_clean else False normalize_relation_type = do_clean.get('normalize_relation_type', True) if do_clean else False if do_clean and (deduplicate or filter_orphans or remove_null_core or normalize_relation_type): triplets, report = clean_triplets( triplets, valid_ids=valid_ids if filter_orphans else None, deduplicate=deduplicate, filter_orphans=filter_orphans, remove_null_core=remove_null_core, normalize_relation_type=normalize_relation_type, ) report_path = _resolve_path(tc.get('report_file', 'report/triplet_cleaning_report.json'), _project_root) os.makedirs(os.path.dirname(report_path), exist_ok=True) with open(report_path, 'w', encoding='utf-8') as f: json.dump(report, f, ensure_ascii=False, indent=2, default=_json_serializer) print(f" 清洗: {report['original_count']} -> {report['final_count']}") print(f" 报告: {report_path}") # 转为输出格式：head, head_type, relation, tail, tail_type from src.triplet_cleaner import transform_triplet_output_format id_to_role = cleaner.get_id_to_role() triplets = transform_triplet_output_format(triplets, id_to_role) output_path = _resolve_path(tc.get('output_file', 'data/triples.json'), _project_root) os.makedirs(os.path.dirname(output_path), exist_ok=True) with open(output_path, 'w', encoding='utf-8') as f: json.dump(triplets, f, ensure_ascii=False, indent=2, default=_json_serializer) print(f" 输出: {output_path}") # ---------- 3. 模块1.3：形式化入库 ---------- formal_cfg = config.get("formal", {}) if source == "db" and formal_cfg.get("enabled", False): print("\n【3/3】形式化模块 读取-补齐-入库") try: result = run_formalization_pipeline(config) if result.get("ok"): stats = result.get("stats", {}) print(f" 批次: {result.get('batch_id')}") print(f" 推理三元组: {stats.get('total_triples', 0)}") print(f" 形式化入库: {stats.get('inserted_formal_count', 0)}") print(f" 属性补齐写回: {stats.get('generated_attrs_count', 0)}") else: print(f" 形式化流程跳过: {result.get('reason', 'unknown')}") except Exception as e: print(f" 形式化流程失败: {e}") raise elif formal_cfg.get("enabled", False): print("\n【3/3】形式化模块 已跳过（仅 source=db 时启用）") print("\n完成。") if __name__ == "__main__": main()