lollms-webui/endpoints/lollms_apps.py

314 lines
12 KiB
Python

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel, Field
from fastapi.responses import FileResponse
from lollms_webui import LOLLMSWebUI
from pydantic import BaseModel
from pathlib import Path
import shutil
import uuid
import os
import requests
import yaml
from lollms.security import check_access, sanitize_path
import os
import subprocess
import yaml
import uuid
import platform
from ascii_colors import ASCIIColors, trace_exception
import pipmaster as pm
if not pm.is_installed("httpx"):
pm.install("httpx")
import httpx
router = APIRouter()
lollmsElfServer: LOLLMSWebUI = LOLLMSWebUI.get_instance()
class AuthRequest(BaseModel):
client_id: str
class AppInfo:
def __init__(self, uid: str, name: str, folder_name: str, icon: str, category:str, description: str, author:str, version:str, model_name:str, disclaimer:str, installed: bool):
self.uid = uid
self.name = name
self.folder_name = folder_name
self.icon = icon
self.category = category
self.description = description
self.author = author
self.version = version
self.model_name = model_name
self.disclaimer = disclaimer
self.installed = installed
@router.get("/apps")
async def list_apps():
apps = []
apps_zoo_path = lollmsElfServer.lollms_paths.apps_zoo_path
for app_name in apps_zoo_path.iterdir():
if app_name.is_dir():
icon_path = app_name / "icon.png"
description_path = app_name / "description.yaml"
description = ""
author = ""
version = ""
model_name = ""
disclaimer = ""
if description_path.exists():
with open(description_path, 'r') as file:
data = yaml.safe_load(file)
application_name = data.get('name', app_name.name)
category = data.get('category', 'generic')
description = data.get('description', '')
author = data.get('author', '')
version = data.get('version', '')
model_name = data.get('model_name', '')
disclaimer = data.get('disclaimer', 'No disclaimer provided.')
installed = True
else:
installed = False
if icon_path.exists():
uid = str(uuid.uuid4())
apps.append(AppInfo(
uid=uid,
name=application_name,
folder_name = app_name.name,
icon=f"/apps/{app_name.name}/icon.png",
category=category,
description=description,
author=author,
version=version,
model_name=model_name,
disclaimer=disclaimer,
installed=installed
))
return apps
class ShowAppsFolderRequest(BaseModel):
client_id: str = Field(...)
@router.post("/show_apps_folder")
async def open_folder_in_vscode(request: ShowAppsFolderRequest):
check_access(lollmsElfServer, request.client_id)
# Get the current operating system
current_os = platform.system()
try:
if current_os == "Windows":
# For Windows
subprocess.run(['explorer', lollmsElfServer.lollms_paths.apps_zoo_path])
elif current_os == "Darwin":
# For macOS
subprocess.run(['open', lollmsElfServer.lollms_paths.apps_zoo_path])
elif current_os == "Linux":
# For Linux
subprocess.run(['xdg-open', lollmsElfServer.lollms_paths.apps_zoo_path])
else:
print("Unsupported operating system.")
except Exception as e:
print(f"An error occurred: {e}")
class OpenFolderRequest(BaseModel):
client_id: str = Field(...)
app_name: str = Field(...)
@router.post("/open_app_in_vscode")
async def open_folder_in_vscode(request: OpenFolderRequest):
check_access(lollmsElfServer, request.client_id)
sanitize_path(request.app_name)
# Construct the folder path
folder_path = lollmsElfServer.lollms_paths.apps_zoo_path/ request.app_name
# Check if the folder exists
if not folder_path.exists():
raise HTTPException(status_code=404, detail="Folder not found")
# Open the folder in VSCode
try:
os.system(f'code -n "{folder_path}"') # This assumes 'code' is in the PATH
return {"message": f"Opened {folder_path} in VSCode."}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to open folder: {str(e)}")
@router.post("/apps/{app_name}/code")
async def get_app_code(app_name: str, auth: AuthRequest):
check_access(lollmsElfServer, auth.client_id)
app_path = lollmsElfServer.lollms_paths.apps_zoo_path / app_name / "index.html"
if not app_path.exists():
raise HTTPException(status_code=404, detail="App not found")
return FileResponse(app_path)
@router.get("/apps/{app_name}/{file}")
async def get_app_file(app_name: str, file: str):
file=sanitize_path(file)
app_path = lollmsElfServer.lollms_paths.apps_zoo_path / app_name / file
if not app_path.exists():
raise HTTPException(status_code=404, detail="App file not found")
return FileResponse(app_path)
@router.post("/install/{app_name}")
async def install_app(app_name: str, auth: AuthRequest):
check_access(lollmsElfServer, auth.client_id)
REPO_DIR = lollmsElfServer.lollms_paths.personal_path/"apps_zoo_repo"
# Create the app directory
app_path = lollmsElfServer.lollms_paths.apps_zoo_path/app_name # Adjust the path as needed
os.makedirs(app_path, exist_ok=True)
# Define the local paths for the files to copy
files_to_copy = {
"icon.png": REPO_DIR/app_name/"icon.png",
"description.yaml": REPO_DIR/app_name/"description.yaml",
"index.html": REPO_DIR/app_name/"index.html"
}
# Copy each file from the local repo
for file_name, local_path in files_to_copy.items():
if local_path.exists():
with open(local_path, 'rb') as src_file:
with open(app_path/file_name, 'wb') as dest_file:
dest_file.write(src_file.read())
else:
raise HTTPException(status_code=404, detail=f"{file_name} not found in the local repository")
return {"message": f"App {app_name} installed successfully."}
@router.post("/uninstall/{app_name}")
async def uninstall_app(app_name: str, auth: AuthRequest):
app_path = lollmsElfServer.lollms_paths.apps_zoo_path / app_name
if app_path.exists():
shutil.rmtree(app_path)
return {"message": f"App {app_name} uninstalled successfully."}
else:
raise HTTPException(status_code=404, detail="App not found")
REPO_URL = "https://github.com/ParisNeo/lollms_apps_zoo.git"
class ProxyRequest(BaseModel):
url: str
@router.post("/api/proxy")
async def proxy(request: ProxyRequest):
try:
async with httpx.AsyncClient() as client:
response = await client.get(request.url)
return {"content": response.text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
def clone_repo():
REPO_DIR = Path(lollmsElfServer.lollms_paths.personal_path) / "apps_zoo_repo"
# Check if the directory exists and if it is empty
if REPO_DIR.exists():
if any(REPO_DIR.iterdir()): # Check if the directory is not empty
print(f"Directory {REPO_DIR} is not empty. Aborting clone.")
return
else:
REPO_DIR.mkdir(parents=True, exist_ok=True) # Create the directory if it doesn't exist
# Clone the repository
subprocess.run(["git", "clone", REPO_URL, str(REPO_DIR)], check=True)
print(f"Repository cloned into {REPO_DIR}")
def pull_repo():
REPO_DIR = lollmsElfServer.lollms_paths.personal_path/"apps_zoo_repo"
subprocess.run(["git", "-C", str(REPO_DIR), "pull"], check=True)
def load_apps_data():
apps = []
REPO_DIR = lollmsElfServer.lollms_paths.personal_path/"apps_zoo_repo"
for item in os.listdir(REPO_DIR):
item_path = os.path.join(REPO_DIR, item)
if os.path.isdir(item_path):
description_path = os.path.join(item_path, "description.yaml")
icon_url = f"https://github.com/ParisNeo/lollms_apps_zoo/blob/main/{item}/icon.png?raw=true"
if os.path.exists(description_path):
with open(description_path, 'r') as file:
description_data = yaml.safe_load(file)
apps.append(AppInfo(
uid=str(uuid.uuid4()),
name=description_data.get("name",item),
folder_name=item,
icon=icon_url,
category=description_data.get('category', 'generic'),
description=description_data.get('description', ''),
author=description_data.get('author', ''),
version=description_data.get('version', ''),
model_name=description_data.get('model_name', ''),
disclaimer=description_data.get('disclaimer', 'No disclaimer provided.'),
installed=True
))
return apps
@router.get("/lollms_js", response_class=PlainTextResponse)
async def lollms_js():
# Define the path to the JSON file using pathlib
file_path = Path(__file__).parent / "lollms_client_js.js"
# Read the JSON file
with file_path.open('r') as file:
data = file.read()
return data
@router.get("/lollms_markdown_renderer", response_class=PlainTextResponse)
async def lollms_markdown_renderer():
# Define the path to the JSON file using pathlib
file_path = Path(__file__).parent / "lollms_markdown_renderer.js"
# Read the JSON file
with file_path.open('r') as file:
data = file.read()
return data
@router.get("/lollms_markdown_renderer_css")
async def lollms_markdown_renderer_css():
# Define the path to the CSS file using pathlib
file_path = Path(__file__).parent / "lollms_markdown_renderer.css"
# Use FileResponse to serve the CSS file
return FileResponse(file_path, media_type="text/css")
@router.get("/template")
async def lollms_js():
return {
"start_header_id_template": lollmsElfServer.config.start_header_id_template,
"end_header_id_template": lollmsElfServer.config.end_header_id_template,
"separator_template": lollmsElfServer.config.separator_template,
"start_user_header_id_template": lollmsElfServer.config.start_user_header_id_template,
"end_user_header_id_template": lollmsElfServer.config.end_user_header_id_template,
"end_user_message_id_template": lollmsElfServer.config.end_user_message_id_template,
"start_ai_header_id_template": lollmsElfServer.config.start_ai_header_id_template,
"end_ai_header_id_template": lollmsElfServer.config.end_ai_header_id_template,
"end_ai_message_id_template": lollmsElfServer.config.end_ai_message_id_template,
"system_message_template": lollmsElfServer.config.system_message_template
}
@router.get("/github/apps")
async def fetch_github_apps():
try:
clone_repo()
pull_repo()
except:
ASCIIColors.error("Couldn't interact with ")
lollmsElfServer.error("Couldn't interact with github.\nPlease verify your internet connection")
apps = load_apps_data()
return {"apps": apps}
@router.get("/apps/{app_name}/icon")
async def get_app_icon(app_name: str):
icon_path = lollmsElfServer.lollms_paths.apps_zoo_path / app_name / "icon.png"
if not icon_path.exists():
raise HTTPException(status_code=404, detail="Icon not found")
return FileResponse(icon_path)