"""Gitea API service""" import httpx from typing import List, Dict, Any, Optional from app.services.base import BaseGitService, FileChange, PRInfo class GiteaService(BaseGitService): """Service for interacting with Gitea API""" def _get_headers(self) -> Dict[str, str]: """Get headers for API requests""" return { "Authorization": f"token {self.token}", "Content-Type": "application/json" } def _get_repo_path(self) -> str: """Get repository API path""" return f"{self.base_url}/api/v1/repos/{self.repo_owner}/{self.repo_name}" async def get_pull_request(self, pr_number: int) -> PRInfo: """Get pull request information from Gitea""" url = f"{self._get_repo_path()}/pulls/{pr_number}" async with httpx.AsyncClient() as client: response = await client.get(url, headers=self._get_headers()) response.raise_for_status() data = response.json() return PRInfo( number=data["number"], title=data["title"], description=data.get("body", ""), author=data["user"]["login"], source_branch=data["head"]["ref"], target_branch=data["base"]["ref"], url=data["html_url"], state=data["state"] ) async def get_pr_files(self, pr_number: int) -> List[FileChange]: """Get list of changed files in PR""" url = f"{self._get_repo_path()}/pulls/{pr_number}/files" async with httpx.AsyncClient() as client: response = await client.get(url, headers=self._get_headers()) response.raise_for_status() files_data = response.json() changes = [] for file in files_data: patch = file.get("patch") # Если patch отсутствует, попробуем получить через diff API if not patch: print(f"⚠️ Patch отсутствует для {file['filename']}, попытка получить через .diff") try: diff_url = f"{self._get_repo_path()}/pulls/{pr_number}.diff" diff_response = await client.get(diff_url, headers=self._get_headers()) if diff_response.status_code == 200: full_diff = diff_response.text # Извлекаем diff для конкретного файла patch = self._extract_file_diff(full_diff, file["filename"]) print(f"✅ Получен diff через .diff API ({len(patch) if patch else 0} символов)") except Exception as e: print(f"❌ Не удалось получить diff: {e}") changes.append(FileChange( filename=file["filename"], status=file["status"], additions=file.get("additions", 0), deletions=file.get("deletions", 0), patch=patch )) return changes def _extract_file_diff(self, full_diff: str, filename: str) -> str: """Extract diff for specific file from full diff""" lines = full_diff.split('\n') file_diff = [] in_file = False for i, line in enumerate(lines): # Начало diff для файла if line.startswith('diff --git') and filename in line: in_file = True file_diff.append(line) continue # Следующий файл - прекращаем if in_file and line.startswith('diff --git') and filename not in line: break if in_file: file_diff.append(line) return '\n'.join(file_diff) if file_diff else None async def get_file_content(self, file_path: str, ref: str) -> str: """Get file content at specific ref""" url = f"{self._get_repo_path()}/contents/{file_path}" async with httpx.AsyncClient() as client: response = await client.get( url, headers=self._get_headers(), params={"ref": ref} ) response.raise_for_status() data = response.json() # Gitea returns base64 encoded content import base64 content = base64.b64decode(data["content"]).decode("utf-8") return content async def get_pr_commits(self, pr_number: int) -> List[Dict[str, Any]]: """Get commits in PR""" url = f"{self._get_repo_path()}/pulls/{pr_number}/commits" async with httpx.AsyncClient() as client: response = await client.get(url, headers=self._get_headers()) response.raise_for_status() return response.json() async def create_review_comment( self, pr_number: int, file_path: str, line_number: int, comment: str, commit_id: str ) -> Dict[str, Any]: """Create a review comment on PR""" url = f"{self._get_repo_path()}/pulls/{pr_number}/reviews" payload = { "body": comment, "commit_id": commit_id, "comments": [{ "path": file_path, "body": comment, "new_position": line_number }] } async with httpx.AsyncClient() as client: response = await client.post( url, headers=self._get_headers(), json=payload ) response.raise_for_status() return response.json() async def create_review( self, pr_number: int, comments: List[Dict[str, Any]], body: str = "", event: str = "COMMENT" ) -> Dict[str, Any]: """Create a review with separate comment for each issue Args: pr_number: PR number comments: List of comments with file_path, line_number, content, severity body: Overall review summary (markdown supported) event: Review event (не используется, для совместимости) Note: Gitea не поддерживает inline комментарии через API, поэтому создаем отдельный комментарий для каждой проблемы. """ print(f"\n📤 Публикация ревью в Gitea PR #{pr_number}") print(f" Комментариев для публикации: {len(comments)}") url = f"{self._get_repo_path()}/issues/{pr_number}/comments" # 1. Сначала публикуем общий summary if body: print(f"\n 📝 Публикация общего summary ({len(body)} символов)...") payload = {"body": body} async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( url, headers=self._get_headers(), json=payload ) response.raise_for_status() print(f" ✅ Summary опубликован!") # 2. Затем публикуем каждую проблему отдельным комментарием if comments: print(f"\n 💬 Публикация {len(comments)} отдельных комментариев...") for i, comment in enumerate(comments, 1): severity_emoji = { "ERROR": "❌", "WARNING": "⚠️", "INFO": "ℹ️" }.get(comment.get("severity", "INFO").upper(), "💬") # Создаем ссылку на строку file_url = f"{self.base_url}/{self.repo_owner}/{self.repo_name}/pulls/{pr_number}/files#L{comment['line_number']}" # Форматируем комментарий comment_body = f"{severity_emoji} **[`{comment['file_path']}:{comment['line_number']}`]({file_url})**\n\n" comment_body += f"**{comment.get('severity', 'INFO').upper()}**: {comment['content']}" payload = {"body": comment_body} try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( url, headers=self._get_headers(), json=payload ) response.raise_for_status() print(f" ✅ {i}/{len(comments)}: {comment['file_path']}:{comment['line_number']}") except Exception as e: print(f" ❌ {i}/{len(comments)}: Ошибка - {e}") print(f"\n 🎉 Все комментарии опубликованы!") return {"summary": "posted", "comments_count": len(comments)}