705 lines
28 KiB
Python
705 lines
28 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
符号数据导入器
|
||
胡汉三千年项目数据导入工具
|
||
|
||
功能：提供符号数据的批量导入、验证和转换功能
|
||
"""
|
||
|
||
import sqlite3
|
||
import pandas as pd
|
||
import json
|
||
import csv
|
||
import os
|
||
from typing import Dict, List, Tuple, Any, Optional
|
||
from datetime import datetime
|
||
import re
|
||
|
||
class SymbolDataImporter:
|
||
"""符号数据导入器"""
|
||
|
||
def __init__(self, db_path: str = "symbols.db"):
|
||
"""初始化导入器"""
|
||
self.db_path = db_path
|
||
self.conn = sqlite3.connect(db_path)
|
||
|
||
def create_tables(self) -> bool:
|
||
"""创建数据库表结构"""
|
||
cursor = self.conn.cursor()
|
||
|
||
try:
|
||
# 创建符号主表
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS symbols (
|
||
symbol_id TEXT PRIMARY KEY,
|
||
symbol_form TEXT NOT NULL,
|
||
symbol_name TEXT NOT NULL,
|
||
yin_yang_attribute TEXT CHECK(yin_yang_attribute IN ('yin', 'yang', 'neutral')),
|
||
engraving_type TEXT CHECK(engraving_type IN ('yin_engraving', 'yang_engraving', 'mixed')),
|
||
origin_civilization TEXT,
|
||
origin_period TEXT,
|
||
geographical_context TEXT,
|
||
functional_context TEXT,
|
||
phonetic_context TEXT,
|
||
semantic_context TEXT,
|
||
discovery_date TEXT,
|
||
discovery_location TEXT,
|
||
material_type TEXT,
|
||
preservation_status TEXT,
|
||
description TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
|
||
# 创建符号属性表
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS symbol_attributes (
|
||
attribute_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
symbol_id TEXT,
|
||
attribute_name TEXT NOT NULL,
|
||
attribute_value TEXT,
|
||
attribute_type TEXT CHECK(attribute_type IN ('text', 'numeric', 'boolean', 'date')),
|
||
confidence_level INTEGER CHECK(confidence_level BETWEEN 1 AND 100),
|
||
source_reference TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
FOREIGN KEY (symbol_id) REFERENCES symbols(symbol_id)
|
||
)
|
||
""")
|
||
|
||
# 创建跨文明关联表
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS cross_civilization_links (
|
||
link_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
source_symbol_id TEXT NOT NULL,
|
||
target_symbol_id TEXT NOT NULL,
|
||
link_type TEXT CHECK(link_type IN ('phonetic', 'semantic', 'morphological', 'functional', 'geographical')),
|
||
confidence_level INTEGER CHECK(confidence_level BETWEEN 1 AND 100),
|
||
evidence_description TEXT,
|
||
source_reference TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
FOREIGN KEY (source_symbol_id) REFERENCES symbols(symbol_id),
|
||
FOREIGN KEY (target_symbol_id) REFERENCES symbols(symbol_id),
|
||
UNIQUE(source_symbol_id, target_symbol_id, link_type)
|
||
)
|
||
""")
|
||
|
||
# 创建符号传播路径表
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS symbol_transmission_paths (
|
||
path_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
path_name TEXT NOT NULL,
|
||
start_civilization TEXT,
|
||
end_civilization TEXT,
|
||
transmission_period TEXT,
|
||
transmission_route TEXT,
|
||
supporting_evidence TEXT,
|
||
confidence_level INTEGER CHECK(confidence_level BETWEEN 1 AND 100),
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
|
||
# 创建路径节点表
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS path_nodes (
|
||
node_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
path_id INTEGER,
|
||
symbol_id TEXT,
|
||
node_sequence INTEGER,
|
||
node_role TEXT,
|
||
transmission_context TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
FOREIGN KEY (path_id) REFERENCES symbol_transmission_paths(path_id),
|
||
FOREIGN KEY (symbol_id) REFERENCES symbols(symbol_id)
|
||
)
|
||
""")
|
||
|
||
self.conn.commit()
|
||
print("数据库表结构创建成功")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"创建表结构失败：{e}")
|
||
self.conn.rollback()
|
||
return False
|
||
|
||
def validate_symbol_data(self, symbol_data: Dict[str, Any]) -> Tuple[bool, List[str]]:
|
||
"""验证符号数据"""
|
||
errors = []
|
||
|
||
# 必填字段检查
|
||
required_fields = ['symbol_id', 'symbol_form', 'symbol_name']
|
||
for field in required_fields:
|
||
if field not in symbol_data or not symbol_data[field]:
|
||
errors.append(f"缺少必填字段：{field}")
|
||
|
||
# 符号ID格式检查
|
||
if 'symbol_id' in symbol_data:
|
||
symbol_id = symbol_data['symbol_id']
|
||
if not re.match(r'^[A-Za-z0-9_]+$', symbol_id):
|
||
errors.append("符号ID只能包含字母、数字和下划线")
|
||
|
||
# 阴阳属性验证
|
||
if 'yin_yang_attribute' in symbol_data and symbol_data['yin_yang_attribute']:
|
||
valid_yin_yang = ['yin', 'yang', 'neutral']
|
||
if symbol_data['yin_yang_attribute'] not in valid_yin_yang:
|
||
errors.append(f"阴阳属性必须是：{', '.join(valid_yin_yang)}")
|
||
|
||
# 刻法类型验证
|
||
if 'engraving_type' in symbol_data and symbol_data['engraving_type']:
|
||
valid_engraving = ['yin_engraving', 'yang_engraving', 'mixed']
|
||
if symbol_data['engraving_type'] not in valid_engraving:
|
||
errors.append(f"刻法类型必须是：{', '.join(valid_engraving)}")
|
||
|
||
# 置信度验证
|
||
if 'confidence_level' in symbol_data and symbol_data['confidence_level']:
|
||
try:
|
||
confidence = int(symbol_data['confidence_level'])
|
||
if not 1 <= confidence <= 100:
|
||
errors.append("置信度必须在1-100之间")
|
||
except ValueError:
|
||
errors.append("置信度必须是整数")
|
||
|
||
return len(errors) == 0, errors
|
||
|
||
def import_symbol_from_dict(self, symbol_data: Dict[str, Any]) -> Tuple[bool, str]:
|
||
"""从字典导入单个符号"""
|
||
# 验证数据
|
||
is_valid, errors = self.validate_symbol_data(symbol_data)
|
||
if not is_valid:
|
||
return False, f"数据验证失败：{'; '.join(errors)}"
|
||
|
||
cursor = self.conn.cursor()
|
||
|
||
try:
|
||
# 检查符号是否已存在
|
||
cursor.execute("SELECT symbol_id FROM symbols WHERE symbol_id = ?",
|
||
(symbol_data['symbol_id'],))
|
||
existing_symbol = cursor.fetchone()
|
||
|
||
if existing_symbol:
|
||
# 更新现有符号
|
||
update_fields = []
|
||
update_values = []
|
||
|
||
for field, value in symbol_data.items():
|
||
if field != 'symbol_id':
|
||
update_fields.append(f"{field} = ?")
|
||
update_values.append(value)
|
||
|
||
update_values.append(symbol_data['symbol_id']) # WHERE条件
|
||
|
||
update_query = f"""
|
||
UPDATE symbols
|
||
SET {', '.join(update_fields)}, updated_at = CURRENT_TIMESTAMP
|
||
WHERE symbol_id = ?
|
||
"""
|
||
|
||
cursor.execute(update_query, update_values)
|
||
action = "更新"
|
||
else:
|
||
# 插入新符号
|
||
fields = list(symbol_data.keys())
|
||
placeholders = ['?'] * len(fields)
|
||
values = list(symbol_data.values())
|
||
|
||
insert_query = f"""
|
||
INSERT INTO symbols ({', '.join(fields)})
|
||
VALUES ({', '.join(placeholders)})
|
||
"""
|
||
|
||
cursor.execute(insert_query, values)
|
||
action = "插入"
|
||
|
||
self.conn.commit()
|
||
return True, f"符号 {symbol_data['symbol_id']} {action}成功"
|
||
|
||
except Exception as e:
|
||
self.conn.rollback()
|
||
return False, f"导入失败：{e}"
|
||
|
||
def import_symbols_from_csv(self, csv_file: str, delimiter: str = ',') -> Dict[str, Any]:
|
||
"""从CSV文件批量导入符号"""
|
||
results = {
|
||
'total': 0,
|
||
'success': 0,
|
||
'failed': 0,
|
||
'errors': []
|
||
}
|
||
|
||
try:
|
||
with open(csv_file, 'r', encoding='utf-8') as file:
|
||
# 检测编码
|
||
try:
|
||
reader = csv.DictReader(file, delimiter=delimiter)
|
||
rows = list(reader)
|
||
except UnicodeDecodeError:
|
||
# 尝试其他编码
|
||
file.seek(0)
|
||
reader = csv.DictReader(file, delimiter=delimiter)
|
||
rows = list(reader)
|
||
|
||
results['total'] = len(rows)
|
||
|
||
for i, row in enumerate(rows, 1):
|
||
# 清理数据
|
||
symbol_data = {}
|
||
for key, value in row.items():
|
||
if value is not None and value.strip():
|
||
symbol_data[key.strip()] = value.strip()
|
||
|
||
# 导入符号
|
||
success, message = self.import_symbol_from_dict(symbol_data)
|
||
|
||
if success:
|
||
results['success'] += 1
|
||
else:
|
||
results['failed'] += 1
|
||
results['errors'].append({
|
||
'row': i,
|
||
'symbol_id': symbol_data.get('symbol_id', '未知'),
|
||
'error': message
|
||
})
|
||
|
||
# 进度显示
|
||
if i % 10 == 0:
|
||
print(f"已处理 {i}/{len(rows)} 行")
|
||
|
||
print(f"CSV导入完成：成功 {results['success']}，失败 {results['failed']}")
|
||
|
||
except Exception as e:
|
||
results['errors'].append({
|
||
'row': '文件级别',
|
||
'symbol_id': 'N/A',
|
||
'error': f"文件读取失败：{e}"
|
||
})
|
||
results['failed'] = results['total']
|
||
|
||
return results
|
||
|
||
def import_symbols_from_json(self, json_file: str) -> Dict[str, Any]:
|
||
"""从JSON文件导入符号"""
|
||
results = {
|
||
'total': 0,
|
||
'success': 0,
|
||
'failed': 0,
|
||
'errors': []
|
||
}
|
||
|
||
try:
|
||
with open(json_file, 'r', encoding='utf-8') as file:
|
||
data = json.load(file)
|
||
|
||
# 支持多种JSON格式
|
||
if isinstance(data, list):
|
||
symbols_list = data
|
||
elif isinstance(data, dict) and 'symbols' in data:
|
||
symbols_list = data['symbols']
|
||
else:
|
||
symbols_list = [data]
|
||
|
||
results['total'] = len(symbols_list)
|
||
|
||
for i, symbol_data in enumerate(symbols_list, 1):
|
||
success, message = self.import_symbol_from_dict(symbol_data)
|
||
|
||
if success:
|
||
results['success'] += 1
|
||
else:
|
||
results['failed'] += 1
|
||
results['errors'].append({
|
||
'index': i,
|
||
'symbol_id': symbol_data.get('symbol_id', '未知'),
|
||
'error': message
|
||
})
|
||
|
||
# 进度显示
|
||
if i % 10 == 0:
|
||
print(f"已处理 {i}/{len(symbols_list)} 个符号")
|
||
|
||
print(f"JSON导入完成：成功 {results['success']}，失败 {results['failed']}")
|
||
|
||
except Exception as e:
|
||
results['errors'].append({
|
||
'index': '文件级别',
|
||
'symbol_id': 'N/A',
|
||
'error': f"文件读取失败：{e}"
|
||
})
|
||
results['failed'] = results['total']
|
||
|
||
return results
|
||
|
||
def import_cross_civilization_links(self, links_data: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""导入跨文明关联"""
|
||
results = {
|
||
'total': 0,
|
||
'success': 0,
|
||
'failed': 0,
|
||
'errors': []
|
||
}
|
||
|
||
cursor = self.conn.cursor()
|
||
|
||
results['total'] = len(links_data)
|
||
|
||
for i, link_data in enumerate(links_data, 1):
|
||
try:
|
||
# 验证必填字段
|
||
required_fields = ['source_symbol_id', 'target_symbol_id', 'link_type']
|
||
for field in required_fields:
|
||
if field not in link_data or not link_data[field]:
|
||
raise ValueError(f"缺少必填字段：{field}")
|
||
|
||
# 验证符号存在
|
||
cursor.execute("SELECT symbol_id FROM symbols WHERE symbol_id = ?",
|
||
(link_data['source_symbol_id'],))
|
||
if not cursor.fetchone():
|
||
raise ValueError(f"源符号不存在：{link_data['source_symbol_id']}")
|
||
|
||
cursor.execute("SELECT symbol_id FROM symbols WHERE symbol_id = ?",
|
||
(link_data['target_symbol_id'],))
|
||
if not cursor.fetchone():
|
||
raise ValueError(f"目标符号不存在：{link_data['target_symbol_id']}")
|
||
|
||
# 验证关联类型
|
||
valid_link_types = ['phonetic', 'semantic', 'morphological', 'functional', 'geographical']
|
||
if link_data['link_type'] not in valid_link_types:
|
||
raise ValueError(f"关联类型必须是：{', '.join(valid_link_types)}")
|
||
|
||
# 验证置信度
|
||
confidence = link_data.get('confidence_level', 50)
|
||
if not 1 <= confidence <= 100:
|
||
raise ValueError("置信度必须在1-100之间")
|
||
|
||
# 插入或更新关联
|
||
insert_query = """
|
||
INSERT OR REPLACE INTO cross_civilization_links
|
||
(source_symbol_id, target_symbol_id, link_type, confidence_level,
|
||
evidence_description, source_reference)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
"""
|
||
|
||
cursor.execute(insert_query, (
|
||
link_data['source_symbol_id'],
|
||
link_data['target_symbol_id'],
|
||
link_data['link_type'],
|
||
confidence,
|
||
link_data.get('evidence_description', ''),
|
||
link_data.get('source_reference', '')
|
||
))
|
||
|
||
results['success'] += 1
|
||
|
||
except Exception as e:
|
||
results['failed'] += 1
|
||
results['errors'].append({
|
||
'index': i,
|
||
'link': f"{link_data.get('source_symbol_id', '?')} -> {link_data.get('target_symbol_id', '?')}",
|
||
'error': str(e)
|
||
})
|
||
|
||
# 进度显示
|
||
if i % 10 == 0:
|
||
print(f"已处理 {i}/{len(links_data)} 个关联")
|
||
|
||
self.conn.commit()
|
||
print(f"关联导入完成：成功 {results['success']}，失败 {results['failed']}")
|
||
|
||
return results
|
||
|
||
def import_transmission_paths(self, paths_data: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""导入传播路径"""
|
||
results = {
|
||
'total': 0,
|
||
'success': 0,
|
||
'failed': 0,
|
||
'errors': []
|
||
}
|
||
|
||
cursor = self.conn.cursor()
|
||
|
||
results['total'] = len(paths_data)
|
||
|
||
for i, path_data in enumerate(paths_data, 1):
|
||
try:
|
||
# 验证必填字段
|
||
if 'path_name' not in path_data or not path_data['path_name']:
|
||
raise ValueError("缺少必填字段：path_name")
|
||
|
||
# 插入传播路径
|
||
insert_path_query = """
|
||
INSERT INTO symbol_transmission_paths
|
||
(path_name, start_civilization, end_civilization, transmission_period,
|
||
transmission_route, supporting_evidence, confidence_level)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
"""
|
||
|
||
cursor.execute(insert_path_query, (
|
||
path_data['path_name'],
|
||
path_data.get('start_civilization', ''),
|
||
path_data.get('end_civilization', ''),
|
||
path_data.get('transmission_period', ''),
|
||
path_data.get('transmission_route', ''),
|
||
path_data.get('supporting_evidence', ''),
|
||
path_data.get('confidence_level', 50)
|
||
))
|
||
|
||
path_id = cursor.lastrowid
|
||
|
||
# 插入路径节点
|
||
if 'nodes' in path_data and isinstance(path_data['nodes'], list):
|
||
for j, node_data in enumerate(path_data['nodes'], 1):
|
||
if 'symbol_id' not in node_data:
|
||
continue
|
||
|
||
insert_node_query = """
|
||
INSERT INTO path_nodes
|
||
(path_id, symbol_id, node_sequence, node_role, transmission_context)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
"""
|
||
|
||
cursor.execute(insert_node_query, (
|
||
path_id,
|
||
node_data['symbol_id'],
|
||
node_data.get('node_sequence', j),
|
||
node_data.get('node_role', ''),
|
||
node_data.get('transmission_context', '')
|
||
))
|
||
|
||
results['success'] += 1
|
||
|
||
except Exception as e:
|
||
results['failed'] += 1
|
||
results['errors'].append({
|
||
'index': i,
|
||
'path_name': path_data.get('path_name', '未知'),
|
||
'error': str(e)
|
||
})
|
||
|
||
# 进度显示
|
||
if i % 5 == 0:
|
||
print(f"已处理 {i}/{len(paths_data)} 个传播路径")
|
||
|
||
self.conn.commit()
|
||
print(f"传播路径导入完成：成功 {results['success']}，失败 {results['failed']}")
|
||
|
||
return results
|
||
|
||
def generate_sample_data(self) -> Dict[str, Any]:
|
||
"""生成示例数据"""
|
||
sample_symbols = [
|
||
{
|
||
'symbol_id': 'P_yin_001',
|
||
'symbol_form': 'P',
|
||
'symbol_name': '太平洋符号',
|
||
'yin_yang_attribute': 'yin',
|
||
'engraving_type': 'yin_engraving',
|
||
'origin_civilization': 'Chinese',
|
||
'origin_period': '上古',
|
||
'geographical_context': '太平洋盆地',
|
||
'functional_context': '阴刻地理标识',
|
||
'phonetic_context': 'pǐ (否卦读音)',
|
||
'semantic_context': '凹陷、阴刻、盆地',
|
||
'description': '代表太平洋盆地的阴刻符号'
|
||
},
|
||
{
|
||
'symbol_id': 'T_yang_001',
|
||
'symbol_form': 'T',
|
||
'symbol_name': '泰山符号',
|
||
'yin_yang_attribute': 'yang',
|
||
'engraving_type': 'yang_engraving',
|
||
'origin_civilization': 'Chinese',
|
||
'origin_period': '上古',
|
||
'geographical_context': '泰山山脉',
|
||
'functional_context': '阳刻地理标识',
|
||
'phonetic_context': 'tài (泰卦读音)',
|
||
'semantic_context': '凸起、阳刻、山脉',
|
||
'description': '代表泰山山脉的阳刻符号'
|
||
},
|
||
{
|
||
'symbol_id': 'M_neutral_001',
|
||
'symbol_form': 'M',
|
||
'symbol_name': '月亮符号',
|
||
'yin_yang_attribute': 'neutral',
|
||
'engraving_type': 'mixed',
|
||
'origin_civilization': 'Multiple',
|
||
'origin_period': '远古',
|
||
'geographical_context': '全球分布',
|
||
'functional_context': '月亮崇拜标识',
|
||
'phonetic_context': 'moon, month',
|
||
'semantic_context': '月亮、月份、循环',
|
||
'description': '代表月亮崇拜的中性符号'
|
||
}
|
||
]
|
||
|
||
sample_links = [
|
||
{
|
||
'source_symbol_id': 'P_yin_001',
|
||
'target_symbol_id': 'T_yang_001',
|
||
'link_type': 'functional',
|
||
'confidence_level': 85,
|
||
'evidence_description': '阴阳对立功能互补',
|
||
'source_reference': '周易卦象分析'
|
||
},
|
||
{
|
||
'source_symbol_id': 'P_yin_001',
|
||
'target_symbol_id': 'M_neutral_001',
|
||
'link_type': 'semantic',
|
||
'confidence_level': 70,
|
||
'evidence_description': '阴刻与月亮关联',
|
||
'source_reference': '跨文明符号研究'
|
||
}
|
||
]
|
||
|
||
sample_paths = [
|
||
{
|
||
'path_name': '阴阳符号传播路径',
|
||
'start_civilization': 'Chinese',
|
||
'end_civilization': 'Global',
|
||
'transmission_period': '上古-现代',
|
||
'transmission_route': '丝绸之路-海上贸易',
|
||
'supporting_evidence': '考古发现与文献记载',
|
||
'confidence_level': 80,
|
||
'nodes': [
|
||
{'symbol_id': 'P_yin_001', 'node_sequence': 1, 'node_role': '起源', 'transmission_context': '中国周易体系'},
|
||
{'symbol_id': 'T_yang_001', 'node_sequence': 2, 'node_role': '发展', 'transmission_context': '阴阳哲学传播'},
|
||
{'symbol_id': 'M_neutral_001', 'node_sequence': 3, 'node_role': '融合', 'transmission_context': '跨文明符号融合'}
|
||
]
|
||
}
|
||
]
|
||
|
||
return {
|
||
'symbols': sample_symbols,
|
||
'links': sample_links,
|
||
'paths': sample_paths
|
||
}
|
||
|
||
def import_sample_data(self) -> Dict[str, Any]:
|
||
"""导入示例数据"""
|
||
sample_data = self.generate_sample_data()
|
||
|
||
results = {}
|
||
|
||
# 导入符号
|
||
print("正在导入示例符号数据...")
|
||
results['symbols'] = self.import_symbols_from_json_data(sample_data['symbols'])
|
||
|
||
# 导入关联
|
||
print("正在导入示例关联数据...")
|
||
results['links'] = self.import_cross_civilization_links(sample_data['links'])
|
||
|
||
# 导入传播路径
|
||
print("正在导入示例传播路径...")
|
||
results['paths'] = self.import_transmission_paths(sample_data['paths'])
|
||
|
||
return results
|
||
|
||
def import_symbols_from_json_data(self, symbols_list: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""从JSON数据导入符号（内部方法）"""
|
||
results = {
|
||
'total': 0,
|
||
'success': 0,
|
||
'failed': 0,
|
||
'errors': []
|
||
}
|
||
|
||
results['total'] = len(symbols_list)
|
||
|
||
for i, symbol_data in enumerate(symbols_list, 1):
|
||
success, message = self.import_symbol_from_dict(symbol_data)
|
||
|
||
if success:
|
||
results['success'] += 1
|
||
else:
|
||
results['failed'] += 1
|
||
results['errors'].append({
|
||
'index': i,
|
||
'symbol_id': symbol_data.get('symbol_id', '未知'),
|
||
'error': message
|
||
})
|
||
|
||
return results
|
||
|
||
def export_database_schema(self, output_file: str = "database_schema.md") -> str:
|
||
"""导出数据库结构文档"""
|
||
cursor = self.conn.cursor()
|
||
|
||
schema_doc = ["# 胡汉三千年项目 - 数据库结构文档", "", f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", ""]
|
||
|
||
# 获取所有表
|
||
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
|
||
tables = cursor.fetchall()
|
||
|
||
for table in tables:
|
||
table_name = table[0]
|
||
schema_doc.append(f"## {table_name} 表")
|
||
schema_doc.append("")
|
||
|
||
# 获取表结构
|
||
cursor.execute(f"PRAGMA table_info({table_name})")
|
||
columns = cursor.fetchall()
|
||
|
||
schema_doc.append("| 字段名 | 类型 | 是否主键 | 是否为空 | 默认值 | 说明 |")
|
||
schema_doc.append("|--------|------|----------|----------|--------|------|")
|
||
|
||
for col in columns:
|
||
col_name = col[1]
|
||
col_type = col[2]
|
||
is_primary = "是" if col[5] == 1 else "否"
|
||
is_nullable = "否" if col[3] == 0 else "是"
|
||
default_value = col[4] if col[4] else "无"
|
||
|
||
schema_doc.append(f"| {col_name} | {col_type} | {is_primary} | {is_nullable} | {default_value} | |")
|
||
|
||
schema_doc.append("")
|
||
|
||
schema_content = '\n'.join(schema_doc)
|
||
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
f.write(schema_content)
|
||
|
||
return f"数据库结构文档已导出至：{output_file}"
|
||
|
||
# 使用示例
|
||
def main():
|
||
"""主函数示例"""
|
||
# 创建导入器
|
||
importer = SymbolDataImporter()
|
||
|
||
# 示例1：创建数据库表结构
|
||
print("=== 创建数据库表结构 ===")
|
||
success = importer.create_tables()
|
||
if success:
|
||
print("表结构创建成功")
|
||
|
||
# 示例2：导入示例数据
|
||
print("\n=== 导入示例数据 ===")
|
||
results = importer.import_sample_data()
|
||
print(f"符号导入：{results['symbols']['success']} 成功，{results['symbols']['failed']} 失败")
|
||
print(f"关联导入：{results['links']['success']} 成功，{results['links']['failed']} 失败")
|
||
print(f"路径导入：{results['paths']['success']} 成功，{results['paths']['failed']} 失败")
|
||
|
||
# 示例3：导出数据库结构文档
|
||
print("\n=== 导出数据库结构文档 ===")
|
||
result = importer.export_database_schema("database_schema.md")
|
||
print(result)
|
||
|
||
# 示例4：从CSV导入（需要示例CSV文件）
|
||
# print("\n=== 从CSV导入数据 ===")
|
||
# if os.path.exists("sample_symbols.csv"):
|
||
# results = importer.import_symbols_from_csv("sample_symbols.csv")
|
||
# print(f"CSV导入结果：{results['success']} 成功，{results['failed']} 失败")
|
||
# else:
|
||
# print("示例CSV文件不存在，跳过测试")
|
||
|
||
# 示例5：从JSON导入（需要示例JSON文件）
|
||
# print("\n=== 从JSON导入数据 ===")
|
||
# if os.path.exists("sample_symbols.json"):
|
||
# results = importer.import_symbols_from_json("sample_symbols.json")
|
||
# print(f"JSON导入结果：{results['success']} 成功，{results['failed']} 失败")
|
||
# else:
|
||
# print("示例JSON文件不存在，跳过测试")
|
||
|
||
if __name__ == "__main__":
|
||
main() |