| """
|
| FilterLM — Gradio UI for PDF criteria search (no LLMs).
|
|
|
| Uses the same quote/text matching as pdf_criteria_search_gui_v2.py. Background jobs,
|
| job list, Refresh, and auto-refresh mirror the execution pattern in psyllm.py.
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import datetime
|
| import html as html_module
|
| import os
|
| import queue
|
| import tempfile
|
| import threading
|
| import time
|
| import uuid
|
| from typing import Any, Dict, List, Tuple
|
|
|
| import gradio as gr
|
|
|
| from search_keywords import CATEGORY_HEADINGS, CATEGORY_STYLES, KEY_TO_CATEGORY
|
| from pdf_criteria_search_gui_v2 import (
|
| QuoteSearchEngine,
|
| best_highlight_phrase_for_group,
|
| export_docx,
|
| parse_terms_text,
|
| phrases_for_group_key,
|
| terms_csv_path,
|
| ensure_terms_csv,
|
| write_terms_to_csv,
|
| load_terms_from_csv,
|
| get_pdf_title_and_first_author,
|
| )
|
|
|
|
|
|
|
|
|
|
|
| jobs: Dict[str, Dict[str, Any]] = {}
|
| results_queue: queue.Queue = queue.Queue()
|
| last_job_id: str | None = None
|
|
|
|
|
| def debug_print(message: str) -> None:
|
| print(f"[{datetime.datetime.now().isoformat()}] {message}", flush=True)
|
|
|
|
|
| def flatten_upload_inputs(files: Any) -> List[Any]:
|
| """Gradio may return a single path, a list, nested lists, or a numpy array."""
|
| if files is None:
|
| return []
|
| try:
|
| import numpy as np
|
|
|
| if isinstance(files, np.ndarray):
|
| files = files.tolist()
|
| except ImportError:
|
| pass
|
| if isinstance(files, (list, tuple)):
|
| acc: List[Any] = []
|
| for item in files:
|
| acc.extend(flatten_upload_inputs(item))
|
| return acc
|
| return [files]
|
|
|
|
|
| def _upload_item_to_path(item: Any) -> str | None:
|
| if item is None:
|
| return None
|
| path: str | None = None
|
| if isinstance(item, str):
|
| path = item
|
| elif isinstance(item, dict):
|
| path = item.get("path") or item.get("name")
|
| elif hasattr(item, "name"):
|
| n = getattr(item, "name")
|
| if isinstance(n, str):
|
| path = n
|
| if path is None:
|
| try:
|
| cand = os.fspath(item)
|
| if isinstance(cand, str):
|
| path = cand
|
| except Exception:
|
| path = None
|
| if path and isinstance(path, str) and os.path.isfile(path) and path.lower().endswith(".pdf"):
|
| return os.path.abspath(path)
|
| return None
|
|
|
|
|
| def normalize_upload_files(files: Any) -> List[str]:
|
| """Resolve every uploaded PDF path (supports multi-select drag-and-drop)."""
|
| out: List[str] = []
|
| for item in flatten_upload_inputs(files):
|
| p = _upload_item_to_path(item)
|
| if p:
|
| out.append(p)
|
| seen: set[str] = set()
|
| uniq: List[str] = []
|
| for p in out:
|
| if p not in seen:
|
| seen.add(p)
|
| uniq.append(p)
|
| return uniq
|
|
|
|
|
| def terms_keys_html(raw: str) -> str:
|
| """Colored HTML preview of keys + category bands (same grouping as Tk GUI)."""
|
| keys = parse_terms_text(raw)
|
| if not keys:
|
| return '<p class="muted"><em>No keys yet — add one group id per line (# = comment).</em></p>'
|
| parts: List[str] = ['<div class="terms-colored">']
|
| last_cat: str | None = None
|
| for key in keys:
|
| cat = KEY_TO_CATEGORY.get(key, "unknown")
|
| if cat != last_cat:
|
| if last_cat is not None:
|
| parts.append('<div style="height:8px"></div>')
|
| title = CATEGORY_HEADINGS.get(cat, cat)
|
| parts.append(
|
| f'<div class="cat-head">{html_module.escape(title)}</div>'
|
| )
|
| last_cat = cat
|
| bg, fg = CATEGORY_STYLES.get(cat, ("#EEEEEE", "#333333"))
|
| parts.append(
|
| f'<span class="term-chip" style="background:{bg};color:{fg}">{html_module.escape(key)}</span><br/>'
|
| )
|
| parts.append("</div>")
|
| return "".join(parts)
|
|
|
|
|
| def loaded_pdfs_html(files: Any) -> str:
|
| """Immediate listing after drag-and-drop or file picker."""
|
| paths = normalize_upload_files(files)
|
| if not paths:
|
| return (
|
| '<p class="muted"><em>No PDFs loaded — drag files here or use a folder path below.</em></p>'
|
| )
|
| rows = []
|
| for i, p in enumerate(paths, start=1):
|
| base = html_module.escape(os.path.basename(p))
|
| full = html_module.escape(p)
|
| rows.append(f"<tr><td>{i}</td><td>{base}</td><td class=\"path-cell\">{full}</td></tr>")
|
| body = "".join(rows)
|
| return (
|
| f'<p><strong>{len(paths)}</strong> PDF(s) ready to search.</p>'
|
| f'<table class="pdf-loaded-table"><thead><tr><th>#</th><th>File</th><th>Path</th></tr></thead>'
|
| f"<tbody>{body}</tbody></table>"
|
| )
|
|
|
|
|
| def excerpt_html_highlighted(group_key: str, excerpt: str) -> str:
|
| """Green `<span>` around the same substring as Tk preview / DOCX highlight."""
|
| ex = excerpt or ""
|
| ph = best_highlight_phrase_for_group(group_key, ex)
|
| if not ph or not ex:
|
| return html_module.escape(ex)
|
| low = ex.lower()
|
| idx = low.find(ph.lower())
|
| if idx < 0:
|
| return html_module.escape(ex)
|
| return (
|
| html_module.escape(ex[:idx])
|
| + '<span class="kw-hit">'
|
| + html_module.escape(ex[idx : idx + len(ph)])
|
| + "</span>"
|
| + html_module.escape(ex[idx + len(ph) :])
|
| )
|
|
|
|
|
| def drain_results_queue() -> None:
|
| try:
|
| while not results_queue.empty():
|
| completed_id, result = results_queue.get_nowait()
|
| if completed_id in jobs:
|
| jobs[completed_id]["status"] = "completed"
|
| jobs[completed_id]["result"] = result
|
| jobs[completed_id]["end_time"] = time.time()
|
| debug_print(f"Job {completed_id} completed and stored")
|
| except queue.Empty:
|
| pass
|
|
|
|
|
| def process_in_background(job_id: str, function, args: Tuple[Any, ...]) -> None:
|
| try:
|
| debug_print(f"Processing job {job_id} in background")
|
| result = function(*args)
|
| results_queue.put((job_id, result))
|
| debug_print(f"Job {job_id} completed and queued")
|
| except Exception as e:
|
| debug_print(f"Error in background job {job_id}: {e}")
|
| results_queue.put((job_id, ("error", str(e))))
|
|
|
|
|
| def format_search_report(
|
| results_by_term: Dict[str, List[Dict[str, Any]]],
|
| criteria_order: List[str],
|
| grouped_pages: bool,
|
| pdf_title_cache: Dict[str, Tuple[str, str]],
|
| ) -> str:
|
| lines: List[str] = []
|
| for term in criteria_order:
|
| blocks = results_by_term.get(term, [])
|
| lines.append(f"## {term}")
|
| if not blocks:
|
| lines.append("(no matches)")
|
| lines.append("")
|
| continue
|
| sorted_blocks = sorted(blocks, key=lambda b: os.path.basename(b["pdf_path"]).lower())
|
| for block in sorted_blocks:
|
| fp = block["pdf_path"]
|
| base = os.path.basename(fp)
|
| if fp not in pdf_title_cache:
|
| pdf_title_cache[fp] = get_pdf_title_and_first_author(fp)
|
| title, fa = pdf_title_cache[fp]
|
| lines.append(f"- **{base}** — {title} — {fa}")
|
| occs = block.get("occurrences", [])
|
| if grouped_pages:
|
| pages = ", ".join(str(o["page"]) for o in occs)
|
| apx = "yes" if any(o.get("approximate") for o in occs) else "no"
|
| lines.append(f" - Pages: {pages} | approximate: {apx}")
|
| for o in occs:
|
| ex = (o.get("excerpt") or "").strip()
|
| if ex:
|
| lines.append(f" p.{o['page']}: {ex}")
|
| else:
|
| for o in occs:
|
| apx = " (approx.)" if o.get("approximate") else ""
|
| ex = (o.get("excerpt") or "").strip()
|
| lines.append(f" - p.{o['page']}{apx}: {ex}")
|
| lines.append("")
|
| lines.append("")
|
| return "\n".join(lines).strip()
|
|
|
|
|
| def format_search_report_html(
|
| results_by_term: Dict[str, List[Dict[str, Any]]],
|
| criteria_order: List[str],
|
| grouped_pages: bool,
|
| pdf_title_cache: Dict[str, Tuple[str, str]],
|
| ) -> str:
|
| """Same content as plain report with category-colored headings."""
|
| chunks: List[str] = ['<div class="results-report">']
|
| for term in criteria_order:
|
| blocks = results_by_term.get(term, [])
|
| cat = KEY_TO_CATEGORY.get(term, "unknown")
|
| bg, fg = CATEGORY_STYLES.get(cat, ("#EEEEEE", "#333333"))
|
| chunks.append(
|
| f'<h3 class="result-term-h3" style="background:{bg};color:{fg}">{html_module.escape(term)}</h3>'
|
| )
|
| if not blocks:
|
| chunks.append('<p class="muted">(no matches)</p>')
|
| continue
|
| sorted_blocks = sorted(blocks, key=lambda b: os.path.basename(b["pdf_path"]).lower())
|
| for block in sorted_blocks:
|
| fp = block["pdf_path"]
|
| base = os.path.basename(fp)
|
| if fp not in pdf_title_cache:
|
| pdf_title_cache[fp] = get_pdf_title_and_first_author(fp)
|
| title, fa = pdf_title_cache[fp]
|
| chunks.append(
|
| "<p><strong>%s</strong> — %s — %s</p>"
|
| % (
|
| html_module.escape(base),
|
| html_module.escape(title),
|
| html_module.escape(fa),
|
| )
|
| )
|
| occs = block.get("occurrences", [])
|
| if grouped_pages:
|
| pages = ", ".join(str(o["page"]) for o in occs)
|
| apx = "yes" if any(o.get("approximate") for o in occs) else "no"
|
| chunks.append(
|
| "<p class=\"indent\">Pages: %s | approximate: %s</p>"
|
| % (html_module.escape(pages), html_module.escape(apx))
|
| )
|
| for o in occs:
|
| ex = (o.get("excerpt") or "").strip()
|
| if ex:
|
| chunks.append(
|
| '<p class="excerpt">p.%s: %s</p>'
|
| % (
|
| html_module.escape(str(o["page"])),
|
| excerpt_html_highlighted(term, ex),
|
| )
|
| )
|
| else:
|
| for o in occs:
|
| apx = " (approx.)" if o.get("approximate") else ""
|
| ex = (o.get("excerpt") or "").strip()
|
| chunks.append(
|
| '<p class="excerpt">p.%s%s: %s</p>'
|
| % (
|
| html_module.escape(str(o["page"])),
|
| html_module.escape(apx),
|
| excerpt_html_highlighted(term, ex),
|
| )
|
| )
|
| chunks.append("")
|
| chunks.append("")
|
| chunks.append("</div>")
|
| return "".join(chunks)
|
|
|
|
|
| def run_pdf_search_job(
|
| job_id: str,
|
| pdf_paths: List[str],
|
| terms: List[str],
|
| grouped_pages: bool,
|
| folder_label: str,
|
| ) -> Tuple[Any, ...]:
|
| """Heavy work; runs in a worker thread."""
|
| engine = QuoteSearchEngine()
|
| results: Dict[str, List[Dict[str, Any]]] = {t: [] for t in terms}
|
| title_cache: Dict[str, Tuple[str, str]] = {}
|
|
|
| for term in terms:
|
| phrases = phrases_for_group_key(term)
|
| for fp in pdf_paths:
|
| occ = engine.search_group_in_pdf(fp, phrases)
|
| if occ:
|
| results[term].append({"pdf_path": fp, "occurrences": occ})
|
|
|
| report_plain = format_search_report(results, terms, grouped_pages, title_cache)
|
| report_html = format_search_report_html(results, terms, grouped_pages, title_cache)
|
| stats = (
|
| f"PDF files scanned: {len(pdf_paths)}\n"
|
| f"Group keys: {len(terms)}\n"
|
| f"Groups with ≥1 hit: {sum(1 for t in terms if results[t])}"
|
| )
|
|
|
| docx_path = ""
|
| try:
|
| import docx
|
|
|
| safe = "".join(c if c.isalnum() else "_" for c in job_id[:12])
|
| docx_path = os.path.join(
|
| tempfile.gettempdir(), f"filterlm_criteria_{safe}_{int(time.time())}.docx"
|
| )
|
| export_docx(docx_path, folder_label, results, terms)
|
| except Exception as ex:
|
| debug_print(f"DOCX export skipped: {ex}")
|
|
|
| return ("ok", report_plain, report_html, docx_path, stats, folder_label)
|
|
|
|
|
| def submit_pdf_search(
|
| folder: str,
|
| files: Any,
|
| terms_raw: str,
|
| grouped_pages: bool,
|
| ) -> Tuple[Any, ...]:
|
| """Enqueue async PDF search; return UI strings + job id."""
|
| global last_job_id
|
|
|
| uploaded = normalize_upload_files(files)
|
| folder = (folder or "").strip()
|
|
|
| if uploaded:
|
| pdf_paths = uploaded
|
| folder_label = f"{len(uploaded)} uploaded PDF(s)"
|
| elif folder and os.path.isdir(folder):
|
| names = sorted(f for f in os.listdir(folder) if f.lower().endswith(".pdf"))
|
| pdf_paths = [os.path.join(folder, f) for f in names]
|
| folder_label = folder
|
| else:
|
| return (
|
| "Drag PDF(s) into the upload area **or** enter a folder path that contains `.pdf` files.",
|
| "",
|
| "",
|
| get_job_list(),
|
| )
|
|
|
| if not pdf_paths:
|
| return (
|
| "No `.pdf` files found — upload PDFs or pick a folder that contains PDFs.",
|
| "",
|
| "",
|
| get_job_list(),
|
| )
|
|
|
| terms = parse_terms_text(terms_raw)
|
| if not terms:
|
| return (
|
| "Add at least one group key (one per line).",
|
| "",
|
| "",
|
| get_job_list(),
|
| )
|
|
|
| job_id = str(uuid.uuid4())
|
| threading.Thread(
|
| target=process_in_background,
|
| args=(
|
| job_id,
|
| run_pdf_search_job,
|
| (job_id, pdf_paths, terms, grouped_pages, folder_label),
|
| ),
|
| daemon=True,
|
| ).start()
|
|
|
| preview = folder_label if len(folder_label) <= 80 else folder_label[:77] + "..."
|
| jobs[job_id] = {
|
| "status": "processing",
|
| "type": "pdf_search",
|
| "start_time": time.time(),
|
| "query": f"{preview} | {len(terms)} term(s)",
|
| "folder": folder_label,
|
| "pdf_paths": list(pdf_paths),
|
| "terms_count": len(terms),
|
| "grouped": grouped_pages,
|
| }
|
| last_job_id = job_id
|
|
|
| msg = (
|
| f"Search submitted in the background.\n\n"
|
| f"**Job ID:** `{job_id}`\n\n"
|
| f"Open **Check Job Status** (sub-tab) and use Refresh or Enable Auto Refresh."
|
| )
|
| summary = (
|
| f"Source: {folder_label}\n"
|
| f"PDFs: {len(pdf_paths)}\n"
|
| f"Terms: {len(terms)}\n"
|
| f"GROUP pages: {grouped_pages}"
|
| )
|
| return msg, job_id, summary, get_job_list()
|
|
|
|
|
| def get_job_list() -> str:
|
| job_list_md = "### Submitted Jobs\n\n"
|
| if not jobs:
|
| return "No jobs yet. Run a PDF criteria search to create jobs."
|
|
|
| sorted_jobs = sorted(
|
| jobs.items(),
|
| key=lambda x: x[1].get("start_time", 0),
|
| reverse=True,
|
| )
|
|
|
| for job_id, job_info in sorted_jobs:
|
| status = job_info.get("status", "unknown")
|
| start_time = job_info.get("start_time", 0)
|
| time_str = datetime.datetime.fromtimestamp(start_time).strftime("%Y-%m-%d %H:%M:%S")
|
| query = job_info.get("query", "")
|
| query_preview = query[:56] + "..." if query and len(query) > 56 else query or "N/A"
|
|
|
| if status == "processing":
|
| status_formatted = f"<span style='color: red'>⏳ {status}</span>"
|
| elif status == "completed":
|
| status_formatted = f"<span style='color: green'>✅ {status}</span>"
|
| else:
|
| status_formatted = f"<span style='color: orange'>❓ {status}</span>"
|
|
|
| job_list_md += (
|
| f"- [{job_id}](javascript:void) — {time_str} — {status_formatted} — {query_preview}\n"
|
| )
|
|
|
| return job_list_md
|
|
|
|
|
| def job_selected(job_id: str) -> Tuple[str, str]:
|
| jid = (job_id or "").strip()
|
| if jid in jobs:
|
| j = jobs[jid]
|
| lines: List[str] = []
|
| paths = j.get("pdf_paths") or []
|
| if paths:
|
| lines.append(f"PDFs: {len(paths)} file(s)")
|
| lines.append(f"Source: {j.get('folder', '')}")
|
| lines.append(f"Terms: {j.get('terms_count', '')}")
|
| return jid, "\n".join(lines)
|
| return jid, "Job not found"
|
|
|
|
|
| def refresh_job_list() -> str:
|
| drain_results_queue()
|
| return get_job_list()
|
|
|
|
|
| def generate_job_status_report(job_id: str, job: Dict[str, Any]) -> str:
|
| lines = [
|
| "## Job status",
|
| "",
|
| f"**Job ID:** `{job_id}`",
|
| f"**Type:** PDF criteria search",
|
| f"**Status:** {job.get('status', '?')}",
|
| f"**Summary:** {job.get('query', '')}",
|
| "",
|
| ]
|
| if job.get("status") == "processing":
|
| elapsed = time.time() - job.get("start_time", 0)
|
| lines.append(f"Elapsed: **{elapsed:.1f}s**")
|
| elif job.get("status") == "completed":
|
| st = job.get("start_time", 0)
|
| en = job.get("end_time", st)
|
| lines.append(f"Duration: **{en - st:.2f}s**")
|
| return "\n".join(lines)
|
|
|
|
|
| def _stats_as_md(stats: str) -> str:
|
| if not (stats or "").strip():
|
| return ""
|
| return "```\n" + stats.strip() + "\n```"
|
|
|
|
|
| def check_job_status(job_id: str) -> Tuple[str, str, str, str | None, str]:
|
| """status_md, results_html, stats_md, docx_path, job_summary."""
|
| drain_results_queue()
|
|
|
| waiting_html = '<p class="muted"><em>Searching PDFs… colored results appear here when finished.</em></p>'
|
|
|
| if not (job_id or "").strip():
|
| return ("Enter a job ID or click one in the list.", "", "", None, "")
|
|
|
| jid = job_id.strip()
|
| if jid not in jobs:
|
| return (
|
| "Job not found. Refresh the list or copy the ID from the submission message.",
|
| "",
|
| "",
|
| None,
|
| "",
|
| )
|
|
|
| job = jobs[jid]
|
|
|
| paths = job.get("pdf_paths") or []
|
| summary = (
|
| f"Source: {job.get('folder', '')}\n"
|
| f"PDFs: {len(paths)}\n"
|
| f"Terms: {job.get('terms_count', '')}"
|
| )
|
|
|
| if job["status"] == "processing":
|
| elapsed = time.time() - job["start_time"]
|
| header = generate_job_status_report(jid, job)
|
| body = f"\n\nStill running… ({elapsed:.1f}s elapsed)."
|
| return header + body, waiting_html, "", None, summary
|
|
|
| if job["status"] == "completed":
|
| raw = job.get("result")
|
|
|
| if isinstance(raw, tuple) and raw and raw[0] == "error":
|
| err = raw[1] if len(raw) > 1 else "unknown error"
|
| err_html = f"<pre class=\"error-pre\">{html_module.escape(str(err))}</pre>"
|
| return (
|
| generate_job_status_report(jid, job) + f"\n\n**Error:** {err}",
|
| err_html,
|
| "",
|
| None,
|
| summary,
|
| )
|
|
|
| if isinstance(raw, tuple) and len(raw) >= 6 and raw[0] == "ok":
|
| _ok, _plain, report_html, docx_path, stats, _folder = raw[:6]
|
| header = generate_job_status_report(jid, job)
|
| dp = docx_path if docx_path and os.path.isfile(docx_path) else None
|
| return header, report_html, _stats_as_md(stats), dp, summary
|
|
|
| fallback = f"<pre class=\"muted\">{html_module.escape(str(raw))}</pre>"
|
| return generate_job_status_report(jid, job), fallback, "", None, summary
|
|
|
| return generate_job_status_report(jid, job), "", "", None, ""
|
|
|
|
|
| def cleanup_old_jobs() -> Tuple[str, str, str]:
|
| current_time = time.time()
|
| to_delete: List[str] = []
|
| for job_id, job in jobs.items():
|
| if job["status"] == "completed" and (current_time - job.get("end_time", 0)) > 86400:
|
| to_delete.append(job_id)
|
| elif job["status"] == "processing" and (current_time - job.get("start_time", 0)) > 172800:
|
| to_delete.append(job_id)
|
| for job_id in to_delete:
|
| del jobs[job_id]
|
| debug_print(f"Cleaned up {len(to_delete)} job(s); {len(jobs)} remaining.")
|
| return f"Removed {len(to_delete)} old job(s).", "", ""
|
|
|
|
|
| def periodic_update_fixed(is_checked: bool):
|
| """Refresh job list + last job status every ~2s when auto-refresh is on."""
|
| jl = refresh_job_list()
|
| if not is_checked:
|
| return jl, gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip()
|
| if last_job_id:
|
| return (jl,) + check_job_status(last_job_id)
|
| return (
|
| jl,
|
| "No job ID yet — submit a search first.",
|
| '<p class="muted">Submit a search first.</p>',
|
| "",
|
| None,
|
| "",
|
| )
|
|
|
|
|
| def save_terms_to_disk(raw: str) -> Tuple[str, str]:
|
| terms = parse_terms_text(raw)
|
| preview = terms_keys_html(raw)
|
| if not terms:
|
| return "Nothing to save (empty terms).", preview
|
| path = terms_csv_path()
|
| ensure_terms_csv(path)
|
| try:
|
| write_terms_to_csv(path, terms)
|
| return f"Saved {len(terms)} keys to `{path}`.", preview
|
| except OSError as e:
|
| return f"Save failed: {e}", preview
|
|
|
|
|
| def load_terms_for_ui() -> str:
|
| path = terms_csv_path()
|
| ensure_terms_csv(path)
|
| keys = load_terms_from_csv(path)
|
| return "\n".join(keys)
|
|
|
|
|
|
|
|
|
|
|
|
|
| custom_css = """
|
| textarea { overflow-y: scroll !important; max-height: 220px; }
|
| .job-list-container a { cursor: pointer; }
|
| .muted { color: #616161; font-style: italic; }
|
| .cat-head {
|
| background: #eceff1;
|
| color: #37474f;
|
| font-weight: 600;
|
| padding: 6px 10px;
|
| margin-top: 10px;
|
| border-radius: 4px;
|
| font-size: 13px;
|
| }
|
| .term-chip {
|
| display: inline-block;
|
| padding: 3px 10px;
|
| margin: 3px 0;
|
| border-radius: 4px;
|
| font-family: Consolas, monospace;
|
| font-size: 13px;
|
| }
|
| .pdf-loaded-table {
|
| width: 100%;
|
| border-collapse: collapse;
|
| font-size: 13px;
|
| margin-top: 8px;
|
| }
|
| .pdf-loaded-table th, .pdf-loaded-table td {
|
| border: 1px solid #cfd8dc;
|
| padding: 6px 8px;
|
| text-align: left;
|
| }
|
| .pdf-loaded-table thead { background: #eceff1; }
|
| .path-cell { word-break: break-all; font-size: 12px; color: #424242; }
|
| .results-report {
|
| font-family: system-ui, Segoe UI, sans-serif;
|
| line-height: 1.45;
|
| font-size: 14px;
|
| }
|
| .result-term-h3 {
|
| padding: 8px 12px;
|
| margin: 14px 0 8px 0;
|
| border-radius: 4px;
|
| font-size: 1.05rem;
|
| }
|
| .results-report .indent { margin-left: 12px; color: #455a64; }
|
| .results-report .excerpt {
|
| margin: 4px 0 4px 16px;
|
| padding-left: 8px;
|
| border-left: 3px solid #b0bec5;
|
| }
|
| .kw-hit {
|
| background: #006400;
|
| color: #ffffff;
|
| padding: 0 3px;
|
| border-radius: 2px;
|
| }
|
| .error-pre {
|
| background: #ffebee;
|
| padding: 10px;
|
| border-radius: 4px;
|
| overflow-x: auto;
|
| }
|
| """
|
|
|
| JS_CLICK_JOB_ID = r"""
|
| document.addEventListener('DOMContentLoaded', function() {
|
| const jobListInterval = setInterval(() => {
|
| const jobLinks = document.querySelectorAll('.job-list-container a');
|
| if (jobLinks.length > 0) {
|
| jobLinks.forEach(link => {
|
| link.addEventListener('click', function(e) {
|
| e.preventDefault();
|
| const jobId = this.textContent.trim().split(/\s/)[0];
|
| const jobIdInput = document.querySelector('.job-id-input input');
|
| if (jobIdInput) {
|
| jobIdInput.value = jobId;
|
| jobIdInput.dispatchEvent(new Event('input', { bubbles: true }));
|
| }
|
| });
|
| });
|
| clearInterval(jobListInterval);
|
| }
|
| }, 500);
|
| });
|
| """
|
|
|
|
|
| with gr.Blocks(css=custom_css, js=JS_CLICK_JOB_ID) as app:
|
| gr.Markdown(
|
| """# DOX — parallel search using multiple keywords in multiple PDFs
|
|
|
| Searches in parallel for keywords (organized in groups) against multiple PDFs using the same normalization / fuzzy logic. No LLMs are used.
|
| Each line is a **group id**. Jobs run **in the background** — use **Refresh** / **Enable Auto Refresh** on the job list below.
|
| """
|
| )
|
|
|
| with gr.Tabs():
|
| with gr.TabItem("PDF Criteria Search"):
|
| submit_btn = gr.Button("Submit PDF search", variant="primary")
|
| submit_msg = gr.Textbox(label="Submission log", lines=4)
|
|
|
| gr.Markdown(
|
| "**PDF sources:** Drag **multiple** PDFs below (listed immediately), **or** enter a folder path. "
|
| "If both are set, **uploaded files are searched**. "
|
| "In the file dialog, use **Ctrl+click** / **Shift+click** to select several PDFs."
|
| )
|
| with gr.Row():
|
| with gr.Column(scale=1):
|
| pdf_upload = gr.File(
|
| label="PDF files (multi-select drag-and-drop or browse)",
|
| file_count="multiple",
|
| file_types=[".pdf"],
|
| )
|
| loaded_files_html = gr.HTML(value=loaded_pdfs_html(None))
|
| with gr.Column(scale=1):
|
| folder_input = gr.Textbox(
|
| label="Folder with PDFs (optional if uploading)",
|
| placeholder=r"C:\path\to\pdfs",
|
| lines=1,
|
| )
|
| terms_box = gr.Textbox(
|
| label="Group keys (one per line; # starts a comment)",
|
| value=load_terms_for_ui(),
|
| lines=14,
|
| )
|
| terms_html_preview = gr.HTML(
|
| label="Keys by category (colors)",
|
| value=terms_keys_html(load_terms_for_ui()),
|
| )
|
| with gr.Row():
|
| group_toggle = gr.Checkbox(label="GROUP page hits per PDF", value=True)
|
| save_csv_btn = gr.Button("Save keys to CSV")
|
|
|
| gr.Markdown("### Jobs & results")
|
| with gr.Row():
|
| with gr.Column(scale=1):
|
| job_list = gr.Markdown(
|
| value="No jobs yet",
|
| label="Job list (click an ID)",
|
| elem_classes=["job-list-container"],
|
| )
|
| refresh_button = gr.Button("Refresh Job List")
|
| auto_refresh_checkbox = gr.Checkbox(
|
| label="Enable Auto Refresh",
|
| value=False,
|
| )
|
| with gr.Column(scale=2):
|
| job_id_input = gr.Textbox(
|
| label="Job ID",
|
| placeholder="Paste ID or click from the list",
|
| lines=1,
|
| elem_classes=["job-id-input"],
|
| )
|
| job_summary = gr.Textbox(
|
| label="Job summary",
|
| lines=3,
|
| interactive=False,
|
| )
|
| check_button = gr.Button("Check Status")
|
| cleanup_button = gr.Button("Cleanup old jobs")
|
|
|
| status_header = gr.Markdown(label="Status")
|
| results_html = gr.HTML(label="Search results (category headings + green keyword span)")
|
| stats_box = gr.Markdown(label="Stats")
|
| docx_file = gr.File(label="Exported DOCX (when ready)", interactive=False)
|
|
|
| pdf_upload.change(loaded_pdfs_html, inputs=[pdf_upload], outputs=[loaded_files_html])
|
| terms_box.change(terms_keys_html, inputs=[terms_box], outputs=[terms_html_preview])
|
|
|
| save_csv_btn.click(save_terms_to_disk, inputs=[terms_box], outputs=[submit_msg, terms_html_preview])
|
|
|
| submit_btn.click(
|
| submit_pdf_search,
|
| inputs=[folder_input, pdf_upload, terms_box, group_toggle],
|
| outputs=[submit_msg, job_id_input, job_summary, job_list],
|
| )
|
|
|
| refresh_button.click(refresh_job_list, outputs=[job_list])
|
|
|
| check_button.click(
|
| check_job_status,
|
| inputs=[job_id_input],
|
| outputs=[
|
| status_header,
|
| results_html,
|
| stats_box,
|
| docx_file,
|
| job_summary,
|
| ],
|
| )
|
|
|
| job_id_input.change(job_selected, inputs=[job_id_input], outputs=[job_id_input, job_summary])
|
|
|
| cleanup_button.click(cleanup_old_jobs, outputs=[status_header, results_html, stats_box])
|
|
|
| auto_refresh_checkbox.change(
|
| fn=lambda chk: periodic_update_fixed(chk),
|
| inputs=[auto_refresh_checkbox],
|
| outputs=[
|
| job_list,
|
| status_header,
|
| results_html,
|
| stats_box,
|
| docx_file,
|
| job_summary,
|
| ],
|
| every=2,
|
| )
|
|
|
| with gr.TabItem("App Management"):
|
| gr.Markdown(
|
| "**Cleanup old jobs** removes completed jobs older than 24h "
|
| "and stale processing entries (48h). Running searches are unaffected unless stale."
|
| )
|
|
|
| def load_app():
|
| drain_results_queue()
|
| keys_txt = load_terms_for_ui()
|
| return get_job_list(), terms_keys_html(keys_txt), loaded_pdfs_html(None)
|
|
|
| app.load(fn=load_app, outputs=[job_list, terms_html_preview, loaded_files_html])
|
|
|
|
|
| if __name__ == "__main__":
|
| debug_print("Launching FilterLM (Gradio).")
|
| app.queue().launch(share=False)
|
|
|
