.
This commit is contained in:
Ross
@@ -0,0 +1,217 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user