Gene Function Lab
Lab access only — enter the password to continue
"""
@app.route("/login")
def login():
return LOGIN_HTML
@app.route("/api/auth", methods=["POST"])
def api_auth():
"""Check password — called by login page JS."""
body = request.get_json(force=True) or {}
pwd = body.get("password", "").strip()
if not APP_PASSWORD:
return jsonify({"ok": True})
if pwd == APP_PASSWORD:
return jsonify({"ok": True})
return jsonify({"ok": False}), 401
# Routes
@app.route("/")
def index():
auth = require_auth()
if auth: return auth
return send_from_directory("templates", "index.html")
@app.route("/api/stats")
def api_stats():
auth = require_auth()
if auth: return auth
try:
_ensure_db_schema()
stats = database.db_stats(_local_db())
stats["last_sync"] = time.strftime(
"%Y-%m-%d %H:%M UTC", time.gmtime(drive_sync._last_sync)
) if drive_sync._last_sync else "never"
return jsonify(stats)
except Exception as e:
return jsonify({
"total_papers": 0, "total_genes": 0, "functional_papers": 0,
"top_genes": [], "all_genes": [], "last_sync": "error", "error": str(e)
})
@app.route("/api/query")
def api_query():
auth = require_auth()
if auth: return auth
genes_raw = request.args.get("genes", "").strip()
min_conf = float(request.args.get("min_conf", 0.0))
cancer_type = request.args.get("cancer_type", "all")
functional = request.args.get("functional", "all")
review_status = request.args.get("review_status", "all")
gene_sort = request.args.get("gene_sort", "selected")
paper_sort = request.args.get("paper_sort", "support_desc")
page = max(1, int(request.args.get("page", 1)))
per_page = min(100, int(request.args.get("per_page", 20)))
if not genes_raw:
return jsonify({"error": "genes parameter is required"}), 400
genes = [g.strip().upper() for g in genes_raw.split(",") if g.strip()]
try:
_ensure_db_schema()
summaries = {
g: database.gene_summary(g, min_conf=min_conf, db_path=_local_db())
for g in genes
}
sorted_genes = _sort_genes_for_display(genes, summaries, gene_sort)
result = database.query_papers(
genes=sorted_genes, cancer_type=cancer_type, functional=functional,
review_status=review_status,
min_conf=min_conf, page=page, per_page=per_page,
paper_sort=paper_sort, db_path=_local_db(),
)
result["summaries"] = summaries
result["genes_order"] = sorted_genes
return jsonify(result)
except Exception as e:
return jsonify({"error": str(e), "rows": [], "total": 0, "summaries": {}}), 200
@app.route("/api/review", methods=["POST"])
def api_review():
auth = require_auth()
if auth: return auth
body = request.get_json(force=True) or {}
gene = (body.get("gene") or "").strip().upper()
pmid = str(body.get("pmid") or "").strip()
if not gene or not pmid:
return jsonify({"error": "gene and pmid are required"}), 400
try:
_ensure_db_schema()
row = database.update_paper_review(
gene=gene,
pmid=pmid,
review_status=body.get("review_status", "unreviewed"),
review_label=body.get("review_label", ""),
review_notes=body.get("review_notes", ""),
reviewed_by=body.get("reviewed_by", request.remote_addr or ""),
db_path=_local_db(),
)
if row is None:
return jsonify({"error": "paper not found"}), 404
uploaded = False
upload_error = ""
try:
uploaded = drive_sync.upload_db_to_drive(_local_db())
except Exception as e:
upload_error = str(e)
return jsonify({"ok": True, "row": row, "uploaded": uploaded, "upload_error": upload_error})
except ValueError as e:
return jsonify({"error": str(e)}), 400
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/api/check")
def api_check():
auth = require_auth()
if auth: return auth
genes_raw = request.args.get("genes", "").strip()
if not genes_raw:
return jsonify({"error": "genes parameter required"}), 400
genes = [g.strip().upper() for g in genes_raw.split(",") if g.strip()]
result = {}
_ensure_db_schema()
for gene in genes:
in_db = database.gene_is_processed(gene, _local_db())
q_status = database.get_queue_status(gene, _local_db())
result[gene] = {
"in_db": in_db,
"queue_status": q_status,
"summary": database.gene_summary(gene, db_path=_local_db()) if in_db else None,
}
return jsonify(result)
@app.route("/api/request", methods=["POST"])
def api_request():
auth = require_auth()
if auth: return auth
body = request.get_json(force=True) or {}
gene = (body.get("gene") or "").strip().upper()
max_papers = int(body.get("max_papers", 300))
if not gene:
return jsonify({"error": "gene is required"}), 400
if not gene.replace("-","").replace("_","").isalnum() or len(gene) > 20:
return jsonify({"error": "Invalid gene name"}), 400
_ensure_db_schema()
if database.gene_is_processed(gene, _local_db()):
return jsonify({
"status": "already_processed", "gene": gene,
"summary": database.gene_summary(gene, db_path=_local_db()),
"message": f"{gene} is already in the database. Search it to see results."
})
existing = database.get_queue_status(gene, _local_db())
if existing and existing["status"] in ("pending", "processing"):
return jsonify({
"status": existing["status"], "gene": gene, "queue": existing,
"message": f"{gene} is already {existing['status']}. Check back soon."
})
try:
entry = drive_sync.write_queue_entry_to_drive(
gene=gene, requested_by=request.remote_addr or "", max_papers=max_papers
)
return jsonify({
"status": "queued", "gene": gene, "queue": entry,
"message": f"{gene} has been queued. Results will appear once the pipeline runs."
})
except Exception as e:
return jsonify({"error": f"Could not queue request: {e}"}), 500
@app.route("/api/queue")
def api_queue():
auth = require_auth()
if auth: return auth
try:
_ensure_db_schema()
queue = database.get_all_queue(_local_db())
pending = [q for q in queue if q["status"] == "pending"]
return jsonify({"queue": queue, "pending_count": len(pending)})
except Exception as e:
return jsonify({"queue": [], "pending_count": 0, "error": str(e)})
@app.route("/api/export")
def api_export():
auth = require_auth()
if auth: return auth
genes_raw = request.args.get("genes", "").strip()
min_conf = float(request.args.get("min_conf", 0.0))
cancer_type = request.args.get("cancer_type", "all")
functional = request.args.get("functional", "all")
review_status = request.args.get("review_status", "all")
gene_sort = request.args.get("gene_sort", "selected")
paper_sort = request.args.get("paper_sort", "support_desc")
genes = [g.strip().upper() for g in genes_raw.split(",") if g.strip()] if genes_raw else None
try:
_ensure_db_schema()
if genes:
summaries = {
g: database.gene_summary(g, min_conf=min_conf, db_path=_local_db())
for g in genes
}
genes = _sort_genes_for_display(genes, summaries, gene_sort)
df = database.export_to_df(
genes=genes, cancer_type=cancer_type,
functional=functional, review_status=review_status,
min_conf=min_conf, paper_sort=paper_sort, db_path=_local_db(),
)
except Exception as e:
return jsonify({"error": str(e)}), 500
buf = io.StringIO()
df.to_csv(buf, index=False)
buf.seek(0)
fname = f"functional_papers_{'_'.join(genes[:3]) if genes else 'all'}.csv"
return Response(
buf.getvalue(), mimetype="text/csv",
headers={"Content-Disposition": f"attachment; filename={fname}"}
)
@app.route("/api/export_gene_summary")
def api_export_gene_summary():
auth = require_auth()
if auth: return auth
genes_raw = request.args.get("genes", "").strip()
min_conf = float(request.args.get("min_conf", 0.0))
gene_sort = request.args.get("gene_sort", "selected")
genes = [g.strip().upper() for g in genes_raw.split(",") if g.strip()] if genes_raw else None
try:
_ensure_db_schema()
if genes:
summaries = {
g: database.gene_summary(g, min_conf=min_conf, db_path=_local_db())
for g in genes
}
genes = _sort_genes_for_display(genes, summaries, gene_sort)
df = database.export_gene_summary_to_df(
genes=genes,
min_conf=min_conf,
db_path=_local_db(),
)
except Exception as e:
return jsonify({"error": str(e)}), 500
buf = io.StringIO()
df.to_csv(buf, index=False)
buf.seek(0)
fname = f"gene_summary_{'_'.join(genes[:3]) if genes else 'all'}.csv"
return Response(
buf.getvalue(), mimetype="text/csv",
headers={"Content-Disposition": f"attachment; filename={fname}"}
)
@app.route("/api/sync", methods=["GET", "POST"])
def api_sync():
auth = require_auth()
if auth: return auth
try:
drive_sync._drive_file_id = None
synced = drive_sync.sync_db_from_drive(force=True)
_ensure_db_schema()
return jsonify({"synced": synced, "message": "DB refreshed from Drive."})
except Exception as e:
return jsonify({"synced": False, "error": str(e)}), 500
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
app.run(host="0.0.0.0", port=port, debug=False, threaded=True)