806 lines
30 KiB
Python
806 lines
30 KiB
Python
#!/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 clear_doc_cache():
|
|
global _docid_map
|
|
_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]]:
|
|
clear_doc_cache()
|
|
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),
|
|
'is_cached': True
|
|
})
|
|
|
|
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),
|
|
'is_cached': True
|
|
})
|
|
seen_paths.add(path)
|
|
else:
|
|
expanded_results.append({
|
|
'path': f"missing_{docid}",
|
|
'title': link['title'],
|
|
'docid': docid,
|
|
'breadcrumbs': [],
|
|
'authors': [],
|
|
'pageKeywords': [],
|
|
'reasons': [f'Linked (Anatomy of {parent_title})'],
|
|
'snippet': '',
|
|
'linked_info': None,
|
|
'is_cached': False
|
|
})
|
|
seen_paths.add(f"missing_{docid}")
|
|
seen_docids.add(docid)
|
|
|
|
# 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),
|
|
'is_cached': True
|
|
})
|
|
seen_paths.add(path)
|
|
else:
|
|
expanded_results.append({
|
|
'path': f"missing_{docid}",
|
|
'title': link['title'],
|
|
'docid': docid,
|
|
'breadcrumbs': [],
|
|
'authors': [],
|
|
'pageKeywords': [],
|
|
'reasons': [f'Linked (Diff Diag of {parent_title})'],
|
|
'snippet': '',
|
|
'linked_info': None,
|
|
'is_cached': False
|
|
})
|
|
seen_paths.add(f"missing_{docid}")
|
|
seen_docids.add(docid)
|
|
|
|
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 ''])
|
|
elif fmt == 'combined_json':
|
|
combined_docs = []
|
|
for r in results:
|
|
path = r.get('path')
|
|
title = r.get('title', 'Untitled')
|
|
docid = r.get('docid', '')
|
|
|
|
content = ""
|
|
if path and os.path.exists(path):
|
|
try:
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
except Exception as e:
|
|
content = f"Error reading file {path}: {e}"
|
|
else:
|
|
content = f"File not found: {path}"
|
|
|
|
# Format using XML tags as a clear separator that LLMs understand
|
|
doc_str = f'<article id="{docid}" title="{title}">\n{content.strip()}\n</article>'
|
|
combined_docs.append(doc_str)
|
|
|
|
combined_document = "\n\n".join(combined_docs)
|
|
|
|
output_data = {
|
|
"combined_document": combined_document,
|
|
"articles": [
|
|
{
|
|
"title": r.get('title'),
|
|
"docid": r.get('docid'),
|
|
"path": r.get('path')
|
|
}
|
|
for r in results
|
|
]
|
|
}
|
|
|
|
with open(out_path, 'w', encoding='utf-8') as f:
|
|
json.dump(output_data, f, indent=2, ensure_ascii=False)
|
|
elif fmt == 'combined_md':
|
|
combined_docs = []
|
|
for r in results:
|
|
path = r.get('path')
|
|
title = r.get('title', 'Untitled')
|
|
docid = r.get('docid', '')
|
|
|
|
content = ""
|
|
if path and os.path.exists(path):
|
|
try:
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
except Exception as e:
|
|
content = f"Error reading file {path}: {e}"
|
|
else:
|
|
content = f"File not found: {path}"
|
|
|
|
# Format using XML tags as a clear separator that LLMs understand
|
|
doc_str = f'<article id="{docid}" title="{title}">\n{content.strip()}\n</article>'
|
|
combined_docs.append(doc_str)
|
|
|
|
combined_document = "\n\n".join(combined_docs)
|
|
|
|
with open(out_path, 'w', encoding='utf-8') as f:
|
|
f.write(combined_document)
|
|
|
|
|
|
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', 'combined_json', 'combined_md'], 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())
|