ss-tools/backend/src/services/git/_url.py

# #region GitServiceUrlMixin [C:3] [TYPE Module] [SEMANTICS git, url, parsing, normalization, host_alignment]
# @LAYER: Infra
# @BRIEF URL helper mixin for GitService — parse, normalize, align, and strip credentials from Git URLs.
# @RELATION USED_BY -> [GitServiceSyncMixin]
# @RELATION USED_BY -> [GitServiceGiteaMixin]
# @RELATION USED_BY -> [GitServiceRemoteMixin]

import os
from typing import Any, Dict, List, Optional
from urllib.parse import quote, urlparse
from fastapi import HTTPException
from src.core.database import SessionLocal
from src.core.logger import logger, belief_scope
from src.models.git import GitRepository, GitServerConfig

# #region GitServiceUrlMixin [C:3] [TYPE Class]
# @BRIEF URL helper methods for GitService — extract host, strip credentials, replace host, align origin, parse remote identity, derive server URL, normalize server URL.
class GitServiceUrlMixin:
    # [DEF:_extract_http_host:Function]
    # @PURPOSE: Extract normalized host[:port] from HTTP(S) URL.
    # @PRE: url_value may be empty.
    # @POST: Returns lowercase host token or None.
    # @RETURN: Optional[str]
    def _extract_http_host(self, url_value: Optional[str]) -> Optional[str]:
        normalized = str(url_value or "").strip()
        if not normalized:
            return None
        try:
            parsed = urlparse(normalized)
        except Exception:
            return None
        if parsed.scheme not in {"http", "https"}:
            return None
        host = parsed.hostname
        if not host:
            return None
        if parsed.port:
            return f"{host.lower()}:{parsed.port}"
        return host.lower()
    # [/DEF:_extract_http_host:Function]

    # [DEF:_strip_url_credentials:Function]
    # @PURPOSE: Remove credentials from URL while preserving scheme/host/path.
    # @PRE: url_value may contain credentials.
    # @POST: Returns URL without username/password.
    # @RETURN: str
    def _strip_url_credentials(self, url_value: str) -> str:
        normalized = str(url_value or "").strip()
        if not normalized:
            return normalized
        try:
            parsed = urlparse(normalized)
        except Exception:
            return normalized
        if parsed.scheme not in {"http", "https"} or not parsed.hostname:
            return normalized
        host = parsed.hostname
        if parsed.port:
            host = f"{host}:{parsed.port}"
        return parsed._replace(netloc=host).geturl()
    # [/DEF:_strip_url_credentials:Function]

    # [DEF:_replace_host_in_url:Function]
    # @PURPOSE: Replace source URL host with host from configured server URL.
    # @PRE: source_url and config_url are HTTP(S) URLs.
    # @POST: Returns source URL with updated host (credentials preserved) or None.
    # @RETURN: Optional[str]
    def _replace_host_in_url(self, source_url: Optional[str], config_url: Optional[str]) -> Optional[str]:
        source = str(source_url or "").strip()
        config = str(config_url or "").strip()
        if not source or not config:
            return None
        try:
            source_parsed = urlparse(source)
            config_parsed = urlparse(config)
        except Exception:
            return None
        if source_parsed.scheme not in {"http", "https"} or config_parsed.scheme not in {"http", "https"}:
            return None
        if not source_parsed.hostname or not config_parsed.hostname:
            return None
        target_host = config_parsed.hostname
        if config_parsed.port:
            target_host = f"{target_host}:{config_parsed.port}"
        auth_part = ""
        if source_parsed.username:
            auth_part = quote(source_parsed.username, safe="")
            if source_parsed.password is not None:
                auth_part = f"{auth_part}:{quote(source_parsed.password, safe='')}"
            auth_part = f"{auth_part}@"
        new_netloc = f"{auth_part}{target_host}"
        return source_parsed._replace(netloc=new_netloc).geturl()
    # [/DEF:_replace_host_in_url:Function]

    # [DEF:_align_origin_host_with_config:Function]
    # @PURPOSE: Auto-align local origin host to configured Git server host when they drift.
    # @PRE: origin remote exists.
    # @POST: origin URL host updated and DB binding normalized when mismatch detected.
    # @RETURN: Optional[str]
    def _align_origin_host_with_config(
        self,
        dashboard_id: int,
        origin,
        config_url: Optional[str],
        current_origin_url: Optional[str],
        binding_remote_url: Optional[str],
    ) -> Optional[str]:
        config_host = self._extract_http_host(config_url)
        source_origin_url = str(current_origin_url or "").strip() or str(binding_remote_url or "").strip()
        origin_host = self._extract_http_host(source_origin_url)
        if not config_host or not origin_host:
            return None
        if config_host == origin_host:
            return None
        aligned_url = self._replace_host_in_url(source_origin_url, config_url)
        if not aligned_url:
            return None
        logger.warning(
            "[_align_origin_host_with_config][Action] Host mismatch for dashboard %s: config_host=%s origin_host=%s, applying origin.set_url",
            dashboard_id, config_host, origin_host,
        )
        try:
            origin.set_url(aligned_url)
        except Exception as e:
            logger.warning(
                "[_align_origin_host_with_config][Coherence:Failed] Failed to set origin URL for dashboard %s: %s",
                dashboard_id, e,
            )
            return None
        try:
            session = SessionLocal()
            try:
                db_repo = (
                    session.query(GitRepository)
                    .filter(GitRepository.dashboard_id == int(dashboard_id))
                    .first()
                )
                if db_repo:
                    db_repo.remote_url = self._strip_url_credentials(aligned_url)
                    session.commit()
            finally:
                session.close()
        except Exception as e:
            logger.warning(
                "[_align_origin_host_with_config][Action] Failed to persist aligned remote_url for dashboard %s: %s",
                dashboard_id, e,
            )
        return aligned_url
    # [/DEF:_align_origin_host_with_config:Function]

    # [DEF:_parse_remote_repo_identity:Function]
    # @PURPOSE: Parse owner/repo from remote URL for Git server API operations.
    # @PRE: remote_url is a valid git URL.
    # @POST: Returns owner/repo tokens.
    # @RETURN: Dict[str, str]
    def _parse_remote_repo_identity(self, remote_url: str) -> Dict[str, str]:
        normalized = str(remote_url or "").strip()
        if not normalized:
            raise HTTPException(status_code=400, detail="Repository remote_url is empty")
        if normalized.startswith("git@"):
            path = normalized.split(":", 1)[1] if ":" in normalized else ""
        else:
            parsed = urlparse(normalized)
            path = parsed.path or ""
        path = path.strip("/")
        if path.endswith(".git"):
            path = path[:-4]
        parts = [segment for segment in path.split("/") if segment]
        if len(parts) < 2:
            raise HTTPException(status_code=400, detail=f"Cannot parse repository owner/name from remote URL: {remote_url}")
        owner = parts[0]
        repo = parts[-1]
        namespace = "/".join(parts[:-1])
        return {"owner": owner, "repo": repo, "namespace": namespace, "full_name": f"{namespace}/{repo}"}
    # [/DEF:_parse_remote_repo_identity:Function]

    # [DEF:_derive_server_url_from_remote:Function]
    # @PURPOSE: Build API base URL from remote repository URL without credentials.
    # @PRE: remote_url may be any git URL.
    # @POST: Returns normalized http(s) base URL or None when derivation is impossible.
    # @RETURN: Optional[str]
    def _derive_server_url_from_remote(self, remote_url: str) -> Optional[str]:
        normalized = str(remote_url or "").strip()
        if not normalized or normalized.startswith("git@"):
            return None
        parsed = urlparse(normalized)
        if parsed.scheme not in {"http", "https"}:
            return None
        if not parsed.hostname:
            return None
        netloc = parsed.hostname
        if parsed.port:
            netloc = f"{netloc}:{parsed.port}"
        return f"{parsed.scheme}://{netloc}".rstrip("/")
    # [/DEF:_derive_server_url_from_remote: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]
# #endregion GitServiceUrlMixin
# #endregion GitServiceUrlMixin