Create app.py

app.py

@@ -0,0 +1,200 @@
+import gradio as gr
+import time
+from datetime import datetime
+from collections import deque
+
+# --- GLOBAL SERVER STATE (In-Memory Only) ---
+# We use a simple dictionary to store the last 50 messages for active channels.
+# When the Space restarts or sleeps, all data is wiped.
+class ChatServer:
+    def __init__(self):
+        # channels = { "channel_name": deque([messages], maxlen=50) }
+        self.channels = {"Main": deque(maxlen=50)}
+        # active_users = { "username": {"last_seen": timestamp, "profile_pic": url} }
+        self.active_users = {}
+        # dms = { "user1-user2": deque(maxlen=50) }
+        self.dms = {}
+
+    def prune_users(self):
+        """Remove users who haven't polled in 30 seconds."""
+        now = time.time()
+        expired = [u for u, data in self.active_users.items() if now - data['last_seen'] > 30]
+        for u in expired:
+            del self.active_users[u]
+
+    def broadcast(self, channel, user, message, profile_pic):
+        if channel not in self.channels:
+            self.channels[channel] = deque(maxlen=50)
+        
+        msg_data = {
+            "user": user,
+            "text": message,
+            "pic": profile_pic,
+            "time": datetime.now().strftime("%H:%M")
+        }
+        self.channels[channel].append(msg_data)
+        
+    def send_dm(self, sender, recipient, message, profile_pic):
+        dm_key = "-".join(sorted([sender, recipient]))
+        if dm_key not in self.dms:
+            self.dms[dm_key] = deque(maxlen=50)
+        
+        msg_data = {
+            "user": sender,
+            "text": message,
+            "pic": profile_pic,
+            "time": datetime.now().strftime("%H:%M")
+        }
+        self.dms[dm_key].append(msg_data)
+
+server = ChatServer()
+
+# --- UTILS ---
+
+def get_messages_html(messages):
+    if not messages:
+        return "<div style='color: gray; text-align: center; margin-top: 20px;'>No messages yet. Start the conversation!</div>"
+    
+    html = "<div style='display: flex; flex-direction: column; gap: 10px;'>"
+    for m in messages:
+        html += f"""
+        <div style="display: flex; align-items: flex-start; gap: 10px;">
+            <img src="{m['pic']}" style="width: 35px; height: 35px; border-radius: 50%; border: 1px solid #ddd;">
+            <div style="background: rgba(100, 100, 100, 0.1); padding: 8px 12px; border-radius: 12px; max-width: 80%;">
+                <div style="font-weight: bold; font-size: 0.8em; margin-bottom: 2px;">{m['user']} <span style="font-weight: normal; color: gray; margin-left: 5px;">{m['time']}</span></div>
+                <div style="font-size: 0.95em;">{m['text']}</div>
+            </div>
+        </div>
+        """
+    html += "</div>"
+    return html
+
+def get_active_users_html(current_user):
+    server.prune_users()
+    html = "<div style='display: flex; flex-direction: column; gap: 5px;'>"
+    for user, data in server.active_users.items():
+        status = "🟢" if user != current_user else "👤 (You)"
+        html += f"""
+        <div style="display: flex; align-items: center; gap: 8px; font-size: 0.9em; padding: 4px;">
+            <img src="{data['pic']}" style="width: 20px; height: 20px; border-radius: 50%;">
+            <span>{user}</span>
+            <span style="font-size: 0.7em;">{status}</span>
+        </div>
+        """
+    html += "</div>"
+    return html
+
+# --- GRADIO INTERFACE ---
+
+with gr.Blocks(css="""
+    #chat-container { height: 500px; overflow-y: auto; border: 1px solid #ddd; padding: 15px; border-radius: 8px; background: var(--body-background-fill); }
+    .user-list { border-left: 1px solid #ddd; padding-left: 15px; }
+    #msg-input input { border-radius: 20px !important; }
+""") as demo:
+    
+    # HF OAuth session data
+    user_session = gr.State(None)
+    current_room = gr.State("Main")
+    dm_target = gr.State(None)
+
+    with gr.Row():
+        with gr.Column(scale=8):
+            gr.Markdown(f"# 🤗 HF Live Community Chat")
+        with gr.Column(scale=2):
+            login_btn = gr.LoginButton()
+
+    with gr.Row(visible=False) as chat_ui:
+        # Sidebar
+        with gr.Column(scale=2, variant="panel"):
+            gr.Markdown("### 🏠 Rooms")
+            room_list = gr.Radio(["Main"], value="Main", label="Join a Channel")
+            new_room_name = gr.Textbox(placeholder="Create room...", label=None, show_label=False)
+            create_room_btn = gr.Button("Create & Join", size="sm")
+            
+            gr.Markdown("### ✉️ Direct Messages")
+            dm_user_input = gr.Textbox(placeholder="Username...", label="DM Someone")
+            start_dm_btn = gr.Button("Open DM", size="sm")
+            
+            gr.Markdown("### 👥 Online Now")
+            active_users_box = gr.HTML()
+
+        # Main Chat Area
+        with gr.Column(scale=8):
+            room_title = gr.Markdown("## # Main")
+            chat_display = gr.HTML(elem_id="chat-container")
+            
+            with gr.Row():
+                msg_input = gr.Textbox(
+                    placeholder="Type a message...", 
+                    show_label=False, 
+                    scale=9,
+                    elem_id="msg-input"
+                )
+                send_btn = gr.Button("Send", scale=1, variant="primary")
+
+    # --- LOGIC ---
+
+    def on_load(profile: gr.OAuthProfile | None):
+        if profile is None:
+            return gr.update(visible=False), None
+        
+        username = profile.username
+        pic = profile.profile_image or "https://huggingface.co/front/assets/huggingface_logo-noborder.svg"
+        
+        # Register user
+        server.active_users[username] = {"last_seen": time.time(), "pic": pic}
+        return gr.update(visible=True), {"name": username, "pic": pic}
+
+    demo.load(on_load, None, [chat_ui, user_session])
+
+    def refresh_chat(session, room, dm_user):
+        if not session: return "", ""
+        
+        # Heartbeat
+        server.active_users[session['name']]['last_seen'] = time.time()
+        
+        if dm_user:
+            dm_key = "-".join(sorted([session['name'], dm_user]))
+            msgs = list(server.dms.get(dm_key, []))
+            title = f"## 💬 DM with {dm_user}"
+        else:
+            msgs = list(server.channels.get(room, []))
+            title = f"## # {room}"
+            
+        return get_messages_html(msgs), get_active_users_html(session['name']), title
+
+    timer = gr.Timer(1)
+    timer.tick(refresh_chat, [user_session, current_room, dm_target], [chat_display, active_users_box, room_title])
+
+    def send_msg(text, session, room, dm_user):
+        if not text or not session: return ""
+        
+        if dm_user:
+            server.send_dm(session['name'], dm_user, text, session['pic'])
+        else:
+            server.broadcast(room, session['name'], text, session['pic'])
+        return ""
+
+    send_btn.click(send_msg, [msg_input, user_session, current_room, dm_target], [msg_input])
+    msg_input.submit(send_msg, [msg_input, user_session, current_room, dm_target], [msg_input])
+
+    def create_and_join(name, current_options):
+        if not name: return gr.update(), "Main"
+        clean_name = name.strip().replace(" ", "-")
+        new_options = list(set(current_options + [clean_name]))
+        return gr.update(choices=new_options, value=clean_name), clean_name, None
+
+    create_room_btn.click(create_and_join, [new_room_name, room_list], [room_list, current_room, dm_target])
+
+    def switch_room(name):
+        return name, None
+    
+    room_list.change(switch_room, room_list, [current_room, dm_target])
+
+    def start_dm(target):
+        if not target: return "Main", None
+        return "Main", target
+    
+    start_dm_btn.click(start_dm, dm_user_input, [current_room, dm_target])
+
+demo.launch()