""" 数据预处理脚本 - 读取黑龙江非遗Excel数据 """ import pandas as pd import json from pathlib import Path from typing import Dict, List, Any import logging class ExcelDataReader: """Excel数据读取器""" def __init__(self, excel_path: str): """ 初始化数据读取器 Args: excel_path: Excel文件路径 """ self.excel_path = Path(excel_path) self.data = None self.logger = self._setup_logger() def _setup_logger(self): """设置日志""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) return logging.getLogger(__name__) def read_excel(self) -> pd.DataFrame: """ 读取Excel文件 Returns: DataFrame: 数据框 """ try: self.logger.info(f"开始读取Excel文件: {self.excel_path}") # 读取Excel文件 self.data = pd.read_excel(self.excel_path) self.logger.info(f"成功读取 {len(self.data)} 行数据") self.logger.info(f"列名: {list(self.data.columns)}") # 显示前5行 self.logger.info("\n前5行数据:") self.logger.info(self.data.head()) return self.data except Exception as e: self.logger.error(f"读取Excel文件失败: {str(e)}") raise def analyze_data(self) -> Dict[str, Any]: """ 分析数据 Returns: Dict: 分析结果 """ if self.data is None: raise ValueError("请先读取Excel文件") analysis = { "total_records": len(self.data), "columns": list(self.data.columns), "column_types": {col: str(dtype) for col, dtype in self.data.dtypes.items()}, "missing_values": self.data.isnull().sum().to_dict(), "statistics": {} } # 分析类别分布 if '类别' in self.data.columns: category_counts = self.data['类别'].value_counts() analysis['category_distribution'] = category_counts.to_dict() # 分析级别分布 if '项目级别' in self.data.columns: level_counts = self.data['项目级别'].value_counts() analysis['level_distribution'] = level_counts.to_dict() # 分析地域分布 if '项目申报单位/地区' in self.data.columns: location_counts = self.data['项目申报单位/地区'].value_counts() analysis['location_distribution'] = location_counts.head(20).to_dict() # 传承人覆盖率 if '代表性传承人' in self.data.columns: has_inheritor = self.data['代表性传承人'].notna().sum() analysis['inheritor_coverage'] = { "total": len(self.data), "has_inheritor": int(has_inheritor), "coverage_rate": float(has_inheritor / len(self.data) * 100) } return analysis def clean_data(self) -> pd.DataFrame: """ 清洗数据 Returns: DataFrame: 清洗后的数据 """ if self.data is None: raise ValueError("请先读取Excel文件") self.logger.info("开始清洗数据") # 去除空行 original_len = len(self.data) self.data = self.data.dropna(how='all') self.logger.info(f"去除空行: {original_len} -> {len(self.data)}") # 填充缺失值 for col in self.data.columns: if self.data[col].dtype == 'object': self.data[col] = self.data[col].fillna('') else: self.data[col] = self.data[col].fillna(0) # 去重（基于项目名称、项目批次、项目保护单位、代表性传承人四个字段） dedup_cols = ['项目名称', '项目批次', '项目保护单位', '代表性传承人'] existing_cols = [col for col in dedup_cols if col in self.data.columns] if existing_cols: before_dedup = len(self.data) self.data = self.data.drop_duplicates(subset=existing_cols, keep='first') duplicate_count = before_dedup - len(self.data) dedup_rate = (duplicate_count / before_dedup * 100) if before_dedup > 0 else 0 self.logger.info(f"去重（基于{len(existing_cols)}个字段: {', '.join(existing_cols)}）: {before_dedup} -> {len(self.data)} (删除{duplicate_count}条，去重率{dedup_rate:.1f}%)") return self.data def convert_to_kg_format(self) -> List[Dict[str, Any]]: """ 转换为知识图谱格式 Returns: List[Dict]: 知识图谱节点列表 """ if self.data is None: raise ValueError("请先读取Excel文件") self.logger.info("转换为知识图谱格式") kg_nodes = [] for idx, row in self.data.iterrows(): try: # 创建非遗项目节点 project_node = { "id": f"ICH-{idx:04d}", "type": "ICH_Project", "properties": { "project_id": f"ICH-{idx:04d}", "name": str(row.get('项目名称', '')), "level": str(row.get('项目级别', '')), "batch": str(row.get('批次号', '')), "category": str(row.get('类别', '')), "declaration_area": str(row.get('项目申报单位/地区', '')), "description": str(row.get('备注', '')), "source_row": idx + 2 # Excel行号（从2开始，第一行是表头） } } kg_nodes.append(project_node) except Exception as e: self.logger.warning(f"转换第{idx}行数据失败: {str(e)}") continue self.logger.info(f"成功转换 {len(kg_nodes)} 个节点") return kg_nodes def extract_inheritors(self) -> List[Dict[str, Any]]: """ 提取传承人信息 Returns: List[Dict]: 传承人节点列表 """ if self.data is None: raise ValueError("请先读取Excel文件") self.logger.info("提取传承人信息") inheritors = [] inheritor_id = 0 for idx, row in self.data.iterrows(): inheritor_names = row.get('代表性传承人', '') if pd.isna(inheritor_names) or not str(inheritor_names).strip(): continue # 处理多个传承人（用顿号分隔） names = str(inheritor_names).replace('、', ',').replace('，', ',').split(',') for name in names: name = name.strip() if not name: continue inheritor_id += 1 inheritor_node = { "id": f"INH-{inheritor_id:04d}", "type": "Inheritor", "properties": { "inheritor_id": f"INH-{inheritor_id:04d}", "name": name, "project_name": str(row.get('项目名称', '')), "source_row": idx + 2 } } inheritors.append(inheritor_node) self.logger.info(f"提取了 {len(inheritors)} 个传承人") return inheritors def extract_relations(self) -> List[Dict[str, Any]]: """ 提取关系 Returns: List[Dict]: 关系列表 """ if self.data is None: raise ValueError("请先读取Excel文件") self.logger.info("提取关系") relations = [] relation_id = 0 for idx, row in self.data.iterrows(): project_id = f"ICH-{idx:04d}" # 项目-类别关系 category = row.get('类别', '') if pd.notna(category) and str(category).strip(): relation_id += 1 relations.append({ "id": f"REL-{relation_id:04d}", "type": "belongs_to", "from": project_id, "to": f"CAT-{str(category)}", "properties": {} }) # 项目-传承人关系 inheritor_names = row.get('代表性传承人', '') if pd.notna(inheritor_names) and str(inheritor_names).strip(): names = str(inheritor_names).replace('、', ',').replace('，', ',').split(',') for name in names: name = name.strip() if name: relation_id += 1 relations.append({ "id": f"REL-{relation_id:04d}", "type": "has_inheritor", "from": project_id, "to": f"INH-{name}", # 简化处理 "properties": {} }) self.logger.info(f"提取了 {len(relations)} 个关系") return relations def save_analysis_report(self, output_path: str): """ 保存分析报告 Args: output_path: 输出路径 """ analysis = self.analyze_data() with open(output_path, 'w', encoding='utf-8') as f: json.dump(analysis, f, indent=2, ensure_ascii=False) self.logger.info(f"分析报告已保存到: {output_path}") def save_kg_data(self, nodes: List[Dict], relations: List[Dict], output_dir: str): """ 保存知识图谱数据 Args: nodes: 节点列表 relations: 关系列表 output_dir: 输出目录 """ output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) # 保存节点 nodes_file = output_path / "nodes.json" with open(nodes_file, 'w', encoding='utf-8') as f: json.dump(nodes, f, indent=2, ensure_ascii=False) self.logger.info(f"节点已保存到: {nodes_file}") # 保存关系 relations_file = output_path / "relations.json" with open(relations_file, 'w', encoding='utf-8') as f: json.dump(relations, f, indent=2, ensure_ascii=False) self.logger.info(f"关系已保存到: {relations_file}") def main(): """主函数""" # 配置 excel_path = r"E:\Project\2026_KG_ICH\data\黑龙江国家级和省级非遗名单.xlsx" output_dir = r"E:\Project\2026_KG_ICH\dofile\kg_project\output" # 创建读取器 reader = ExcelDataReader(excel_path) # 读取数据 reader.read_excel() # 分析数据 print("\n" + "="*60) print("数据分析") print("="*60) analysis = reader.analyze_data() print(f"\n总记录数: {analysis['total_records']}") print(f"\n类别分布:") for cat, count in analysis.get('category_distribution', {}).items(): print(f" {cat}: {count}") print(f"\n传承人覆盖率: {analysis.get('inheritor_coverage', {}).get('coverage_rate', 0):.1f}%") # 保存分析报告 Path(output_dir).mkdir(parents=True, exist_ok=True) reader.save_analysis_report(f"{output_dir}/data_analysis.json") # 清洗数据 print("\n" + "="*60) print("数据清洗") print("="*60) reader.clean_data() # 转换为知识图谱格式 print("\n" + "="*60) print("转换为知识图谱格式") print("="*60) nodes = reader.convert_to_kg_format() inheritors = reader.extract_inheritors() relations = reader.extract_relations() # 合并所有节点 all_nodes = nodes + inheritors print(f"\n节点统计:") print(f" 非遗项目节点: {len(nodes)}") print(f" 传承人节点: {len(inheritors)}") print(f" 总节点数: {len(all_nodes)}") print(f" 关系数: {len(relations)}") # 保存知识图谱数据 reader.save_kg_data(all_nodes, relations, output_dir) print("\n" + "="*60) print("处理完成！") print("="*60) if __name__ == '__main__': main()