Initial version

app.py

@@ -0,0 +1,183 @@
+import gradio as gr
+import os
+import threading
+from datetime import datetime, timedelta
+from huggingface_hub import InferenceClient
+
+ROWS = [chr(i) for i in range(ord("A"), ord("Z") + 1)]
+SEATS_PER_ROW = 26
+SECTIONS = {
+    "left": range(1, 9),
+    "middle": range(9, 19),
+    "right": range(19, 27),
+}
+
+lock = threading.Lock()
+
+seat_state = {
+    row: {seat: "available" for seat in range(1, 27)}
+    for row in ROWS
+}
+
+holds = {}
+bookings = {}
+lock_owner = None
+lock_expires_at = None
+
+client = InferenceClient(
+    base_url="https://router.huggingface.co/v1",
+    token=os.getenv("HF_TOKEN")
+)
+
+def seat_label(row, seat):
+    return f"{row}{seat}"
+
+def section_for_seat(seat):
+    if 1 <= seat <= 8:
+        return "left"
+    if 9 <= seat <= 18:
+        return "middle"
+    return "right"
+
+def available_seats():
+    out = []
+    for row in ROWS:
+        for seat in range(1, 27):
+            if seat_state[row][seat] == "available":
+                out.append(seat_label(row, seat))
+    return out
+
+def find_best_block(n):
+    for row in ROWS:
+        for section_name, seats in SECTIONS.items():
+            seats = list(seats)
+            free_runs = []
+            current = []
+            for s in seats:
+                if seat_state[row][s] == "available":
+                    current.append(s)
+                else:
+                    if current:
+                        free_runs.append(current)
+                    current = []
+            if current:
+                free_runs.append(current)
+
+            for run in free_runs:
+                if len(run) >= n:
+                    return [seat_label(row, s) for s in run[:n]]
+    return None
+
+def acquire_lock(user_id):
+    global lock_owner, lock_expires_at
+    now = datetime.utcnow()
+    if lock_owner and lock_expires_at and now < lock_expires_at and lock_owner != user_id:
+        return False, f"Locked by {lock_owner}"
+    lock_owner = user_id
+    lock_expires_at = now + timedelta(minutes=10)
+    return True, f"Lock granted to {user_id}"
+
+def release_lock(user_id):
+    global lock_owner, lock_expires_at
+    if lock_owner == user_id:
+        lock_owner = None
+        lock_expires_at = None
+        return True
+    return False
+
+def hold_seats(user_id, count):
+    with lock:
+        ok, msg = acquire_lock(user_id)
+        if not ok:
+            return msg, ""
+        seats = find_best_block(count)
+        if not seats:
+            return "Not enough adjacent seats available.", ""
+        hold_id = f"H{len(holds)+1}"
+        for seat in seats:
+            row = seat[0]
+            num = int(seat[1:])
+            seat_state[row][num] = "held"
+        holds[hold_id] = {
+            "user_id": user_id,
+            "seats": seats,
+            "expires_at": datetime.utcnow() + timedelta(minutes=5),
+        }
+        return f"Hold created: {hold_id}", ", ".join(seats)
+
+def confirm_hold(user_id, hold_id):
+    with lock:
+        h = holds.get(hold_id)
+        if not h:
+            return "Hold not found."
+        if h["user_id"] != user_id:
+            return "You do not own this hold."
+        for seat in h["seats"]:
+            row = seat[0]
+            num = int(seat[1:])
+            seat_state[row][num] = "booked"
+        booking_id = f"B{len(bookings)+1}"
+        bookings[booking_id] = h
+        del holds[hold_id]
+        release_lock(user_id)
+        return f"Booked successfully: {booking_id}"
+
+def chat_with_ai(message, history):
+    messages = [{"role": "system", "content": "You are a ticketing assistant that helps users book venue seats."}]
+    for item in history:
+        messages.append({"role": "user", "content": item[0]})
+        messages.append({"role": "assistant", "content": item[1]})
+    messages.append({"role": "user", "content": message})
+
+    if os.getenv("HF_TOKEN"):
+        resp = client.chat.completions.create(
+            model="meta-llama/Llama-3.1-8B-Instruct",
+            messages=messages,
+            max_tokens=200,
+            temperature=0.2,
+        )
+        return resp.choices[0].message.content
+
+    return "HF_TOKEN is not configured."
+
+def render_map():
+    rows = []
+    for row in ROWS:
+        vals = []
+        for seat in range(1, 27):
+            status = seat_state[row][seat]
+            vals.append(f"{seat_label(row, seat)}:{status[0]}")
+        rows.append(" | ".join(vals))
+    return "\n".join(rows)
+
+def ui_action(user_id, group_size):
+    hold_msg, seats = hold_seats(user_id, int(group_size))
+    return render_map(), hold_msg, seats
+
+def confirm_action(user_id, hold_id):
+    return render_map(), confirm_hold(user_id, hold_id)
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Venue Ticketing System")
+
+    with gr.Row():
+        user_id = gr.Textbox(label="User ID", value="user1")
+        group_size = gr.Number(label="Seats needed", value=2, precision=0)
+
+    with gr.Row():
+        btn_hold = gr.Button("Hold Best Seats")
+        hold_id = gr.Textbox(label="Hold ID")
+        btn_confirm = gr.Button("Confirm Hold")
+
+    seat_map = gr.Textbox(label="Seat map", lines=18)
+    hold_result = gr.Textbox(label="Hold result")
+    booked_result = gr.Textbox(label="Booking result")
+    ai_chat = gr.ChatInterface(fn=chat_with_ai, type="messages", title="AI Assistant")
+
+    btn_hold.click(ui_action, inputs=[user_id, group_size], outputs=[seat_map, hold_result, hold_id])
+    btn_confirm.click(confirm_action, inputs=[user_id, hold_id], outputs=[seat_map, booked_result])
+
+    demo.load(render_map, outputs=seat_map)
+    demo.queue(default_concurrency_limit=1)
+
+demo.launch()