Ross
218 lines
7.9 KiB
Python
218 lines
7.9 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
|
|
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
|
|
from typing import List, Dict, Any
|
|
|
|
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 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
|
|
|
|
|
|
def run_search(root: str, qkey: str, qval: str) -> List[Dict[str, Any]]:
|
|
out = []
|
|
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 = read_frontmatter(path)
|
|
if not fm:
|
|
continue
|
|
matched = False
|
|
if qkey == 'breadcrumbs':
|
|
# qval format: A>B>C
|
|
query_parts = [p.strip() for p in qval.split('>') if p.strip()]
|
|
bcs = fm.get('breadcrumbs') or []
|
|
# support both list-of-strings and list-of-dicts
|
|
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
|
|
matched = match_breadcrumbs(bcs_list, query_parts)
|
|
elif qkey == 'authors':
|
|
authors = fm.get('authors') or []
|
|
matched = match_author(authors, qval)
|
|
elif qkey == 'pageKeywords':
|
|
pk = fm.get('pageKeywords') or ''
|
|
# pageKeywords often is a comma-separated string; split into tokens
|
|
tokens: List[str]
|
|
if isinstance(pk, list):
|
|
tokens = [str(x) for x in pk]
|
|
else:
|
|
tokens = [t.strip() for t in str(pk).split(',') if t.strip()]
|
|
matched = False
|
|
if '*' in qval:
|
|
for t in tokens:
|
|
if fnmatch.fnmatchcase(t.lower(), qval.lower()):
|
|
matched = True
|
|
break
|
|
else:
|
|
for t in tokens:
|
|
if qval.strip().lower() in t.lower():
|
|
matched = True
|
|
break
|
|
else:
|
|
v = fm.get(qkey)
|
|
if v is None:
|
|
matched = False
|
|
else:
|
|
# If v is a list, check each element for a match
|
|
if isinstance(v, list):
|
|
matched = False
|
|
for elem in v:
|
|
s = str(elem)
|
|
if '*' in qval:
|
|
if fnmatch.fnmatchcase(s.lower(), qval.lower()):
|
|
matched = True
|
|
break
|
|
else:
|
|
if qval.strip().lower() in s.lower():
|
|
matched = True
|
|
break
|
|
else:
|
|
s = str(v)
|
|
if '*' in qval:
|
|
matched = fnmatch.fnmatchcase(s.lower(), qval.lower())
|
|
else:
|
|
matched = qval.strip().lower() in s.lower()
|
|
|
|
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'),
|
|
})
|
|
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 in the form key:value where key is breadcrumbs/authors/pageKeywords/...')
|
|
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')
|
|
args = p.parse_args(argv)
|
|
|
|
if ':' not in args.query:
|
|
print('Query must be key:value')
|
|
return 2
|
|
key, val = args.query.split(':', 1)
|
|
key = key.strip()
|
|
val = val.strip()
|
|
|
|
results = run_search(args.root, key, val)
|
|
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)
|
|
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())
|