slug first logic

2026-03-01 13:17:05 +03:00
parent f24200d52a
commit 80b28ac371
20 changed files with 1739 additions and 379 deletions
--- a/backend/src/services/git_service.py
+++ b/backend/src/services/git_service.py
@@ -13,7 +13,7 @@ import os
 import httpx
 from git import Repo
 from fastapi import HTTPException
-from typing import List
+from typing import Any, Dict, List, Optional
 from datetime import datetime
 from src.core.logger import logger, belief_scope
 from src.models.git import GitProvider
@@ -460,5 +460,280 @@ class GitService:
                return False
    # [/DEF:test_connection:Function]

+    # [DEF:_normalize_git_server_url:Function]
+    # @PURPOSE: Normalize Git server URL for provider API calls.
+    # @PRE: raw_url is non-empty.
+    # @POST: Returns URL without trailing slash.
+    # @RETURN: str
+    def _normalize_git_server_url(self, raw_url: str) -> str:
+        normalized = (raw_url or "").strip()
+        if not normalized:
+            raise HTTPException(status_code=400, detail="Git server URL is required")
+        return normalized.rstrip("/")
+    # [/DEF:_normalize_git_server_url:Function]
+
+    # [DEF:_gitea_headers:Function]
+    # @PURPOSE: Build Gitea API authorization headers.
+    # @PRE: pat is provided.
+    # @POST: Returns headers with token auth.
+    # @RETURN: Dict[str, str]
+    def _gitea_headers(self, pat: str) -> Dict[str, str]:
+        token = (pat or "").strip()
+        if not token:
+            raise HTTPException(status_code=400, detail="Git PAT is required for Gitea operations")
+        return {
+            "Authorization": f"token {token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+    # [/DEF:_gitea_headers:Function]
+
+    # [DEF:_gitea_request:Function]
+    # @PURPOSE: Execute HTTP request against Gitea API with stable error mapping.
+    # @PRE: method and endpoint are valid.
+    # @POST: Returns decoded JSON payload.
+    # @RETURN: Any
+    async def _gitea_request(
+        self,
+        method: str,
+        server_url: str,
+        pat: str,
+        endpoint: str,
+        payload: Optional[Dict[str, Any]] = None,
+    ) -> Any:
+        base_url = self._normalize_git_server_url(server_url)
+        url = f"{base_url}/api/v1{endpoint}"
+        headers = self._gitea_headers(pat)
+        try:
+            async with httpx.AsyncClient(timeout=20.0) as client:
+                response = await client.request(
+                    method=method,
+                    url=url,
+                    headers=headers,
+                    json=payload,
+                )
+        except Exception as e:
+            logger.error(f"[gitea_request][Coherence:Failed] Network error: {e}")
+            raise HTTPException(status_code=503, detail=f"Gitea API is unavailable: {str(e)}")
+
+        if response.status_code >= 400:
+            detail = response.text
+            try:
+                parsed = response.json()
+                detail = parsed.get("message") or parsed.get("error") or detail
+            except Exception:
+                pass
+            logger.error(
+                f"[gitea_request][Coherence:Failed] method={method} endpoint={endpoint} status={response.status_code} detail={detail}"
+            )
+            raise HTTPException(
+                status_code=response.status_code,
+                detail=f"Gitea API error: {detail}",
+            )
+
+        if response.status_code == 204:
+            return None
+        return response.json()
+    # [/DEF:_gitea_request:Function]
+
+    # [DEF:get_gitea_current_user:Function]
+    # @PURPOSE: Resolve current Gitea user for PAT.
+    # @PRE: server_url and pat are valid.
+    # @POST: Returns current username.
+    # @RETURN: str
+    async def get_gitea_current_user(self, server_url: str, pat: str) -> str:
+        payload = await self._gitea_request("GET", server_url, pat, "/user")
+        username = payload.get("login") or payload.get("username")
+        if not username:
+            raise HTTPException(status_code=500, detail="Failed to resolve Gitea username")
+        return str(username)
+    # [/DEF:get_gitea_current_user:Function]
+
+    # [DEF:list_gitea_repositories:Function]
+    # @PURPOSE: List repositories visible to authenticated Gitea user.
+    # @PRE: server_url and pat are valid.
+    # @POST: Returns repository list from Gitea.
+    # @RETURN: List[dict]
+    async def list_gitea_repositories(self, server_url: str, pat: str) -> List[dict]:
+        payload = await self._gitea_request(
+            "GET",
+            server_url,
+            pat,
+            "/user/repos?limit=100&page=1",
+        )
+        if not isinstance(payload, list):
+            return []
+        return payload
+    # [/DEF:list_gitea_repositories:Function]
+
+    # [DEF:create_gitea_repository:Function]
+    # @PURPOSE: Create repository in Gitea for authenticated user.
+    # @PRE: name is non-empty and PAT has repo creation permission.
+    # @POST: Returns created repository payload.
+    # @RETURN: dict
+    async def create_gitea_repository(
+        self,
+        server_url: str,
+        pat: str,
+        name: str,
+        private: bool = True,
+        description: Optional[str] = None,
+        auto_init: bool = True,
+        default_branch: Optional[str] = "main",
+    ) -> Dict[str, Any]:
+        payload = {
+            "name": name,
+            "private": bool(private),
+            "auto_init": bool(auto_init),
+        }
+        if description:
+            payload["description"] = description
+        if default_branch:
+            payload["default_branch"] = default_branch
+        created = await self._gitea_request(
+            "POST",
+            server_url,
+            pat,
+            "/user/repos",
+            payload=payload,
+        )
+        if not isinstance(created, dict):
+            raise HTTPException(status_code=500, detail="Unexpected Gitea response while creating repository")
+        return created
+    # [/DEF:create_gitea_repository:Function]
+
+    # [DEF:delete_gitea_repository:Function]
+    # @PURPOSE: Delete repository in Gitea.
+    # @PRE: owner and repo_name are non-empty.
+    # @POST: Repository deleted on Gitea server.
+    async def delete_gitea_repository(
+        self,
+        server_url: str,
+        pat: str,
+        owner: str,
+        repo_name: str,
+    ) -> None:
+        if not owner or not repo_name:
+            raise HTTPException(status_code=400, detail="owner and repo_name are required")
+        await self._gitea_request(
+            "DELETE",
+            server_url,
+            pat,
+            f"/repos/{owner}/{repo_name}",
+        )
+    # [/DEF:delete_gitea_repository:Function]
+
+    # [DEF:create_github_repository:Function]
+    # @PURPOSE: Create repository in GitHub or GitHub Enterprise.
+    # @PRE: PAT has repository create permission.
+    # @POST: Returns created repository payload.
+    # @RETURN: dict
+    async def create_github_repository(
+        self,
+        server_url: str,
+        pat: str,
+        name: str,
+        private: bool = True,
+        description: Optional[str] = None,
+        auto_init: bool = True,
+        default_branch: Optional[str] = "main",
+    ) -> Dict[str, Any]:
+        base_url = self._normalize_git_server_url(server_url)
+        if "github.com" in base_url:
+            api_url = "https://api.github.com/user/repos"
+        else:
+            api_url = f"{base_url}/api/v3/user/repos"
+        headers = {
+            "Authorization": f"token {pat.strip()}",
+            "Content-Type": "application/json",
+            "Accept": "application/vnd.github+json",
+        }
+        payload: Dict[str, Any] = {
+            "name": name,
+            "private": bool(private),
+            "auto_init": bool(auto_init),
+        }
+        if description:
+            payload["description"] = description
+        # GitHub API does not reliably support setting default branch on create without template/import.
+        if default_branch:
+            payload["default_branch"] = default_branch
+        try:
+            async with httpx.AsyncClient(timeout=20.0) as client:
+                response = await client.post(api_url, headers=headers, json=payload)
+        except Exception as e:
+            raise HTTPException(status_code=503, detail=f"GitHub API is unavailable: {str(e)}")
+
+        if response.status_code >= 400:
+            detail = response.text
+            try:
+                parsed = response.json()
+                detail = parsed.get("message") or detail
+            except Exception:
+                pass
+            raise HTTPException(status_code=response.status_code, detail=f"GitHub API error: {detail}")
+        return response.json()
+    # [/DEF:create_github_repository:Function]
+
+    # [DEF:create_gitlab_repository:Function]
+    # @PURPOSE: Create repository(project) in GitLab.
+    # @PRE: PAT has api scope.
+    # @POST: Returns created repository payload.
+    # @RETURN: dict
+    async def create_gitlab_repository(
+        self,
+        server_url: str,
+        pat: str,
+        name: str,
+        private: bool = True,
+        description: Optional[str] = None,
+        auto_init: bool = True,
+        default_branch: Optional[str] = "main",
+    ) -> Dict[str, Any]:
+        base_url = self._normalize_git_server_url(server_url)
+        api_url = f"{base_url}/api/v4/projects"
+        headers = {
+            "PRIVATE-TOKEN": pat.strip(),
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+        payload: Dict[str, Any] = {
+            "name": name,
+            "visibility": "private" if private else "public",
+            "initialize_with_readme": bool(auto_init),
+        }
+        if description:
+            payload["description"] = description
+        if default_branch:
+            payload["default_branch"] = default_branch
+        try:
+            async with httpx.AsyncClient(timeout=20.0) as client:
+                response = await client.post(api_url, headers=headers, json=payload)
+        except Exception as e:
+            raise HTTPException(status_code=503, detail=f"GitLab API is unavailable: {str(e)}")
+
+        if response.status_code >= 400:
+            detail = response.text
+            try:
+                parsed = response.json()
+                if isinstance(parsed, dict):
+                    detail = parsed.get("message") or detail
+            except Exception:
+                pass
+            raise HTTPException(status_code=response.status_code, detail=f"GitLab API error: {detail}")
+
+        data = response.json()
+        # Normalize clone URL key to keep route response stable.
+        if "clone_url" not in data:
+            data["clone_url"] = data.get("http_url_to_repo")
+        if "html_url" not in data:
+            data["html_url"] = data.get("web_url")
+        if "ssh_url" not in data:
+            data["ssh_url"] = data.get("ssh_url_to_repo")
+        if "full_name" not in data:
+            data["full_name"] = data.get("path_with_namespace") or data.get("name")
+        return data
+    # [/DEF:create_gitlab_repository:Function]
+
 # [/DEF:GitService:Class]
 # [/DEF:backend.src.services.git_service:Module]