clinical_laboratory/gitea_cli_helper.py
Luis Ernesto Portillo Zaldivar 9c32be4c28 feat: Agregar comando list-open-issues a gitea_cli_helper.py
- Nuevo comando para listar todos los issues abiertos
- Muestra número, título, autor, fecha y URL
- Útil para verificar el estado de los issues
- Documentado en GEMINI.md con ejemplos
2025-07-15 00:06:37 -06:00

246 lines
10 KiB
Python

import requests
import json
import os
import argparse
from dotenv import load_dotenv
# Cargar variables del archivo .env
load_dotenv()
# --- Configuración (obtenida de variables de entorno) ---
GITEA_API_KEY = os.getenv("GITEA_API_KEY")
GITEA_API_KEY_URL = os.getenv("GITEA_API_KEY_URL")
GITEA_USERNAME = os.getenv("GITEA_USERNAME")
GITEA_REPO_NAME = os.getenv("GITEA_REPO_NAME")
# Extraer la URL base de Gitea de GITEA_API_KEY_URL
GITEA_BASE_URL = GITEA_API_KEY_URL.split('/api/v1/')[0]
def _make_gitea_request(method, endpoint, payload=None):
"""Helper function to make authenticated requests to Gitea API."""
api_url = f"{GITEA_BASE_URL}/api/v1/{endpoint}"
headers = {
"Accept": "application/json",
"Authorization": f"token {GITEA_API_KEY}",
"Content-Type": "application/json"
}
try:
if method == "GET":
response = requests.get(api_url, headers=headers)
elif method == "POST":
response = requests.post(api_url, headers=headers, data=json.dumps(payload))
elif method == "PATCH":
response = requests.patch(api_url, headers=headers, data=json.dumps(payload))
else:
raise ValueError(f"Unsupported HTTP method: {method}")
response.raise_for_status()
# Some endpoints return empty responses (like merge)
if response.text:
return response.json()
return {}
except requests.exceptions.RequestException as e:
print(f"Error en la solicitud {method} a {api_url}: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"Respuesta del servidor: {e.response.text}")
raise
def create_issue(title, body_file):
"""Creates a new issue in the Gitea repository."""
try:
with open(body_file, 'r', encoding='utf-8') as f:
body = f.read()
except FileNotFoundError:
print(f"Error: El archivo de cuerpo '{body_file}' no fue encontrado.")
return
endpoint = f"repos/{GITEA_USERNAME}/{GITEA_REPO_NAME}/issues"
payload = {
"title": title,
"body": body,
"closed": False
}
print(f"Creando issue: '{title}' desde archivo '{body_file}'...")
issue_data = _make_gitea_request("POST", endpoint, payload)
print(f"Issue creado exitosamente:")
print(f" Número: {issue_data['number']}")
print(f" Título: {issue_data['title']}")
print(f" URL: {issue_data['html_url']}")
def create_pull_request(head_branch, base_branch, title, body_file):
"""Creates a new pull request in the Gitea repository."""
try:
with open(body_file, 'r', encoding='utf-8') as f:
body = f.read()
except FileNotFoundError:
print(f"Error: El archivo de cuerpo '{body_file}' no fue encontrado.")
return
endpoint = f"repos/{GITEA_USERNAME}/{GITEA_REPO_NAME}/pulls"
payload = {
"head": head_branch,
"base": base_branch,
"title": title,
"body": body
}
print(f"Creando Pull Request: '{title}' de '{head_branch}' a '{base_branch}' desde archivo '{body_file}'...")
pr_data = _make_gitea_request("POST", endpoint, payload)
print(f"Pull Request creado exitosamente:")
print(f" Número: {pr_data['number']}")
print(f" Título: {pr_data['title']}")
print(f" URL: {pr_data['html_url']}")
def comment_issue(issue_number, body):
"""Adds a comment to an existing issue."""
endpoint = f"repos/{GITEA_USERNAME}/{GITEA_REPO_NAME}/issues/{issue_number}/comments"
payload = {
"body": body
}
print(f"Añadiendo comentario al issue #{issue_number}...")
_make_gitea_request("POST", endpoint, payload)
print(f"Comentario añadido al issue #{issue_number} exitosamente.")
def close_issue(issue_number):
"""Closes an existing issue."""
endpoint = f"repos/{GITEA_USERNAME}/{GITEA_REPO_NAME}/issues/{issue_number}"
payload = {
"state": "closed"
}
print(f"Cerrando issue #{issue_number}...")
_make_gitea_request("PATCH", endpoint, payload)
print(f"Issue #{issue_number} cerrado exitosamente.")
def list_open_issues():
"""Lists all open issues in the repository."""
endpoint = f"repos/{GITEA_USERNAME}/{GITEA_REPO_NAME}/issues"
print(f"Obteniendo issues abiertos del repositorio {GITEA_REPO_NAME}...")
try:
# Get open issues (state=open is default)
issues = _make_gitea_request("GET", endpoint + "?state=open")
if not issues:
print("No hay issues abiertos en este momento.")
return
print(f"\nIssues abiertos ({len(issues)}):")
print("-" * 80)
for issue in issues:
# Format issue information
number = issue.get('number', 'N/A')
title = issue.get('title', 'Sin título')
author = issue.get('user', {}).get('login', 'Desconocido')
created = issue.get('created_at', '').split('T')[0] if issue.get('created_at') else 'N/A'
labels = [label.get('name', '') for label in issue.get('labels', [])]
labels_str = f" [{', '.join(labels)}]" if labels else ""
print(f"#{number}: {title}{labels_str}")
print(f" Autor: {author} | Creado: {created}")
print(f" URL: {issue.get('html_url', 'N/A')}")
print()
print("-" * 80)
print(f"Total: {len(issues)} issues abiertos")
except Exception as e:
print(f"Error al obtener los issues: {e}")
def merge_pull_request(pr_number, merge_method="merge"):
"""Merges a pull request (only allowed to dev branch)."""
# First, get PR information to check the base branch
endpoint = f"repos/{GITEA_USERNAME}/{GITEA_REPO_NAME}/pulls/{pr_number}"
print(f"Obteniendo información del PR #{pr_number}...")
pr_info = _make_gitea_request("GET", endpoint)
# Check if PR is already merged
if pr_info.get('merged', False):
print(f"INFO: El PR #{pr_number} ya fue mergeado.")
return
# Check if PR is closed
if pr_info.get('state') == 'closed':
print(f"ERROR: El PR #{pr_number} está cerrado y no se puede mergear.")
return
# Check if base branch is 'dev'
base_branch = pr_info.get('base', {}).get('ref', '')
if base_branch != 'dev':
print(f"ERROR: Solo se permite hacer merge a la rama 'dev'.")
print(f"Este PR tiene como destino la rama '{base_branch}'.")
return
# Check if PR is mergeable
if not pr_info.get('mergeable', False):
print(f"ERROR: El PR #{pr_number} no se puede mergear. Puede tener conflictos.")
return
# Proceed with merge
merge_endpoint = f"repos/{GITEA_USERNAME}/{GITEA_REPO_NAME}/pulls/{pr_number}/merge"
payload = {
"do": merge_method,
"merge_title_field": pr_info.get('title', ''),
"merge_message_field": f"Merge pull request #{pr_number} from {pr_info.get('head', {}).get('ref', '')}\n\n{pr_info.get('title', '')}"
}
print(f"Haciendo merge del PR #{pr_number} a la rama 'dev' usando método '{merge_method}'...")
try:
_make_gitea_request("POST", merge_endpoint, payload)
print(f"PR #{pr_number} mergeado exitosamente a la rama 'dev'.")
except Exception as e:
print(f"Error al hacer merge del PR #{pr_number}: {e}")
def main():
parser = argparse.ArgumentParser(description="Helper CLI para interactuar con la API de Gitea.")
subparsers = parser.add_subparsers(dest="command", help="Comandos disponibles")
# Subparser para crear issue
create_issue_parser = subparsers.add_parser("create-issue", help="Crea un nuevo issue.")
create_issue_parser.add_argument("--title", required=True, help="Título del issue.")
create_issue_parser.add_argument("--body-file", required=True, help="Ruta al archivo de texto con el cuerpo/descripción del issue.")
# Subparser para crear pull request
create_pr_parser = subparsers.add_parser("create-pr", help="Crea un nuevo pull request.")
create_pr_parser.add_argument("--head", required=True, help="Rama de origen (head branch).")
create_pr_parser.add_argument("--base", required=True, help="Rama de destino (base branch).")
create_pr_parser.add_argument("--title", required=True, help="Título del pull request.")
create_pr_parser.add_argument("--body-file", required=True, help="Ruta al archivo de texto con el cuerpo/descripción del pull request.")
# Subparser para comentar issue
comment_issue_parser = subparsers.add_parser("comment-issue", help="Añade un comentario a un issue existente.")
comment_issue_parser.add_argument("--issue-number", type=int, required=True, help="Número del issue.")
comment_issue_parser.add_argument("--body", required=True, help="Cuerpo del comentario (soporta multilínea con \n).")
# Subparser para cerrar issue
close_issue_parser = subparsers.add_parser("close-issue", help="Cierra un issue existente.")
close_issue_parser.add_argument("--issue-number", type=int, required=True, help="Número del issue a cerrar.")
# Subparser para merge PR
merge_pr_parser = subparsers.add_parser("merge-pr", help="Hace merge de un pull request (solo a rama dev).")
merge_pr_parser.add_argument("--pr-number", type=int, required=True, help="Número del pull request.")
merge_pr_parser.add_argument("--merge-method", choices=["merge", "squash", "rebase"], default="merge",
help="Método de merge: merge, squash o rebase (default: merge).")
# Subparser para listar issues abiertos
list_issues_parser = subparsers.add_parser("list-open-issues", help="Lista todos los issues abiertos del repositorio.")
args = parser.parse_args()
if args.command == "create-issue":
create_issue(args.title, args.body_file)
elif args.command == "create-pr":
create_pull_request(args.head, args.base, args.title, args.body_file)
elif args.command == "comment-issue":
comment_issue(args.issue_number, args.body)
elif args.command == "close-issue":
close_issue(args.issue_number)
elif args.command == "merge-pr":
merge_pull_request(args.pr_number, args.merge_method)
elif args.command == "list-open-issues":
list_open_issues()
else:
parser.print_help()
if __name__ == "__main__":
main()