host/open-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Добавлен Gitea Manager для Proxmox (192.168.31.57:3000)

Browse Source
This commit is contained in:
host
2026-08-16 14:59:14 +03:00
parent 7f5a4e8ed2
commit 89f6345923
1 changed files with 336 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_manager_proxmox.py
+336
View File
@@ -0,0 +1,336 @@
"""
title: Gitea Manager — Proxmox
author: Claude
version: 1.3.0
description: Инструмент для управления второй Gitea (Proxmox, 192.168.31.57:3000) из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток, загрузка и безопасная распаковка ZIP-архивов.
requirements: requests
"""
import base64
import json
import shutil
import stat
import tempfile
import zipfile
from pathlib import Path, PurePosixPath
from typing import Optional
from urllib.parse import quote
import requests
from pydantic import BaseModel, Field
class Tools:
"""Инструмент управления Gitea (Proxmox) и локальными файлами рабочего пространства."""
class Valves(BaseModel):
GITEA_URL: str = Field(default="http://192.168.31.57:3000", description="Базовый URL Gitea (Proxmox) без слэша в конце.")
GITEA_TOKEN: str = Field(default="", description="API-токен Gitea. Заполните в настройках инструмента.")
TIMEOUT: int = Field(default=15, description="Таймаут запросов в секундах.")
ENABLED: bool = Field(default=True, description="Включить инструмент Gitea Manager — Proxmox.")
WORKING_DIRECTORY: str = Field(default="/tmp/gitea-manager-proxmox-workdir", description="Корневая рабочая директория.")
MAX_ARCHIVE_FILES: int = Field(default=10000, description="Максимальное количество элементов ZIP.")
MAX_ARCHIVE_SIZE_MB: int = Field(default=2048, description="Максимальный распакованный размер ZIP в МБ.")
def __init__(self):
self.valves = self.Valves()
def _check(self) -> Optional[str]:
"""Проверить доступность инструмента и наличие токена."""
if not self.valves.ENABLED:
return "Инструмент Gitea Manager — Proxmox выключен."
if not self.valves.GITEA_TOKEN:
return "API-токен Gitea не задан в настройках инструмента."
return None
def _request(self, method: str, path: str, **kwargs) -> str:
"""Выполнить запрос к Gitea (Proxmox) и вернуть JSON или сообщение об ошибке."""
error = self._check()
if error:
return error
headers = {"Authorization": f"token {self.valves.GITEA_TOKEN}", "Content-Type": "application/json"}
try:
response = requests.request(
method,
f"{self.valves.GITEA_URL.rstrip('/')}/api/v1{path}",
headers=headers,
timeout=self.valves.TIMEOUT,
**kwargs,
)
if not response.ok:
return f"Ошибка {response.status_code}: {response.text}"
return response.text or "Операция выполнена успешно."
except requests.RequestException as exc:
return f"Ошибка соединения с Gitea (Proxmox): {exc}"
def create_repository(self, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str:
"""Создать репозиторий текущего пользователя."""
return self._request("POST", "/user/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init})
def create_repository_in_org(self, org: str, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str:
"""Создать репозиторий внутри организации."""
return self._request("POST", f"/orgs/{quote(org, safe='')}/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init})
def list_repositories(
self,
limit: int = 50,
page: int = 1,
sort: str = "updated",
fields: Optional[str] = "name,private,language,html_url,description,updated_at,default_branch",
) -> str:
"""
Получить список репозиториев текущего пользователя в компактном виде.
По умолчанию возвращает не весь JSON Gitea, а только краткий набор полей —
иначе ответ раздувается из-за permissions/url/настроек и обрезается в чате.
Если репозиториев больше, чем limit, в конце добавляется подсказка про page+1.
Args:
limit: максимум репозиториев на странице (1..100).
page: номер страницы (начиная с 1).
sort: сортировка — updated | created | alpha | size | id.
fields: список полей через запятую, которые нужно оставить.
Пустая строка — вернуть весь JSON как есть.
"""
safe_limit = max(1, min(int(limit or 50), 100))
safe_page = max(1, int(page or 1))
allowed_sort = {"updated", "created", "alpha", "size", "id"}
safe_sort = sort if sort in allowed_sort else "updated"
raw = self._request(
"GET",
"/user/repos",
params={"limit": safe_limit, "page": safe_page, "sort": safe_sort},
)
# Если Gitea вернула ошибку — _request отдаёт строку "Ошибка ...".
# Парсить как JSON нечего — возвращаем как есть.
if not raw or not raw.lstrip().startswith("["):
return raw
try:
repos = json.loads(raw)
except (ValueError, TypeError):
return raw
if not isinstance(repos, list):
return raw
# Пустой fields — режим «как раньше», отдаём весь JSON.
if not fields:
return raw
keep = {f.strip() for f in fields.split(",") if f.strip()}
compact = [
{k: r.get(k) for k in keep if isinstance(r, dict) and k in r}
for r in repos
if isinstance(r, dict)
]
hint = ""
if len(repos) >= safe_limit:
hint = (
f"\n\n[Есть ещё репозитории — вызови "
f"list_repositories(limit={safe_limit}, page={safe_page + 1})]"
)
return json.dumps(compact, ensure_ascii=False, indent=2) + hint
def get_repository_info(self, owner: str, repo: str) -> str:
"""Получить информацию о репозитории."""
return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}")
def delete_repository(self, owner: str, repo: str) -> str:
"""Удалить репозиторий. Операция необратима."""
return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}")
def create_organization(self, org_name: str, description: str = "", visibility: str = "public") -> str:
"""Создать организацию."""
return self._request("POST", "/orgs", json={"username": org_name, "description": description, "visibility": visibility})
def list_organizations(self) -> str:
"""Получить организации текущего пользователя."""
return self._request("GET", "/user/orgs")
def list_files(self, owner: str, repo: str, path: str = "", branch: str = "") -> str:
"""Получить список файлов и папок."""
params = {"ref": branch} if branch else {}
return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(path, safe='/')}", params=params)
def get_file_content(self, owner: str, repo: str, file_path: str, branch: str = "") -> str:
"""Получить содержимое файла и его SHA."""
params = {"ref": branch} if branch else {}
result = self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", params=params)
try:
data = json.loads(result)
content = base64.b64decode(data["content"]).decode("utf-8", errors="replace")
return f"SHA: {data['sha']}\n\n{content}"
except (ValueError, KeyError, TypeError):
return result
def create_file(self, owner: str, repo: str, file_path: str, content: str, commit_message: str = "Добавлен файл", branch: str = "") -> str:
"""Создать файл в репозитории."""
payload = {"content": base64.b64encode(content.encode()).decode(), "message": commit_message}
if branch:
payload["branch"] = branch
return self._request("POST", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", json=payload)
def update_file(self, owner: str, repo: str, file_path: str, content: str, sha: str, commit_message: str = "Обновление файла", branch: str = "") -> str:
"""Обновить существующий файл по SHA."""
payload = {"content": base64.b64encode(content.encode()).decode(), "message": commit_message, "sha": sha}
if branch:
payload["branch"] = branch
return self._request("PUT", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", json=payload)
def delete_file(self, owner: str, repo: str, file_path: str, sha: str, commit_message: str = "Удаление файла", branch: str = "") -> str:
"""Удалить файл по SHA."""
payload = {"message": commit_message, "sha": sha}
if branch:
payload["branch"] = branch
return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}", json=payload)
def list_branches(self, owner: str, repo: str) -> str:
"""Получить список веток репозитория."""
return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches")
def download_repository_file_to_workspace(self, owner: str, repo: str, file_path: str, destination_name: str = "", branch: str = "", overwrite: bool = False) -> str:
"""Скачать файл из Gitea (Proxmox) в локальную рабочую директорию Open WebUI."""
error = self._check()
if error:
return error
try:
root = Path(self.valves.WORKING_DIRECTORY).expanduser().resolve()
root.mkdir(parents=True, exist_ok=True)
name = destination_name or Path(file_path).name
destination = (root / name).resolve()
if not self._is_inside(destination, root) or destination == root:
return "Ошибка безопасности: destination_name должен быть обычным именем внутри WORKING_DIRECTORY."
if destination.exists() and not overwrite:
return f"Ошибка: файл уже существует: {destination}. Укажи overwrite=true."
params = {"ref": branch} if branch else {}
path = f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/contents/{quote(file_path, safe='/')}"
headers = {"Authorization": f"token {self.valves.GITEA_TOKEN}"}
response = requests.get(
f"{self.valves.GITEA_URL.rstrip('/')}/api/v1{path}",
headers=headers,
params=params,
timeout=self.valves.TIMEOUT,
)
if not response.ok:
return f"Ошибка {response.status_code}: {response.text}"
data = response.json()
encoded = data.get("content", "").replace("\n", "")
raw = base64.b64decode(encoded, validate=True)
max_size = max(1, self.valves.MAX_ARCHIVE_SIZE_MB) * 1024 * 1024
if len(raw) > max_size:
return f"Ошибка безопасности: размер файла превышает лимит {self.valves.MAX_ARCHIVE_SIZE_MB} МБ."
temporary = destination.with_name(f".{destination.name}.tmp")
temporary.write_bytes(raw)
temporary.replace(destination)
return json.dumps({"status": "ok", "path": str(destination), "size": len(raw)}, ensure_ascii=False, indent=2)
except (OSError, ValueError, KeyError, requests.RequestException) as exc:
return f"Ошибка загрузки файла из Gitea (Proxmox): {exc}"
def extract_zip(self, zip_path: str, destination_dir: str = "", overwrite: bool = False) -> str:
"""Безопасно распаковать ZIP в рабочую директорию."""
try:
archive = Path(zip_path).expanduser().resolve(strict=True)
if not archive.is_file():
return "Ошибка: zip_path должен указывать на обычный файл."
working_root = Path(self.valves.WORKING_DIRECTORY).expanduser().resolve()
working_root.mkdir(parents=True, exist_ok=True)
if destination_dir:
destination = Path(destination_dir).expanduser()
if not destination.is_absolute():
destination = working_root / destination
destination = destination.resolve()
else:
destination = working_root / archive.stem
if not self._is_inside(destination, working_root):
return "Ошибка безопасности: destination_dir находится за пределами WORKING_DIRECTORY."
if destination.exists() and not overwrite:
return f"Ошибка: каталог назначения уже существует: {destination}. Укажи overwrite=true для слияния."
max_files = max(1, self.valves.MAX_ARCHIVE_FILES)
max_size = max(1, self.valves.MAX_ARCHIVE_SIZE_MB) * 1024 * 1024
total_size = 0
with zipfile.ZipFile(archive, "r") as zip_file:
members = zip_file.infolist()
if len(members) > max_files:
return f"Ошибка безопасности: архив содержит {len(members)} элементов, лимит — {max_files}."
for info in members:
relative = PurePosixPath(info.filename)
if relative.is_absolute() or ".." in relative.parts or (relative.parts and ":" in relative.parts[0]):
return f"Ошибка безопасности: запрещённый путь в архиве: {info.filename}"
mode = (info.external_attr >> 16) & 0xFFFF
if stat.S_ISLNK(mode):
return f"Ошибка безопасности: символическая ссылка в архиве запрещена: {info.filename}"
total_size += info.file_size
if total_size > max_size:
return f"Ошибка безопасности: распакованный размер превышает лимит {self.valves.MAX_ARCHIVE_SIZE_MB} МБ."
temporary_dir = Path(tempfile.mkdtemp(prefix=".zip-extract-", dir=working_root))
try:
for info in members:
relative = PurePosixPath(info.filename)
target = (temporary_dir / Path(*relative.parts)).resolve()
if not self._is_inside(target, temporary_dir):
raise ValueError(f"Запрещённый путь в архиве: {info.filename}")
if info.is_dir() or info.filename.endswith("/"):
target.mkdir(parents=True, exist_ok=True)
continue
target.parent.mkdir(parents=True, exist_ok=True)
with zip_file.open(info, "r") as source, target.open("wb") as output:
shutil.copyfileobj(source, output, length=1024 * 1024)
if destination.exists():
if not destination.is_dir() or destination.is_symlink():
return f"Ошибка: путь назначения не является безопасным каталогом: {destination}"
for item in temporary_dir.iterdir():
target = destination / item.name
if target.is_symlink():
target.unlink()
if target.exists() and target.is_dir() and item.is_dir():
self._merge_directory(item, target)
else:
if target.exists() and target.is_dir():
shutil.rmtree(target)
elif target.exists():
target.unlink()
shutil.move(str(item), str(target))
else:
temporary_dir.rename(destination)
temporary_dir = None
return json.dumps({"status": "ok", "message": "ZIP-архив успешно распакован.", "archive": str(archive), "destination": str(destination), "items": len(members), "uncompressed_size": total_size}, ensure_ascii=False, indent=2)
finally:
if temporary_dir is not None and temporary_dir.exists():
shutil.rmtree(temporary_dir, ignore_errors=True)
except (OSError, ValueError, zipfile.BadZipFile) as exc:
return f"Ошибка распаковки ZIP: {exc}"
@staticmethod
def _is_inside(path: Path, root: Path) -> bool:
"""Проверить, находится ли путь внутри корневого каталога."""
try:
path.relative_to(root)
return True
except ValueError:
return False
def _merge_directory(self, source: Path, destination: Path) -> None:
"""Рекурсивно объединить временный каталог с существующим."""
destination.mkdir(parents=True, exist_ok=True)
for item in source.iterdir():
target = destination / item.name
if target.is_symlink():
target.unlink()
if item.is_dir():
if target.exists() and not target.is_dir():
target.unlink()
self._merge_directory(item, target)
else:
if target.exists():
target.unlink()
shutil.move(str(item), str(target))