"""
return html
def _ls_render_nav(state):
"""Render the pagination bar HTML."""
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 display: always show up to 10, then "..."
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': [], 'is_admin': False,
}
def ls_init_search(topic, sources, start_year, end_year, is_admin=False):
"""Kick off a new search; returns (results_html, nav_html, new_state)."""
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 topic or not topic.strip():
err = ("
"
"⚠️ Please enter a valid search topic.
")
return err, "", _ls_empty_state()
state = {
'topic': topic.strip(), 'sources': list(sources),
'start_year': start_year, 'end_year': end_year,
'page': 1, 'pool': [], 'offsets': {}, 'exhausted': [],
'seen_dois': [], 'seen_titles': [], 'is_admin': bool(is_admin),
}
# Fetch first two batches to ensure a full first page even with heavy deduplication
_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 “"
+ topic +
"” in the selected year range across chosen databases.
"
)
return empty_html, "", state
reviewed_dois = load_reviewed_dois()
return _ls_render_page(state, reviewed_dois), _ls_render_nav(state), state
def ls_prev_page(state):
"""Navigate to the previous page."""
reviewed_dois = load_reviewed_dois()
if not state or state['page'] <= 1:
return _ls_render_page(state, reviewed_dois), _ls_render_nav(state), state
state = {**state, 'page': state['page'] - 1}
return _ls_render_page(state, reviewed_dois), _ls_render_nav(state), state
def ls_next_page(state):
"""Navigate to next page, fetching more results from APIs if needed."""
if not state:
return "", "", state
state = dict(state)
next_page = state['page'] + 1
needed = next_page * LS_PAGE_SIZE
# Keep fetching until we have enough or all sources are exhausted
max_batches = 6 # safety limit per click
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
# Advance only if results exist for next page
if len(state['pool']) >= (next_page - 1) * LS_PAGE_SIZE + 1:
state['page'] = next_page
reviewed_dois = load_reviewed_dois()
return _ls_render_page(state, reviewed_dois), _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."]})
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": [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:
")
def check_upload_credentials_generator(username, password):
global _admin_active
if not AUTH_USER or not AUTH_PASS:
_admin_active = False
yield (gr.Column(visible=True), gr.Column(visible=False), "Configuration Error: Secrets missing.", BLANK_TEMPLATE_DF, gr.Column(visible=True), gr.Column(visible=False), gr.HTML(value=generate_editor_html()), False, gr.Column(visible=False), gr.Column(visible=False))
return
if username == AUTH_USER and password == AUTH_PASS:
_admin_active = True
yield (gr.Column(visible=False), gr.Column(visible=True), "Authenticated. Loading data...", BLANK_TEMPLATE_DF, gr.Column(visible=False), gr.Column(visible=True), gr.HTML(value=generate_editor_html()), True, gr.Column(visible=True), gr.Column(visible=True))
entire_table_df = load_entire_table()
yield (gr.Column(visible=False), gr.Column(visible=True), "Admin Dashboard Ready.", entire_table_df, gr.Column(visible=False), gr.Column(visible=True), gr.HTML(value=generate_editor_html()), True, gr.Column(visible=True), gr.Column(visible=True))
else:
_admin_active = False
yield (gr.Column(visible=True), gr.Column(visible=False), "Invalid credentials. Access denied.", BLANK_TEMPLATE_DF, gr.Column(visible=True), gr.Column(visible=False), gr.HTML(value=generate_view_html()), False, gr.Column(visible=False), gr.Column(visible=False))
def upload_data_to_supabase(file_obj, species, tissue, dataset_type):
if not supabase:
return "Configuration Error: Supabase connection is inactive."
if file_obj is None:
return "Please upload a valid CSV or Excel file first."
try:
if file_obj.name.endswith('.csv'):
uploaded_df = pd.read_csv(file_obj.name)
elif file_obj.name.endswith(('.xls', '.xlsx')):
uploaded_df = pd.read_excel(file_obj.name)
else:
return "Unsupported format. Please upload a .csv or .xlsx file."
uploaded_df.columns = [c.strip() for c in uploaded_df.columns]
table_name = "sra_database" if dataset_type == "SRA" else "geo_database"
for target_col in EXACT_MATCH_COLUMNS:
if target_col not in uploaded_df.columns:
uploaded_df[target_col] = None
db_ready_df = uploaded_df[EXACT_MATCH_COLUMNS].copy().dropna(subset=["Accession"])
canonical_organism = "Bos taurus" if normalize_organism(species) == "bos taurus" else species.strip()
db_ready_df["Organism"] = canonical_organism
db_ready_df["tissue"] = tissue.strip()
db_ready_df["dataset"] = dataset_type.strip()
db_ready_df = sanitize_dataframe(db_ready_df)
new_records = db_ready_df.to_dict(orient="records")
if not new_records:
return "No rows with valid Accession numbers found."
db_response = supabase.table(table_name).select("Accession,Organism,tissue").execute()
existing = db_response.data if db_response.data else []
exact_keys = set()
accession_only = {}
for r in existing:
acc = normalize_val(r.get("Accession"))
org = normalize_organism(r.get("Organism", ""))
tis = normalize_val(r.get("tissue", ""))
exact_keys.add((acc, org, tis))
if acc not in accession_only:
accession_only[acc] = []
accession_only[acc].append((r.get("Organism", ""), r.get("tissue", "")))
incoming_org_norm = normalize_organism(canonical_organism)
incoming_tis_norm = normalize_val(tissue)
true_dupes = []
cross_category_warnings = []
final_records = []
for r in new_records:
acc = normalize_val(r.get("Accession"))
key = (acc, incoming_org_norm, incoming_tis_norm)
if key in exact_keys:
true_dupes.append(acc)
else:
if acc in accession_only:
for (ex_org, ex_tis) in accession_only[acc]:
cross_category_warnings.append(
f" • {acc} already exists as Organism='{ex_org}', Tissue='{ex_tis}'"
)
final_records.append(r)
msg_parts = []
if cross_category_warnings:
warn_lines = "\n".join(cross_category_warnings)
msg_parts.append(
f"⚠️ Warning — are you sure you inputted the right parameters?\n"
f"The following accessions already exist in a different category:\n{warn_lines}"
)
if true_dupes:
msg_parts.append(f"Skipped {len(true_dupes)} exact duplicate(s) (same accession + organism + tissue).")
if not final_records:
msg_parts.append("No new records to insert after duplicate check.")
return "\n".join(msg_parts)
supabase.table(table_name).insert(final_records).execute()
msg_parts.append(f"✅ Inserted {len(final_records)} record(s) into {table_name}.")
return "\n".join(msg_parts)
except Exception as e:
return f"Upload failed: {str(e)}"
def handle_row_selection(evt: gr.SelectData, current_df):
try:
row_idx = evt.index[0]
if "Accession" in current_df.columns:
accession_id = str(current_df.iloc[row_idx].get("Accession", "")).strip()
if accession_id and accession_id != "N/A":
return accession_id, f"Selected: {accession_id}"
return "", "Selected row has no valid Accession ID."
except Exception as e:
return "", f"Selection error: {str(e)}"
def delete_record_from_supabase(accession_id, current_df):
if not supabase or not accession_id.strip() or accession_id == "N/A":
return "No valid record selected.", current_df, accession_id
try:
response = supabase.table("sra_database").delete().eq("Accession", accession_id.strip()).execute()
if not response.data:
response = supabase.table("geo_database").delete().eq("Accession", accession_id.strip()).execute()
if response.data:
return f"Deleted: {accession_id}", load_entire_table(), ""
return f"Record {accession_id} not found.", current_df, accession_id
except Exception as e:
return f"Deletion error: {str(e)}", current_df, accession_id
def _df_to_html_table(df):
if df is None or df.empty:
return "
No records found.
"
cols = list(df.columns)
header = "".join(f'
{c}
' for c in cols)
rows = ""
for i, row in df.iterrows():
bg = "#ffffff" if i % 2 == 0 else "#f8fafc"
cells = "".join(f'
{str(v)}
' for v in row)
rows += f'
{cells}
'
return f'''
{header}
{rows}
'''
def update_homepage_string(html_content):
log_msg = save_homepage_content(html_content)
return log_msg, generate_view_html()
# ── Reviewed-Article helpers ──────────────────────────────────────────────────
# Global flag so card-button API calls can verify admin is active
_admin_active = False
# In-memory fallback when Supabase reviewed_articles table is unavailable
_reviewed_cache: set = set()
_supabase_reviewed_ok: bool | None = None # None = not yet tested
def _check_reviewed_table():
"""Return True if the reviewed_articles table is accessible."""
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():
"""Return set of all reviewed article IDs (DOI or URL, lower-cased)."""
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()
return f"✅ Marked as reviewed"
except Exception as e:
pass
# fallback: in-memory
_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):
"""REST endpoint called from card-button fetch(). api_name='mark_article_api'."""
global _admin_active
if not _admin_active:
return "⚠ Admin login required"
article_id = (doi or "").strip().lower()
if not article_id:
return "⚠ No identifier"
if action == "mark":
return save_reviewed_article(article_id)
if action == "unmark":
return remove_reviewed_article(article_id)
return "⚠ Unknown action"
def save_reviewed_article(doi_str):
"""Mark an article DOI as reviewed (upsert)."""
doi_str = (doi_str or "").strip().lower()
if not supabase or not doi_str:
return "⚠️ Please paste a DOI first."
try:
supabase.table("reviewed_articles").upsert({"doi": doi_str}).execute()
return f"✅ Marked as reviewed: {doi_str}"
except Exception as e:
return f"❌ Error: {e}"
def remove_reviewed_article(doi_str):
"""Remove a DOI from the reviewed set."""
doi_str = (doi_str or "").strip().lower()
if not supabase or not doi_str:
return "⚠️ Please paste a DOI first."
try:
supabase.table("reviewed_articles").delete().eq("doi", doi_str).execute()
return f"🗑️ Removed from reviewed: {doi_str}"
except Exception as e:
return f"❌ Error: {e}"
def list_reviewed_html():
"""Render an HTML list of all reviewed article IDs for the admin panel."""
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
# Merge with in-memory cache
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"
"
f"✓"
f"{display}"
+ (f"{at}" if at else "") +
f"
"
)
for c in cache_only:
display = c[:80] + ('...' if len(c) > 80 else '')
html += (
f"