import gradio as gr import tempfile, os, json import pandas as pd from PIL import Image from scorer.au_detector import compute_au_genuineness from scorer.emotion import compute_emotion_confidence from scorer.blender import blend_score from bridge import submit_score, get_leaderboard, get_leaderboard_by_best_score, get_total_earned, get_history, get_weekly_trend from cloud_storage import upload_laugh_photo, get_recent_posts def score_laugh(image, wallet): if image is None: return json.dumps({"error": "No image provided"}, indent=2) if not wallet or not wallet.startswith("0x") or len(wallet) != 42: return json.dumps({"error": "Invalid wallet address"}, indent=2) with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp: if isinstance(image, str): tmp_path = image else: Image.fromarray(image).save(tmp.name, format="JPEG") tmp_path = tmp.name try: au_score, au_details = compute_au_genuineness(tmp_path) em_score, em_details = compute_emotion_confidence(Image.fromarray(image)) result = blend_score(au_score, em_score) result["au_details"] = au_details result["emotion_details"] = em_details score = result.get("score", 0) if score >= 50: result["payout"] = submit_score(wallet, int(score)) else: result["payout"] = {"success": False, "reason": f"Score {score} below minimum 50"} # Upload regardless of tier so "Recent Posts" reflects every attempt, # not just ones that earned a reward. Upload failure never blocks # the scoring/payout result itself — worst case, photo_url is null. tx_hash = result["payout"].get("tx_hash", "") photo_url = upload_laugh_photo(tmp_path, wallet, int(score), tx_hash) result["photo_url"] = photo_url return json.dumps(result, indent=2) finally: if not isinstance(image, str): try: os.unlink(tmp_path) except: pass def refresh_leaderboard(): data = get_leaderboard(top_n=10) if isinstance(data, dict) and "error" in data: return [[f"Error: {data['error']}", "", ""]] if not data: return [["No submissions yet", "", ""]] return [ [i + 1, row["wallet"], f"{row['total_earned_eth']:.6f} ETH"] for i, row in enumerate(data) ] def refresh_best_score_leaderboard(): data = get_leaderboard_by_best_score(top_n=10) if isinstance(data, dict) and "error" in data: return [[f"Error: {data['error']}", "", ""]] if not data: return [["No submissions yet", "", ""]] return [ [i + 1, row["wallet"], row["best_score"]] for i, row in enumerate(data) ] def compute_streak(entries): """Consecutive-day streak ending today or yesterday, derived from on-chain submission timestamps. Not stored anywhere — recomputed fresh each time from get_history(), since it's cheap and always correct rather than risking a stale cached counter. """ if not entries: return 0 import datetime days = sorted({ datetime.datetime.utcfromtimestamp(e["timestamp"]).date() for e in entries }, reverse=True) today = datetime.datetime.utcnow().date() if days[0] not in (today, today - datetime.timedelta(days=1)): return 0 # streak is broken if there's no submission today or yesterday streak = 1 for i in range(1, len(days)): if (days[i - 1] - days[i]).days == 1: streak += 1 else: break return streak def refresh_dashboard(wallet): if not wallet or not wallet.startswith("0x") or len(wallet) != 42: return "Enter a valid wallet address (0x...)", "0", [], pd.DataFrame({"day": [], "score": []}), "[]", "[]" earned = get_total_earned(wallet) if "error" in earned: return f"Error: {earned['error']}", "0", [], pd.DataFrame({"day": [], "score": []}), "[]", "[]" history = get_history(wallet) if isinstance(history, dict) and "error" in history: history = [] streak = compute_streak(history) recent = get_recent_posts(wallet, limit=3) if isinstance(recent, dict) and "error" in recent: recent = [] gallery_items = [ (post["url"], f"Score: {post.get('score', '?')}") for post in recent if post.get("url") ] # Structured version for the frontend — score as a real numeric field, # not just embedded in the gallery's display caption text. recent_posts_structured = [ { "url": post["url"], "score": int(post["score"]) if post.get("score") is not None else None, "created_at": post.get("created_at"), } for post in recent if post.get("url") ] trend = get_weekly_trend(wallet) if isinstance(trend, dict) and "error" in trend: trend = [] trend_df = pd.DataFrame( [{"day": day["day_label"], "score": day["avg_score"]} for day in trend] ) if trend else pd.DataFrame({"day": [], "score": []}) # Raw per-day data (including submission_count) for a 7-day checkmark row — # separate from trend_df since gr.BarPlot only wants the two chart columns. weekly_checkmarks = [ {"day": day["day_label"], "has_laugh": day["submission_count"] > 0} for day in trend ] summary = f"{earned['total_earned_eth']:.6f} ETH lifetime earned" return summary, str(streak), gallery_items, trend_df, json.dumps(weekly_checkmarks), json.dumps(recent_posts_structured) with gr.Blocks(title="Hasya — Laugh to Earn") as demo: gr.Markdown("# 😄 Hasya — Laugh to Earn\nUpload a genuine laugh photo. Score 50+ earns real Sepolia ETH.") with gr.Tab("Score a laugh"): with gr.Row(): img_input = gr.Image(label="Laugh photo", type="numpy") wallet_input = gr.Textbox(label="Your wallet address (0x...)", placeholder="0x65A5...") submit_btn = gr.Button("Score my laugh 🚀", variant="primary") output = gr.Textbox(label="Result", lines=20) submit_btn.click(fn=score_laugh, inputs=[img_input, wallet_input], outputs=output) with gr.Tab("Dashboard"): gr.Markdown("### 😄 Your Hasya Dashboard") dash_wallet_input = gr.Textbox(label="Your wallet address (0x...)", placeholder="0x65A5...") dash_refresh_btn = gr.Button("Load dashboard 🔄", variant="primary") with gr.Row(): earned_display = gr.Textbox(label="Happiness Balance", interactive=False) streak_display = gr.Textbox(label="Laughter Streak (days)", interactive=False) gr.Markdown("#### Recent Laughter") recent_gallery = gr.Gallery(label="Last 3 laughs", columns=3, height=200) gr.Markdown("#### Positive Affect Trend (last 7 days)") trend_chart = gr.BarPlot( x="day", y="score", x_title="", y_title="Avg score", width=500, height=250, ) # Not rendered visibly here — exists so the API exposes per-day # submission booleans for the frontend's own checkmark row UI. weekly_checkmarks_json = gr.Textbox(visible=False) recent_posts_json = gr.Textbox(visible=False) dash_refresh_btn.click( fn=refresh_dashboard, inputs=[dash_wallet_input], outputs=[earned_display, streak_display, recent_gallery, trend_chart, weekly_checkmarks_json, recent_posts_json], ) with gr.Tab("Top Earners"): gr.Markdown("### 🏆 Top Earners (cumulative)") leaderboard_table = gr.Dataframe( headers=["Rank", "Wallet", "Total Earned"], datatype=["number", "str", "str"], value=refresh_leaderboard(), ) refresh_btn = gr.Button("Refresh leaderboard 🔄") refresh_btn.click(fn=refresh_leaderboard, outputs=leaderboard_table) with gr.Tab("Best Laughs"): gr.Markdown("### 😂 Best Single Laugh Score") best_score_table = gr.Dataframe( headers=["Rank", "Wallet", "Best Score"], datatype=["number", "str", "number"], value=refresh_best_score_leaderboard(), ) refresh_best_btn = gr.Button("Refresh leaderboard 🔄") refresh_best_btn.click(fn=refresh_best_score_leaderboard, outputs=best_score_table) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)