#!/usr/bin/env python3 """ Search extracted markdown articles by YAML frontmatter metadata and export results. Usage: python tools/search_md.py --root docs_md/articles --query "breadcrumbs:Brain>Diagnosis" --out results.json Supported query keys: breadcrumbs, authors, pageKeywords, category, title, enhancedTitle, type You can also use key `global` to search across all relevant fields. Breadcrumbs query format: 'A>B>C' (matches files whose breadcrumbs contain that sequence in order) Authors query: substring match against author.value Keywords: substring match against pageKeywords Exports: JSON (default) or CSV with columns: path, title, docid, breadcrumbs, authors, pageKeywords Optionally copy matched files into an output directory. """ from __future__ import annotations import argparse import csv import json import os import re from typing import List, Dict, Any, Tuple, Optional import yaml import fnmatch def read_frontmatter(path: str) -> Dict[str, Any]: """Read YAML frontmatter from a markdown file and return as dict.""" try: with open(path, 'r', encoding='utf-8') as f: text = f.read() except Exception: return {} if text.startswith('---'): parts = text.split('---', 2) if len(parts) >= 3: fm = parts[1] try: data = yaml.safe_load(fm) if isinstance(data, dict): return data except Exception: return {} return {} def read_md_file(path: str) -> tuple[Dict[str, Any], str]: """Read frontmatter and content from a markdown file.""" try: with open(path, 'r', encoding='utf-8') as f: text = f.read() except Exception: return {}, "" fm = {} content = text if text.startswith('---'): parts = text.split('---', 2) if len(parts) >= 3: fm_text = parts[1] content = parts[2] try: data = yaml.safe_load(fm_text) if isinstance(data, dict): fm = data except Exception: pass return fm, content import re def tokenize(text: str) -> List[str]: """Tokenize text into lowercase alphanumeric words.""" if not text: return [] return re.findall(r'\b[a-zA-Z0-9]+\b', text.lower()) def stem_word(word: str) -> str: """A simple suffix-stripping stemmer for common English and medical terms.""" word = word.lower().strip() if len(word) <= 3: return word # Standard plural stripping if word.endswith('s') and not word.endswith('ss'): if word.endswith('ies'): word = word[:-3] + 'y' elif word.endswith('es'): if any(word.endswith(x) for x in ['ches', 'shes', 'xes', 'zes']): word = word[:-2] else: word = word[:-1] else: word = word[:-1] # Common suffixes suffixes = [ ('ectomy', 6), ('otomy', 5), ('pathy', 5), ('itis', 4), ('osis', 4), ('tion', 4), ('sion', 4), ('ness', 4), ('ment', 4), ('able', 4), ('ible', 4), ('ical', 4), ('ing', 3), ('ive', 3), ('est', 3), ('ism', 3), ('ity', 3), ('oma', 3), ('ed', 2), ('ly', 2), ('al', 2), ('ic', 2), ('er', 2), ('or', 2), ('ia', 2), ('ar', 2), ('y', 1) ] for suffix, length in suffixes: if word.endswith(suffix) and len(word) - length >= 3: return word[:-length] return word def stem_text(text: str) -> Set[str]: """Tokenize and stem all words in a text block.""" return {stem_word(w) for w in tokenize(text)} def match_stemming(query_text: str, target_text: str) -> bool: if not query_text or not target_text: return False q_stems = {stem_word(w) for w in tokenize(query_text)} t_stems = stem_text(target_text) return q_stems.issubset(t_stems) def edit_distance(s1: str, s2: str) -> int: # Levenshtein distance if len(s1) < len(s2): s1, s2 = s2, s1 if len(s2) == 0: return len(s1) previous_row = list(range(len(s2) + 1)) for i, c1 in enumerate(s1): current_row = [i + 1] for j, c2 in enumerate(s2): insertions = previous_row[j + 1] + 1 deletions = current_row[j] + 1 substitutions = previous_row[j] + (c1 != c2) current_row.append(min(insertions, deletions, substitutions)) previous_row = current_row return previous_row[-1] def is_fuzzy_match_word(q_word: str, t_word: str) -> bool: if q_word == t_word or q_word in t_word or t_word in q_word: return True len_q, len_t = len(q_word), len(t_word) if abs(len_q - len_t) > 2: return False max_dist = 0 if len_q >= 6: max_dist = 2 elif len_q >= 4: max_dist = 1 return edit_distance(q_word, t_word) <= max_dist def match_fuzzy(query_text: str, target_text: str) -> bool: if not query_text or not target_text: return False q_words = tokenize(query_text) t_words = set(tokenize(target_text)) for qw in q_words: found_match = False for tw in t_words: if is_fuzzy_match_word(qw, tw): found_match = True break if not found_match: return False return True def get_content_snippet(content: str, query: str) -> str: """Extract a small snippet from content around the first match of query terms.""" if not content or not query: return "" q_words = tokenize(query) if not q_words: return content[:150] + "..." if len(content) > 150 else content content_lower = content.lower() first_pos = -1 for qw in q_words: pos = content_lower.find(qw) if pos != -1: if first_pos == -1 or pos < first_pos: first_pos = pos if first_pos == -1: return content[:150] + "..." if len(content) > 150 else content start = max(0, first_pos - 60) end = min(len(content), first_pos + 90) snippet = content[start:end] prefix = "..." if start > 0 else "" suffix = "..." if end < len(content) else "" snippet_clean = re.sub(r'\s+', ' ', snippet).strip() return f"{prefix}{snippet_clean}{suffix}" def match_breadcrumbs(bcs: List[str], query: List[str]) -> bool: """Return True if breadcrumbs list contains the query sequence (in order). Supports '*' wildcard in query elements. """ if not bcs or not query: return False qi = 0 for b in bcs: if qi >= len(query): break qpart = query[qi].strip() if not qpart: qi += 1 continue # case-insensitive matching with wildcard support pattern = qpart try: if fnmatch.fnmatchcase(str(b).strip().lower(), pattern.lower()): qi += 1 except Exception: if pattern.lower() == str(b).strip().lower(): qi += 1 return qi == len(query) def match_author(authors: List[Dict[str, Any]], q: str) -> bool: if not authors or not q: return False q = q.strip() # wildcard support has_wild = '*' in q for a in authors: v = a.get('value') if isinstance(a, dict) else str(a) if not v: continue if has_wild: if fnmatch.fnmatchcase(str(v).strip().lower(), q.lower()): return True else: if q.lower() in str(v).lower(): return True return False _docid_map = {} def get_doc_by_id(root: str, identifier: str) -> Tuple[Optional[Dict[str, Any]], Optional[str], Optional[str]]: global _docid_map if not _docid_map: for dirpath, dirnames, filenames in os.walk(root): for fn in filenames: if fn.endswith('.md'): path = os.path.join(dirpath, fn) fm, content = read_md_file(path) basename = os.path.splitext(fn)[0] if fm: docid = fm.get('docid') doc_info = (fm, content, path) if docid: _docid_map[docid] = doc_info _docid_map[basename] = doc_info return _docid_map.get(identifier, (None, None, None)) def extract_section_content(content: str, section_type: str) -> str: lines = content.split('\n') section_content = [] in_section = False section_level = 0 if section_type == 'anatomy': pattern = re.compile(r'^#+\s+.*anatomy.*', re.IGNORECASE) elif section_type == 'differential': pattern = re.compile(r'^#+\s+.*differential.*', re.IGNORECASE) else: return "" for line in lines: if in_section: if line.startswith('#'): current_level = len(line) - len(line.lstrip('#')) if current_level <= section_level: in_section = False else: section_content.append(line) else: section_content.append(line) elif pattern.match(line): in_section = True section_level = len(line) - len(line.lstrip('#')) return '\n'.join(section_content) def find_links_in_text(text: str) -> List[Dict[str, str]]: links = [] if not text: return links # 1. Standard markdown links [Label](/document/slug/uuid) markdown_pattern = re.compile(r'\[([^\]]+)\]\((/document/[^/]+/([a-f0-9\-]{36}))\)') for m in markdown_pattern.finditer(text): links.append({ 'title': m.group(1), 'docid': m.group(3) }) # 2. Raw ANATOMY links (e.g. Brain/ANATOMY:uuid or ANATOMY:uuid) anatomy_pattern = re.compile(r'(?:[a-zA-Z0-9\-]+/)?ANATOMY:([a-f0-9\-]{36})', re.IGNORECASE) for m in anatomy_pattern.finditer(text): docid = m.group(1) if not any(l['docid'] == docid for l in links): links.append({ 'title': 'Anatomy Document', 'docid': docid }) # 3. Raw DDX links (e.g. DDX:uuid) ddx_pattern = re.compile(r'DDX:([a-f0-9\-]{36})', re.IGNORECASE) for m in ddx_pattern.finditer(text): docid = m.group(1) if not any(l['docid'] == docid for l in links): links.append({ 'title': 'Differential Diagnosis', 'docid': docid }) return links def check_linked_sections(content: str, root: str = 'docs_md/articles') -> Dict[str, Any]: anatomy_text = extract_section_content(content, 'anatomy') anatomy_exists = bool(re.search(r'(?i)^#+\s+.*anatomy.*', content, re.MULTILINE)) diff_text = extract_section_content(content, 'differential') diff_exists = bool(re.search(r'(?i)^#+\s+.*differential.*', content, re.MULTILINE)) raw_anatomy_links = find_links_in_text(anatomy_text) raw_diff_links = find_links_in_text(diff_text) anatomy_links = [] for link in raw_anatomy_links: docid = link['docid'] fm, _, _ = get_doc_by_id(root, docid) title = fm.get('title') or fm.get('pageTitle') if fm else link['title'] anatomy_links.append({ 'title': title, 'docid': docid }) diff_links = [] for link in raw_diff_links: docid = link['docid'] fm, _, _ = get_doc_by_id(root, docid) title = fm.get('title') or fm.get('pageTitle') if fm else link['title'] diff_links.append({ 'title': title, 'docid': docid }) return { 'anatomy': { 'exists': anatomy_exists, 'links': anatomy_links }, 'differential': { 'exists': diff_exists, 'links': diff_links } } def run_search(root: str, qkey: str, qval: str, mode: str = 'exact', targets: List[str] = None, or_queries: List[Dict[str, Any]] = None, expand_links: bool = False) -> List[Dict[str, Any]]: out = [] # Normalize queries list if not or_queries: or_queries = [{ 'qkey': qkey, 'qval': qval, 'mode': mode, 'targets': targets }] normalized_specs = [] for spec in or_queries: qk = spec.get('qkey') or 'global' qv = spec.get('qval') or '' m = spec.get('mode') or 'exact' t = spec.get('targets') normalized_specs.append({ 'qkey': qk, 'qval': qv, 'mode': m, 'targets': t }) # Helper to test if a string/list/dict matches under a specific mode def value_matches(query: str, target_val: Any, mode_val: str) -> bool: if target_val is None: return False if isinstance(target_val, list): for item in target_val: if isinstance(item, dict): for k, v in item.items(): if value_matches(query, v, mode_val): return True else: if value_matches(query, item, mode_val): return True return False elif isinstance(target_val, dict): for k, v in target_val.items(): if value_matches(query, v, mode_val): return True return False s_val = str(target_val).strip() if not s_val: return False if mode_val == 'stemming': return match_stemming(query, s_val) elif mode_val == 'fuzzy': return match_fuzzy(query, s_val) else: # Exact substring or wildcard match q_lower = query.lower().strip() s_lower = s_val.lower() if '*' in q_lower: return fnmatch.fnmatchcase(s_lower, q_lower) else: return q_lower in s_lower for dirpath, dirnames, filenames in os.walk(root): for fn in filenames: if not fn.endswith('.md'): continue path = os.path.join(dirpath, fn) fm, content = read_md_file(path) if not fm and not content: continue matched = False reasons = [] snippet = "" for spec in normalized_specs: qk = spec['qkey'] qv = spec['qval'] m = spec['mode'] t = spec['targets'] if not qv: continue # skip empty queries spec_matched = False spec_reasons = [] spec_snippet = "" k_lower = (qk or '').strip().lower() if not k_lower or k_lower in ('global', 'all', 'any'): # Determine which targets to test if t: target_set = {x.lower() for x in t} else: target_set = {'title', 'docid', 'breadcrumbs', 'authors', 'keywords', 'category', 'type', 'content'} # 1. Title if 'title' in target_set: for f in ['title', 'enhancedTitle', 'pageTitle']: if value_matches(qv, fm.get(f), m): spec_matched = True spec_reasons.append('Title') break # 2. DocID if 'docid' in target_set: if value_matches(qv, fm.get('docid'), m): spec_matched = True spec_reasons.append('DocID') # 3. Breadcrumbs if 'breadcrumbs' in target_set: bcs = fm.get('breadcrumbs') or [] if isinstance(bcs, list) and bcs and isinstance(bcs[0], dict): bcs_list = [d.get('name') or d.get('slug') or '' for d in bcs] else: bcs_list = bcs if '>' in qv: query_parts = [p.strip() for p in qv.split('>') if p.strip()] if match_breadcrumbs(bcs_list, query_parts): spec_matched = True spec_reasons.append('Breadcrumbs') else: if value_matches(qv, bcs_list, m): spec_matched = True spec_reasons.append('Breadcrumbs') # 4. Authors if 'authors' in target_set: if value_matches(qv, fm.get('authors'), m): spec_matched = True spec_reasons.append('Authors') # 5. PageKeywords if 'keywords' in target_set or 'pagekeywords' in target_set: if value_matches(qv, fm.get('pageKeywords'), m): spec_matched = True spec_reasons.append('Keywords') # 6. Category if 'category' in target_set: if value_matches(qv, fm.get('category'), m): spec_matched = True spec_reasons.append('Category') # 7. Type if 'type' in target_set: if value_matches(qv, fm.get('type'), m): spec_matched = True spec_reasons.append('Type') # 8. Content if 'content' in target_set: if value_matches(qv, content, m): spec_matched = True spec_reasons.append('Content') spec_snippet = get_content_snippet(content, qv) else: # Specific key search if k_lower == 'breadcrumbs': bcs = fm.get('breadcrumbs') or [] if bcs and isinstance(bcs, list) and bcs and isinstance(bcs[0], dict): bcs_list = [d.get('name') or d.get('slug') or '' for d in bcs] else: bcs_list = bcs if '>' in qv: query_parts = [p.strip() for p in qv.split('>') if p.strip()] spec_matched = match_breadcrumbs(bcs_list, query_parts) else: spec_matched = value_matches(qv, bcs_list, m) if spec_matched: spec_reasons.append('Breadcrumbs') elif k_lower == 'authors': spec_matched = value_matches(qv, fm.get('authors'), m) if spec_matched: spec_reasons.append('Authors') elif k_lower == 'pagekeywords': spec_matched = value_matches(qv, fm.get('pageKeywords'), m) if spec_matched: spec_reasons.append('Keywords') elif k_lower == 'content': spec_matched = value_matches(qv, content, m) if spec_matched: spec_reasons.append('Content') spec_snippet = get_content_snippet(content, qv) elif k_lower in ('title', 'enhancedtitle', 'pagetitle'): for f in ['title', 'enhancedTitle', 'pageTitle']: if value_matches(qv, fm.get(f), m): spec_matched = True spec_reasons.append('Title') break else: spec_matched = value_matches(qv, fm.get(qk), m) if spec_matched: spec_reasons.append(qk.capitalize()) if spec_matched: matched = True spec_reasons_clean = [str(x) for x in spec_reasons] reasons.extend(spec_reasons_clean) if spec_snippet and not snippet: snippet = spec_snippet if matched: out.append({ 'path': path, 'title': fm.get('title') or fm.get('pageTitle') or os.path.splitext(fn)[0], 'docid': fm.get('docid'), 'breadcrumbs': fm.get('breadcrumbs'), 'authors': fm.get('authors'), 'pageKeywords': fm.get('pageKeywords'), 'reasons': sorted(list(set(reasons))), 'snippet': snippet, 'linked_info': check_linked_sections(content, root) }) if expand_links: seen_docids = {r.get('docid') for r in out if r.get('docid')} seen_paths = {r['path'] for r in out} expanded_results = [] for r in out: info = r.get('linked_info') if not info: continue parent_title = r['title'] # Anatomy links for link in info['anatomy']['links']: docid = link['docid'] if docid and docid not in seen_docids: fm, content, path = get_doc_by_id(root, docid) if fm: linked_fm_title = fm.get('title') or fm.get('pageTitle') or link['title'] expanded_results.append({ 'path': path, 'title': linked_fm_title, 'docid': docid, 'breadcrumbs': fm.get('breadcrumbs'), 'authors': fm.get('authors'), 'pageKeywords': fm.get('pageKeywords'), 'reasons': [f'Linked (Anatomy of {parent_title})'], 'snippet': content[:300] + '...' if content else '', 'linked_info': check_linked_sections(content, root) }) seen_docids.add(docid) seen_paths.add(path) # Differential links for link in info['differential']['links']: docid = link['docid'] if docid and docid not in seen_docids: fm, content, path = get_doc_by_id(root, docid) if fm: linked_fm_title = fm.get('title') or fm.get('pageTitle') or link['title'] expanded_results.append({ 'path': path, 'title': linked_fm_title, 'docid': docid, 'breadcrumbs': fm.get('breadcrumbs'), 'authors': fm.get('authors'), 'pageKeywords': fm.get('pageKeywords'), 'reasons': [f'Linked (Diff Diag of {parent_title})'], 'snippet': content[:300] + '...' if content else '', 'linked_info': check_linked_sections(content, root) }) seen_docids.add(docid) seen_paths.add(path) out.extend(expanded_results) return out def write_output(results: List[Dict[str, Any]], out_path: str, fmt: str = 'json'): if fmt == 'json': with open(out_path, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2, ensure_ascii=False) elif fmt == 'csv': with open(out_path, 'w', encoding='utf-8', newline='') as f: w = csv.writer(f) w.writerow(['path', 'title', 'docid', 'breadcrumbs', 'authors', 'pageKeywords']) for r in results: w.writerow([r['path'], r['title'], r.get('docid') or '', json.dumps(r.get('breadcrumbs') or []), json.dumps(r.get('authors') or []), r.get('pageKeywords') or '']) def main(argv=None): p = argparse.ArgumentParser() p.add_argument('--root', default='docs_md/articles', help='Root dir with extracted markdown') p.add_argument('--query', required=True, help='Query. Can be key:value or a raw search string') p.add_argument('--mode', choices=['exact', 'fuzzy', 'stemming'], default='exact', help='Search matching mode') p.add_argument('--out', default='results.json', help='Output file') p.add_argument('--format', choices=['json','csv'], default='json') p.add_argument('--copy-to', help='Optional: copy matched markdown files to this dir') p.add_argument("--copy-to-clear", default=True, action="store_true", help="Overwrite files when copying to --copy-to") args = p.parse_args(argv) if ':' in args.query: key, val = args.query.split(':', 1) key = key.strip() val = val.strip() else: key = 'global' val = args.query.strip() results = run_search(args.root, key, val, args.mode) write_output(results, args.out, args.format) print(f'Found {len(results)} matches. Wrote {args.out}') if args.copy_to and results: os.makedirs(args.copy_to, exist_ok=True) if args.copy_to_clear: # clear existing files in target dir for existing_fn in os.listdir(args.copy_to): existing_path = os.path.join(args.copy_to, existing_fn) try: if os.path.isfile(existing_path): os.remove(existing_path) except Exception: pass for r in results: try: dst = os.path.join(args.copy_to, os.path.basename(r['path'])) with open(r['path'], 'rb') as srcf, open(dst, 'wb') as dstf: dstf.write(srcf.read()) except Exception: pass if __name__ == '__main__': raise SystemExit(main())