151 lines
6.2 KiB
Python
151 lines
6.2 KiB
Python
|
import aiohttp
|
||
|
|
import os
|
||
|
|
import logging
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
from typing import Optional, List
|
||
|
|
|
||
|
|
# Logger interne mis à jour pour Forgejo
|
||
|
|
forgejo_internal_logger = logging.getLogger("ForgejoClient")
|
||
|
|
forgejo_internal_logger.setLevel(logging.INFO)
|
||
|
|
|
||
|
|
class ForgejoClient:
|
||
|
|
def __init__(self):
|
||
|
|
self.token = os.getenv("FORGEJO_TOKEN")
|
||
|
|
self.owner = os.getenv("FORGEJO_OWNER") # Exemple: "Gameur"
|
||
|
|
self.repo = os.getenv("FORGEJO_REPO") # Exemple: "kuby"
|
||
|
|
self.assignee = os.getenv("FORGEJO_ASSIGNEE") # Nom d'utilisateur (optionnel)
|
||
|
|
|
||
|
|
# URL de base automatique via ton Tailscale
|
||
|
|
self.base_url = os.getenv("FORGEJO_URL", "https://omegakubeserv.tail951d2f.ts.net").rstrip("/")
|
||
|
|
|
||
|
|
# Construction de l'URL de l'API v1 standard de Forgejo/Gitea
|
||
|
|
if "api/v1" not in self.base_url:
|
||
|
|
self.api_url = f"{self.base_url}/api/v1/repos/{self.owner}/{self.repo}/issues"
|
||
|
|
else:
|
||
|
|
self.api_url = f"{self.base_url}/repos/{self.owner}/{self.repo}/issues"
|
||
|
|
|
||
|
|
async def create_issue(self, title: str, description: str, labels: Optional[List[str]] = None) -> Optional[dict]:
|
||
|
|
"""
|
||
|
|
Crée un ticket sur Forgejo.
|
||
|
|
"""
|
||
|
|
if not self.token or not self.owner or not self.repo:
|
||
|
|
if not hasattr(self, '_config_warned'):
|
||
|
|
print("⚠️ L'intégration Forgejo n'est pas configurée.")
|
||
|
|
self._config_warned = True
|
||
|
|
return None
|
||
|
|
|
||
|
|
headers = {
|
||
|
|
"Authorization": f"token {self.token}",
|
||
|
|
"Content-Type": "application/json"
|
||
|
|
}
|
||
|
|
|
||
|
|
# Payload au format Forgejo (JSON)
|
||
|
|
payload = {
|
||
|
|
"title": title,
|
||
|
|
"body": description, # 'description' devient 'body'
|
||
|
|
"labels": labels if labels else [] # Devient une vraie liste []
|
||
|
|
}
|
||
|
|
|
||
|
|
if self.assignee:
|
||
|
|
payload["assignees"] = [self.assignee] # Liste de pseudos strings
|
||
|
|
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
try:
|
||
|
|
# Forgejo attend du JSON strict (json=payload)
|
||
|
|
async with session.post(self.api_url, headers=headers, json=payload) as response:
|
||
|
|
if response.status == 201:
|
||
|
|
result = await response.json()
|
||
|
|
# Forgejo renvoie l'URL dans 'html_url' au lieu de 'web_url'
|
||
|
|
forgejo_internal_logger.info(f"✅ Forgejo issue créée : {result.get('html_url')}")
|
||
|
|
return result
|
||
|
|
else:
|
||
|
|
error_text = await response.text()
|
||
|
|
forgejo_internal_logger.error(f"❌ Échec de la création : {response.status} - {error_text}")
|
||
|
|
return None
|
||
|
|
except Exception as e:
|
||
|
|
forgejo_internal_logger.error(f"❌ Erreur de connexion à l'API Forgejo : {e}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
async def find_issue_by_title(self, title: str, state: str = "open") -> Optional[dict]:
|
||
|
|
"""
|
||
|
|
Recherche un ticket par son titre exact et son état (open/closed).
|
||
|
|
"""
|
||
|
|
if not self.token or not self.owner or not self.repo:
|
||
|
|
return None
|
||
|
|
|
||
|
|
headers = {"Authorization": f"token {self.token}"}
|
||
|
|
# GitLab utilise 'opened', Forgejo utilise 'open'
|
||
|
|
forgejo_state = "open" if state == "opened" else state
|
||
|
|
|
||
|
|
params = {
|
||
|
|
"q": title,
|
||
|
|
"state": forgejo_state
|
||
|
|
}
|
||
|
|
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
try:
|
||
|
|
async with session.get(self.api_url, headers=headers, params=params) as response:
|
||
|
|
if response.status == 200:
|
||
|
|
issues = await response.json()
|
||
|
|
for issue in issues:
|
||
|
|
if issue.get("title") == title:
|
||
|
|
return issue
|
||
|
|
return None
|
||
|
|
else:
|
||
|
|
return None
|
||
|
|
except Exception as e:
|
||
|
|
forgejo_internal_logger.error(f"❌ Erreur lors de la recherche Forgejo : {e}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
async def get_closed_issues(self) -> List[dict]:
|
||
|
|
"""
|
||
|
|
Récupère les tickets récemment fermés (depuis 24h).
|
||
|
|
"""
|
||
|
|
if not self.token or not self.owner or not self.repo:
|
||
|
|
return []
|
||
|
|
|
||
|
|
headers = {"Authorization": f"token {self.token}"}
|
||
|
|
# Forgejo utilise 'since' au format ISO pour filtrer par date
|
||
|
|
yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||
|
|
|
||
|
|
params = {
|
||
|
|
"state": "closed",
|
||
|
|
"since": yesterday
|
||
|
|
}
|
||
|
|
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
try:
|
||
|
|
async with session.get(self.api_url, headers=headers, params=params) as response:
|
||
|
|
if response.status == 200:
|
||
|
|
return await response.json()
|
||
|
|
else:
|
||
|
|
error_text = await response.text()
|
||
|
|
forgejo_internal_logger.error(f"❌ Erreur récupération tickets fermés : {response.status} - {error_text}")
|
||
|
|
return []
|
||
|
|
except Exception as e:
|
||
|
|
forgejo_internal_logger.error(f"❌ Erreur connexion API Forgejo : {e}")
|
||
|
|
return []
|
||
|
|
|
||
|
|
async def get_issue(self, issue_number: int) -> Optional[dict]:
|
||
|
|
"""
|
||
|
|
Récupère un ticket spécifique via son numéro (ID interne).
|
||
|
|
"""
|
||
|
|
if not self.token or not self.owner or not self.repo:
|
||
|
|
return None
|
||
|
|
|
||
|
|
headers = {"Authorization": f"token {self.token}"}
|
||
|
|
url = f"{self.api_url}/{issue_number}"
|
||
|
|
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
try:
|
||
|
|
async with session.get(url, headers=headers) as response:
|
||
|
|
if response.status == 200:
|
||
|
|
return await response.json()
|
||
|
|
else:
|
||
|
|
return None
|
||
|
|
except Exception as e:
|
||
|
|
forgejo_internal_logger.error(f"❌ Erreur connexion API Forgejo : {e}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
# Instance unique
|
||
|
|
gitlab_client = ForgejoClient() # Nom conservé pour éviter de réécrire les imports du bot
|