139 lines
5.5 KiB
Python
139 lines
5.5 KiB
Python
import requests
|
|
import base64
|
|
import os
|
|
import logging
|
|
from urllib.parse import quote
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class GiteaAPI:
|
|
def __init__(self, owner="butterfly", repo="ragdocs"):
|
|
self.base_url = "https://git.client.guacamolebox.net/api/v1"
|
|
self.owner = owner
|
|
self.repo = repo
|
|
# Hardcoded for portability in standalone binary
|
|
self.token = "f8e1300d871e905e27ce54c3455a3343104d9b04"
|
|
self.headers = {"Authorization": f"token {self.token}", "Content-Type": "application/json"}
|
|
|
|
def delete_file(self, filepath, sha, branch="main", message="Automated Cleanup"):
|
|
"""Delete a file from the repository via the Gitea API."""
|
|
url = f"{self.base_url}/repos/{self.owner}/{self.repo}/contents/{quote(filepath, safe='/')}"
|
|
data = {
|
|
"branch": branch,
|
|
"sha": sha,
|
|
"message": message
|
|
}
|
|
try:
|
|
resp = requests.delete(url, headers=self.headers, json=data, timeout=15)
|
|
if resp.status_code == 200:
|
|
logger.info(f"Deleted remote file: {filepath}")
|
|
return True
|
|
else:
|
|
logger.error(f"Failed to delete {filepath}: {resp.text}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Error deleting {filepath}: {e}")
|
|
return False
|
|
|
|
def list_files(self, path="", recursive=True):
|
|
"""Recursively list all files in the repository."""
|
|
url = f"{self.base_url}/repos/{self.owner}/{self.repo}/contents/{quote(path, safe='/')}"
|
|
try:
|
|
resp = requests.get(url, headers=self.headers, timeout=15)
|
|
if resp.status_code == 404:
|
|
return []
|
|
resp.raise_for_status()
|
|
items = resp.json()
|
|
|
|
# Gitea returns a list for directories, but a dict for single files
|
|
if isinstance(items, dict):
|
|
items = [items]
|
|
|
|
files = []
|
|
for item in items:
|
|
if item["type"] == "file":
|
|
files.append({
|
|
"path": item["path"],
|
|
"download_url": item["download_url"],
|
|
"size": item["size"],
|
|
"sha": item["sha"]
|
|
})
|
|
elif item["type"] == "dir" and recursive:
|
|
files.extend(self.list_files(item["path"], recursive=True))
|
|
return files
|
|
except Exception as e:
|
|
logger.debug(f"Note: Failed to list files at {path}: {e}")
|
|
return []
|
|
|
|
def upload_file(self, local_path, remote_path, branch="main", message="Added new document"):
|
|
"""Upload a file to the repository via the Gitea API (POST for Create, PUT for Update)."""
|
|
url = f"{self.base_url}/repos/{self.owner}/{self.repo}/contents/{quote(remote_path, safe='/')}"
|
|
|
|
try:
|
|
with open(local_path, "rb") as f:
|
|
content = base64.b64encode(f.read()).decode("utf-8")
|
|
|
|
# 1. Existing Check
|
|
sha = None
|
|
exists = False
|
|
try:
|
|
existing_resp = requests.get(url, headers=self.headers, timeout=10)
|
|
if existing_resp.status_code == 200:
|
|
sha = existing_resp.json().get("sha")
|
|
exists = True
|
|
elif existing_resp.status_code == 404:
|
|
exists = False
|
|
except:
|
|
exists = False # Assume new if check fails
|
|
|
|
data = {
|
|
"branch": branch,
|
|
"content": content,
|
|
"message": message
|
|
}
|
|
if sha:
|
|
data["sha"] = sha
|
|
|
|
# 2. Method selection (POST for Create, PUT for Update)
|
|
method = requests.put if exists else requests.post
|
|
resp = method(url, headers=self.headers, json=data, timeout=15)
|
|
|
|
if resp.status_code not in [200, 201]:
|
|
logger.error(f"Failed to upload {remote_path}: {resp.text}")
|
|
return None
|
|
|
|
return resp.json()["content"]["download_url"]
|
|
except Exception as e:
|
|
logger.error(f"Failed to upload {local_path} to {remote_path}: {e}")
|
|
return None
|
|
|
|
def download_file(self, remote_path, local_path, is_binary=True):
|
|
"""Download a file from the repository to a local path."""
|
|
url = self.get_raw_url(remote_path)
|
|
try:
|
|
resp = requests.get(url, headers=self.headers, timeout=30)
|
|
resp.raise_for_status()
|
|
|
|
mode = "wb" if is_binary else "w"
|
|
content = resp.content if is_binary else resp.text
|
|
|
|
with open(local_path, mode) as f:
|
|
f.write(content)
|
|
logger.info(f"Downloaded {remote_path} to {local_path}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to download {remote_path}: {e}")
|
|
return False
|
|
|
|
def get_raw_url(self, filepath, branch="main"):
|
|
"""Generate the raw URL for a given file path."""
|
|
return f"https://git.client.guacamolebox.net/{self.owner}/{self.repo}/raw/branch/{branch}/{quote(filepath, safe='/')}"
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO)
|
|
api = GiteaAPI()
|
|
files = api.list_files()
|
|
print(f"Found {len(files)} files.")
|
|
for f in files[:3]:
|
|
print(f" - {f['path']} at {f['download_url']}")
|