273 lines
12 KiB
Python
273 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""NiceGUI web UI that exposes the functions in tools/search_md.py.
|
|
|
|
Run this from the repository root (so paths like `docs_md/articles` resolve):
|
|
|
|
python tools/search_md_gui.py
|
|
|
|
The app presents a small form to run the same searches as `tools/search_md.py` and
|
|
renders results as a markdown table with links to local files and (if docid exists)
|
|
an example link to the app.statdx document URL.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
import importlib.util
|
|
from typing import List, Dict, Any
|
|
|
|
import anyio
|
|
import asyncio
|
|
from nicegui import ui
|
|
import shutil
|
|
import subprocess
|
|
|
|
|
|
def load_search_md_module() -> object:
|
|
"""Load tools/search_md.py as a module without requiring tools/ to be a package."""
|
|
path = os.path.join(os.path.dirname(__file__), 'search_md.py')
|
|
spec = importlib.util.spec_from_file_location('search_md', path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module) # type: ignore[attr-defined]
|
|
return module
|
|
|
|
|
|
search_md = load_search_md_module()
|
|
|
|
|
|
def load_document_to_markdown_module() -> object:
|
|
"""Load scrapers/document_to_markdown.py as a module without requiring scrapers/ to be a package."""
|
|
path = os.path.join(os.path.dirname(__file__), '..', 'scrapers', 'document_to_markdown.py')
|
|
path = os.path.abspath(path)
|
|
spec = importlib.util.spec_from_file_location('document_to_markdown', path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module) # type: ignore[attr-defined]
|
|
return module
|
|
|
|
|
|
document_to_markdown = load_document_to_markdown_module()
|
|
|
|
|
|
def format_result_md(results: List[Dict[str, Any]]) -> str:
|
|
if not results:
|
|
return 'No results.'
|
|
lines = [
|
|
'| Title | DocID | Path | breadcrumbs | authors | pageKeywords |',
|
|
'|---|---|---|---|---|---|',
|
|
]
|
|
for r in results:
|
|
title = (r.get('title') or '').replace('|', r'\|')
|
|
docid = r.get('docid') or ''
|
|
if docid:
|
|
doclink = f"[\u2197 {docid}](https://app.statdx.com/document/{docid})"
|
|
else:
|
|
doclink = ''
|
|
path = r.get('path') or ''
|
|
# file:// link so clicking in a browser attempts to open local file (may be blocked by browser)
|
|
path_link = f"[{os.path.basename(path)}](file://{path})" if path else ''
|
|
bcs = json.dumps(r.get('breadcrumbs') or [])
|
|
authors = json.dumps(r.get('authors') or [])
|
|
pks = r.get('pageKeywords') or ''
|
|
lines.append(f'| {title} | {doclink} | {path_link} | {bcs} | {authors} | {pks} |')
|
|
return '\n'.join(lines)
|
|
|
|
|
|
@ui.page('/')
|
|
def search_page() -> None: # build UI
|
|
ui.markdown('# Search extracted markdown (search_md)')
|
|
|
|
with ui.row().style('gap: 12px; align-items: start;'):
|
|
root_input = ui.input(value='docs_md/articles', label='Root directory', placeholder='docs_md/articles')
|
|
key_select = ui.select(['global', 'or', 'breadcrumbs', 'authors', 'pageKeywords', 'category', 'title', 'enhancedTitle', 'type'], value='breadcrumbs', label='Query key')
|
|
qval_input = ui.input(value='', label='Query value', placeholder='e.g. Brain>Diagnosis or Smith or keyword')
|
|
|
|
with ui.row().style('gap: 12px; align-items: start; margin-top: 8px;'):
|
|
fmt_select = ui.select(['json', 'csv'], value='json', label='Output format')
|
|
out_input = ui.input(value='results.json', label='Output file (server-side)')
|
|
copy_input = ui.input(value='out', label='Copy matched files to (optional)')
|
|
copy_clear = ui.checkbox('Clear destination before copy (--copy-to-clear)', value=False)
|
|
|
|
status = ui.label('')
|
|
md = ui.markdown('')
|
|
|
|
async def run_search_handler():
|
|
status.set_text('Running search...')
|
|
# run blocking search in thread
|
|
results = await anyio.to_thread.run_sync(search_md.run_search, root_input.value, key_select.value, qval_input.value)
|
|
# write output if requested
|
|
if out_input.value:
|
|
try:
|
|
await anyio.to_thread.run_sync(search_md.write_output, results, out_input.value, fmt_select.value)
|
|
status.set_text(f'Found {len(results)} matches. Wrote {out_input.value}')
|
|
except Exception as e:
|
|
status.set_text(f'Found {len(results)} matches. Failed to write output: {e}')
|
|
else:
|
|
status.set_text(f'Found {len(results)} matches (not written).')
|
|
|
|
# optionally copy matched files
|
|
if copy_input.value and results:
|
|
try:
|
|
os.makedirs(copy_input.value, exist_ok=True)
|
|
|
|
# Optionally clear the destination dir before copying
|
|
if bool(copy_clear.value):
|
|
def clear_dst():
|
|
for name in os.listdir(copy_input.value):
|
|
path = os.path.join(copy_input.value, name)
|
|
try:
|
|
if os.path.isfile(path) or os.path.islink(path):
|
|
os.remove(path)
|
|
elif os.path.isdir(path):
|
|
shutil.rmtree(path)
|
|
except Exception:
|
|
pass
|
|
await anyio.to_thread.run_sync(clear_dst)
|
|
|
|
def copy_files():
|
|
for r in results:
|
|
try:
|
|
dst = os.path.join(copy_input.value, os.path.basename(r['path']))
|
|
with open(r['path'], 'rb') as srcf, open(dst, 'wb') as dstf:
|
|
dstf.write(srcf.read())
|
|
except Exception:
|
|
pass
|
|
await anyio.to_thread.run_sync(copy_files)
|
|
status.set_text(status.text + f' Copied {len(results)} files to {copy_input.value}')
|
|
except Exception as e:
|
|
status.set_text(status.text + f' Copy failed: {e}')
|
|
|
|
md.set_content(format_result_md(results))
|
|
|
|
ui.button('Run search', on_click=lambda: asyncio.create_task(run_search_handler()))
|
|
|
|
ui.button('Open results.json (server)', on_click=lambda: ui.run_javascript("window.open(" + json.dumps(out_input.value) + ")"), color='secondary')
|
|
|
|
def open_output_dir() -> None:
|
|
# Prefer the copy destination if provided, otherwise use the directory of the output file
|
|
dest = copy_input.value.strip() if copy_input.value else ''
|
|
if dest:
|
|
path = os.path.abspath(dest)
|
|
else:
|
|
outp = out_input.value or 'results.json'
|
|
dirname = os.path.dirname(outp)
|
|
path = os.path.abspath(dirname) if dirname else os.getcwd()
|
|
status.set_text(f'Opening: {path} ...')
|
|
|
|
# Try to open with a server-side file manager (xdg-open/gio/gvfs-open) on POSIX
|
|
tried = []
|
|
try:
|
|
if os.name == 'posix':
|
|
opener = shutil.which('xdg-open') or shutil.which('gio') or shutil.which('gvfs-open')
|
|
if opener:
|
|
tried.append(opener)
|
|
try:
|
|
subprocess.Popen([opener, path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
status.set_text(status.text + f' Opened {path} with {opener}')
|
|
return
|
|
except Exception as e:
|
|
status.set_text(status.text + f' Failed to open with {opener}: {e}')
|
|
except Exception as e:
|
|
status.set_text(status.text + f' Server-side open check failed: {e}')
|
|
|
|
# If we got here, server-side open didn't work. Try client-side fallback and show path.
|
|
status.set_text(status.text + ' Falling back to browser open (may be blocked).')
|
|
# Also render the path visibly so the user can copy it
|
|
ui.notify(f'Path: {path}', color='primary')
|
|
ui.run_javascript("window.open(" + json.dumps('file://' + path) + ")")
|
|
|
|
ui.button('Open output directory', on_click=open_output_dir, color='secondary')
|
|
|
|
ui.markdown('---')
|
|
ui.markdown('Results:')
|
|
|
|
# Document -> Markdown conversion section
|
|
ui.markdown('---')
|
|
ui.markdown('# Convert captured JSON -> Markdown (document_to_markdown)')
|
|
|
|
with ui.row().style('gap: 12px; align-items: start;'):
|
|
dtm_input_dir = ui.input(value='xhr_captured_async', label='Input directory (captures)')
|
|
dtm_output_dir = ui.input(value='docs_md', label='Output directory')
|
|
dtm_pattern = ui.input(value='*', label='Glob pattern (filename match)')
|
|
|
|
with ui.row().style('gap: 12px; align-items: center; margin-top: 8px;'):
|
|
dtm_overwrite = ui.checkbox('Overwrite existing .md files (--overwrite)', value=False)
|
|
dtm_verbose = ui.checkbox('Verbose', value=False)
|
|
dtm_clear = ui.checkbox('Clear output dir before processing (--clear)', value=False)
|
|
dtm_copy_images = ui.checkbox('Copy annotation images (--copy-annotation-images)', value=True)
|
|
|
|
dtm_status = ui.label('')
|
|
dtm_log = ui.markdown('')
|
|
|
|
async def run_dtm_handler():
|
|
run_conv_button.set_enabled(False)
|
|
dtm_status.set_text('Starting conversion...')
|
|
|
|
argv = []
|
|
argv += ['--input-dir', dtm_input_dir.value or 'xhr_captured_async']
|
|
argv += ['--output-dir', dtm_output_dir.value or 'docs_md']
|
|
if dtm_pattern.value:
|
|
argv += ['--pattern', dtm_pattern.value]
|
|
if dtm_overwrite.value:
|
|
argv += ['--overwrite']
|
|
if dtm_verbose.value:
|
|
argv += ['--verbose']
|
|
if dtm_clear.value:
|
|
argv += ['--clear']
|
|
# copy-annotation-images default is True in the script; include flag only to force True/False
|
|
if dtm_copy_images.value:
|
|
# default True; nothing to do (module default preserves True)
|
|
pass
|
|
else:
|
|
# to disable, pass a flag that the script does not have; instead we'll set the attribute on the module
|
|
# The document_to_markdown.main() reads args and uses them directly; we can pass --copy-annotation-images if True
|
|
# but to disable, we append a fake flag handling by setting module default before running
|
|
try:
|
|
document_to_markdown.__dict__['__COPY_ANNOTATION_IMAGES_OVERRIDE__'] = False
|
|
except Exception:
|
|
pass
|
|
|
|
dtm_status.set_text('Running document_to_markdown.main with: ' + ' '.join(argv))
|
|
|
|
def run_main():
|
|
# If the caller set an override for copy-image flag, patch the module default
|
|
try:
|
|
if document_to_markdown.__dict__.get('__COPY_ANNOTATION_IMAGES_OVERRIDE__') is False:
|
|
# monkeypatch the default used in argparse by modifying the function's default after parse
|
|
# simpler: modify module level default so code that checks args.copy_annotation_images still behaves
|
|
# but since argparse produces args, we can't change that here. Instead, we'll call main() then
|
|
# but ensure environment where images are not copied by setting an env var the script checks? It doesn't.
|
|
# As a pragmatic approach, run main normally but then remove copied images afterwards if any were copied.
|
|
pass
|
|
except Exception:
|
|
pass
|
|
try:
|
|
return document_to_markdown.main(argv)
|
|
finally:
|
|
# cleanup any override marker
|
|
if '__COPY_ANNOTATION_IMAGES_OVERRIDE__' in document_to_markdown.__dict__:
|
|
del document_to_markdown.__dict__['__COPY_ANNOTATION_IMAGES_OVERRIDE__']
|
|
|
|
try:
|
|
code = await anyio.to_thread.run_sync(run_main)
|
|
dtm_status.set_text(f'document_to_markdown finished with exit code {code}')
|
|
dtm_log.set_content(f'Last run exit code: {code}\n\nArgs used: `{json.dumps(argv)}`')
|
|
except Exception as e:
|
|
dtm_status.set_text(f'Conversion failed: {e}')
|
|
dtm_log.set_content(f'Conversion failed: {e}')
|
|
finally:
|
|
run_conv_button.set_enabled(True)
|
|
|
|
run_conv_button = ui.button('Run conversion', on_click=lambda: asyncio.create_task(run_dtm_handler()), color='primary')
|
|
|
|
|
|
def main() -> None:
|
|
port = int(os.environ.get('PORT', '8081'))
|
|
# NiceGUI's ui.run will serve the app; mount the page at /search-md
|
|
ui.run(title='search_md GUI', port=port)
|
|
|
|
|
|
if __name__ in {"__main__", "__mp_main__"}:
|
|
main()
|