Удалить gitea_manager_proxmox.py
This commit is contained in:
host committed
1 parent
8a30cc8229
commit
35417d2622
1 file changed
-337
@@ -1,337 +0,0 @@
|
|||||||
"""
|
|
||||||
title: Gitea Manager (Proxmox)
|
|
||||||
author: Claude
|
|
||||||
version: 1.4.0
|
|
||||||
description: Инструмент для управления Gitea на Proxmox (192.168.31.57:3000) из Open WebUI — создание и удаление репозиториев, организаций, файлов, веток, загрузка и безопасная распаковка ZIP-архивов. Все методы имеют суффикс _proxmox, чтобы не путаться с инструментом для основной Gitea.
|
|
||||||
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 (192.168.31.57:3000) и локальными файлами рабочего пространства.
|
|
||||||
|
|
||||||
ВАЖНО: это инструмент для Gitea на Proxmox. Для основного сервера Gitea (192.168.31.100)
|
|
||||||
используется отдельный инструмент "Gitea Manager (Main)" с суффиксом _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 на Proxmox. Заполните в настройках инструмента.")
|
|
||||||
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]:
|
|
||||||
"""Проверить доступность инструмента и наличие токена (Gitea на Proxmox)."""
|
|
||||||
if not self.valves.ENABLED:
|
|
||||||
return "Инструмент Gitea Manager (Proxmox) выключен."
|
|
||||||
if not self.valves.GITEA_TOKEN:
|
|
||||||
return "API-токен Gitea на Proxmox не задан в настройках инструмента."
|
|
||||||
return None
|
|
||||||
|
|
||||||
def _request(self, method: str, path: str, **kwargs) -> str:
|
|
||||||
"""Выполнить запрос к Gitea на Proxmox (192.168.31.57) и вернуть 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 (192.168.31.57): {exc}"
|
|
||||||
|
|
||||||
def create_repository_proxmox(self, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str:
|
|
||||||
"""Создать репозиторий текущего пользователя на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
return self._request("POST", "/user/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init})
|
|
||||||
|
|
||||||
def create_repository_in_org_proxmox(self, org: str, name: str, description: str = "", private: bool = False, auto_init: bool = True) -> str:
|
|
||||||
"""Создать репозиторий внутри организации на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
return self._request("POST", f"/orgs/{quote(org, safe='')}/repos", json={"name": name, "description": description, "private": private, "auto_init": auto_init})
|
|
||||||
|
|
||||||
def list_repositories_proxmox(
|
|
||||||
self,
|
|
||||||
limit: int = 50,
|
|
||||||
page: int = 1,
|
|
||||||
sort: str = "updated",
|
|
||||||
fields: Optional[str] = "name,private,language,html_url,description,updated_at,default_branch",
|
|
||||||
) -> str:
|
|
||||||
"""
|
|
||||||
Получить список репозиториев текущего пользователя на Gitea на PROXMOX (192.168.31.57) в компактном виде.
|
|
||||||
|
|
||||||
По умолчанию возвращает не весь 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},
|
|
||||||
)
|
|
||||||
|
|
||||||
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
|
|
||||||
|
|
||||||
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_proxmox(limit={safe_limit}, page={safe_page + 1})]"
|
|
||||||
)
|
|
||||||
return json.dumps(compact, ensure_ascii=False, indent=2) + hint
|
|
||||||
|
|
||||||
def get_repository_info_proxmox(self, owner: str, repo: str) -> str:
|
|
||||||
"""Получить информацию о репозитории на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}")
|
|
||||||
|
|
||||||
def delete_repository_proxmox(self, owner: str, repo: str) -> str:
|
|
||||||
"""Удалить репозиторий на Gitea на PROXMOX (192.168.31.57). Операция необратима."""
|
|
||||||
return self._request("DELETE", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}")
|
|
||||||
|
|
||||||
def create_organization_proxmox(self, org_name: str, description: str = "", visibility: str = "public") -> str:
|
|
||||||
"""Создать организацию на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
return self._request("POST", "/orgs", json={"username": org_name, "description": description, "visibility": visibility})
|
|
||||||
|
|
||||||
def list_organizations_proxmox(self) -> str:
|
|
||||||
"""Получить организации текущего пользователя на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
return self._request("GET", "/user/orgs")
|
|
||||||
|
|
||||||
def list_files_proxmox(self, owner: str, repo: str, path: str = "", branch: str = "") -> str:
|
|
||||||
"""Получить список файлов и папок на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
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_proxmox(self, owner: str, repo: str, file_path: str, branch: str = "") -> str:
|
|
||||||
"""Получить содержимое файла и его SHA на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
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_proxmox(self, owner: str, repo: str, file_path: str, content: str, commit_message: str = "Добавлен файл", branch: str = "") -> str:
|
|
||||||
"""Создать файл в репозитории на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
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_proxmox(self, owner: str, repo: str, file_path: str, content: str, sha: str, commit_message: str = "Обновление файла", branch: str = "") -> str:
|
|
||||||
"""Обновить существующий файл по SHA на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
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_proxmox(self, owner: str, repo: str, file_path: str, sha: str, commit_message: str = "Удаление файла", branch: str = "") -> str:
|
|
||||||
"""Удалить файл по SHA на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
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_proxmox(self, owner: str, repo: str) -> str:
|
|
||||||
"""Получить список веток репозитория на Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
return self._request("GET", f"/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches")
|
|
||||||
|
|
||||||
def download_repository_file_to_workspace_proxmox(self, owner: str, repo: str, file_path: str, destination_name: str = "", branch: str = "", overwrite: bool = False) -> str:
|
|
||||||
"""Скачать файл из Gitea на PROXMOX (192.168.31.57) в локальную рабочую директорию 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_proxmox(self, zip_path: str, destination_dir: str = "", overwrite: bool = False) -> str:
|
|
||||||
"""Безопасно распаковать ZIP в рабочую директорию инструмента Gitea на PROXMOX (192.168.31.57)."""
|
|
||||||
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))
|
|
||||||
Reference in new issue
Block a user