import os
import gradio as gr
import pandas as pd
import numpy as np
import datetime
import urllib.parse
import urllib.request
import urllib.error
import json
import time
import re
import html as html_lib
from supabase import create_client, Client
# ── Helper: build query from terms and operators ──────────────
def build_boolean_query(term1, op1, term2, op2, term3):
"""
Build a boolean query from up to three terms with operators.
"""
terms = []
if term1 and term1.strip():
terms.append(term1.strip())
if term2 and term2.strip():
terms.append(term2.strip())
if term3 and term3.strip():
terms.append(term3.strip())
if len(terms) == 0:
return ""
if len(terms) == 1:
return terms[0]
# Build query with operators
query_parts = []
for i, t in enumerate(terms):
# Put quotes around phrases with spaces
if ' ' in t and not (t.startswith('"') and t.endswith('"')):
t = f'"{t}"'
query_parts.append(t)
# Insert operators between terms
if len(query_parts) == 2:
return f"{query_parts[0]} {op1} {query_parts[1]}"
else:
return f"{query_parts[0]} {op1} {query_parts[1]} {op2} {query_parts[2]}"
def normalize_boolean_query(query: str) -> str:
"""
Replace lowercase 'and' and 'or' with uppercase operators.
"""
if not query:
return query
pattern_and = re.compile(r'(? bool:
"""Return True if query contains AND or OR operators."""
q = normalize_boolean_query(query)
return ' AND ' in q or ' OR ' in q
# ── Role-based admin authentication using session tracking ─────────
# Roles: junior = DOI review only; senior = full administration.
_admin_sessions = {} # session_id -> role
def set_admin_active(session_id, role="senior"):
if session_id:
_admin_sessions[session_id] = role
return True
def get_admin_role(session_id):
return _admin_sessions.get(session_id)
def is_admin_active(session_id):
return bool(session_id and session_id in _admin_sessions)
def has_admin_role(session_id, *roles):
role = get_admin_role(session_id)
return bool(role and role in roles)
def request_has_role(request, *roles):
session_hash = getattr(request, "session_hash", None)
return has_admin_role(session_hash, *roles)
# ── Supabase credentials ──────────────────────────────────────────
url: str = os.environ.get("SUPABASE_URL")
key: str = os.environ.get("SUPABASE_KEY")
# Existing credentials become the junior-admin credentials.
AUTH_USER = os.environ.get("UPLOAD_USER")
AUTH_PASS = os.environ.get("UPLOAD_PASS")
# New senior-admin credentials. Set these as Hugging Face Space Secrets.
SENIOR_ADMIN_USER = os.environ.get("s_admin_user")
SENIOR_ADMIN_PASS = os.environ.get("s_admin_pass")
if not url or not key:
print("Warning: Missing Supabase credentials in Space Secrets!")
supabase = None
else:
supabase: Client = create_client(url, key)
HOMEPAGE_FILE = "homepage_content.html"
HOME_CONTACT_HTML = """
"""
def search_by_doi(doi):
"""Look up a DOI across CrossRef, Europe PMC and PubMed."""
doi = _extract_doi(doi)
if not doi or not re.match(r'^10\.\d{4,9}/\S+$', doi):
return "
Please enter a valid DOI, for example 10.1093/nar/gkac123.
"
# CrossRef — primary DOI metadata source
cr_url = "https://api.crossref.org/works/" + urllib.parse.quote(doi, safe='')
cr = _safe_request(cr_url)
if cr and cr.get('message'):
item = cr['message']
titles = item.get('title') or ['Untitled']
title = titles[0] if titles else 'Untitled'
auth = item.get('author') or []
names = []
for a in auth[:5]:
name = ' '.join(x for x in [a.get('given',''), a.get('family','')] if x).strip() or a.get('name','')
if name:
names.append(name)
if len(auth) > 5:
names.append('et al.')
journals = item.get('container-title') or []
journal = journals[0] if journals else ''
date_parts = (item.get('published-print') or item.get('published-online') or item.get('published') or {}).get('date-parts', [[]])
parts = date_parts[0] if date_parts else []
date = '-'.join(str(x) for x in parts) if parts else ''
url = item.get('URL') or f"https://doi.org/{doi}"
return _doi_html_result(doi, title, ', '.join(names), journal, date, url, 'CrossRef')
# Europe PMC fallback
epmc = _safe_request(
"https://www.ebi.ac.uk/europepmc/webservices/rest/search?query=" +
urllib.parse.quote(f'DOI:"{doi}"') + "&resultType=core&pageSize=5&format=json"
)
if epmc:
items = epmc.get('resultList', {}).get('result', [])
if items:
item = items[0]
authors_list = item.get('authorList', {}).get('author', [])
names = [a.get('fullName','') for a in authors_list[:5] if a.get('fullName')]
if len(authors_list) > 5:
names.append('et al.')
pmid = item.get('pmid','')
url = f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" if pmid else f"https://doi.org/{doi}"
return _doi_html_result(
doi,
item.get('title','Untitled'),
', '.join(names),
item.get('journalTitle',''),
item.get('firstPublicationDate', str(item.get('pubYear',''))),
url,
'Europe PMC'
)
# PubMed final fallback
pm_url = (
"https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi?db=pubmed&retmode=json&retmax=5&term=" +
urllib.parse.quote(f'"{doi}"[AID] OR "{doi}"[DOI]')
)
pm = _safe_request(pm_url)
ids = (pm or {}).get('esearchresult', {}).get('idlist', [])
if ids:
sm = _safe_request(
"https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi?db=pubmed&retmode=json&id=" + ','.join(ids)
)
if sm:
result = sm.get('result', {})
uid = result.get('uids', [ids[0]])[0]
item = result.get(uid, {})
raw = item.get('authors', [])
names = [a.get('name','') for a in raw[:5] if a.get('name')]
if len(raw) > 5:
names.append('et al.')
return _doi_html_result(
doi,
item.get('title','Untitled'),
', '.join(names),
item.get('fulljournalname', item.get('source','')),
item.get('pubdate',''),
f"https://pubmed.ncbi.nlm.nih.gov/{uid}/",
'PubMed'
)
return "
No publication metadata found for " + doi + ".
"
def get_reviewed_table_df():
"""Return all reviewed DOIs as a DataFrame for the admin console."""
columns = ['DOI', 'Reviewed At', 'Link']
rows = []
if _check_reviewed_table():
try:
res = supabase.table('reviewed_articles').select('doi,reviewed_at').order('reviewed_at', desc=True).execute()
for r in (res.data or []):
doi = (r.get('doi') or '').strip()
if doi:
rows.append({'DOI': doi, 'Reviewed At': r.get('reviewed_at') or '', 'Link': f'https://doi.org/{doi}'})
except Exception:
pass
db_ids = {r['DOI'].lower() for r in rows}
for doi in sorted(_reviewed_cache):
if doi not in db_ids:
rows.append({'DOI': doi, 'Reviewed At': 'Session only', 'Link': f'https://doi.org/{doi}'})
return pd.DataFrame(rows, columns=columns)
def reviewed_row_selection(evt: gr.SelectData, current_df):
try:
row_idx = evt.index[0]
doi = str(current_df.iloc[row_idx].get('DOI', '')).strip()
return doi, f'Selected reviewed DOI: {doi}' if doi else 'No DOI selected.'
except Exception as e:
return '', f'Selection error: {e}'
def remove_selected_reviewed_doi(selected_doi, request: gr.Request):
if not request_has_role(request, "senior"):
return 'Senior admin access required.', get_reviewed_table_df(), ''
if not selected_doi or not selected_doi.strip():
return ' Select a marked article first.', get_reviewed_table_df(), ''
msg = remove_reviewed_article(selected_doi)
return msg, get_reviewed_table_df(), ''
def _fetch_biorxiv(query, since_date_str, max_results=100):
"""Fetch from bioRxiv/medRxiv – basic keyword filtering (simple AND on split terms)."""
results = []
try:
today_str = datetime.date.today().strftime("%Y-%m-%d")
since_bxr = since_date_str.replace('/', '-')
# Extract keywords (remove quotes and split)
keywords = [w.strip('"').lower() for w in query.split() if len(w) > 2 and w not in ('AND', 'OR', 'NOT')]
# If query has quotes, keep them as phrase?
# We'll keep it simple: use the raw query as a phrase for title/abstract checking.
for server in ['biorxiv', 'medrxiv']:
url = f"https://api.biorxiv.org/details/{server}/{since_bxr}/{today_str}/0/json"
data = _safe_request(url)
if not data:
continue
collection = data.get('collection', [])
for item in collection:
title = item.get('title', 'Untitled')
abstract = item.get('abstract', '').lower()
# Check if all keywords appear (simple AND)
match = True
for kw in keywords:
if kw not in title.lower() and kw not in abstract:
match = False
break
if not match:
continue
doi = item.get('doi', '')
authors_str = item.get('authors', 'Unknown Authors')
author_list = [a.strip() for a in authors_str.split(';')]
names = author_list[:3]
if len(author_list) > 3:
names.append('et al.')
results.append({
'title': title,
'authors': '; '.join(names),
'journal': f"{server.capitalize()} (Preprint)",
'date': item.get('date', 'N/A'),
'url': f"https://doi.org/{doi}" if doi else f"https://www.{server}.org/",
'doi': doi,
'source': server.capitalize(),
'source_color': '#0891b2',
'source_bg': '#ecfeff',
})
if len(results) >= max_results:
break
if len(results) >= max_results:
break
return results, f"bioRxiv/medRxiv: {len(results)} results"
except Exception as e:
return results, f"bioRxiv/medRxiv: error – {str(e)}"
def _deduplicate(all_results):
"""Remove duplicates by DOI (if present) or by normalised title."""
seen_dois = set()
seen_titles = set()
unique = []
for r in all_results:
doi = r.get('doi', '').strip().lower()
title_key = r.get('title', '').strip().lower()[:80]
if doi and doi in seen_dois:
continue
if title_key and title_key in seen_titles:
continue
if doi:
seen_dois.add(doi)
if title_key:
seen_titles.add(title_key)
unique.append(r)
return unique
def _sort_results(results):
"""Sort results by date descending, with unparseable dates last."""
def date_key(r):
d = r.get('date', '') or ''
for fmt in ('%Y-%m-%d', '%Y/%m/%d', '%Y %b %d', '%Y %b', '%Y'):
try:
return datetime.datetime.strptime(d.strip()[:len(fmt)+2], fmt)
except Exception:
pass
parts = d.strip().split()
if parts:
try:
return datetime.datetime(int(parts[0]), 1, 1)
except Exception:
pass
return datetime.datetime.min
return sorted(results, key=date_key, reverse=True)
def _render_source_badge(source, color, bg):
return (
f'{source}'
)
def _run_multi_db_search(query, selected_sources, since_date, since_date_dash, window_label):
"""
Shared engine for running a multi-database search given an explicit
'since' date window, and rendering the resulting HTML.
"""
if not query or not query.strip():
return "
Please enter a valid search query.
"
# If query contains boolean operators, force only PubMed
if _is_boolean_query(query):
selected_sources = ['PubMed'] # override
all_results = []
source_logs = []
source_map = {
'PubMed': lambda: _fetch_pubmed(query, since_date),
'Europe PMC': lambda: _fetch_europe_pmc(query, since_date_dash),
'CrossRef': lambda: _fetch_crossref(query, since_date_dash),
'Semantic Scholar': lambda: _fetch_semantic_scholar(query, since_date_dash),
'bioRxiv / medRxiv': lambda: _fetch_biorxiv(query, since_date_dash),
}
for src_name, fetch_fn in source_map.items():
if src_name not in selected_sources:
continue
try:
res, log = fetch_fn()
all_results.extend(res)
source_logs.append(log)
except Exception as e:
source_logs.append(f"{src_name}: unexpected error – {str(e)}")
if not all_results:
log_html = ' | '.join(source_logs)
return f"""
No papers found for "{query}"
{window_label}. {log_html}
"""
unique_results = _deduplicate(all_results)
sorted_results = _sort_results(unique_results)
source_counts = {}
for r in sorted_results:
src = r.get('source', 'Unknown')
source_counts[src] = source_counts.get(src, 0) + 1
badge_row = ' '.join(
f''
f'{src}: {cnt}'
for src, cnt in source_counts.items()
)
log_html = ' | '.join(source_logs)
reviewed_dois = load_reviewed_dois()
html = f"""
"""
# No reviewed DOIs or mark buttons - removed entirely
for r in sorted_results:
src_badge = _render_source_badge(r['source'], r['source_color'], r['source_bg'])
title = r.get('title', 'Untitled')
url = r.get('url', '')
authors = r.get('authors', 'Unknown Authors')
journal = r.get('journal', '')
date = r.get('date', 'N/A')
doi = _extract_doi(r.get('doi', ''))
reviewed_badge = (
' Read'
if doi and doi in reviewed_dois else ''
)
doi_link = (
f'DOI: {doi}'
if doi else ''
)
title_html = (
f'{title}'
if url else
f'{title}'
)
html += f"""
"
return html
def fetch_multi_db_papers(query, selected_sources):
"""
Entry-point for the recent (past 6 months) multi-database literature search.
"""
six_months_ago = datetime.date.today() - datetime.timedelta(days=180)
since_date = six_months_ago.strftime("%Y/%m/%d")
since_date_dash = six_months_ago.strftime("%Y-%m-%d")
return _run_multi_db_search(query, selected_sources, since_date, since_date_dash, "past 6 months")
# ── Paged All-Time Literature Search ─────────────────────────────────────────
LS_FETCH_SIZE = 50
LS_PAGE_SIZE = 20
LS_YEAR_MIN = 1900
LS_YEAR_MAX = datetime.date.today().year
def _fetch_pubmed_paged(query, start_year, end_year, offset):
results, exhausted = [], False
try:
query = normalize_boolean_query(query)
query_enc = urllib.parse.quote(query)
search_url = (
f"https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi"
f"?db=pubmed&retmax={LS_FETCH_SIZE}&retstart={offset}"
f"&retmode=json&sort=date&term={query_enc}"
f"&mindate={start_year}/01/01&maxdate={end_year}/12/31"
)
search_data = _safe_request(search_url)
if not search_data:
return results, True
esresult = search_data.get('esearchresult', {})
total = int(esresult.get('count', 0))
ids = esresult.get('idlist', [])
if not ids:
return results, True
exhausted = (offset + len(ids)) >= total
summary_url = (
f"https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi"
f"?db=pubmed&retmode=json&id={','.join(ids)}"
)
summary_data = _safe_request(summary_url)
if not summary_data:
return results, exhausted
result_dict = summary_data.get('result', {})
for uid in result_dict.get('uids', []):
a = result_dict.get(uid, {})
raw_auth = a.get('authors', [])
names = [x.get('name', '') for x in raw_auth[:3]]
if len(raw_auth) > 3:
names.append('et al.')
results.append({
'title': a.get('title', 'Untitled'),
'authors': ', '.join(names) if names else 'Unknown Authors',
'journal': a.get('fulljournalname', a.get('source', '')),
'date': a.get('pubdate', 'N/A'),
'url': f"https://pubmed.ncbi.nlm.nih.gov/{uid}/",
'doi': a.get('elocationid', '').replace('doi: ', '').strip(),
'source': 'PubMed',
'source_color': '#dc2626', 'source_bg': '#fef2f2',
})
return results, exhausted
except Exception:
return results, True
def _fetch_europe_pmc_paged(query, start_year, end_year, offset):
results, exhausted = [], False
try:
query = normalize_boolean_query(query)
page_num = offset // LS_FETCH_SIZE
query_enc = urllib.parse.quote(query)
url = (
f"https://www.ebi.ac.uk/europepmc/webservices/rest/search"
f"?query={query_enc}&resultType=core&pageSize={LS_FETCH_SIZE}"
f"&page={page_num}&format=json&sort=P_PDATE_D%20desc"
f"&dateFrom={start_year}-01-01&dateTo={end_year}-12-31"
)
data = _safe_request(url)
if not data:
return results, True
hit_count = int(data.get('hitCount', 0))
articles = data.get('resultList', {}).get('result', [])
if not articles:
return results, True
exhausted = (offset + len(articles)) >= hit_count
for a in articles:
auth_list = a.get('authorList', {}).get('author', [])
names = [au.get('fullName', '') for au in auth_list[:3]]
if len(auth_list) > 3:
names.append('et al.')
doi = a.get('doi', '')
pmid = a.get('pmid', '')
article_url = (
f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" if pmid
else (f"https://doi.org/{doi}" if doi
else f"https://europepmc.org/article/{a.get('source','')}/{a.get('id','')}")
)
results.append({
'title': a.get('title', 'Untitled').rstrip('.'),
'authors': ', '.join(names) if names else 'Unknown Authors',
'journal': a.get('journalTitle', ''),
'date': a.get('firstPublicationDate', a.get('pubYear', 'N/A')),
'url': article_url, 'doi': doi,
'source': 'Europe PMC',
'source_color': '#7c3aed', 'source_bg': '#f5f3ff',
})
return results, exhausted
except Exception:
return results, True
def _fetch_crossref_paged(query, start_year, end_year, offset):
results, exhausted = [], False
try:
query_enc = urllib.parse.quote(query)
url = (
f"https://api.crossref.org/works"
f"?query={query_enc}"
f"&filter=from-pub-date:{start_year}-01-01,until-pub-date:{end_year}-12-31"
f"&rows={LS_FETCH_SIZE}&offset={offset}&sort=published&order=desc"
f"&select=DOI,title,author,container-title,published,type"
f"&mailto=biohub@example.com"
)
data = _safe_request(url)
if not data:
return results, True
msg = data.get('message', {})
total = msg.get('total-results', 0)
items = msg.get('items', [])
if not items:
return results, True
exhausted = (offset + len(items)) >= total
for item in items:
if item.get('type', '') not in ('journal-article', 'proceedings-article', 'posted-content'):
continue
titles = item.get('title', ['Untitled'])
title = titles[0] if titles else 'Untitled'
authors_raw = item.get('author', [])
names = []
for au in authors_raw[:3]:
fn = au.get('given', ''); ln = au.get('family', '')
names.append(f"{fn} {ln}".strip() if fn or ln else au.get('name', ''))
if len(authors_raw) > 3:
names.append('et al.')
journals = item.get('container-title', [])
journal = journals[0] if journals else ''
doi = item.get('DOI', '')
parts = (item.get('published', {}).get('date-parts', [[]])[0] or [])
pub_date = '-'.join(str(p) for p in parts) if parts else 'N/A'
results.append({
'title': title,
'authors': ', '.join(n for n in names if n) or 'Unknown Authors',
'journal': journal, 'date': pub_date,
'url': f"https://doi.org/{doi}" if doi else '',
'doi': doi, 'source': 'CrossRef',
'source_color': '#059669', 'source_bg': '#ecfdf5',
})
return results, exhausted
except Exception:
return results, True
def _fetch_semantic_scholar_paged(query, start_year, end_year, offset):
results, exhausted = [], False
try:
query_enc = urllib.parse.quote(query)
url = (
f"https://api.semanticscholar.org/graph/v1/paper/search"
f"?query={query_enc}"
f"&publicationDateOrYear={start_year}:{end_year}"
f"&fields=title,authors,venue,year,publicationDate,externalIds,openAccessPdf"
f"&limit={min(LS_FETCH_SIZE, 100)}&offset={offset}"
)
data = _safe_request(url, headers={'x-api-key': ''})
if not data:
return results, True
total = data.get('total', 0)
papers = data.get('data', [])
if not papers:
return results, True
exhausted = (offset + len(papers)) >= total
for p in papers:
authors_raw = p.get('authors', [])
names = [a.get('name', '') for a in authors_raw[:3]]
if len(authors_raw) > 3:
names.append('et al.')
ext_ids = p.get('externalIds', {})
doi = ext_ids.get('DOI', '')
paper_id = p.get('paperId', '')
pdf_info = p.get('openAccessPdf') or {}
article_url = (
pdf_info.get('url') or
(f"https://doi.org/{doi}" if doi else '') or
(f"https://www.semanticscholar.org/paper/{paper_id}" if paper_id else '')
)
pub_date = p.get('publicationDate') or str(p.get('year', 'N/A'))
results.append({
'title': p.get('title', 'Untitled'),
'authors': ', '.join(n for n in names if n) or 'Unknown Authors',
'journal': p.get('venue', ''), 'date': pub_date,
'url': article_url, 'doi': doi,
'source': 'Semantic Scholar',
'source_color': '#d97706', 'source_bg': '#fffbeb',
})
return results, exhausted
except Exception:
return results, True
def _fetch_biorxiv_paged(query, start_year, end_year, cursor):
results, exhausted = [], False
try:
start_dt = f"{start_year}-01-01"
end_dt = f"{end_year}-12-31"
# Use normalized query for keyword extraction
norm_query = normalize_boolean_query(query)
keywords = [w.strip('"').lower() for w in norm_query.split() if len(w) > 2 and w not in ('AND', 'OR', 'NOT')]
for server in ['biorxiv', 'medrxiv']:
url = f"https://api.biorxiv.org/details/{server}/{start_dt}/{end_dt}/{cursor}/json"
data = _safe_request(url)
if not data:
continue
collection = data.get('collection', [])
if not collection:
exhausted = True
continue
if len(collection) < 100:
exhausted = True
for item in collection:
title = item.get('title', 'Untitled')
abstract = item.get('abstract', '').lower()
match = True
for kw in keywords:
if kw not in title.lower() and kw not in abstract:
match = False
break
if not match:
continue
doi = item.get('doi', '')
authors_str = item.get('authors', 'Unknown Authors')
author_list = [a.strip() for a in authors_str.split(';')]
names = author_list[:3]
if len(author_list) > 3:
names.append('et al.')
results.append({
'title': title,
'authors': '; '.join(names),
'journal': f"{server.capitalize()} (Preprint)",
'date': item.get('date', 'N/A'),
'url': f"https://doi.org/{doi}" if doi else f"https://www.{server}.org/",
'doi': doi,
'source': server.capitalize(),
'source_color': '#0891b2', 'source_bg': '#ecfeff',
})
return results, exhausted
except Exception:
return results, True
def _ls_fetch_batch(state):
"""Fetch one batch from all non-exhausted sources and merge into pool."""
topic = state['topic']
start_year = state['start_year']
end_year = state['end_year']
sources = state['sources']
exhausted = set(state['exhausted'])
seen_dois = set(state['seen_dois'])
seen_titles = set(state['seen_titles'])
fetcher_map = {
'PubMed': lambda off: _fetch_pubmed_paged(topic, start_year, end_year, off),
'Europe PMC': lambda off: _fetch_europe_pmc_paged(topic, start_year, end_year, off),
'CrossRef': lambda off: _fetch_crossref_paged(topic, start_year, end_year, off),
'Semantic Scholar': lambda off: _fetch_semantic_scholar_paged(topic, start_year, end_year, off),
'bioRxiv / medRxiv': lambda off: _fetch_biorxiv_paged(topic, start_year, end_year, off),
}
new_results = []
for src_name, fetcher in fetcher_map.items():
if src_name not in sources or src_name in exhausted:
continue
offset = state['offsets'].get(src_name, 0)
try:
res, is_exhausted = fetcher(offset)
if is_exhausted:
exhausted.add(src_name)
state['offsets'][src_name] = offset + LS_FETCH_SIZE
new_results.extend(res)
except Exception:
exhausted.add(src_name)
for r in new_results:
doi = r.get('doi', '').strip().lower()
title_key = r.get('title', '').strip().lower()[:80]
if doi and doi in seen_dois:
continue
if title_key and title_key in seen_titles:
continue
if doi:
seen_dois.add(doi)
if title_key:
seen_titles.add(title_key)
state['pool'].append(r)
state['pool'] = _sort_results(state['pool'])
state['exhausted'] = list(exhausted)
state['seen_dois'] = list(seen_dois)
state['seen_titles'] = list(seen_titles)
def _ls_render_page(state):
# No reviewed DOIs or mark buttons - removed entirely
page = state['page']
pool = state['pool']
start = (page - 1) * LS_PAGE_SIZE
end = start + LS_PAGE_SIZE
items = pool[start:end]
if not items:
return ("
No results on this page.
")
reviewed_dois = load_reviewed_dois()
html = ""
for r in items:
src_badge = _render_source_badge(r['source'], r['source_color'], r['source_bg'])
title = r.get('title', 'Untitled')
url = r.get('url', '')
authors = r.get('authors', 'Unknown Authors')
journal = r.get('journal', '')
date = r.get('date', 'N/A')
doi = _extract_doi(r.get('doi', ''))
reviewed_badge = (
' Read'
if doi and doi in reviewed_dois else ''
)
doi_link = (
f'DOI: {doi}'
if doi else ''
)
title_html = (
f'{title}'
if url else
f'{title}'
)
html += f"""
"""
return html
def _ls_render_nav(state):
page = state['page']
pool = state['pool']
exhausted = state['exhausted']
sources = state['sources']
total_pool = len(pool)
max_known_page = max(1, (total_pool + LS_PAGE_SIZE - 1) // LS_PAGE_SIZE)
all_exhausted = all(s in exhausted for s in sources) if sources else True
has_more = not all_exhausted
pages_to_show = list(range(1, min(max_known_page, 10) + 1))
show_ellipsis = (max_known_page > 10) or has_more
btns = ""
for p in pages_to_show:
if p == page:
btns += (f'{p}')
else:
btns += (f'{p}')
if show_ellipsis:
btns += ('…')
src_counts = {}
for r in pool:
s = r.get('source', '')
src_counts[s] = src_counts.get(s, 0) + 1
badges = ' '.join(
f'{s}: {c}'
for s, c in src_counts.items()
)
status_line = (
f'
'
f'Page {page} · {total_pool} results loaded'
+ (' · fetching more…' if has_more else ' · all sources loaded')
+ f'
'
)
return f"""
{status_line}
{btns}
{badges}
"""
def _ls_empty_state():
return {
'topic': '', 'sources': [], 'start_year': LS_YEAR_MIN, 'end_year': LS_YEAR_MAX,
'page': 1, 'pool': [], 'offsets': {}, 'exhausted': [],
'seen_dois': [], 'seen_titles': [],
}
def ls_init_search(query, sources, start_year, end_year):
start_year = int(start_year) if start_year else LS_YEAR_MIN
end_year = int(end_year) if end_year else LS_YEAR_MAX
if not query or not query.strip():
err = ("
"
" Please enter a valid search query.
")
return err, "", _ls_empty_state()
# If boolean query, restrict to PubMed
if _is_boolean_query(query):
sources = ['PubMed']
state = {
'topic': query.strip(), 'sources': list(sources),
'start_year': start_year, 'end_year': end_year,
'page': 1, 'pool': [], 'offsets': {}, 'exhausted': [],
'seen_dois': [], 'seen_titles': [],
}
_ls_fetch_batch(state)
if len(state['pool']) < LS_PAGE_SIZE:
_ls_fetch_batch(state)
if not state['pool']:
empty_html = (
"
"
" No papers found for “"
+ query +
"” in the selected year range across chosen databases.
"
)
return empty_html, "", state
return _ls_render_page(state), _ls_render_nav(state), state
def ls_prev_page(state):
if not state or state['page'] <= 1:
return _ls_render_page(state), _ls_render_nav(state), state
state = {**state, 'page': state['page'] - 1}
return _ls_render_page(state), _ls_render_nav(state), state
def ls_next_page(state):
if not state:
return "", "", state
state = dict(state)
next_page = state['page'] + 1
needed = next_page * LS_PAGE_SIZE
max_batches = 6
batches_done = 0
all_exhausted = lambda: all(s in state['exhausted'] for s in state['sources'])
while len(state['pool']) < needed and not all_exhausted() and batches_done < max_batches:
_ls_fetch_batch(state)
batches_done += 1
if len(state['pool']) >= (next_page - 1) * LS_PAGE_SIZE + 1:
state['page'] = next_page
return _ls_render_page(state), _ls_render_nav(state), state
# ── Database functions ────────────────────────────────────────────────────────
def search_supabase(species, tissue, tool, accession=""):
if not supabase:
return pd.DataFrame({"Error": ["Supabase keys are missing. Connection failed."]})
try:
table_name = "sra_database" if tool == "SRA" else "geo_database"
response = supabase.table(table_name).select("*").execute()
raw_data = response.data
if not raw_data:
return pd.DataFrame({"Status": [f"No records found in {table_name}."]})
df = pd.DataFrame(raw_data)
if 'created_at' in df.columns:
df = df.drop(columns=['created_at'])
if accession and accession.strip():
if "Accession" in df.columns:
df = df[df["Accession"].astype(str).str.lower().str.strip() == accession.lower().strip()]
if df.empty:
return pd.DataFrame({"Status": [f"No matching record for Accession '{accession}'."]})
else:
if "Organism" in df.columns:
df = df[df["Organism"].apply(normalize_organism) == normalize_organism(species)]
if "tissue" in df.columns:
df = df[df["tissue"].astype(str).str.lower().str.strip() == tissue.lower().strip()]
if df.empty:
return pd.DataFrame({"Status": [f"No records for '{species}' in {tool}."]})
display_cols = [c for c in ALL_UNIQUE_COLUMNS if c in df.columns]
return df[display_cols].fillna("N/A")
except Exception as e:
return pd.DataFrame({"Database Error": [f"Failed to fetch data: {str(e)}"]})
def load_entire_table():
if not supabase:
return pd.DataFrame({"Error": ["Supabase connection is inactive."]})
try:
sra_res = supabase.table("sra_database").select("*").execute()
df_sra = pd.DataFrame(sra_res.data) if sra_res.data else pd.DataFrame()
if 'created_at' in df_sra.columns:
df_sra = df_sra.drop(columns=['created_at'])
geo_res = supabase.table("geo_database").select("*").execute()
df_geo = pd.DataFrame(geo_res.data) if geo_res.data else pd.DataFrame()
if 'created_at' in df_geo.columns:
df_geo = df_geo.drop(columns=['created_at'])
if df_sra.empty and df_geo.empty:
return pd.DataFrame(columns=ALL_UNIQUE_COLUMNS)
combined_df = pd.concat([df_sra, df_geo], ignore_index=True)
ordered_cols = [c for c in ALL_UNIQUE_COLUMNS if c in combined_df.columns]
return combined_df[ordered_cols].fillna("N/A")
except Exception as e:
return pd.DataFrame({"Error": [str(e)]})
def generate_tools_launchpad_html(dataset_type="GEO"):
tools = SRA_TOOLS_DATA if dataset_type == "SRA" else GEO_TOOLS_DATA
icon = "" if dataset_type == "SRA" else ""
label = "SRA Download & QC" if dataset_type == "SRA" else "GEO Analytical"
accent = "#0891b2" if dataset_type == "SRA" else "#2563eb"
html = f"""
{icon} Biological Data Located. Launch a {label} Suite below:
Integrated Bioinformatics Platform for Animal Reproduction Research
ReproOmics Hub is an integrated bioinformatics platform developed to support research in
Animal Reproductive Technologies (ART) with a primary focus on cattle and buffalo.
The portal provides seamless access to public omics datasets, literature resources,
AI-assisted analytical tools, and bioinformatics workflows to accelerate research in
reproductive biology, embryo development, fertility, and livestock improvement.
About ReproOmics Hub
The portal integrates biological databases, literature resources, and computational tools
into a single platform for researchers working in:
Researchers can search public repositories such as
GEO and SRA, explore reproductive omics datasets,
perform downstream analyses, and access AI-assisted resources through a unified interface.
"""
# ── Reviewed-Article helpers (kept for admin DOI box) ──────────────────────────────────
_admin_active = False
_reviewed_cache: set = set()
_supabase_reviewed_ok: bool | None = None
def _check_reviewed_table():
global _supabase_reviewed_ok
if _supabase_reviewed_ok is not None:
return _supabase_reviewed_ok
if not supabase:
_supabase_reviewed_ok = False
return False
try:
supabase.table("reviewed_articles").select("doi").limit(1).execute()
_supabase_reviewed_ok = True
except Exception:
_supabase_reviewed_ok = False
return _supabase_reviewed_ok
def load_reviewed_dois():
if _check_reviewed_table():
try:
res = supabase.table("reviewed_articles").select("doi").execute()
return {r['doi'].strip().lower() for r in (res.data or []) if r.get('doi')}
except Exception:
pass
return set(_reviewed_cache)
def save_reviewed_article(article_id):
article_id = (article_id or "").strip().lower()
if not article_id:
return " No identifier provided"
if _check_reviewed_table():
try:
supabase.table("reviewed_articles").upsert({"doi": article_id}).execute()
_reviewed_cache.add(article_id)
return f" Marked as reviewed"
except Exception:
pass
_reviewed_cache.add(article_id)
return " Marked as reviewed (session only — create reviewed_articles table in Supabase for persistence)"
def remove_reviewed_article(article_id):
article_id = (article_id or "").strip().lower()
if not article_id:
return " No identifier provided"
if _check_reviewed_table():
try:
supabase.table("reviewed_articles").delete().eq("doi", article_id).execute()
_reviewed_cache.discard(article_id)
return "Removed from reviewed"
except Exception:
pass
_reviewed_cache.discard(article_id)
return "Removed from reviewed (session only)"
def mark_article_api(doi, action, session_hash=None):
role = get_admin_role(session_hash)
if role is None:
return "Admin login required. Please log in first."
if action == "unmark" and role != "senior":
return "Senior admin access required to remove reviewed status."
article_id = (doi or "").strip().lower()
if not article_id:
return " No identifier provided"
if action == "mark":
return save_reviewed_article(article_id)
elif action == "unmark":
return remove_reviewed_article(article_id)
else:
return " Unknown action"
def list_reviewed_html():
rows = []
if _check_reviewed_table():
try:
res = supabase.table("reviewed_articles").select("doi,reviewed_at").order("reviewed_at", desc=True).execute()
rows = res.data or []
except Exception:
pass
db_ids = {r.get('doi', '').lower() for r in rows}
cache_only = [c for c in _reviewed_cache if c not in db_ids]
if not rows and not cache_only:
note = "" if _check_reviewed_table() else (
"
"
" Create a reviewed_articles (doi TEXT PRIMARY KEY, reviewed_at TIMESTAMPTZ DEFAULT NOW())"
" table in Supabase for persistent storage. Currently using session memory.
"
)
return "
No articles marked yet.
" + note
html = ""
for r in rows:
article_id = r.get('doi', '')
at = (r.get('reviewed_at') or '')[:10]
display = article_id[:80] + ('...' if len(article_id) > 80 else '')
link = f"https://doi.org/{article_id}" if not article_id.startswith('http') else article_id
html += (
f"
"
terms_df = get_pubcrawler_terms_df()
terms = []
if not terms_df.empty:
terms = [str(x).strip() for x in terms_df['Term'].tolist() if str(x).strip()]
if not terms:
return "
No PubCrawler terms have been saved yet.
"
since = datetime.date.today() - datetime.timedelta(days=30)
since_date = since.strftime('%Y/%m/%d')
since_dash = since.strftime('%Y-%m-%d')
all_results = []
term_logs = []
for term in terms:
source_map = {
'PubMed': lambda t=term: _fetch_pubmed(t, since_date, max_results=200),
'Europe PMC': lambda t=term: _fetch_europe_pmc(t, since_dash, max_results=200),
'CrossRef': lambda t=term: _fetch_crossref(t, since_dash, max_results=100),
'Semantic Scholar': lambda t=term: _fetch_semantic_scholar(t, since_dash, max_results=100),
'bioRxiv / medRxiv': lambda t=term: _fetch_biorxiv(t, since_dash, max_results=100),
}
for source_name, fetch_fn in source_map.items():
try:
results, log = fetch_fn()
for result in results:
result = dict(result)
result['_crawler_term'] = term
all_results.append(result)
if results:
term_logs.append(f"{term}: {source_name} {len(results)}")
except Exception as e:
term_logs.append(f"{term}: {source_name} error")
if not all_results:
return f"""
No publications matching the saved PubCrawler terms were found in the past 30 days.
"""
# Deduplicate while retaining all saved terms that matched each article.
merged = {}
for r in all_results:
doi = _extract_doi(r.get('doi', ''))
title_key = str(r.get('title', '')).strip().lower()
key = ('doi', doi.lower()) if doi else ('title', title_key[:140])
if key not in merged:
merged[key] = dict(r)
merged[key]['_crawler_terms'] = [r.get('_crawler_term', '')]
else:
hit = r.get('_crawler_term', '')
if hit and hit not in merged[key]['_crawler_terms']:
merged[key]['_crawler_terms'].append(hit)
if not merged[key].get('doi') and doi:
merged[key]['doi'] = doi
results = _sort_results(list(merged.values()))
reviewed_dois = load_reviewed_dois()
source_counts = {}
for r in results:
src = r.get('source', 'Unknown')
source_counts[src] = source_counts.get(src, 0) + 1
count_badges = ''.join(
f"{html_lib.escape(src)}: {cnt}"
for src, cnt in source_counts.items()
)
html = f"""
"""
for r in results:
title = html_lib.escape(str(r.get('title', 'Untitled')))
url = html_lib.escape(str(r.get('url', '') or ''))
authors = html_lib.escape(str(r.get('authors', 'Unknown Authors')))
journal = html_lib.escape(str(r.get('journal', '')))
date = html_lib.escape(str(r.get('date', 'N/A')))
doi = _extract_doi(r.get('doi', ''))
doi_esc = html_lib.escape(doi)
src = html_lib.escape(str(r.get('source', 'Unknown')))
src_color = r.get('source_color', '#2563eb')
src_bg = r.get('source_bg', '#eff6ff')
terms_hit = r.get('_crawler_terms', []) or []
term_html = ''.join(
f"{html_lib.escape(t)}"
for t in terms_hit
)
reviewed_badge = (
' Read'
if doi and doi in reviewed_dois else ''
)
title_html = (
f'{title}'
if url else f'
{title}
'
)
html += f"""
{title_html}
{src}
{reviewed_badge}
{term_html}
{date}
Authors: {authors}
{f' •{journal}' if journal else ''}
"
return html
def refresh_pubcrawler_articles(request: gr.Request):
"""Refresh the PubCrawler article feed for the authenticated admin."""
return get_pubcrawler_articles_html(request)
def delete_pubcrawler_term(selected_id, request: gr.Request):
"""Delete a selected PubCrawler term for an authenticated admin."""
if not _pubcrawler_admin_ok(request):
return ' Admin login required. Please log in first.', get_pubcrawler_terms_df(), '', get_pubcrawler_articles_html(request)
if not selected_id:
return ' Select a term first.', get_pubcrawler_terms_df(), '', get_pubcrawler_articles_html(request)
if not supabase:
return ' Supabase connection is unavailable.', get_pubcrawler_terms_df(), '', get_pubcrawler_articles_html(request)
try:
supabase.table('pubcrawler').delete().eq('id', int(selected_id)).execute()
return 'PubCrawler term removed.', get_pubcrawler_terms_df(), '', get_pubcrawler_articles_html(request)
except Exception as e:
return f' Could not remove term: {e}', get_pubcrawler_terms_df(), str(selected_id), get_pubcrawler_articles_html(request)
# ── Layout ────────────────────────────────────────────────────────────────────
ALL_SOURCES = ['PubMed', 'Europe PMC', 'CrossRef', 'Semantic Scholar', 'bioRxiv / medRxiv']
with gr.Blocks() as demo:
# ── Institute header ──────────────────────────────────────────────────────
gr.HTML("""
भाकृअनुप-राष्ठ्रीय डेरी अनुसंधान संस्थान
ICAR – National Dairy Research Institute
KARNAL, HARYANA
ReproOmics Hub — Bioinformatics Hub for Animal Reproduction Research
""")
gr.HTML("""
RH
ReproOmics Hub
Research workspace for animal reproduction data, literature, and analysis
GEOSRALiteraturePubCrawler
""")
# ── Paper Alert ──
gr.HTML(value=generate_paper_alert_html())
# ── Shared state ──
is_authenticated = gr.State(False)
selected_accession = gr.State("")
hidden_canvas_data = gr.Textbox(visible=False)
# ── Hidden textbox for session_hash ──
session_hash_state = gr.Textbox(visible=False, value="")
# ── JavaScript to set session_hash_state ──
gr.HTML("""
""")
with gr.Tabs() as app_tabs:
# ── TAB 1: Home ────────────────────────────────────────────────────────
with gr.TabItem("Home"):
with gr.Column(visible=True) as home_locked_panel:
home_view_component = gr.HTML(value=generate_view_html())
gr.HTML(HOME_CONTACT_HTML)
with gr.Column(visible=False) as home_editor_panel:
gr.HTML("""
Site editor
Homepage Content
Edit the homepage content below. Changes are saved to the local homepage file.
""")
home_editor_component = gr.HTML(value=generate_editor_html())
with gr.Row():
save_home_btn = gr.Button("Save Homepage", variant="primary")
home_save_status = gr.Markdown("")
# ── TAB 2: Query Database Hub ──────────────────────────────────────────
with gr.TabItem("Query Database Hub"):
with gr.Column(elem_id="qt-search-panel", visible=True) as qt_search_panel:
gr.HTML("""
Dataset search
Query Biological Records
Search your GEO and SRA records using animal, tissue, source, and accession filters.
Search the most recent six months across PubMed, Europe PMC, CrossRef, Semantic Scholar, and bioRxiv/medRxiv. Results are deduplicated and sorted by date.
")
with gr.Row():
ls_prev_btn = gr.Button("Previous", variant="secondary")
ls_next_btn = gr.Button("Next", variant="primary")
with gr.Column(scale=1, min_width=220, visible=False) as ls_doi_col:
with gr.Group():
gr.HTML("
Save one or more terms, then refresh the article feed.
")
# ── TAB 7: Administrative Portal ───────────────────────────────────────
with gr.TabItem("Administrative Portal"):
with gr.Column(visible=True, elem_classes=["admin-login"]) as login_panel:
gr.HTML("""
Restricted access
Administrative Portal
Sign in with a junior or senior administrator account. Junior access is limited to DOI review; senior access includes database management, PubCrawler, reviewed-article management, and homepage editing.
Your account can mark literature DOIs as read. The database administration console, homepage editor, PubCrawler, and reviewed-article deletion require senior administrator access.
""")
with gr.Column(visible=False) as upload_panel:
gr.HTML("
Administration
Admin Console
Manage dataset records, reviewed DOIs, and site content from one workspace.
")
with gr.Row():
with gr.Column(scale=1, min_width=290):
with gr.Group(elem_classes=["admin-section"]):
gr.HTML("
Data ingestion
Upload Records
Upload a CSV or Excel dataset and assign its biological metadata.