File size: 5,830 Bytes
ee62ce4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
import os
import sys
import time
from typing import Dict
import gradio as gr

# Ensure submodules are importable
sys.path.append(os.path.dirname(__file__))

from common import PoetryModel

# Try to import sub-app factories
APPS: Dict[str, callable] = {}
try:
    from poem_generator.app import create_app as create_poem_generator
    APPS["Poem Generator"] = create_poem_generator
except Exception:
    pass

try:
    from collaborative_writer.app import create_app as create_collaborative_writer
    APPS["Collaborative Writer"] = create_collaborative_writer
except Exception:
    pass

try:
    from poem_tutor.app import create_app as create_poem_tutor
    APPS["Poem Tutor"] = create_poem_tutor
except Exception:
    pass

try:
    from vocabulary_game.app import create_app as create_vocabulary_game
    APPS["Vocabulary Game"] = create_vocabulary_game
except Exception:
    pass


def create_launcher(model: PoetryModel):
    """Create the main launcher with a simple login and rate limiting."""

    VALID_USER = os.getenv("APP_USERNAME", "demo")
    VALID_PASS = os.getenv("APP_PASSWORD", "demo123")

    # basic per-session rate limiting: max N actions per T seconds
    MAX_ACTIONS = int(os.getenv("RATE_LIMIT_MAX_ACTIONS", "30"))
    WINDOW_SEC = int(os.getenv("RATE_LIMIT_WINDOW_SEC", "300"))  # 5 minutes

    def check_login(username, password, state):
        if not isinstance(state, dict):
            state = {}
        ok = (username or "").strip() == VALID_USER and (password or "").strip() == VALID_PASS
        if ok:
            state.update({"authed": True, "actions": [], "username": username.strip()})
            return {
                login_status: gr.update(value="✅ Login successful", visible=True),
                login_row: gr.update(visible=False),
                app_tabs: gr.update(visible=True),
                session_state: state,
            }
        else:
            return {
                login_status: gr.update(value="❌ Invalid credentials", visible=True),
                session_state: state,
            }

    def rate_gate(state):
        # Called before tab loads and occasionally via a button to enforce limits
        if not isinstance(state, dict) or not state.get("authed"):
            return {
                login_row: gr.update(visible=True),
                app_tabs: gr.update(visible=False),
                login_status: gr.update(value="🔒 Please log in", visible=True),
                session_state: state if isinstance(state, dict) else {},
            }
        now = time.time()
        actions = [t for t in state.get("actions", []) if now - t < WINDOW_SEC]
        state["actions"] = actions
        if len(actions) >= MAX_ACTIONS:
            remaining = int(WINDOW_SEC - (now - actions[0])) if actions else WINDOW_SEC
            msg = f"⏱️ Rate limit exceeded. Try again in ~{remaining}s."
            return {
                limit_status: gr.update(value=msg, visible=True),
                app_tabs: gr.update(visible=False),
                session_state: state,
            }
        # allow access
        return {
            limit_status: gr.update(visible=False),
            app_tabs: gr.update(visible=True),
            session_state: state,
        }

    def record_action(state):
        if isinstance(state, dict) and state.get("authed"):
            now = time.time()
            actions = [t for t in state.get("actions", []) if now - t < WINDOW_SEC]
            actions.append(now)
            state["actions"] = actions
        return {session_state: state}

    with gr.Blocks(title="AI Poetry Apps Launcher") as demo:
        gr.Markdown("# 🎭 AI Poetry Apps Launcher")
        gr.Markdown("Login to access the apps.")

        login_status = gr.Markdown("", visible=False)
        limit_status = gr.Markdown("", visible=False)

        with gr.Row(visible=True) as login_row:
            username_in = gr.Textbox(label="Username", placeholder="Enter username", lines=1)
            password_in = gr.Textbox(label="Password", placeholder="Enter password", type="password", lines=1)
            login_btn = gr.Button("Login", variant="primary")

        session_state = gr.State({})

        with gr.Tabs(visible=False) as app_tabs:
            # Render each available sub-app as a tab
            for title, factory in APPS.items():
                with gr.Tab(title):
                    try:
                        subapp = factory(model)
                        # Add a lightweight "ping" button to count actions when user triggers generation
                        gr.Markdown("Accessing app… actions are rate-limited.")
                        ping_btn = gr.Button("Record Action")
                        ping_btn.click(record_action, inputs=[session_state], outputs=[session_state])
                        # Mount the existing Blocks app inside
                        gr.HTML("<hr>")
                        subapp.render()
                    except Exception as e:
                        gr.Markdown(f"⚠️ Failed to load {title}: {e}")

        login_btn.click(
            check_login,
            inputs=[username_in, password_in, session_state],
            outputs=[login_status, login_row, app_tabs, session_state],
        ).then(
            rate_gate,
            inputs=[session_state],
            outputs=[limit_status, app_tabs, session_state],
        )

        # Gate access on load as well
        gr.Button("Re-check Access").click(
            rate_gate,
            inputs=[session_state],
            outputs=[limit_status, app_tabs, session_state],
        )

    return demo


if __name__ == "__main__":
    from dotenv import load_dotenv
    load_dotenv()
    model = PoetryModel()
    app = create_launcher(model)
    # Spaces expects `app.py` to launch
    app.launch()