113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
import os
|
|
import httpx
|
|
import logging
|
|
from urllib.parse import quote
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def _gitea_headers() -> dict:
|
|
"""Constructs headers for Gitea API requests."""
|
|
return {"Authorization": f"token {os.environ.get('GITEA_TOKEN')}", "Accept": "application/json"}
|
|
|
|
async def handle_get_file_content(p: dict, default_repo: str) -> dict:
|
|
"""Gets the raw content of a file from the Gitea repository."""
|
|
gitea_url = os.environ.get("GITEA_URL")
|
|
if not gitea_url:
|
|
raise ValueError("GITEA_URL environment variable is not set.")
|
|
|
|
path = p.get("path")
|
|
if not path:
|
|
raise ValueError("Missing required parameter: 'path' for get_file_content")
|
|
|
|
repo_id = p.get("repo", default_repo)
|
|
ref = p.get("ref", "main")
|
|
|
|
url = f"{gitea_url}/api/v1/repos/{repo_id}/raw/{path}?ref={ref}"
|
|
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(url, headers=_gitea_headers())
|
|
r.raise_for_status()
|
|
return {"path": path, "content": r.text, "encoding": "text"}
|
|
|
|
async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
|
|
"""Lists files and directories in a given path in the Gitea repository."""
|
|
gitea_url = os.environ.get("GITEA_URL")
|
|
if not gitea_url:
|
|
raise ValueError("GITEA_URL environment variable is not set.")
|
|
|
|
if not isinstance(p, dict):
|
|
raise ValueError("Invalid list_repo_files input.")
|
|
|
|
if set(p.keys()) - {"path"}:
|
|
raise ValueError("Unsupported list_repo_files input field.")
|
|
|
|
path = p.get("path")
|
|
|
|
if path is None or path == "" or path == ".":
|
|
normalized_path = ""
|
|
else:
|
|
if not isinstance(path, str):
|
|
raise ValueError("Invalid repository path.")
|
|
normalized_path = path
|
|
|
|
if normalized_path.startswith('/') or \
|
|
'\\' in normalized_path or \
|
|
any(c in normalized_path for c in (':', '?', '#')):
|
|
raise ValueError("Invalid repository path.")
|
|
|
|
parts = normalized_path.split('/')
|
|
if any(part in ('.', '..') for part in parts):
|
|
raise ValueError("Invalid repository path.")
|
|
if '' in parts and normalized_path != '': # check for empty segments
|
|
raise ValueError("Invalid repository path.")
|
|
|
|
repo_id = default_repo
|
|
ref = "main"
|
|
|
|
safe_path = quote(normalized_path, safe='/')
|
|
|
|
url = f"{gitea_url.rstrip('/')}/api/v1/repos/{repo_id}/contents/{safe_path}"
|
|
|
|
async with httpx.AsyncClient(timeout=15) as c:
|
|
r = await c.get(url, headers=_gitea_headers(), params={"ref": ref})
|
|
r.raise_for_status()
|
|
|
|
response_data = r.json()
|
|
|
|
if isinstance(response_data, dict):
|
|
raise ValueError("Path is a file, not a directory. Use get_file_content instead.")
|
|
|
|
if not isinstance(response_data, list):
|
|
raise ValueError("Invalid upstream response.")
|
|
|
|
result_files = []
|
|
for entry in response_data:
|
|
if not isinstance(entry, dict):
|
|
raise ValueError("Invalid upstream response.")
|
|
|
|
name = entry.get("name")
|
|
entry_path = entry.get("path")
|
|
entry_type = entry.get("type")
|
|
|
|
if not all([name, entry_path, entry_type]):
|
|
raise ValueError("Invalid upstream response.")
|
|
|
|
result_entry = {
|
|
"name": name,
|
|
"path": entry_path,
|
|
"type": entry_type
|
|
}
|
|
|
|
if "sha" in entry and entry["sha"] is not None:
|
|
result_entry["sha"] = entry["sha"]
|
|
|
|
if "size" in entry and entry["size"] is not None:
|
|
result_entry["size"] = entry["size"]
|
|
|
|
result_files.append(result_entry)
|
|
|
|
return {
|
|
"path": normalized_path,
|
|
"files": result_files
|
|
}
|