510 lines
17 KiB
Python
510 lines
17 KiB
Python
# moopui.py — Popup manajemen Model Option (moop) untuk TUI.
|
|
#
|
|
# Menyediakan:
|
|
# * model_selector_popup — pilih default model set tipe 'llm'
|
|
# * sets_manage_popup — CRUD model set
|
|
# * set_detail_popup — kelola type dalam sebuah set
|
|
# * set_models_popup — kelola model (provider/model) dalam sebuah set
|
|
# * providers_manage_popup — CRUD provider (moop_api)
|
|
# * provider_detail_popup — kelola API key per provider
|
|
|
|
import curses
|
|
|
|
from . import keycodes, theme
|
|
from .agent import log
|
|
from lib import moop
|
|
|
|
MODEL_TYPES = moop.MODEL_TYPES
|
|
|
|
|
|
def _mask_key(k):
|
|
if not k:
|
|
return "(empty)"
|
|
if len(k) <= 8:
|
|
return "*" * len(k)
|
|
return k[:4] + "..." + k[-4:]
|
|
|
|
|
|
def _parse_priority(text):
|
|
if text is None or text.strip() == "":
|
|
return None
|
|
try:
|
|
return int(text.strip())
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _text_input(app, stdscr, title, initial="", width=60):
|
|
pw = min(width, app.w - 4)
|
|
ph = 3
|
|
px = (app.w - pw) // 2
|
|
py = app.h // 2 - 1
|
|
|
|
win = curses.newwin(ph, pw, py, px)
|
|
win.keypad(True)
|
|
|
|
buf = initial
|
|
pos = len(buf)
|
|
max_chars = pw - 4
|
|
result = None
|
|
|
|
while True:
|
|
win.erase()
|
|
win.box()
|
|
try:
|
|
win.addstr(0, 2, f" {title} ", curses.A_BOLD)
|
|
except curses.error:
|
|
pass
|
|
start = max(0, pos - (max_chars - 1)) if pos > max_chars - 1 else 0
|
|
shown = buf[start:start + max_chars]
|
|
try:
|
|
win.addstr(1, 2, shown.ljust(max_chars))
|
|
except curses.error:
|
|
pass
|
|
try:
|
|
win.move(1, 2 + (pos - start))
|
|
except curses.error:
|
|
pass
|
|
win.refresh()
|
|
|
|
key = keycodes.read_key(stdscr, -1)
|
|
if key in (curses.KEY_ENTER, 10, 13):
|
|
result = buf
|
|
break
|
|
elif key in (27,):
|
|
break
|
|
elif key in (curses.KEY_BACKSPACE, 127):
|
|
if pos > 0:
|
|
buf = buf[:pos - 1] + buf[pos:]
|
|
pos -= 1
|
|
elif key == curses.KEY_DC:
|
|
if pos < len(buf):
|
|
buf = buf[:pos] + buf[pos + 1:]
|
|
elif key == curses.KEY_LEFT:
|
|
pos = max(0, pos - 1)
|
|
elif key == curses.KEY_RIGHT:
|
|
pos = min(len(buf), pos + 1)
|
|
elif key == curses.KEY_HOME:
|
|
pos = 0
|
|
elif key == curses.KEY_END:
|
|
pos = len(buf)
|
|
elif 32 <= key <= 255:
|
|
ch = chr(key)
|
|
buf = buf[:pos] + ch + buf[pos:]
|
|
pos += 1
|
|
|
|
del win
|
|
stdscr.touchwin()
|
|
stdscr.refresh()
|
|
if result is None:
|
|
return None
|
|
return result.strip()
|
|
|
|
|
|
def _confirm(app, stdscr, message, width=60):
|
|
pw = min(width, app.w - 4)
|
|
ph = 4
|
|
px = (app.w - pw) // 2
|
|
py = app.h // 2 - 1
|
|
|
|
win = curses.newwin(ph, pw, py, px)
|
|
win.box()
|
|
try:
|
|
win.addstr(0, 2, " Konfirmasi ", curses.A_BOLD)
|
|
except curses.error:
|
|
pass
|
|
display = message if len(message) <= pw - 4 else message[:pw - 7] + "..."
|
|
try:
|
|
win.addstr(1, 2, display)
|
|
except curses.error:
|
|
pass
|
|
try:
|
|
win.addstr(2, 2, " [Y]es [N]o ")
|
|
except curses.error:
|
|
pass
|
|
win.refresh()
|
|
|
|
result = False
|
|
while True:
|
|
key = win.getch()
|
|
if key in (ord("y"), ord("Y")):
|
|
result = True
|
|
break
|
|
elif key in (ord("n"), ord("N"), 27):
|
|
break
|
|
del win
|
|
stdscr.touchwin()
|
|
stdscr.refresh()
|
|
return result
|
|
|
|
|
|
def _run_list_popup(app, stdscr, title, items, footer="", current_idx=0, active_flags=None):
|
|
"""Popup daftar generik. Mengembalikan (idx, key).
|
|
|
|
key: KEY_ENTER / 10/13 (pilih), 27 (tutup), atau kode char aksi (a/d/e/r/dst).
|
|
items: list label; gunakan "---" untuk separator.
|
|
active_flags: list bool sejajar items untuk memberi gaya A_DIM."""
|
|
if not items:
|
|
items = ["(kosong)"]
|
|
pw = min(60, app.w - 4)
|
|
ph = min(len(items) + 4, app.h - 4)
|
|
if ph < 5:
|
|
return -1, 27
|
|
px = (app.w - pw) // 2
|
|
py = (app.h - ph) // 2
|
|
|
|
win = curses.newwin(ph, pw, py, px)
|
|
win.keypad(True)
|
|
|
|
while True:
|
|
win.erase()
|
|
win.box()
|
|
try:
|
|
win.addstr(0, 2, f" {title} ", curses.A_BOLD)
|
|
except curses.error:
|
|
pass
|
|
|
|
visible_h = ph - 2
|
|
total = len(items)
|
|
scroll = max(0, min(current_idx - visible_h // 2, total - visible_h))
|
|
|
|
for i in range(visible_h):
|
|
idx = scroll + i
|
|
if idx >= total:
|
|
break
|
|
y = 1 + i
|
|
label = items[idx]
|
|
if label == "---":
|
|
try:
|
|
win.addstr(y, 2, "\u2500" * max(0, pw - 4),
|
|
curses.color_pair(theme.C_MENU_DISABLED))
|
|
except curses.error:
|
|
pass
|
|
continue
|
|
is_cur = (idx == current_idx)
|
|
active = True
|
|
if active_flags and idx < len(active_flags):
|
|
active = bool(active_flags[idx])
|
|
if is_cur:
|
|
prefix = " \u25b6 " if active else " > "
|
|
else:
|
|
prefix = " "
|
|
if is_cur:
|
|
attr = curses.A_REVERSE
|
|
elif not active:
|
|
attr = curses.A_DIM
|
|
else:
|
|
attr = curses.A_NORMAL
|
|
display = prefix + label
|
|
if len(display) > pw - 4:
|
|
display = display[:pw - 4]
|
|
try:
|
|
win.addstr(y, 2, display.ljust(pw - 4), attr)
|
|
except curses.error:
|
|
pass
|
|
|
|
if footer:
|
|
try:
|
|
win.addstr(ph - 1, 2, footer[:pw - 4], curses.A_DIM)
|
|
except curses.error:
|
|
pass
|
|
|
|
win.refresh()
|
|
key = keycodes.read_key(stdscr, -1)
|
|
|
|
if key in (curses.KEY_UP,):
|
|
current_idx = max(0, current_idx - 1)
|
|
elif key in (curses.KEY_DOWN,):
|
|
current_idx = min(total - 1, current_idx + 1)
|
|
elif key in (27,):
|
|
del win
|
|
stdscr.touchwin()
|
|
stdscr.refresh()
|
|
return current_idx, 27
|
|
elif key in (curses.KEY_ENTER, 10, 13):
|
|
del win
|
|
stdscr.touchwin()
|
|
stdscr.refresh()
|
|
return current_idx, key
|
|
elif 32 <= key <= 255:
|
|
del win
|
|
stdscr.touchwin()
|
|
stdscr.refresh()
|
|
return current_idx, key
|
|
|
|
|
|
def _first_model_desc(set_id):
|
|
models = moop.list_models(None, set_id)
|
|
for m in models:
|
|
if m["priority"] is not None:
|
|
return f"{m['provider']} / {m['model']}"
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------- select model ---
|
|
|
|
def model_selector_popup(app, stdscr):
|
|
while True:
|
|
sets = moop.list_sets(None)
|
|
ids = []
|
|
items = []
|
|
flags = []
|
|
current = 0
|
|
default_id = moop.get_default_set_id(None, "llm")
|
|
|
|
for s in sets:
|
|
if "llm" not in s["types"]:
|
|
continue
|
|
ids.append(s["id"])
|
|
label = s["name"]
|
|
if s["id"] == default_id:
|
|
label += " [default]"
|
|
first = _first_model_desc(s["id"])
|
|
if first:
|
|
label += f" ({first})"
|
|
items.append(label)
|
|
flags.append(s["id"] == default_id)
|
|
if s["id"] == default_id:
|
|
current = len(items) - 1
|
|
|
|
if not items:
|
|
log(app, "error", "Tidak ada model set tipe 'llm'. Atur via Model > Manage Sets.")
|
|
return
|
|
|
|
idx, key = _run_list_popup(
|
|
app, stdscr, "Pilih Model Set (LLM)", items,
|
|
footer=" \u2191\u2193 nav \u21b5 select esc close ",
|
|
current_idx=current, active_flags=flags,
|
|
)
|
|
if key == 27:
|
|
return
|
|
elif key in (10, 13, curses.KEY_ENTER):
|
|
app.switch_model_set(ids[idx])
|
|
return
|
|
|
|
|
|
# ------------------------------------------------------------ manage sets ---
|
|
|
|
def sets_manage_popup(app, stdscr):
|
|
while True:
|
|
sets = moop.list_sets(None)
|
|
ids = [s["id"] for s in sets]
|
|
items = []
|
|
for s in sets:
|
|
t = ",".join(s["types"]) if s["types"] else "-"
|
|
d = " \u2605" if s["default_for"] else ""
|
|
items.append(f"{s['name']} [{t}]{d}")
|
|
|
|
idx, key = _run_list_popup(
|
|
app, stdscr, "Model Sets", items,
|
|
footer=" \u2191\u2193 nav \u21b5 detail a add r rename d delete esc close ",
|
|
)
|
|
if key == 27:
|
|
return
|
|
elif key == ord("a"):
|
|
name = _text_input(app, stdscr, "Nama model set:")
|
|
if name:
|
|
sid = moop.create_set(None, name)
|
|
log(app, "system", f"Model set '{name}' dibuat.")
|
|
if sid is not None:
|
|
set_detail_popup(app, stdscr, sid)
|
|
elif key == ord("r") and ids:
|
|
name = _text_input(app, stdscr, "Rename model set:", initial=sets[idx]["name"])
|
|
if name:
|
|
moop.rename_set(None, ids[idx], name)
|
|
log(app, "system", "Model set di-rename.")
|
|
elif key == ord("d") and ids:
|
|
if _confirm(app, stdscr, f"Hapus model set '{sets[idx]['name']}'?"):
|
|
moop.delete_set(None, ids[idx])
|
|
log(app, "system", "Model set dihapus.")
|
|
elif key in (10, 13, curses.KEY_ENTER) and ids:
|
|
set_detail_popup(app, stdscr, ids[idx])
|
|
|
|
|
|
def set_detail_popup(app, stdscr, set_id):
|
|
while True:
|
|
s = moop.get_set(None, set_id)
|
|
if not s:
|
|
return
|
|
|
|
items = []
|
|
vals = []
|
|
for t in MODEL_TYPES:
|
|
if t in s["types"]:
|
|
star = " \u2605 default" if t in s["default_for"] else ""
|
|
n = len(moop.list_models(None, set_id))
|
|
items.append(f"{t}{star} ({n} model)")
|
|
vals.append(t)
|
|
|
|
idx, key = _run_list_popup(
|
|
app, stdscr, f"Set: {s['name']}", items,
|
|
footer=" \u2191\u2193 \u21b5 models a add type d del type s default r rename esc close ",
|
|
)
|
|
if key == 27:
|
|
return
|
|
elif key == ord("a"):
|
|
available = [t for t in MODEL_TYPES if t not in s["types"]]
|
|
if not available:
|
|
log(app, "system", "Semua tipe sudah ada di set ini.")
|
|
continue
|
|
tidx, tkey = _run_list_popup(
|
|
app, stdscr, "Tambah Tipe", list(available),
|
|
footer=" \u21b5 pilih esc batal ",
|
|
)
|
|
if tkey in (10, 13, curses.KEY_ENTER):
|
|
tname = available[tidx]
|
|
moop.add_type(None, set_id, tname)
|
|
log(app, "system", f"Tipe '{tname}' ditambahkan.")
|
|
elif key == ord("s") and vals:
|
|
moop.set_default(None, vals[idx], set_id)
|
|
log(app, "system", f"Set ini menjadi default untuk '{vals[idx]}'.")
|
|
elif key == ord("d") and vals:
|
|
if _confirm(app, stdscr, f"Hapus tipe '{vals[idx]}' dari set ini?"):
|
|
moop.remove_type(None, set_id, vals[idx])
|
|
log(app, "system", f"Tipe '{vals[idx]}' dihapus dari set.")
|
|
elif key == ord("r"):
|
|
newname = _text_input(app, stdscr, "Rename model set:", initial=s["name"])
|
|
if newname:
|
|
moop.rename_set(None, set_id, newname)
|
|
elif key in (10, 13, curses.KEY_ENTER) and vals:
|
|
set_models_popup(app, stdscr, set_id, vals[idx])
|
|
|
|
|
|
def _pick_provider(app, stdscr):
|
|
apis = moop.list_apis(None)
|
|
if not apis:
|
|
log(app, "system", "Belum ada provider. Tambah via Model > Manage Providers.")
|
|
return None
|
|
items = [f"{a['name']} ({a['baseurl']})" for a in apis]
|
|
idx, key = _run_list_popup(
|
|
app, stdscr, "Pilih Provider", items,
|
|
footer=" \u2191\u2193 nav \u21b5 pilih esc batal ",
|
|
)
|
|
if key in (10, 13, curses.KEY_ENTER):
|
|
return apis[idx]["id"]
|
|
return None
|
|
|
|
|
|
def _add_model_popup(app, stdscr, set_id):
|
|
api_id = _pick_provider(app, stdscr)
|
|
if api_id is None:
|
|
return
|
|
model = _text_input(app, stdscr, "Nama model:")
|
|
if not model:
|
|
return
|
|
pri = _text_input(app, stdscr, "Priority (kosong = disabled):")
|
|
moop.add_model(None, set_id, api_id, model, _parse_priority(pri))
|
|
log(app, "system", f"Model '{model}' ditambahkan.")
|
|
|
|
|
|
def _edit_model_popup(app, stdscr, m):
|
|
model = _text_input(app, stdscr, "Nama model:", initial=m["model"])
|
|
if not model:
|
|
return
|
|
pri = _text_input(
|
|
app, stdscr, "Priority (kosong = disabled):",
|
|
initial="" if m["priority"] is None else str(m["priority"]),
|
|
)
|
|
moop.update_model(None, m["model_id"], model, _parse_priority(pri))
|
|
log(app, "system", "Model di-update.")
|
|
|
|
|
|
def set_models_popup(app, stdscr, set_id, type_name):
|
|
while True:
|
|
models = moop.list_models(None, set_id)
|
|
ids = [m["model_id"] for m in models]
|
|
items = []
|
|
for m in models:
|
|
pri = str(m["priority"]) if m["priority"] is not None else "off"
|
|
items.append(f"{m['provider']} / {m['model']} (pri {pri})")
|
|
|
|
idx, key = _run_list_popup(
|
|
app, stdscr, f"Models ({type_name})", items,
|
|
footer=" \u2191\u2193 a add e edit d delete esc close ",
|
|
)
|
|
if key == 27:
|
|
return
|
|
elif key == ord("a"):
|
|
_add_model_popup(app, stdscr, set_id)
|
|
elif key == ord("e") and ids:
|
|
_edit_model_popup(app, stdscr, models[idx])
|
|
elif key == ord("d") and ids:
|
|
if _confirm(app, stdscr, f"Hapus model '{models[idx]['model']}'?"):
|
|
moop.delete_model(None, ids[idx])
|
|
log(app, "system", "Model dihapus.")
|
|
|
|
|
|
# ---------------------------------------------------------- manage provider ---
|
|
|
|
def providers_manage_popup(app, stdscr):
|
|
while True:
|
|
apis = moop.list_apis(None)
|
|
ids = [a["id"] for a in apis]
|
|
items = [f"{a['name']} ({a['baseurl']}) keys:{a['key_count']}" for a in apis]
|
|
|
|
idx, key = _run_list_popup(
|
|
app, stdscr, "Providers (API)", items,
|
|
footer=" \u2191\u2193 \u21b5 keys a add e edit d delete esc close ",
|
|
)
|
|
if key == 27:
|
|
return
|
|
elif key == ord("a"):
|
|
name = _text_input(app, stdscr, "Nama provider:")
|
|
if not name:
|
|
continue
|
|
baseurl = _text_input(app, stdscr, "Base URL:")
|
|
if baseurl:
|
|
moop.create_api(None, name, baseurl)
|
|
log(app, "system", f"Provider '{name}' ditambahkan.")
|
|
elif key == ord("e") and ids:
|
|
name = _text_input(app, stdscr, "Nama provider:", initial=apis[idx]["name"])
|
|
baseurl = _text_input(app, stdscr, "Base URL:", initial=apis[idx]["baseurl"])
|
|
if name and baseurl:
|
|
moop.update_api(None, ids[idx], name, baseurl)
|
|
log(app, "system", "Provider di-update.")
|
|
elif key == ord("d") and ids:
|
|
if _confirm(app, stdscr,
|
|
f"Hapus provider '{apis[idx]['name']}'?\n(keys & model terkait ikut terhapus)"):
|
|
moop.delete_api(None, ids[idx])
|
|
log(app, "system", "Provider dihapus.")
|
|
elif key in (10, 13, curses.KEY_ENTER) and ids:
|
|
provider_detail_popup(app, stdscr, ids[idx])
|
|
|
|
|
|
def provider_detail_popup(app, stdscr, api_id):
|
|
api = moop.get_api(None, api_id)
|
|
if not api:
|
|
return
|
|
while True:
|
|
keys = moop.list_keys(None, api_id)
|
|
ids = [k["id"] for k in keys]
|
|
items = []
|
|
for k in keys:
|
|
pri = str(k["priority"]) if k["priority"] is not None else "off"
|
|
items.append(f"{_mask_key(k['key'])} (pri {pri})")
|
|
|
|
idx, key = _run_list_popup(
|
|
app, stdscr, f"API Keys: {api['name']}", items,
|
|
footer=" \u2191\u2193 a add e edit d delete esc close ",
|
|
)
|
|
if key == 27:
|
|
return
|
|
elif key == ord("a"):
|
|
k = _text_input(app, stdscr, "API Key:")
|
|
if k:
|
|
pri = _text_input(app, stdscr, "Priority (kosong = disabled):")
|
|
moop.add_key(None, api_id, k, _parse_priority(pri))
|
|
log(app, "system", "API key ditambahkan.")
|
|
elif key == ord("e") and ids:
|
|
k = _text_input(app, stdscr, "API Key:", initial=keys[idx]["key"])
|
|
pri = _text_input(
|
|
app, stdscr, "Priority (kosong = disabled):",
|
|
initial="" if keys[idx]["priority"] is None else str(keys[idx]["priority"]),
|
|
)
|
|
if k:
|
|
moop.update_key(None, ids[idx], k, _parse_priority(pri))
|
|
log(app, "system", "API key di-update.")
|
|
elif key == ord("d") and ids:
|
|
if _confirm(app, stdscr, "Hapus API key ini?"):
|
|
moop.delete_key(None, ids[idx])
|