|
|
""" |
|
|
API Module |
|
|
All API endpoints for external access. |
|
|
""" |
|
|
import json |
|
|
from datetime import datetime |
|
|
|
|
|
from .config import VERSION |
|
|
from .logger import logger |
|
|
from .events import event_manager |
|
|
from .file_manager import create_folder, create_file, read_file, write_file, delete_item, rename_item |
|
|
from .system import get_system_info, get_space_info, get_storage_stats |
|
|
|
|
|
|
|
|
def ping() -> dict: |
|
|
"""Health check endpoint""" |
|
|
return {"status": "ok", "version": VERSION, "timestamp": datetime.now().isoformat()} |
|
|
|
|
|
|
|
|
def get_api_list() -> str: |
|
|
"""List all available API endpoints""" |
|
|
return json.dumps({ |
|
|
"system": ["ping()", "get_system_info()", "get_system_dashboard()", "get_space_info()", "get_storage_stats()"], |
|
|
"files": ["create_folder(name)", "create_file(name,content)", "read_file(name)", "write_file(name,content)", |
|
|
"delete_item(name)", "rename_item(old,new)", "search_files(query)"], |
|
|
"hub": ["search_hf_models(query,task,library,sort)", "download_model_file(model_id,filename)", |
|
|
"download_from_url(url,filename)", "list_downloaded_models()"], |
|
|
"events": ["api_create_event(type,path,hours,enabled,desc)", "api_delete_event(id)", |
|
|
"api_toggle_event(id)", "api_run_event(id)", "api_get_events()"], |
|
|
"logs": ["api_get_logs(type,limit,level,category)", "api_clear_logs(type)", "api_export_logs(type)"] |
|
|
}, indent=2) |
|
|
|
|
|
|
|
|
|
|
|
def api_create_event(event_type: str, target_path: str, interval_hours: float, enabled: bool, description: str) -> str: |
|
|
"""Create a scheduled event""" |
|
|
try: |
|
|
event_id = event_manager.create_event( |
|
|
event_type=event_type, |
|
|
target_path=target_path, |
|
|
interval_hours=float(interval_hours), |
|
|
enabled=enabled, |
|
|
description=description |
|
|
) |
|
|
return json.dumps({"success": True, "event_id": event_id}) |
|
|
except Exception as e: |
|
|
return json.dumps({"success": False, "error": str(e)}) |
|
|
|
|
|
|
|
|
def api_delete_event(event_id: str) -> str: |
|
|
"""Delete a scheduled event""" |
|
|
success = event_manager.delete_event(event_id) |
|
|
return json.dumps({"success": success}) |
|
|
|
|
|
|
|
|
def api_toggle_event(event_id: str) -> str: |
|
|
"""Toggle event enabled status""" |
|
|
success = event_manager.toggle_event(event_id) |
|
|
if success and event_id in event_manager.events: |
|
|
return json.dumps({"success": True, "enabled": event_manager.events[event_id].enabled}) |
|
|
return json.dumps({"success": False}) |
|
|
|
|
|
|
|
|
def api_run_event(event_id: str) -> str: |
|
|
"""Run event immediately""" |
|
|
success = event_manager.run_now(event_id) |
|
|
return json.dumps({"success": success}) |
|
|
|
|
|
|
|
|
def api_get_events() -> str: |
|
|
"""Get all scheduled events""" |
|
|
return json.dumps({"events": event_manager.get_events()}, indent=2) |
|
|
|
|
|
|
|
|
|
|
|
def api_get_logs(log_type: str = "all", limit: int = 100, level: str = "", category: str = "") -> str: |
|
|
"""Get logs""" |
|
|
logs = logger.get_logs( |
|
|
log_type=log_type, |
|
|
limit=int(limit), |
|
|
level=level if level else None, |
|
|
category=category if category else None |
|
|
) |
|
|
return json.dumps({"count": len(logs), "logs": logs}, indent=2) |
|
|
|
|
|
|
|
|
def api_clear_logs(log_type: str = "all") -> str: |
|
|
"""Clear logs""" |
|
|
success = logger.clear_logs(log_type) |
|
|
return json.dumps({"success": success}) |
|
|
|
|
|
|
|
|
def api_export_logs(log_type: str = "all") -> str: |
|
|
"""Export logs""" |
|
|
return logger.export_logs(log_type) |
|
|
|
