minchyeom commited on
Commit
3fb65e6
·
1 Parent(s): 97678da

refactor: remove frontend.py file and associated Gradio UI code

Browse files
Files changed (1) hide show
  1. frontend.py +0 -235
frontend.py DELETED
@@ -1,235 +0,0 @@
1
- """Gradio UI for llm-backend."""
2
-
3
- from __future__ import annotations
4
-
5
- import asyncio
6
- import os
7
- from pathlib import Path
8
- from typing import Iterator, List, Optional
9
- import typing
10
-
11
- import gradio as gr
12
- from huggingface_hub import HfApi, login
13
-
14
- from src.config import UPLOAD_DIR
15
- from src.db import list_sessions_info, reset_history
16
- from src.team import TeamChatSession
17
-
18
-
19
- # ---------------------------------------------------------------------------
20
- # Authentication helpers
21
- # ---------------------------------------------------------------------------
22
-
23
- def hf_login(token: str) -> str:
24
- """Login to HuggingFace and return the username."""
25
- login(token=token, new_session=True)
26
- info = HfApi().whoami(token=token)
27
- return info.get("name") or info.get("email")
28
-
29
-
30
- def get_env_token() -> str | None:
31
- """Return an available HuggingFace token from environment variables."""
32
- for name in (
33
- "HF_TOKEN",
34
- "HF_SPACE_TOKEN",
35
- "HF_API_TOKEN",
36
- "HF_ACCESS_TOKEN",
37
- ):
38
- if token := os.getenv(name):
39
- return token
40
- return None
41
-
42
-
43
- # ---------------------------------------------------------------------------
44
- # File helpers
45
- # ---------------------------------------------------------------------------
46
-
47
- def _vm_host_path(user: str, vm_path: str) -> Path:
48
- rel = Path(vm_path).relative_to("/data")
49
- base = Path(UPLOAD_DIR) / user
50
- target = (base / rel).resolve()
51
- if not str(target).startswith(str(base)):
52
- raise ValueError("Invalid path")
53
- return target
54
-
55
-
56
- def list_directory(user: str, path: str) -> list[dict[str, str | bool]]:
57
- target = _vm_host_path(user, path)
58
- if not target.exists() or not target.is_dir():
59
- raise FileNotFoundError(path)
60
- entries: list[dict[str, str | bool]] = []
61
- for entry in sorted(target.iterdir()):
62
- entries.append({"name": entry.name, "is_dir": entry.is_dir()})
63
- return entries
64
-
65
-
66
- def read_file(user: str, path: str) -> str:
67
- target = _vm_host_path(user, path)
68
- if not target.exists() or target.is_dir():
69
- raise FileNotFoundError(path)
70
- return target.read_text()
71
-
72
-
73
- def write_file(user: str, path: str, content: str) -> None:
74
- target = _vm_host_path(user, path)
75
- target.parent.mkdir(parents=True, exist_ok=True)
76
- target.write_text(content)
77
-
78
-
79
- def delete_path(user: str, path: str) -> None:
80
- target = _vm_host_path(user, path)
81
- if target.is_dir():
82
- for child in target.iterdir():
83
- if child.is_file():
84
- child.unlink()
85
- else:
86
- delete_path(user, str(Path("/data") / child.relative_to(target)))
87
- target.rmdir()
88
- elif target.exists():
89
- target.unlink()
90
-
91
-
92
- # ---------------------------------------------------------------------------
93
- # Chat helpers
94
- # ---------------------------------------------------------------------------
95
-
96
- async def chat_generator(user: str, session: str, prompt: str) -> Iterator[List[List[str]]]:
97
- history: List[List[str]] = []
98
- async with TeamChatSession(user=user, session=session) as chat:
99
- history.append([prompt, ""])
100
- resp = ""
101
- async for part in chat.chat_stream(prompt):
102
- resp += part
103
- history[-1][1] = resp
104
- yield history
105
-
106
-
107
- # ---------------------------------------------------------------------------
108
- # Gradio callbacks
109
- # ---------------------------------------------------------------------------
110
-
111
- async def send_message(user: str, session: str, message: str, history: list[list[str]] | None) -> typing.Iterator[tuple[list[list[str]], str]]:
112
- history = history or []
113
- async with TeamChatSession(user=user, session=session) as chat:
114
- history.append([message, ""])
115
- resp = ""
116
- async for part in chat.chat_stream(message):
117
- resp += part
118
- history[-1][1] = resp
119
- yield history, ""
120
-
121
- def load_sessions(user: str) -> list[str]:
122
- info = list_sessions_info(user)
123
- return [s["name"] for s in info]
124
-
125
-
126
- def remove_session(user: str, session: str) -> None:
127
- reset_history(user, session)
128
-
129
-
130
- # ---------------------------------------------------------------------------
131
- # UI construction
132
- # ---------------------------------------------------------------------------
133
-
134
- def build_ui() -> gr.Blocks:
135
- token = get_env_token()
136
- auto_user: str | None = None
137
- auto_sessions: list[str] = []
138
- if token:
139
- try:
140
- auto_user = hf_login(token)
141
- auto_sessions = load_sessions(auto_user)
142
- except Exception as exc: # pragma: no cover - best effort
143
- print(f"Automatic login failed: {exc}")
144
-
145
- with gr.Blocks(theme=gr.themes.Soft()) as demo:
146
- user_state = gr.State(auto_user or str)
147
- session_state = gr.State("default")
148
- history_state = gr.State([])
149
-
150
- with gr.Column(visible=auto_user is None) as login_col:
151
- gr.Markdown("## HuggingFace Login")
152
- token_box = gr.Textbox(type="password", label="HuggingFace token")
153
- login_btn = gr.Button("Login")
154
- login_status = gr.Markdown()
155
-
156
- with gr.Row(visible=auto_user is not None) as main_row:
157
- with gr.Column(scale=3):
158
- session_drop = gr.Dropdown(
159
- label="Session",
160
- interactive=True,
161
- choices=auto_sessions,
162
- value=auto_sessions[0] if auto_sessions else None,
163
- )
164
- new_session = gr.Textbox(label="New Session Name")
165
- create_btn = gr.Button("Create Session")
166
- chatbox = gr.Chatbot(type="messages")
167
- msg = gr.Textbox(label="Message")
168
- send_btn = gr.Button("Send")
169
-
170
- with gr.Column(scale=2):
171
- gr.Markdown("### Files")
172
- dir_path = gr.Textbox(value="/data", label="Path")
173
- refresh_btn = gr.Button("List")
174
- file_list = gr.Dataframe(headers=["Name", "Is Dir"], datatype=["str", "bool"], interactive=False)
175
- open_path = gr.Textbox(label="File Path")
176
- open_btn = gr.Button("Open")
177
- file_editor = gr.Textbox(label="Content", lines=10)
178
- save_btn = gr.Button("Save")
179
- delete_btn = gr.Button("Delete")
180
-
181
- def do_login(token: str):
182
- user = hf_login(token)
183
- sessions = load_sessions(user)
184
- return {
185
- user_state: user,
186
- session_drop: gr.update(choices=sessions),
187
- login_col: gr.update(visible=False),
188
- main_row: gr.update(visible=True),
189
- }
190
-
191
- login_btn.click(do_login, inputs=token_box, outputs=[user_state, session_drop, login_col, main_row])
192
-
193
- def create_session(user: str, name: str):
194
- if not name:
195
- return gr.update()
196
- reset_history(user, name)
197
- sessions = load_sessions(user)
198
- return gr.update(value=name, choices=sessions)
199
-
200
- create_btn.click(create_session, inputs=[user_state, new_session], outputs=session_drop)
201
-
202
- def refresh_files(user: str, path: str):
203
- entries = list_directory(user, path)
204
- data = [[e["name"], e["is_dir"]] for e in entries]
205
- return {file_list: gr.update(value=data), dir_path: path}
206
-
207
- refresh_btn.click(refresh_files, inputs=[user_state, dir_path], outputs=[file_list, dir_path])
208
-
209
- def open_file(user: str, path: str):
210
- content = read_file(user, path)
211
- return {file_editor: content, open_path: path}
212
-
213
- open_btn.click(open_file, inputs=[user_state, open_path], outputs=[file_editor, open_path])
214
-
215
- def save_file(user: str, path: str, content: str):
216
- write_file(user, path, content)
217
- return gr.update()
218
-
219
- save_btn.click(save_file, inputs=[user_state, open_path, file_editor], outputs=file_editor)
220
-
221
- def delete_file(user: str, path: str):
222
- delete_path(user, path)
223
- return gr.update(value="")
224
-
225
- delete_btn.click(delete_file, inputs=[user_state, open_path], outputs=open_path)
226
-
227
- send_btn.click(send_message, inputs=[user_state, session_drop, msg, history_state], outputs=[chatbox, history_state, msg])
228
- msg.submit(send_message, inputs=[user_state, session_drop, msg, history_state], outputs=[chatbox, history_state, msg])
229
-
230
- return demo
231
-
232
-
233
- if __name__ == "__main__":
234
- ui = build_ui()
235
- ui.launch()