|
|
""" |
|
|
UI Helpers Module |
|
|
Helper functions for Gradio UI interactions. |
|
|
""" |
|
|
import json |
|
|
import gradio as gr |
|
|
|
|
|
from .file_manager import ( |
|
|
get_folder_contents, get_breadcrumb, get_status, get_quick_access_folders, |
|
|
create_folder, create_file, delete_item, rename_item, write_file, |
|
|
navigate_to_folder, get_file_preview |
|
|
) |
|
|
from .events import event_manager |
|
|
from .logger import logger |
|
|
from .api import api_create_event, api_delete_event, api_toggle_event, api_run_event |
|
|
|
|
|
|
|
|
def ui_result(result_str: str) -> str: |
|
|
"""Parse result JSON and return status message""" |
|
|
try: |
|
|
r = json.loads(result_str) |
|
|
return "β
Success" if r.get("success") else f"β {r.get('error', 'Error')}" |
|
|
except: |
|
|
return result_str |
|
|
|
|
|
|
|
|
def ui_create_folder(name): |
|
|
result = create_folder(name) |
|
|
return get_folder_contents(), get_breadcrumb(), ui_result(result), get_quick_access_folders() |
|
|
|
|
|
|
|
|
def ui_create_file(name, content): |
|
|
result = create_file(name, content) |
|
|
return get_folder_contents(), get_breadcrumb(), ui_result(result), get_quick_access_folders() |
|
|
|
|
|
|
|
|
def ui_delete(name): |
|
|
if not name: |
|
|
return get_folder_contents(), get_breadcrumb(), "β οΈ Select item", get_quick_access_folders() |
|
|
result = delete_item(name) |
|
|
return get_folder_contents(), get_breadcrumb(), ui_result(result), get_quick_access_folders() |
|
|
|
|
|
|
|
|
def ui_rename(old, new): |
|
|
result = rename_item(old, new) |
|
|
return get_folder_contents(), get_breadcrumb(), ui_result(result), get_quick_access_folders() |
|
|
|
|
|
|
|
|
def ui_save(name, content): |
|
|
if not name: |
|
|
return "β οΈ Select file" |
|
|
result = write_file(name, content) |
|
|
return ui_result(result) |
|
|
|
|
|
|
|
|
def on_folder_tree_select(evt: gr.SelectData, data): |
|
|
"""Handle folder tree selection""" |
|
|
if evt.index[0] < len(data): |
|
|
folder_path = data[evt.index[0]][2] |
|
|
return navigate_to_folder(folder_path) |
|
|
return get_folder_contents(), get_breadcrumb(), get_status(), get_quick_access_folders() |
|
|
|
|
|
|
|
|
def navigate_from_main(evt: gr.SelectData, data): |
|
|
"""Handle click on main content - navigate into folder or show preview""" |
|
|
from .file_manager import current_path |
|
|
import modules.file_manager as fm |
|
|
|
|
|
if evt.index[0] < len(data): |
|
|
name = data[evt.index[0]][1] |
|
|
item_type = data[evt.index[0]][2] |
|
|
|
|
|
if item_type == "Folder": |
|
|
new_path = fm.current_path / name |
|
|
if new_path.exists() and new_path.is_dir(): |
|
|
fm.current_path = new_path |
|
|
return ( |
|
|
get_folder_contents(), |
|
|
get_breadcrumb(), |
|
|
get_status(), |
|
|
get_quick_access_folders(), |
|
|
name, |
|
|
None, "", "" |
|
|
) |
|
|
else: |
|
|
img, txt, info = get_file_preview(name) |
|
|
return ( |
|
|
data, |
|
|
get_breadcrumb(), |
|
|
get_status(), |
|
|
get_quick_access_folders(), |
|
|
name, |
|
|
img, txt or "", info |
|
|
) |
|
|
|
|
|
return (data, get_breadcrumb(), get_status(), get_quick_access_folders(), "", None, "", "") |
|
|
|
|
|
|
|
|
|
|
|
def get_events_table(): |
|
|
events = event_manager.get_events() |
|
|
rows = [] |
|
|
for e in events: |
|
|
rows.append([ |
|
|
e["event_id"], |
|
|
e["event_type"], |
|
|
e["target_path"][:30] + "..." if len(e["target_path"]) > 30 else e["target_path"], |
|
|
f"{e['interval_hours']}h", |
|
|
"β
" if e["enabled"] else "β", |
|
|
e.get("last_run", "Never")[:16] if e.get("last_run") else "Never", |
|
|
str(e["run_count"]) |
|
|
]) |
|
|
return rows |
|
|
|
|
|
|
|
|
def get_logs_display(log_type: str = "all", limit: int = 50): |
|
|
logs = logger.get_logs(log_type, limit) |
|
|
if not logs: |
|
|
return "No logs found" |
|
|
|
|
|
lines = [] |
|
|
for log in logs: |
|
|
level_icon = {"INFO": "βΉοΈ", "ERROR": "β", "WARNING": "β οΈ", "EVENT": "π", "DEBUG": "π"}.get(log["level"], "β’") |
|
|
lines.append(f"[{log['timestamp'][:19]}] {level_icon} [{log['category']}] {log['message']}") |
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
def ui_create_event(event_type, target_path, interval_hours, enabled, description): |
|
|
result = api_create_event(event_type, target_path, interval_hours, enabled, description) |
|
|
return get_events_table(), ui_result(result) |
|
|
|
|
|
|
|
|
def ui_delete_event(event_id): |
|
|
result = api_delete_event(event_id) |
|
|
return get_events_table(), ui_result(result) |
|
|
|
|
|
|
|
|
def ui_toggle_event(event_id): |
|
|
result = api_toggle_event(event_id) |
|
|
return get_events_table(), ui_result(result) |
|
|
|
|
|
|
|
|
def ui_run_event(event_id): |
|
|
result = api_run_event(event_id) |
|
|
return get_events_table(), get_logs_display("events", 20), ui_result(result) |
|
|
|
|
|
|
|
|
def ui_refresh_logs(log_type, limit): |
|
|
return get_logs_display(log_type, int(limit)) |
|
|
|
|
|
|
|
|
def ui_clear_logs(log_type): |
|
|
from .api import api_clear_logs |
|
|
api_clear_logs(log_type) |
|
|
return get_logs_display(log_type, 50), "β
Logs cleared" |
|
|
|
|
|
|
|
|
def ui_copy_logs(log_type): |
|
|
return logger.export_logs(log_type) |
|
|
|