#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ TKE UI/UX Pro Max Core - BM25 search engine for TKE UI standards """ import csv import re import pickle from pathlib import Path from math import log from collections import defaultdict # ============ CONFIGURATION ============ DATA_DIR = Path(__file__).parent.parent / "data" CACHE_DIR = Path(__file__).parent.parent / ".cache" CACHE_DIR.mkdir(exist_ok=True) MAX_RESULTS = 3 CSV_CONFIG = { "component": { "file": "tke-components.csv", "search_cols": ["Category", "Component", "Keywords", "Usage", "Use Case", "Context"], "output_cols": ["Category", "Component", "Keywords", "Class Names", "Usage", "Use Case", "Minimal Example", "Do", "Don't", "Severity"] }, "form": { "file": "tke-forms.csv", "search_cols": ["Category", "Element", "Keywords", "Layout", "Use Case", "Context"], "output_cols": ["Category", "Element", "Keywords", "Layout", "Use Case", "Minimal Example", "Best Practice", "Common Mistake", "Severity"] }, "table": { "file": "tke-tables.csv", "search_cols": ["Category", "Type", "Keywords", "Use Case", "Context"], "output_cols": ["Category", "Type", "Keywords", "Class Names", "Use Case", "Minimal Example", "Configuration", "Best Practice", "Severity"] }, "layout": { "file": "tke-layouts.csv", "search_cols": ["Pattern", "Keywords", "Use Case", "Context", "Structure"], "output_cols": ["Pattern", "Keywords", "Structure", "Use Case", "Minimal Example", "Notes"] }, "color": { "file": "tke-colors.csv", "search_cols": ["Variable", "Value", "Type", "Class", "Usage", "Keywords", "Description"], "output_cols": ["Variable", "Value", "Hex", "Type", "Class", "Usage", "Keywords", "Description", "Example"] }, "typography": { "file": "tke-typography.csv", "search_cols": ["Element", "Class", "Size", "Font", "Keywords", "Usage"], "output_cols": ["Element", "Class", "Size", "Font", "Keywords", "Usage", "Example"] }, "practice": { "file": "tke-practices.csv", "search_cols": ["Category", "Issue", "Keywords", "Description"], "output_cols": ["Category", "Issue", "Description", "Do", "Don't", "Code Good", "Code Bad", "Severity"] }, "pattern": { "file": "tke-patterns.csv", "search_cols": ["Pattern", "Components", "Keywords", "Use Case", "Context", "Structure"], "output_cols": ["Pattern", "Components", "Keywords", "Structure", "Use Case", "Minimal Example", "Notes"] }, "icon": { "file": "tke-icons.csv", "search_cols": ["Icon Name", "Icon Class", "Category", "Keywords", "Usage"], "output_cols": ["Icon Name", "Icon Class", "Category", "Keywords", "Usage", "Example", "Notes"] }, "responsive": { "file": "tke-responsive.csv", "search_cols": ["Breakpoint", "Class Prefix", "Width", "Use Case", "Keywords", "Notes"], "output_cols": ["Breakpoint", "Class Prefix", "Width", "Use Case", "Example", "Keywords", "Notes"] } } AVAILABLE_DOMAINS = list(CSV_CONFIG.keys()) # ============ QUERY EXPANSION (SYNONYMS) ============ SYNONYM_DICT = { "button": ["btn", "button", "按钮"], "icon": ["icon", "图标", "symbol"], "modal": ["modal", "dialog", "弹窗", "弹出框"], "form": ["form", "表单", "formulary"], "table": ["table", "表格", "datatable", "data-table"], "action": ["action", "操作", "operation", "operate"], "operation": ["operation", "操作", "action", "操作列", "操作按钮"], "layout": ["layout", "布局", "structure", "结构"], "card": ["card", "卡片", "panel"], "input": ["input", "输入框", "textfield", "text-field"], "select": ["select", "选择框", "dropdown", "下拉框"], "checkbox": ["checkbox", "复选框", "check-box"], "radio": ["radio", "单选框", "radio-button"], "required": ["required", "必填", "mandatory", "必需"], "validation": ["validation", "验证", "validate", "校验"], "hover": ["hover", "悬停", "鼠标悬停"], "click": ["click", "点击", "tap"], "responsive": ["responsive", "响应式", "adaptive", "适配"], "mobile": ["mobile", "手机", "移动端"], "desktop": ["desktop", "桌面", "pc端"], "primary": ["primary", "主要", "主要色"], "secondary": ["secondary", "次要", "次要色"], "color": ["color", "颜色", "colour", "色彩"], "font": ["font", "字体", "typography", "字型"], "size": ["size", "尺寸", "sizing", "大小"], "spacing": ["spacing", "间距", "margin", "padding"], "label": ["label", "标签", "标记"], "field": ["field", "字段", "域"], "element": ["element", "元素", "组件"], "component": ["component", "组件", "元件"], "pattern": ["pattern", "模式", "样式"], "style": ["style", "样式", "风格"], "timeline": ["timeline", "时间线", "time line", "timeline-box", "步骤", "步骤条", "流程", "process", "step", "history", "log", "record", "chronological", "事件", "历史记录"], "step": ["step", "步骤", "步骤条", "process", "流程", "阶段", "stage"], "history": ["history", "历史", "历史记录", "log", "记录", "record", "chronological", "时间顺序"], # 添加缺失的同义词 "datepicker": ["datepicker", "日期选择器", "日历", "date picker", "daterangepicker", "datepicker", "日期"], "pagination": ["pagination", "分页", "page navigation", "翻页", "page"], "breadcrumb": ["breadcrumb", "面包屑", "导航路径", "breadcrumb"], # 确保中文查询也能扩展（反向映射） "卡片": ["card", "卡片", "panel"], "弹窗": ["modal", "弹窗", "dialog", "弹出框"], "日期选择器": ["datepicker", "日期选择器", "日历", "date picker"], "分页": ["pagination", "分页", "page navigation"], "表格": ["table", "表格", "datatable", "data-table"], "操作": ["action", "operation", "操作", "操作列", "操作按钮"], "操作列": ["action column", "operation column", "操作列", "表格操作列"], "工单": ["work order", "order", "工单", "任务单", "维护工单"], "审批": ["approval", "approve", "审批", "审核", "批准"], "批量": ["batch", "bulk", "批量", "批量操作"], "表单": ["form", "表单", "formulary"], "图标": ["icon", "图标", "symbol"], "颜色": ["color", "颜色", "colour", "色彩"], "字体": ["typography", "字体", "font", "字型"], "响应式": ["responsive", "响应式", "adaptive", "适配"], # 添加中文词组映射 "表单布局": ["form layout", "表单布局", "form", "layout", "标签宽度", "列宽"], "卡片分块": ["card blocking", "卡片分块", "功能模块", "分块", "模块", "card grouping"], "状态徽章": ["badge", "状态徽章", "状态标签", "徽章", "标签", "status badge"], "标签宽度": ["label width", "标签宽度", "列宽", "column width"], "设备图标": ["equipment icon", "设备图标", "电梯图标", "设备", "电梯", "unit icon"], "维护图标": ["maintenance icon", "维护图标", "维修图标", "维护", "维修", "repair icon"], "详情页": ["detail page", "详情页", "详情页布局", "页面布局", "detail layout"], "报表布局": ["report layout", "报表布局", "统计页面", "dashboard", "statistics page"], "状态颜色": ["status color", "状态颜色", "成功颜色", "状态", "成功", "success color"], "时间线": ["timeline", "时间线", "time line", "timeline-box", "步骤条", "流程"], # 添加拖拽相关同义词 "拖拽": ["drag", "拖拽", "drop", "sortable", "reorder", "move"], "drag": ["drag", "拖拽", "drop", "sortable", "reorder"], "drop": ["drop", "拖拽", "drag", "sortable"], "sortable": ["sortable", "拖拽", "drag drop", "reorder", "move"], # 添加下载相关同义词 "下载": ["download", "下载", "export", "导出", "save"], "download": ["download", "下载", "export", "导出"], "export": ["export", "导出", "download", "下载"], # 添加时间相关同义词 "时间": ["time", "时间", "date", "日期", "datetime", "timestamp"], "日期": ["date", "日期", "time", "时间", "datetime"], "datetime": ["datetime", "时间", "日期", "date time", "timestamp"], # 添加表头排序相关同义词 "表头": ["header", "表头", "thead", "table header", "column header"], "header": ["header", "表头", "thead", "table header", "column header"], "排序": ["sort", "排序", "ordering", "order", "列排序", "列头排序"], "sort": ["sort", "排序", "ordering", "order", "列排序"], "ordering": ["ordering", "排序", "sort", "order", "列排序"], "列排序": ["column sort", "列排序", "列头排序", "ordering", "sort", "header sort"], "列头排序": ["column header sort", "列头排序", "列排序", "header sort", "ordering"], } # ============ FIELD WEIGHTS ============ # Higher weight means more important for search ranking FIELD_WEIGHTS = { "Keywords": 3.0, # Keywords field is most important "Component": 2.0, # Component name "Element": 2.0, # Element name "Pattern": 2.0, # Pattern name "Icon Class": 2.0, # Icon class name "Icon Name": 2.0, # Icon name "Usage": 1.5, # Usage description "Use Case": 1.5, # Use case description "Category": 1.0, # Category (standard weight) "Context": 0.8, # Context (lower weight) "Description": 0.8, # Description (lower weight) "Structure": 0.8, # Structure (lower weight) "Notes": 0.8, # Notes (lower weight) # Default weight for unspecified fields: 1.0 } # ============ QUERY PREPROCESSING ============ def normalize_query(query): """Normalize query: lowercase, handle common abbreviations""" query = query.lower() # Normalize common abbreviations while preserving TKE-specific prefixes replacements = { r'\bbtn\b': 'button', # "btn" -> "button" but keep "btn-primary" intact } for pattern, replacement in replacements.items(): # Only replace standalone words (not part of class names) query = re.sub(pattern, replacement, query) return query def expand_query_with_synonyms(query_tokens, max_expansions=3): """Expand query tokens with synonyms from SYNONYM_DICT""" expanded = set(query_tokens) for token in query_tokens: if token in SYNONYM_DICT: expanded.update(SYNONYM_DICT[token][:max_expansions]) return list(expanded) def preprocess_query(query, expand_synonyms=True): """Full query preprocessing: normalize + optionally expand synonyms""" normalized = normalize_query(query) if not expand_synonyms: return normalized # Tokenize to get individual words tokens = re.findall(r'\w+', normalized) # Expand with synonyms expanded_tokens = expand_query_with_synonyms(tokens) # Join back to query string (BM25 will tokenize again, but we want synonyms included) expanded_query = " ".join(expanded_tokens) return expanded_query # ============ BM25 IMPLEMENTATION ============ class BM25: """BM25 ranking algorithm for text search""" def __init__(self, k1=1.5, b=0.75): self.k1 = k1 self.b = b self.corpus = [] self.doc_lengths = [] self.avgdl = 0 self.idf = {} self.doc_freqs = defaultdict(int) self.N = 0 def tokenize(self, text): """Lowercase, split, remove punctuation, filter short words (but keep Chinese)""" # Include Chinese characters (Unicode range \u4e00-\u9fff) in word matching text = re.sub(r'[^\w\s\u4e00-\u9fff]', ' ', str(text).lower()) words = [] for w in text.split(): # Keep words longer than 2 chars OR Chinese characters if len(w) > 2 or re.search(r'[\u4e00-\u9fff]', w): words.append(w) return words def fit(self, documents): """Build BM25 index from documents""" self.corpus = [self.tokenize(doc) for doc in documents] self.N = len(self.corpus) if self.N == 0: return self.doc_lengths = [len(doc) for doc in self.corpus] self.avgdl = sum(self.doc_lengths) / self.N for doc in self.corpus: seen = set() for word in doc: if word not in seen: self.doc_freqs[word] += 1 seen.add(word) for word, freq in self.doc_freqs.items(): self.idf[word] = log((self.N - freq + 0.5) / (freq + 0.5) + 1) def score(self, query): """Score all documents against query""" query_tokens = self.tokenize(query) scores = [] for idx, doc in enumerate(self.corpus): score = 0 doc_len = self.doc_lengths[idx] term_freqs = defaultdict(int) for word in doc: term_freqs[word] += 1 for token in query_tokens: if token in self.idf: tf = term_freqs[token] idf = self.idf[token] numerator = tf * (self.k1 + 1) denominator = tf + self.k1 * (1 - self.b + self.b * doc_len / self.avgdl) score += idf * numerator / denominator scores.append((idx, score)) return sorted(scores, key=lambda x: x[1], reverse=True) # ============ WEIGHTED BM25 IMPLEMENTATION ============ class WeightedBM25: """BM25 with field-level weighting support""" def __init__(self, k1=1.5, b=0.75, field_weights=None): self.k1 = k1 self.b = b self.field_weights = field_weights or {} self.corpus_dict = [] # List of dicts, each dict represents one document's fields self.field_corpus = {} # Dict of field_name -> list of tokenized field values self.field_avgdl = {} # Dict of field_name -> average document length self.field_idf = {} # Dict of field_name -> dict of word -> idf self.field_doc_freqs = {} # Dict of field_name -> dict of word -> document frequency self.N = 0 def tokenize(self, text): """Lowercase, split, remove punctuation, filter short words (but keep Chinese)""" # Include Chinese characters (Unicode range \u4e00-\u9fff) in word matching text = re.sub(r'[^\w\s\u4e00-\u9fff]', ' ', str(text).lower()) words = [] for w in text.split(): # Keep words longer than 2 chars OR Chinese characters if len(w) > 2 or re.search(r'[\u4e00-\u9fff]', w): words.append(w) return words def fit(self, documents_dict, search_cols): """Build weighted BM25 index from documents (list of dicts)""" self.corpus_dict = documents_dict self.N = len(documents_dict) if self.N == 0: return # Build field-level corpus for col in search_cols: self.field_corpus[col] = [ self.tokenize(str(doc.get(col, ""))) for doc in documents_dict ] # Calculate average document length for this field field_lengths = [len(doc) for doc in self.field_corpus[col]] self.field_avgdl[col] = sum(field_lengths) / self.N if self.N > 0 else 0 # Calculate document frequencies for this field self.field_doc_freqs[col] = defaultdict(int) for doc in self.field_corpus[col]: seen = set() for word in doc: if word not in seen: self.field_doc_freqs[col][word] += 1 seen.add(word) # Calculate IDF for this field self.field_idf[col] = {} for word, freq in self.field_doc_freqs[col].items(): self.field_idf[col][word] = log((self.N - freq + 0.5) / (freq + 0.5) + 1) def score(self, query): """Score all documents against query using weighted BM25""" query_tokens = self.tokenize(query) scores = [] for idx, doc_dict in enumerate(self.corpus_dict): total_score = 0 # Score each field separately, then weight and sum for col, field_corpus in self.field_corpus.items(): field_weight = self.field_weights.get(col, 1.0) field_doc = field_corpus[idx] field_len = len(field_doc) field_avgdl = self.field_avgdl[col] field_idf = self.field_idf[col] # Calculate BM25 score for this field field_score = 0 term_freqs = defaultdict(int) for word in field_doc: term_freqs[word] += 1 for token in query_tokens: if token in field_idf: tf = term_freqs[token] idf = field_idf[token] numerator = tf * (self.k1 + 1) denominator = tf + self.k1 * (1 - self.b + self.b * field_len / field_avgdl) if field_avgdl > 0 else tf + self.k1 field_score += idf * numerator / denominator # Apply field weight total_score += field_score * field_weight scores.append((idx, total_score)) return sorted(scores, key=lambda x: x[1], reverse=True) # ============ SEARCH FUNCTIONS ============ def _load_csv(filepath): """Load CSV and return list of dicts""" with open(filepath, 'r', encoding='utf-8') as f: return list(csv.DictReader(f)) def _get_cache_file(filepath): """Generate cache file path based on CSV file path""" cache_name = filepath.stem + "_weighted_bm25.pkl" return CACHE_DIR / cache_name def _load_cached_index(filepath): """Load cached BM25 index if available and valid""" cache_file = _get_cache_file(filepath) if not cache_file.exists(): return None try: with open(cache_file, 'rb') as f: cache_data = pickle.load(f) # Check if cache is valid (file hasn't been modified) file_mtime = filepath.stat().st_mtime if cache_data.get('mtime') == file_mtime: return cache_data['index'] except Exception: # If cache is corrupted, return None to rebuild pass return None def _save_cached_index(filepath, index): """Save BM25 index to cache""" cache_file = _get_cache_file(filepath) try: file_mtime = filepath.stat().st_mtime with open(cache_file, 'wb') as f: pickle.dump({ 'index': index, 'mtime': file_mtime }, f) except Exception: # If cache save fails, continue without cache pass def _search_csv(filepath, search_cols, output_cols, query, max_results, use_weights=True, use_cache=True, expand_synonyms=True, explain=False): """ Core search function using weighted BM25 with caching Args: filepath: Path to CSV file search_cols: List of column names to search output_cols: List of column names to output query: Search query string max_results: Maximum number of results to return use_weights: Whether to use field weighting (default: True) use_cache: Whether to use cached index (default: True) expand_synonyms: Whether to expand query with synonyms (default: True) explain: Whether to return explanation details (default: False) Returns: If explain=False: List of result dicts If explain=True: Tuple of (results, explanations) where explanations is a list of explain dicts """ if not filepath.exists(): return [] if not explain else ([], []) data = _load_csv(filepath) # Preprocess query (normalize + optionally expand synonyms) original_query = query expanded_query = preprocess_query(query, expand_synonyms=expand_synonyms) # Track synonym expansions for explanation synonym_expansions = {} if expand_synonyms: original_tokens = set(re.findall(r'\w+', original_query.lower())) expanded_tokens = set(re.findall(r'\w+', expanded_query.lower())) for token in original_tokens: if token in SYNONYM_DICT: synonyms = [s for s in SYNONYM_DICT[token] if s in expanded_tokens and s != token] if synonyms: synonym_expansions[token] = synonyms[:3] # Limit to 3 # Try to load cached index index = None if use_cache: index = _load_cached_index(filepath) # Build or use cached index if index is None: if use_weights: # Build documents as dicts for weighted BM25 documents_dict = data index = WeightedBM25(field_weights=FIELD_WEIGHTS) index.fit(documents_dict, search_cols) else: # Build documents as strings for standard BM25 documents = [" ".join(str(row.get(col, "")) for col in search_cols) for row in data] index = BM25() index.fit(documents) # Save to cache if use_cache: _save_cached_index(filepath, index) # Perform search ranked = index.score(expanded_query) # Tokenize query for explanation query_tokens = index.tokenize(expanded_query) if hasattr(index, 'tokenize') else re.findall(r'\w+', expanded_query.lower()) # Get top results with score > 0 results = [] explanations = [] for idx, score in ranked[:max_results]: if score > 0: row = data[idx] result = {col: row.get(col, "") for col in output_cols if col in row} results.append(result) if explain: # Build explanation for this result explain_info = { "score": round(score, 2), "matched_keywords": [], "matched_fields": {}, "synonym_expansions": synonym_expansions if idx == 0 else {} # Only show for first result } # Find matched keywords in each field if use_weights and hasattr(index, 'field_corpus'): # Weighted BM25: check each field for col in search_cols: if col in index.field_corpus: field_doc = index.field_corpus[col][idx] field_text = " ".join(field_doc).lower() matched = [token for token in query_tokens if token in field_text] if matched: explain_info["matched_fields"][col] = matched[:3] # Top 3 matches explain_info["matched_keywords"].extend(matched[:2]) # Top 2 per field else: # Standard BM25: check combined text doc_text = " ".join(str(row.get(col, "")) for col in search_cols).lower() matched = [token for token in query_tokens if token in doc_text] explain_info["matched_keywords"] = matched[:5] # Top 5 matches # Remove duplicates explain_info["matched_keywords"] = list(set(explain_info["matched_keywords"]))[:5] explanations.append(explain_info) if explain: return results, explanations return results def detect_domain(query, expand_synonyms=True): """ Auto-detect the most relevant domain from query Args: query: Search query string expand_synonyms: Whether to expand query with synonyms before detection Returns: Tuple of (primary_domain, confidence_score, candidate_domains) - primary_domain: The most likely domain - confidence_score: 0.0 to 1.0, higher means more confident - candidate_domains: List of (domain, score) tuples, sorted by score """ # Preprocess query with synonyms for better detection if expand_synonyms: query_expanded = preprocess_query(query, expand_synonyms=True) query_tokens = set(re.findall(r'\w+', query_expanded.lower())) original_tokens = set(re.findall(r'\w+', query.lower())) query_tokens.update(original_tokens) # Include original tokens else: query_tokens = set(re.findall(r'\w+', query.lower())) query_lower = " ".join(query_tokens) # Enhanced domain keywords with priority weights domain_keywords = { "component": { "keywords": ["button", "card", "modal", "alert", "tab", "progress", "component", "cursor", "hover", "transition", "border-radius", "border radius", "popover", "tooltip", "dropdown", "drag", "drop", "sortable", "卡片", "弹窗", "按钮", "警告", "标签页", "拖拽", "拖放"], "priority": 1.0 }, "form": { "keywords": ["form", "input", "select", "datepicker", "checkbox", "radio", "required", "validation", "label", "star_mark", "star mark", "textarea", "file", "upload", "date", "time", "datetime", "表单", "输入框", "选择框", "日期选择器", "日历", "复选框", "单选框", "必填", "日期", "时间"], "priority": 1.0 }, "table": { "keywords": ["table", "datatable", "pagination", "row", "column", "data", "performance", "server-side", "sort", "filter", "ordering", "order", "header", "thead", "column sort", "header sort", "表格", "分页", "翻页", "数据表", "排序", "表头", "列排序", "列头排序"], "priority": 1.0 }, "layout": { "keywords": ["layout", "search", "action", "button group", "form layout", "horizontal", "vertical", "responsive", "truncate", "title", "collapsible", "navigation", "left right", "sidebar", "grid", "flex", "timeline", "time line", "步骤", "步骤条", "流程", "process", "step", "history", "log", "record", "chronological", "事件", "时间线"], "priority": 1.0, # Increased from 0.8 to match form/component priority "exact_match_keywords": ["timeline", "时间线", "time line", "timeline-box"] # Exact match keywords get higher weight }, "color": { "keywords": ["color", "palette", "hex", "#", "rgb", "primary", "secondary", "gray", "orange", "text-primary", "bg-primary", "text-secondary", "bg-secondary", "颜色", "色彩", "主要色", "次要色"], "priority": 1.0 }, "typography": { "keywords": ["font", "typography", "heading", "fs-1", "fs-2", "fs-3", "fs-4", "fs-5", "fs-6", "text", "h1", "h2", "h3", "h4", "h5", "h6", "字体", "字型", "标题"], "priority": 1.0 }, "practice": { "keywords": ["accessibility", "aria-label", "semantic html", "code organization"], "priority": 0.8 }, "pattern": { "keywords": ["pattern", "combination", "modal form", "card form", "search form"], "priority": 0.6 }, "icon": { "keywords": ["icon", "tk-icon", "icon-tk-", "visual", "symbol", "emoji", "svg", "download", "export", "excel", "图标", "下载", "导出"], "priority": 1.0 }, "responsive": { "keywords": ["responsive", "breakpoint", "col-sm", "col-md", "col-lg", "col-xl", "mobile", "tablet", "desktop", "viewport", "响应式", "适配", "移动端", "手机"], "priority": 0.9 } } # Calculate scores with priority weights and exact match bonus scores = {} for domain, config in domain_keywords.items(): keywords = config["keywords"] priority = config.get("priority", 1.0) exact_match_keywords = config.get("exact_match_keywords", []) # Exact match bonus: check if query contains exact match keywords (higher weight) exact_matches = 0 for exact_kw in exact_match_keywords: if exact_kw in query_lower or exact_kw in query: exact_matches += 1 # Regular keyword matches keyword_matches = sum(1 for kw in keywords if kw in query_lower) # Check token-level matches (more precise) token_matches = sum(1 for token in query_tokens if any(kw in token or token in kw for kw in keywords)) # Combined score: exact matches get 3x weight, keyword matches get 1x, token matches get 0.5x # This ensures exact matches (like "timeline", "时间线") have higher priority than partial matches (like "time") scores[domain] = (exact_matches * 3.0 + keyword_matches + token_matches * 0.5) * priority # Sort candidates by score candidate_domains = sorted(scores.items(), key=lambda x: x[1], reverse=True) # Get best domain best_domain, best_score = candidate_domains[0] if candidate_domains else ("component", 0.0) # Calculate confidence (normalize by max possible score) max_possible_score = len(query_tokens) * 1.0 # Assuming one keyword per token max confidence = min(best_score / max(max_possible_score, 1), 1.0) # If confidence is very low, consider multiple candidates if confidence < 0.3 and len(candidate_domains) > 1: # Low confidence - might need multi-domain search return best_domain, confidence, candidate_domains return best_domain, confidence, candidate_domains def search(query, domain=None, max_results=MAX_RESULTS, use_weights=True, use_cache=True, expand_synonyms=True, multi_domain=False, auto_fallback=True, min_results=1, explain=False): """ Main search function with auto-domain detection and multi-domain fallback Args: query: Search query string domain: Domain to search (auto-detected if None) max_results: Maximum number of results to return per domain use_weights: Whether to use field weighting (default: True) use_cache: Whether to use cached index (default: True) expand_synonyms: Whether to expand query with synonyms (default: True) multi_domain: If True, search multiple domains when confidence is low (default: False) auto_fallback: If True, automatically search other domains if results < min_results (default: True) min_results: Minimum number of results required before fallback (default: 1) Returns: Dict with search results: { "domain": detected domain(s), "query": original query, "file": CSV filename(s), "count": number of results, "results": list of result dicts, "confidence": confidence score (0.0-1.0), "searched_domains": list of domains searched } """ searched_domains = [] all_results = [] all_explanations = [] confidence = 1.0 domain_was_auto_detected = (domain is None) candidates = [] if domain is None: # Auto-detect domain domain, confidence, candidates = detect_domain(query, expand_synonyms=expand_synonyms) searched_domains.append(domain) # If confidence is low and multi_domain is enabled, search top candidates if multi_domain and confidence < 0.5 and len(candidates) > 1: # Search top 2-3 candidate domains domains_to_search = [d for d, score in candidates[:3] if score > 0] else: domains_to_search = [domain] else: # Explicit domain specified domains_to_search = [domain] searched_domains.append(domain) # Search primary domain(s) for search_domain in domains_to_search: config = CSV_CONFIG.get(search_domain) if not config: continue filepath = DATA_DIR / config["file"] if not filepath.exists(): continue search_result = _search_csv(filepath, config["search_cols"], config["output_cols"], query, max_results, use_weights, use_cache, expand_synonyms, explain=explain) if explain: domain_results, domain_explanations = search_result else: domain_results = search_result domain_explanations = [] # Add domain info to each result for result in domain_results: result["_domain"] = search_domain result["_file"] = config["file"] all_results.extend(domain_results) all_explanations.extend(domain_explanations) # Remove duplicates (same Component/Element/Pattern name) seen = set() unique_results = [] for result in all_results: # Create a unique key key = (result.get("Component") or result.get("Element") or result.get("Pattern") or result.get("Icon Name") or "", result.get("Keywords", "")) if key not in seen: seen.add(key) unique_results.append(result) # Sort by score (if available) or keep original order # Limit to max_results final_results = unique_results[:max_results] # Auto-fallback: if results are insufficient, search additional domains # Only fallback if domain was auto-detected (not explicitly specified) if auto_fallback and len(final_results) < min_results and domain_was_auto_detected: # Use existing candidates or get new ones if not candidates: _, _, candidates = detect_domain(query, expand_synonyms=expand_synonyms) remaining_domains = [d for d, score in candidates if d not in searched_domains and score > 0] # Try searching other domains for fallback_domain in remaining_domains[:3]: # Try up to 3 additional domains config = CSV_CONFIG.get(fallback_domain) if not config: continue filepath = DATA_DIR / config["file"] if not filepath.exists(): continue searched_domains.append(fallback_domain) fallback_result = _search_csv(filepath, config["search_cols"], config["output_cols"], query, max_results, use_weights, use_cache, expand_synonyms, explain=explain) if explain: fallback_results, fallback_explanations = fallback_result else: fallback_results = fallback_result fallback_explanations = [] # Add domain info for result in fallback_results: result["_domain"] = fallback_domain result["_file"] = config["file"] # Merge results, avoiding duplicates existing_keys = {(r.get("Component") or r.get("Element") or r.get("Pattern") or "", r.get("Keywords", "")) for r in final_results} for result in fallback_results: key = (result.get("Component") or result.get("Element") or result.get("Pattern") or "", result.get("Keywords", "")) if key not in existing_keys: final_results.append(result) existing_keys.add(key) if explain: all_explanations.append(fallback_explanations[len(all_explanations) - len(final_results) + len(existing_keys) - 1] if len(all_explanations) < len(final_results) else {}) if len(final_results) >= max_results: break if len(final_results) >= min_results: break # Clean up internal fields before returning for result in final_results: result.pop("_domain", None) result.pop("_file", None) # Determine domain string for output if len(searched_domains) == 1: domain_str = searched_domains[0] file_str = CSV_CONFIG[searched_domains[0]]["file"] else: domain_str = f"{searched_domains[0]}+{len(searched_domains)-1}more" file_str = f"{CSV_CONFIG[searched_domains[0]]['file']} (+{len(searched_domains)-1} more)" result_dict = { "domain": domain_str, "query": query, "file": file_str, "count": len(final_results), "results": final_results, "confidence": confidence, "searched_domains": searched_domains } if explain: # Match explanations to final results (by index) final_explanations = [] for i, result in enumerate(final_results): if i < len(all_explanations): final_explanations.append(all_explanations[i]) else: final_explanations.append({}) result_dict["explanations"] = final_explanations return result_dict