seancherry/hyperlink_exclusion_check
0
1import os2import io3import re4import time5from typing import List, Tuple, Dict, Generator6 7import gradio as gr8import pandas as pd9import requests10from bs4 import BeautifulSoup11from openpyxl import load_workbook12from openpyxl.styles import PatternFill13 14# =========================15# Admin auth (Basic Auth)16# =========================17ADMIN_USER = os.getenv("ADMIN_USER", "admin") # set in Space Variables18ADMIN_PASS = os.getenv("ADMIN_PASS", "change") # set in Space Variables19 20def check_auth(username: str, password: str) -> bool:21 return (username == ADMIN_USER) and (password == ADMIN_PASS)22 23# =========================24# Scanner config25# =========================26DEFAULT_DELAY = 0.6 # polite delay between requests (seconds)27REQUEST_TIMEOUT = 20 # HTTP timeout per request28HIGHLIGHT_COLOR = "FFF59D" # soft yellow (ARGB hex) for True cells29 30# =========================31# Helpers32# =========================33def compile_patterns(raw_terms: List[str], whole_word: bool) -> List[Tuple[str, re.Pattern]]:34 patterns: List[Tuple[str, re.Pattern]] = []35 for t in raw_terms:36 t = t.strip()37 if not t:38 continue39 rx = r"\b" + re.escape(t) + r"\b" if whole_word else re.escape(t)40 patterns.append((t, re.compile(rx, flags=re.IGNORECASE)))41 return patterns42 43def extract_links_from_excel(xlsx_bytes: bytes, sheet_name: str, col_name_or_letter: str) -> pd.DataFrame:44 """45 Returns a DataFrame with columns: Poster, Title, URL, SourceOrder46 """47 bio = io.BytesIO(xlsx_bytes)48 wb = load_workbook(bio, data_only=True, read_only=False)49 50 if sheet_name not in wb.sheetnames:51 raise ValueError(f"Sheet '{sheet_name}' not found. Available: {wb.sheetnames}")52 ws = wb[sheet_name]53 54 # Determine hyperlink column index55 if re.fullmatch(r"[A-Za-z]+", col_name_or_letter.strip()):56 from openpyxl.utils import column_index_from_string57 col_idx = column_index_from_string(col_name_or_letter.strip())58 else:59 headers = [ws.cell(row=1, column=c).value for c in range(1, ws.max_column + 1)]60 name_to_idx = {str(h).strip(): c for c, h in enumerate(headers, start=1) if h is not None}61 if col_name_or_letter not in name_to_idx:62 raise ValueError(f"Column '{col_name_or_letter}' not found. Headers: {list(name_to_idx.keys())}")63 col_idx = name_to_idx[col_name_or_letter]64 65 # Optional poster number column (if present)66 poster_idx = None67 headers = [ws.cell(row=1, column=c).value for c in range(1, ws.max_column + 1)]68 for c, h in enumerate(headers, start=1):69 if isinstance(h, str) and "Session/Poster Number" in h:70 poster_idx = c71 break72 73 rows = []74 order = 075 for r in range(2, ws.max_row + 1):76 cell = ws.cell(row=r, column=col_idx)77 title = cell.value78 url = None79 if cell.hyperlink is not None:80 url = cell.hyperlink.target81 elif isinstance(title, str) and title.startswith(("http://", "https://")):82 url = title83 if url:84 poster = ws.cell(row=r, column=poster_idx).value if poster_idx else None85 rows.append({"Poster": poster, "Title": title, "URL": url, "SourceOrder": order})86 order += 187 88 return pd.DataFrame(rows)89 90def fetch_url(url: str) -> Tuple[str, str]:91 headers = {"User-Agent": "Mozilla/5.0 (compatible; AbstractScanner/1.0)"}92 try:93 r = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT)94 r.raise_for_status()95 except Exception as e:96 return "", f"ERROR: fetch failed: {e}"97 98 ct = (r.headers.get("Content-Type") or "").lower()99 if "application/pdf" in ct or r.url.lower().endswith(".pdf"):100 try:101 from io import BytesIO102 from pdfminer.high_level import extract_text103 text = extract_text(BytesIO(r.content)) or ""104 except Exception as e:105 return "", f"ERROR: pdf parse failed: {e}"106 return text, "OK (pdf)"107 108 # HTML109 soup = BeautifulSoup(r.text, "lxml")110 for tag in soup(["script", "style", "nav", "header", "footer", "aside"]):111 tag.decompose()112 text = soup.get_text(separator=" ", strip=True)113 if not text:114 return "", "ERROR: empty HTML text"115 return text, "OK (html)"116 117def scan_text(text: str, patterns: List[Tuple[str, re.Pattern]]) -> Tuple[Dict[str, bool], Dict[str, str]]:118 hits: Dict[str, bool] = {}119 snippets: Dict[str, str] = {}120 for term, rx in patterns:121 m = rx.search(text)122 if m:123 hits[term] = True124 start = max(m.start() - 60, 0)125 end = min(m.end() + 60, len(text))126 snippets[term] = text[start:end].replace("\n", " ")127 else:128 hits[term] = False129 snippets[term] = ""130 return hits, snippets131 132def highlight_true_cells(workbook_bytes: bytes) -> bytes:133 bio = io.BytesIO(workbook_bytes)134 from openpyxl import load_workbook as _load_wb135 wb = _load_wb(bio)136 ws = wb.active137 fill = PatternFill(start_color=HIGHLIGHT_COLOR, end_color=HIGHLIGHT_COLOR, fill_type="solid")138 139 has_cols = []140 for c in range(1, ws.max_column + 1):141 h = ws.cell(row=1, column=c).value142 if isinstance(h, str) and h.startswith("has::"):143 has_cols.append(c)144 145 for r in range(2, ws.max_row + 1):146 for c in has_cols:147 cell = ws.cell(row=r, column=c)148 if str(cell.value) == "True":149 cell.fill = fill150 151 out = io.BytesIO()152 wb.save(out)153 return out.getvalue()154 155# =========================156# Streaming pipeline157# =========================158def run_pipeline_stream(159 xlsx_file, sheet_name: str, col_name_or_letter: str,160 keywords_text: str, whole_word: bool, delay: float161) -> Generator:162 progress = gr.Progress()163 status_prefix = "Status"164 165 try:166 with open(xlsx_file, "rb") as f:167 xlsx_bytes = f.read()168 except Exception as e:169 raise gr.Error(f"Could not read the uploaded file: {e}")170 171 raw_terms: List[str] = []172 for line in (keywords_text or "").splitlines():173 for chunk in line.split(","):174 t = chunk.strip()175 if t:176 raw_terms.append(t)177 178 yield (None, f"**{status_prefix}:** Parsing Excel and extracting links…")179 df = extract_links_from_excel(xlsx_bytes, sheet_name.strip(), col_name_or_letter.strip())180 total = len(df)181 if total == 0:182 raise gr.Error("No hyperlinks found in the specified column/sheet.")183 184 patterns = compile_patterns(raw_terms, whole_word)185 186 out_rows = []187 for i, row in enumerate(df.itertuples(index=False), start=1):188 url = getattr(row, "URL")189 poster = getattr(row, "Poster", None)190 title = getattr(row, "Title", None)191 source_order = getattr(row, "SourceOrder", i - 1)192 193 progress(i / total)194 yield (None, f"**{status_prefix}:** [{i}/{total}] Fetching:<br>{url}")195 196 text, status = fetch_url(url)197 if text:198 hits, snippets = scan_text(text, patterns)199 else:200 hits = {t: False for t, _ in patterns}201 snippets = {t: "" for t, _ in patterns}202 203 found_terms = [term for term, _ in patterns if hits.get(term)]204 summary_str = ", ".join(found_terms) if found_terms else "None"205 206 rec = {207 "Poster": poster,208 "Title": title,209 "URL": url,210 "Summary": summary_str,211 "status": status,212 "hit_count": sum(1 for v in hits.values() if v),213 "SourceOrder": source_order,214 }215 for term, _ in patterns:216 rec[f"has::{term}"] = hits[term]217 rec[f"snippet::{term}"] = snippets[term]218 out_rows.append(rec)219 220 time.sleep(max(0.0, float(delay)))221 222 yield (None, f"**{status_prefix}:** Compiling results…")223 out_df = pd.DataFrame(out_rows)224 225 if "SourceOrder" in out_df.columns:226 out_df.sort_values("SourceOrder", inplace=True, kind="stable")227 out_df.drop(columns=["SourceOrder"], inplace=True)228 229 cols = list(out_df.columns)230 if "Summary" in cols and "URL" in cols:231 cols.remove("Summary")232 url_idx = cols.index("URL")233 cols.insert(url_idx + 1, "Summary")234 out_df = out_df[cols]235 236 bio = io.BytesIO()237 with pd.ExcelWriter(bio, engine="openpyxl") as writer:238 out_df.to_excel(writer, index=False)239 240 highlighted = highlight_true_cells(bio.getvalue())241 242 import base64243 b64 = base64.b64encode(highlighted).decode("ascii")244 filename = f"abstract_scan_results_{int(time.time())}.xlsx"245 html_link = (246 f'<a href="data:application/vnd.openxmlformats-officedocument.spreadsheetml.sheet;base64,{b64}" '247 f'download="{filename}">⬇️ Download results ({filename})</a>'248 )249 250 yield (html_link, f"✅ Done! Processed **{total}** abstracts.")251 252# =========================253# Gradio UI254# =========================255DEFAULT_KEYWORDS = """patient-reported outcomes256quality of life257histology258biopsy259pharmacokinetics260pharmacodynamic261meta-analysis262systematic review263retrospective264imaging265murine266in vitro267ex vivo268"""269 270with gr.Blocks(title="Abstract Scanner", css="footer {visibility: hidden}") as demo:271 gr.Markdown("## Abstract Scanner (Admin Access)")272 with gr.Row():273 with gr.Column():274 xlsx = gr.File(label="Upload Excel (.xlsx)", file_types=[".xlsx"], type="filepath")275 sheet = gr.Textbox(label="Sheet/Tab name", value="ACG Hierarchy")276 col = gr.Textbox(label="Column header or letter for hyperlinks", value="Abstract Title")277 kws = gr.Textbox(label="Keywords", lines=10, value=DEFAULT_KEYWORDS.strip())278 ww = gr.Checkbox(label="Whole-word match", value=False)279 delay = gr.Slider(label="Delay (seconds)", minimum=0.2, maximum=2.0, value=DEFAULT_DELAY, step=0.1)280 run_btn = gr.Button("Run")281 with gr.Column():282 out = gr.HTML(label="Download results")283 status = gr.Markdown(value="Ready.", elem_id="status_box")284 285 def _run(xlsx_file, sheet_name, col_name_or_letter, keywords, whole_word, delay):286 if xlsx_file is None:287 raise gr.Error("Please upload an Excel file.")288 for payload in run_pipeline_stream(xlsx_file, sheet_name, col_name_or_letter, keywords, whole_word, delay):289 yield payload290 291 run_btn.click(292 fn=_run,293 inputs=[xlsx, sheet, col, kws, ww, delay],294 outputs=[out, status],295 api_name="run",296 )297 298# =========================299# Launch300# =========================301os.environ["GRADIO_DEFAULT_LANGUAGE"] = "en"302 303if __name__ == "__main__":304 demo.queue().launch(305 auth=check_auth,306 ssr_mode=False,307 show_error=True,308 )