#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 符号查询系统 胡汉三千年项目查询工具 功能：提供符号数据库的快速查询、过滤、搜索和分析功能 """ import sqlite3 import pandas as pd from typing import Dict, List, Tuple, Any, Optional from datetime import datetime import re class SymbolQuerySystem: """符号查询系统""" def __init__(self, db_path: str = "symbols.db"): """初始化查询系统""" self.db_path = db_path self.conn = sqlite3.connect(db_path) def search_by_keyword(self, keyword: str, search_fields: List[str] = None) -> List[Dict]: """ 根据关键词搜索符号 Args: keyword: 搜索关键词 search_fields: 搜索字段列表（可选，默认搜索所有字段） Returns: 匹配的符号列表 """ if search_fields is None: search_fields = ['symbol_id', 'symbol_form', 'symbol_name', 'yin_yang_attribute', 'engraving_type', 'origin_civilization', 'origin_period', 'geographical_context', 'functional_context', 'phonetic_context', 'semantic_context'] cursor = self.conn.cursor() # 构建查询条件 conditions = [] params = [] for field in search_fields: conditions.append(f"{field} LIKE ?") params.append(f"%{keyword}%") where_clause = " OR ".join(conditions) query = f""" SELECT * FROM symbols WHERE {where_clause} ORDER BY symbol_id """ cursor.execute(query, params) results = cursor.fetchall() # 转换为字典列表 columns = [desc[0] for desc in cursor.description] symbols = [] for row in results: symbol_dict = dict(zip(columns, row)) symbols.append(symbol_dict) return symbols def filter_by_attributes(self, filters: Dict[str, Any]) -> List[Dict]: """ 根据属性过滤符号 Args: filters: 过滤条件字典 - yin_yang_attribute: 阴阳属性 - engraving_type: 刻法类型 - origin_civilization: 起源文明 - origin_period: 起源时期 - geographical_context: 地理背景 - functional_context: 功能背景 Returns: 匹配的符号列表 """ cursor = self.conn.cursor() # 构建查询条件 conditions = [] params = [] for field, value in filters.items(): if value is not None: conditions.append(f"{field} = ?") params.append(value) if not conditions: where_clause = "1=1" else: where_clause = " AND ".join(conditions) query = f""" SELECT * FROM symbols WHERE {where_clause} ORDER BY symbol_id """ cursor.execute(query, params) results = cursor.fetchall() # 转换为字典列表 columns = [desc[0] for desc in cursor.description] symbols = [] for row in results: symbol_dict = dict(zip(columns, row)) symbols.append(symbol_dict) return symbols def find_related_symbols(self, symbol_id: str, max_depth: int = 3) -> Dict[str, Any]: """ 查找相关符号（传播路径） Args: symbol_id: 符号ID max_depth: 最大搜索深度 Returns: 相关符号信息 """ cursor = self.conn.cursor() # 查找直接关联的符号 cursor.execute(""" SELECT l.link_id, l.link_type, l.confidence_level, s1.symbol_id as source_id, s1.symbol_name as source_name, s2.symbol_id as target_id, s2.symbol_name as target_name FROM cross_civilization_links l JOIN symbols s1 ON l.source_symbol_id = s1.symbol_id JOIN symbols s2 ON l.target_symbol_id = s2.symbol_id WHERE s1.symbol_id = ? OR s2.symbol_id = ? """, (symbol_id, symbol_id)) direct_links = cursor.fetchall() # 查找传播路径 cursor.execute(""" WITH RECURSIVE symbol_path AS ( SELECT source_symbol_id, target_symbol_id, 1 as depth, source_symbol_id || '->' || target_symbol_id as path FROM cross_civilization_links WHERE source_symbol_id = ? UNION ALL SELECT sp.source_symbol_id, l.target_symbol_id, sp.depth + 1, sp.path || '->' || l.target_symbol_id FROM cross_civilization_links l JOIN symbol_path sp ON l.source_symbol_id = sp.target_symbol_id WHERE sp.depth < ? ) SELECT * FROM symbol_path ORDER BY depth, path """, (symbol_id, max_depth)) transmission_paths = cursor.fetchall() # 获取符号基本信息 cursor.execute("SELECT * FROM symbols WHERE symbol_id = ?", (symbol_id,)) symbol_info = cursor.fetchone() if symbol_info: columns = [desc[0] for desc in cursor.description] symbol_dict = dict(zip(columns, symbol_info)) else: symbol_dict = {} return { 'symbol_info': symbol_dict, 'direct_links': direct_links, 'transmission_paths': transmission_paths } def compare_symbols(self, symbol_ids: List[str]) -> Dict[str, Any]: """ 比较多个符号 Args: symbol_ids: 符号ID列表 Returns: 比较结果 """ cursor = self.conn.cursor() comparison = {} # 获取符号基本信息 symbols_data = [] for symbol_id in symbol_ids: cursor.execute("SELECT * FROM symbols WHERE symbol_id = ?", (symbol_id,)) result = cursor.fetchone() if result: columns = [desc[0] for desc in cursor.description] symbol_dict = dict(zip(columns, result)) symbols_data.append(symbol_dict) comparison['symbols'] = symbols_data # 比较阴阳属性 yin_yang_values = [s.get('yin_yang_attribute', '') for s in symbols_data] comparison['yin_yang_comparison'] = { 'values': yin_yang_values, 'is_same': len(set(yin_yang_values)) == 1 } # 比较刻法类型 engraving_values = [s.get('engraving_type', '') for s in symbols_data] comparison['engraving_comparison'] = { 'values': engraving_values, 'is_same': len(set(engraving_values)) == 1 } # 比较起源文明 civilization_values = [s.get('origin_civilization', '') for s in symbols_data] comparison['civilization_comparison'] = { 'values': civilization_values, 'is_same': len(set(civilization_values)) == 1 } # 查找共同关联 if len(symbol_ids) >= 2: placeholders = ','.join(['?'] * len(symbol_ids)) cursor.execute(f""" SELECT DISTINCT l.link_type, COUNT(*) as link_count FROM cross_civilization_links l WHERE l.source_symbol_id IN ({placeholders}) OR l.target_symbol_id IN ({placeholders}) GROUP BY l.link_type ORDER BY link_count DESC """, symbol_ids * 2) common_links = cursor.fetchall() comparison['common_links'] = common_links return comparison def analyze_symbol_family(self, family_pattern: str) -> Dict[str, Any]: """ 分析符号家族 Args: family_pattern: 家族模式（如 "P_*" 或 "T_*"） Returns: 家族分析结果 """ cursor = self.conn.cursor() # 查找匹配的符号 cursor.execute(""" SELECT * FROM symbols WHERE symbol_id LIKE ? ORDER BY symbol_id """, (family_pattern.replace('*', '%'),)) family_symbols = cursor.fetchall() if not family_symbols: return {'error': '未找到匹配的符号家族'} # 转换为字典列表 columns = [desc[0] for desc in cursor.description] symbols_list = [] for row in family_symbols: symbol_dict = dict(zip(columns, row)) symbols_list.append(symbol_dict) # 家族统计分析 family_stats = { 'total_count': len(symbols_list), 'yin_yang_distribution': {}, 'engraving_distribution': {}, 'civilization_distribution': {}, 'period_distribution': {} } for symbol in symbols_list: # 阴阳属性分布 yin_yang = symbol.get('yin_yang_attribute', 'unknown') family_stats['yin_yang_distribution'][yin_yang] = \ family_stats['yin_yang_distribution'].get(yin_yang, 0) + 1 # 刻法类型分布 engraving = symbol.get('engraving_type', 'unknown') family_stats['engraving_distribution'][engraving] = \ family_stats['engraving_distribution'].get(engraving, 0) + 1 # 文明分布 civilization = symbol.get('origin_civilization', 'unknown') family_stats['civilization_distribution'][civilization] = \ family_stats['civilization_distribution'].get(civilization, 0) + 1 # 时期分布 period = symbol.get('origin_period', 'unknown') family_stats['period_distribution'][period] = \ family_stats['period_distribution'].get(period, 0) + 1 # 查找家族内部关联 symbol_ids = [s['symbol_id'] for s in symbols_list] placeholders = ','.join(['?'] * len(symbol_ids)) cursor.execute(f""" SELECT l.link_id, l.link_type, l.confidence_level, s1.symbol_id as source_id, s1.symbol_name as source_name, s2.symbol_id as target_id, s2.symbol_name as target_name FROM cross_civilization_links l JOIN symbols s1 ON l.source_symbol_id = s1.symbol_id JOIN symbols s2 ON l.target_symbol_id = s2.symbol_id WHERE s1.symbol_id IN ({placeholders}) AND s2.symbol_id IN ({placeholders}) ORDER BY l.confidence_level DESC """, symbol_ids * 2) internal_links = cursor.fetchall() return { 'family_symbols': symbols_list, 'family_stats': family_stats, 'internal_links': internal_links } def advanced_search(self, query_params: Dict[str, Any]) -> List[Dict]: """ 高级搜索 Args: query_params: 查询参数 - keywords: 关键词列表 - yin_yang: 阴阳属性列表 - engraving: 刻法类型列表 - civilization: 文明列表 - period: 时期列表 - min_confidence: 最小置信度 - link_type: 关联类型 Returns: 匹配的符号列表 """ cursor = self.conn.cursor() # 构建查询条件 conditions = [] params = [] # 关键词搜索 if 'keywords' in query_params and query_params['keywords']: keyword_conditions = [] for keyword in query_params['keywords']: keyword_conditions.extend([ "symbol_id LIKE ?", "symbol_form LIKE ?", "symbol_name LIKE ?", "geographical_context LIKE ?", "functional_context LIKE ?", "phonetic_context LIKE ?", "semantic_context LIKE ?" ]) params.extend([f"%{keyword}%"] * 7) conditions.append(f"({' OR '.join(keyword_conditions)})") # 阴阳属性过滤 if 'yin_yang' in query_params and query_params['yin_yang']: placeholders = ','.join(['?'] * len(query_params['yin_yang'])) conditions.append(f"yin_yang_attribute IN ({placeholders})") params.extend(query_params['yin_yang']) # 刻法类型过滤 if 'engraving' in query_params and query_params['engraving']: placeholders = ','.join(['?'] * len(query_params['engraving'])) conditions.append(f"engraving_type IN ({placeholders})") params.extend(query_params['engraving']) # 文明过滤 if 'civilization' in query_params and query_params['civilization']: placeholders = ','.join(['?'] * len(query_params['civilization'])) conditions.append(f"origin_civilization IN ({placeholders})") params.extend(query_params['civilization']) # 时期过滤 if 'period' in query_params and query_params['period']: placeholders = ','.join(['?'] * len(query_params['period'])) conditions.append(f"origin_period IN ({placeholders})") params.extend(query_params['period']) # 构建完整查询 if conditions: where_clause = " AND ".join(conditions) else: where_clause = "1=1" query = f""" SELECT DISTINCT s.* FROM symbols s LEFT JOIN cross_civilization_links l ON s.symbol_id = l.source_symbol_id OR s.symbol_id = l.target_symbol_id WHERE {where_clause} """ # 关联类型过滤 if 'link_type' in query_params and query_params['link_type']: query += " AND l.link_type = ?" params.append(query_params['link_type']) # 置信度过滤 if 'min_confidence' in query_params and query_params['min_confidence']: query += " AND l.confidence_level >= ?" params.append(query_params['min_confidence']) query += " ORDER BY s.symbol_id" cursor.execute(query, params) results = cursor.fetchall() # 转换为字典列表 columns = [desc[0] for desc in cursor.description] symbols = [] for row in results: symbol_dict = dict(zip(columns, row)) symbols.append(symbol_dict) return symbols def export_search_results(self, symbols: List[Dict], output_file: str) -> str: """ 导出搜索结果 Args: symbols: 符号列表 output_file: 输出文件路径 Returns: 导出文件路径 """ import csv if not symbols: return "没有数据可导出" # 获取所有字段 fieldnames = symbols[0].keys() with open(output_file, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for symbol in symbols: writer.writerow(symbol) return f"搜索结果已导出至：{output_file}" def create_search_report(self, search_params: Dict[str, Any], output_file: str = None) -> str: """ 创建搜索报告 Args: search_params: 搜索参数 output_file: 输出文件路径（可选） Returns: 报告内容 """ # 执行搜索 results = self.advanced_search(search_params) # 生成报告 report = [] report.append("# 符号搜索报告") report.append(f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") report.append(f"搜索参数：{search_params}") report.append(f"\n## 搜索结果摘要") report.append(f"- 找到 {len(results)} 个匹配的符号") if results: # 统计信息 yin_yang_counts = {} engraving_counts = {} civilization_counts = {} for symbol in results: yin_yang = symbol.get('yin_yang_attribute', 'unknown') engraving = symbol.get('engraving_type', 'unknown') civilization = symbol.get('origin_civilization', 'unknown') yin_yang_counts[yin_yang] = yin_yang_counts.get(yin_yang, 0) + 1 engraving_counts[engraving] = engraving_counts.get(engraving, 0) + 1 civilization_counts[civilization] = civilization_counts.get(civilization, 0) + 1 report.append("\n### 阴阳属性分布") for yin_yang, count in yin_yang_counts.items(): percentage = (count / len(results)) * 100 report.append(f"- {yin_yang}：{count} ({percentage:.1f}%)") report.append("\n### 刻法类型分布") for engraving, count in engraving_counts.items(): percentage = (count / len(results)) * 100 report.append(f"- {engraving}：{count} ({percentage:.1f}%)") report.append("\n### 文明分布") for civilization, count in civilization_counts.items(): percentage = (count / len(results)) * 100 report.append(f"- {civilization}：{count} ({percentage:.1f}%)") # 前10个结果 report.append("\n### 前10个匹配符号") for i, symbol in enumerate(results[:10]): report.append(f"\n**{i+1}. {symbol.get('symbol_id', 'N/A')} - {symbol.get('symbol_name', 'N/A')}**") report.append(f" - 阴阳属性：{symbol.get('yin_yang_attribute', 'N/A')}") report.append(f" - 刻法类型：{symbol.get('engraving_type', 'N/A')}") report.append(f" - 起源文明：{symbol.get('origin_civilization', 'N/A')}") report.append(f" - 起源时期：{symbol.get('origin_period', 'N/A')}") report_content = '\n'.join(report) if output_file: with open(output_file, 'w', encoding='utf-8') as f: f.write(report_content) print(f"搜索报告已保存至：{output_file}") return report_content # 使用示例 def main(): """主函数示例""" # 创建查询系统 query_system = SymbolQuerySystem() # 示例1：关键词搜索 print("=== 关键词搜索示例 ===") results = query_system.search_by_keyword("P") print(f"找到 {len(results)} 个包含 'P' 的符号") # 示例2：属性过滤 print("\n=== 属性过滤示例 ===") filters = { 'yin_yang_attribute': 'yin', 'engraving_type': 'yin_engraving' } results = query_system.filter_by_attributes(filters) print(f"找到 {len(results)} 个阴属性阴刻符号") # 示例3：查找相关符号 print("\n=== 查找相关符号示例 ===") related = query_system.find_related_symbols("P_yin_001") print(f"符号信息：{related['symbol_info'].get('symbol_name', 'N/A')}") print(f"直接关联：{len(related['direct_links'])} 个") # 示例4：符号比较 print("\n=== 符号比较示例 ===") comparison = query_system.compare_symbols(["P_yin_001", "T_yang_001"]) print(f"阴阳属性相同：{comparison['yin_yang_comparison']['is_same']}") # 示例5：符号家族分析 print("\n=== 符号家族分析示例 ===") family = query_system.analyze_symbol_family("P_*") if 'error' not in family: print(f"P家族符号数量：{family['family_stats']['total_count']}") # 示例6：高级搜索 print("\n=== 高级搜索示例 ===") search_params = { 'keywords': ['P', 'yin'], 'yin_yang': ['yin'], 'civilization': ['Chinese', 'Greek'] } results = query_system.advanced_search(search_params) print(f"高级搜索找到 {len(results)} 个符号") # 示例7：创建搜索报告 print("\n=== 创建搜索报告示例 ===") report = query_system.create_search_report(search_params, "search_report.md") print("搜索报告已生成") if __name__ == "__main__": main()