import os
import re
import zipfile
import hashlib
import requests
from requests.auth import HTTPBasicAuth
from typing import List, Dict, Any, Optional
class NexusClient:
def __init__(self, url: str, username: str, password: str):
self.base_url = url.rstrip("/")
self.auth = HTTPBasicAuth(username, password)
# ----------------- Репозитории -----------------
def get_repositories(self) -> List[str]:
"""Получить список всех репозиториев"""
url = f"{self.base_url}/service/rest/v1/repositories"
resp = requests.get(url, auth=self.auth)
resp.raise_for_status()
return [repo["name"] for repo in resp.json()]
# ----------------- Компоненты / артефакты -----------------
def get_components(self, repo_name: str) -> List[Dict[str, Any]]:
"""Получить список компонентов (артефактов) в репозитории"""
url = f"{self.base_url}/service/rest/v1/components?repository={repo_name}"
results = []
continuation_token = None
while True:
full_url = url
if continuation_token:
full_url += f"&continuationToken={continuation_token}"
resp = requests.get(full_url, auth=self.auth)
resp.raise_for_status()
data = resp.json()
results.extend(data.get("items", []))
continuation_token = data.get("continuationToken")
if not continuation_token:
break
return results
# ----------------- Raw файлы -----------------
def upload_raw_file(self, repo_name: str, path_in_repo: str, file_path: str) -> None:
"""Загрузить файл в Raw репозиторий"""
url = f"{self.base_url}/repository/{repo_name}/{path_in_repo}"
with open(file_path, "rb") as f:
resp = requests.put(
url,
data=f,
auth=self.auth,
headers={"Content-Type": "application/octet-stream"}
)
resp.raise_for_status()
print(f"Файл {file_path} успешно загружен в {repo_name}/{path_in_repo}")
def download_file(self, download_url: str, dest_path: str) -> None:
"""Скачать один файл"""
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
resp = requests.get(download_url, auth=self.auth, stream=True)
resp.raise_for_status()
with open(dest_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=8192):
f.write(chunk)
print(f"Файл скачан: {dest_path}")
# ----------------- Каталоги -----------------
def list_assets(self, repo_name: str) -> List[Dict[str, Any]]:
"""Список всех файлов (assets) в репозитории с пагинацией"""
url = f"{self.base_url}/service/rest/v1/assets"
assets = []
continuation_token = None
while True:
params = {"repository": repo_name}
if continuation_token:
params["continuationToken"] = continuation_token
resp = requests.get(url, auth=self.auth, params=params)
resp.raise_for_status()
data = resp.json()
assets.extend(data.get("items", []))
continuation_token = data.get("continuationToken")
if not continuation_token:
break
return assets
def download_catalog_to_dir(
self,
repo_name: str,
prefix: str,
dest_dir: str,
ext: Optional[str] = None,
regex: Optional[str] = None
):
"""Скачать каталог из репозитория в локальную папку с фильтрацией"""
assets = self.list_assets(repo_name)
pattern = re.compile(regex) if regex else None
for asset in assets:
path = asset["path"]
download_url = asset["downloadUrl"]
if not path.startswith(prefix):
continue
if ext and not path.endswith(ext):
continue
if pattern and not pattern.search(path):
continue
local_path = os.path.join(dest_dir, os.path.relpath(path, prefix))
os.makedirs(os.path.dirname(local_path), exist_ok=True)
self.download_file(download_url, local_path)
print(f"Каталог {prefix} из репозитория {repo_name} сохранён в {dest_dir}")
def download_catalog_as_zip(
self,
repo_name: str,
prefix: str,
zip_path: str,
ext: Optional[str] = None,
regex: Optional[str] = None
):
"""Скачать каталог из репозитория и сохранить в zip-архив с фильтром"""
assets = self.list_assets(repo_name)
pattern = re.compile(regex) if regex else None
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
for asset in assets:
path = asset["path"]
download_url = asset["downloadUrl"]
if not path.startswith(prefix):
continue
if ext and not path.endswith(ext):
continue
if pattern and not pattern.search(path):
continue
resp = requests.get(download_url, auth=self.auth, stream=True)
resp.raise_for_status()
arcname = os.path.relpath(path, prefix)
zipf.writestr(arcname, resp.content)
print(f"Каталог {prefix} из репозитория {repo_name} сохранён в архив {zip_path}")
# ----------------- Синхронизация -----------------
def sync_directory(
self,
repo_name: str,
prefix: str,
dest_dir: str,
ext: Optional[str] = None,
regex: Optional[str] = None
):
"""Синхронизировать локальную папку с каталогом в Nexus"""
assets = self.list_assets(repo_name)
pattern = re.compile(regex) if regex else None
for asset in assets:
path = asset["path"]
download_url = asset["downloadUrl"]
if not path.startswith(prefix):
continue
if ext and not path.endswith(ext):
continue
if pattern and not pattern.search(path):
continue
local_path = os.path.join(dest_dir, os.path.relpath(path, prefix))
# Проверка: если файл существует и не изменился — пропускаем
if os.path.exists(local_path):
local_hash = self._file_hash(local_path)
remote_hash = self._get_remote_hash(download_url)
if local_hash == remote_hash:
print(f"Пропускаем {local_path} (не изменился)")
continue
self.download_file(download_url, local_path)
print(f"Синхронизация каталога {prefix} завершена")
# ----------------- Вспомогательные -----------------
def _file_hash(self, path: str) -> str:
"""MD5 локального файла"""
h = hashlib.md5()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
h.update(chunk)
return h.hexdigest()
def _get_remote_hash(self, download_url: str) -> str:
"""MD5 удалённого файла"""
resp = requests.get(download_url, auth=self.auth, stream=True)
resp.raise_for_status()
h = hashlib.md5()
for chunk in resp.iter_content(chunk_size=8192):
h.update(chunk)
return h.hexdigest()
# ----------------- Удаление компонентов -----------------
def delete_component(self, component_id: str) -> None:
url = f"{self.base_url}/service/rest/v1/components/{component_id}"
resp = requests.delete(url, auth=self.auth)
resp.raise_for_status()
print(f"Компонент {component_id} удалён")
# ----------------- Пример использования -----------------
if __name__ == "__main__":
nexus = NexusClient("https://nexus.example.com", "admin", "admin123")
# Получить репозитории
print("Репозитории:", nexus.get_repositories())
# Скачивание одного файла
nexus.download_file(
"https://nexus.example.com/repository/my-raw-repo/test_folder/readme.txt",
"local_readme.txt"
)
# Скачать каталог в папку с фильтром по расширению
nexus.download_catalog_to_dir("my-raw-repo", "test_folder", "./downloaded_sql", ext=".sql")
# Скачать каталог в zip с фильтром по регулярке
nexus.download_catalog_as_zip("my-raw-repo", "test_folder", "tasks.zip", regex=r"task_123.*\.sql")
# Синхронизировать каталог (только новые или обновлённые файлы)
nexus.sync_directory("my-raw-repo", "test_folder", "./synced", ext=".sql", regex=r"task_123.*")