pengxiao
834cad729f
- data/: 非遗地理编码数据（GIS shapefile + CSV） - dofile/kg_project/: 知识图谱构建代码（纳入主仓库） - dofile/visulization/: 可视化数据与路线图 - officefile/: 文献、草稿、bib 文档 - officefile/latex/: Overleaf 同步目录（独立管理，不纳入） - output/: 输出目录 - logs/: 日志目录
376 lines
12 KiB
Python
376 lines
12 KiB
Python
"""
|
||
数据预处理脚本 - 读取黑龙江非遗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()
|