"""蛐蛐江湖 · Cricket Commons — 千人共养的电子蛐蛐. Gradio 应用入口。D1 最小出生线: Tab1 喂养闭环 + 状态展示 + 落盘。 Tab2 日记 / Tab3 回放 在 D2 填充;Tab4 斗蛐蛐受 BATTLES_ENABLED 控制(默认关)。 """ from __future__ import annotations import hashlib import os import time import gradio as gr import brain import render import state as S import traits as T BATTLES_ENABLED = os.environ.get("BATTLES_ENABLED", "false").lower() == "true" STATE = S.load_state() S.start_scheduler() def _diary_daemon(): """每 10 分钟检查: 昨天(UTC+8)有事件但还没写日记 → 后台补写并立即持久化.""" import threading import time as _t from datetime import datetime, timedelta def loop(): while True: try: yesterday = (datetime.now(S.CST) - timedelta(days=1)).strftime("%Y-%m-%d") have = any(d.get("date") == yesterday for d in STATE["diary"]) evs = S.events_for_day(STATE, yesterday) if not have and evs: entry = brain.write_diary( yesterday, STATE["name"], STATE["traits"], STATE["sick"], evs ) with S.LOCK: if not any(d.get("date") == yesterday for d in STATE["diary"]): STATE["diary"].append(entry) S.append_event(STATE, "diary", yesterday, entry["zh"][:40]) S.save_state(STATE) S.trigger_commit() print(f"[diary] wrote {yesterday}") except Exception as e: print(f"[diary] error: {e}") _t.sleep(600) threading.Thread(target=loop, daemon=True).start() _diary_daemon() DISCLOSURE = ( "🪶 你喂的每一句话都会成为它公开一生的一部分(存档于公开 Dataset)。" "Every sentence you feed becomes part of its public life." ) def _fingerprint(request: gr.Request) -> str: ip = "" try: ip = (request.headers.get("x-forwarded-for") or "").split(",")[0].strip() except Exception: pass raw = f"{ip}|{getattr(request, 'session_hash', '')}" return hashlib.sha1(raw.encode()).hexdigest()[:16] def status_md() -> str: age = S.age_days(STATE) stage = T.molt_stage(STATE["feed_count"]) nxt = T.feeds_to_next_molt(STATE["feed_count"]) sick = "🤒 生病中" if STATE["sick"] else "😊 健康" nxt_s = f"还差 **{nxt}** 次喂养蜕皮!" if nxt else "已是最终形态" tr = " · ".join( f"{T.TRAIT_NAMES_ZH[k]} {round(STATE['traits'][k] * 10, 1)}" for k in T.TRAIT_KEYS ) return ( f"**{STATE['name']}** | 第 **{stage}** 形态 | 出生 **{age:.1f}** 天 | " f"被喂 **{STATE['feed_count']}** 次 | {sick}\n\n性格: {tr}\n\n{nxt_s}" ) def current_svg() -> str: return render.render_svg( STATE["traits"], STATE["feed_count"], STATE["sick"], STATE.get("mood", "calm") ) def feed(text: str, request: gr.Request): """喂养主流程: 流式 yield (svg, 反应文本, 状态条).""" text = (text or "").strip()[:200] if not text: yield current_svg(), "(它眨了眨眼,等着你说点什么……)", status_md() return with S.LOCK: allowed = S.check_and_count_feed(STATE, _fingerprint(request)) if not allowed: yield current_svg(), "🍽️ 它今天吃你喂的够多啦,明天再来吧(每人每日 10 句)", status_md() return stage_before = T.molt_stage(STATE["feed_count"]) reaction_text, final = "", None for ev in brain.feed_stream( text, STATE["traits"], STATE["sick"], stage_before, STATE["feed_count"] ): if ev["type"] == "partial": reaction_text = ev["reaction"] yield current_svg(), f"🦗 {reaction_text}▌", status_md() else: final = ev result = final["result"] with S.LOCK: STATE["traits"], STATE["momentum"] = T.update_traits( STATE["traits"], STATE["momentum"], result["trait_delta"] ) STATE["mood"] = result["mood"] STATE["feed_count"] += 1 sick_ch = T.sick_update(STATE, text, result["mood"])["changed"] S.append_event(STATE, "feed", text, result["reaction"]) if sick_ch: S.append_event(STATE, sick_ch, text, "") stage_after = T.molt_stage(STATE["feed_count"]) molted = stage_after > stage_before if molted: S.append_event(STATE, "molt", f"stage {stage_before}->{stage_after}", "") S.append_snapshot( STATE, render.svg_params(STATE["traits"], STATE["feed_count"], STATE["sick"], STATE["mood"]), ) S.save_state(STATE) if sick_ch or molted: S.trigger_commit() # 关键事件立即持久化 msg = f"🦗 {result['reaction']}" if molted: msg += f"\n\n✨✨ **蜕皮了!进入第 {stage_after} 形态!** ✨✨" if sick_ch == "sick": msg += "\n\n🤒 它好像被这句话喂病了……(它会记在日记里的)" if sick_ch == "recover": msg += "\n\n💚 它康复了!谢谢大家的好话。" yield current_svg(), msg, status_md() def diary_md() -> str: if not STATE["diary"]: return "📓 日记还空着——它出生不久,等它睡一觉就会写第一篇。" out = [] for d in reversed(STATE["diary"][-14:]): out.append(f"### {d['date']}\n{d['zh']}\n\n> {d.get('en', '')}") return "\n\n---\n\n".join(out) def replay(idx: int): snaps = STATE["snapshots"] if not snaps: return current_svg(), "还没有历史快照" i = max(0, min(int(idx), len(snaps) - 1)) s = snaps[i] svg = render.render_svg(s["traits"], s["feed_count"], s["sick"], s["mood"]) when = time.strftime("%m-%d %H:%M", time.localtime(s["ts"])) return svg, f"第 {i + 1}/{len(snaps)} 帧 · {when} · 第{s['stage']}形态" with gr.Blocks(title="蛐蛐江湖 · Cricket Commons") as demo: gr.Markdown("# 🦗 蛐蛐江湖 · Cricket Commons\n一只被全网共养的电子蛐蛐 · A cricket raised by the whole internet") gr.Markdown(DISCLOSURE) with gr.Tab("🏡 蛐蛐之家 Home"): svg_box = gr.HTML(current_svg()) status_box = gr.Markdown(status_md()) reaction_box = gr.Markdown("(它在等第一句话……it's waiting for your words)") inp = gr.Textbox(label="喂它一句话 Feed it a sentence", max_length=200, placeholder="夸它、逗它、教它点什么……") btn = gr.Button("🪱 喂!Feed", variant="primary") btn.click(feed, inputs=[inp], outputs=[svg_box, reaction_box, status_box], concurrency_limit=1) # 全局串行: 单模型实例 inp.submit(feed, inputs=[inp], outputs=[svg_box, reaction_box, status_box], concurrency_limit=1) with gr.Tab("📓 成长日记 Diary"): diary_box = gr.Markdown(diary_md()) gr.Button("刷新 Refresh").click(lambda: diary_md(), outputs=[diary_box]) with gr.Tab("⏪ 一生回放 Life Replay"): # 注意: min 必须 < max,新生蛐蛐 snapshots 为空时用 1 兜底 slider = gr.Slider(0, max(len(STATE["snapshots"]) - 1, 1), step=1, label="时间轴 Timeline") replay_svg = gr.HTML() replay_info = gr.Markdown() slider.change(replay, inputs=[slider], outputs=[replay_svg, replay_info]) gr.Button("刷新时间轴 Refresh").click( lambda: gr.update(maximum=max(len(STATE["snapshots"]) - 1, 1)), outputs=[slider] ) if BATTLES_ENABLED: with gr.Tab("⚔️ 斗蛐蛐 Battle"): gr.Markdown("(彩蛋施工中)") if __name__ == "__main__": demo.queue(default_concurrency_limit=1).launch()