class ProviderType(Enum):
    GITHUB = 'github'
    GITLAB = 'gitlab'
    BITBUCKET = 'bitbucket'
    BITBUCKET_DATA_CENTER = 'bitbucket_data_center'
    FORGEJO = 'forgejo'
    AZURE_DEVOPS = 'azure_devops'
    ENTERPRISE_SSO = 'enterprise_sso'

class GitService(Protocol):
    """Protocol defining the interface for Git service providers"""

    def __init__(
        self,
        user_id: str | None = None,
        token: SecretStr | None = None,
        external_auth_id: str | None = None,
        external_auth_token: SecretStr | None = None,
        external_token_manager: bool = False,
        base_domain: str | None = None,
    ) -> None:
        """Initialize the service with authentication details"""
        ...

    async def get_latest_token(self) -> SecretStr | None: ...
    async def get_user(self) -> User: ...
    async def search_repositories(...) -> list[Repository]: ...
    async def get_all_repositories(...) -> list[Repository]: ...
    async def get_paginated_repos(...) -> list[Repository]: ...
    async def get_suggested_tasks(...) -> list[SuggestedTask]: ...
    async def get_repository_details_from_repo_name(...) -> Repository: ...
    async def get_branches(...) -> list[Branch]: ...
    async def get_paginated_branches(...) -> PaginatedBranchesResponse: ...
    async def search_branches(...) -> list[Branch]: ...
    async def get_pr_details(...) -> dict: ...
    async def is_pr_open(...) -> bool: ...

class GitHubService(
    GitHubBranchesMixin,
    GitHubFeaturesMixin,
    GitHubPRsMixin,
    GitHubReposMixin,
    GitHubResolverMixin,
    BaseGitService,
    GitService,
    InstallationsService,
):
    """
    Assembled GitHub service class combining mixins by feature area.
    """

integrations/
├── github/
│   ├── github_service.py          # 组合入口
│   ├── queries.py                 # GraphQL 查询
│   └── service/
│       ├── base.py                # 基础 Mixin
│       ├── repos.py               # 仓库 Mixin
│       ├── branches_prs.py        # 分支 + PR Mixin
│       ├── features.py            # 建议任务 Mixin
│       ├── prs.py                 # PR Mixin
│       └── resolver.py            # 解析器 Mixin
├── gitlab/
│   ├── gitlab_service.py          # 结构相同
│   └── service/                   # 结构相同
├── bitbucket/
│   ├── bitbucket_service.py       # 结构相同
│   └── service/                   # 结构相同
├── bitbucket_data_center/
│   ├── bitbucket_dc_service.py    # 结构相同
│   └── service/                   # 结构相同
├── forgejo/
│   ├── forgejo_service.py         # 结构相同
│   └── service/                   # 结构相同
└── azure_devops/
    ├── azure_devops_service.py    # 结构相同
    └── service/                   # 结构相同

class ProviderHandler:
    PROVIDER_DOMAINS: dict[ProviderType, str] = {
        ProviderType.GITHUB: 'github.com',
        ProviderType.GITLAB: GITLAB_HOST,
        ProviderType.BITBUCKET: 'bitbucket.org',
        ProviderType.FORGEJO: 'codeberg.org',
        ProviderType.AZURE_DEVOPS: 'dev.azure.com',
    }

    def __init__(
        self,
        provider_tokens: PROVIDER_TOKEN_TYPE,
        external_auth_id: str | None = None,
        ...
    ):
        self.service_class_map: dict[ProviderType, type[GitService]] = {
            ProviderType.GITHUB: GithubServiceImpl,
            ProviderType.GITLAB: GitLabServiceImpl,
            ProviderType.BITBUCKET: BitBucketServiceImpl,
            ProviderType.BITBUCKET_DATA_CENTER: BitbucketDCServiceImpl,
            ProviderType.FORGEJO: ForgejoServiceImpl,
            ProviderType.AZURE_DEVOPS: AzureDevOpsServiceImpl,
        }

    def get_service(self, provider: ProviderType) -> GitService:
        """Helper method to instantiate a service for a given provider"""
        token = self.provider_tokens[provider]
        service_class = self.service_class_map[provider]
        return service_class(
            user_id=token.user_id,
            external_auth_id=self.external_auth_id,
            external_auth_token=self.external_auth_token,
            token=token.token,
            external_token_manager=self.external_token_manager,
            base_domain=token.host,
        )

    if not isinstance(provider_tokens, MappingProxyType):
        raise TypeError(
            f'provider_tokens must be a MappingProxyType, got {type(provider_tokens).__name__}'
        )

    async def verify_repo_provider(
        self,
        repository: str,
        specified_provider: ProviderType | None = None,
        is_optional: bool = False,
    ) -> Repository:
        errors = []
        if specified_provider:
            try:
                service = self.get_service(specified_provider)
                return await service.get_repository_details_from_repo_name(repository)
            except Exception as e:
                errors.append(f'{specified_provider.value}: {str(e)}')
        for provider in self.provider_tokens:
            try:
                service = self.get_service(provider)
                return await service.get_repository_details_from_repo_name(repository)
            except Exception as e:
                errors.append(f'{provider.value}: {str(e)}')
        # ... 详细错误日志
        raise AuthenticationError(f'Unable to access repo {repository}{detail}')

@router.get('/installations/search')
async def search_user_installations(
    provider: ProviderType,
    page_id: str | None = None,
    limit: int = 100,
    user_context: UserContext = user_context_dependency,
) -> InstallationPage:
    provider_tokens = await user_context.get_provider_tokens()
    if not provider_tokens:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail='Git provider token required (such as GitHub).',
        )
    user_id = await user_context.get_user_id()
    client = ProviderHandler(
        provider_tokens=MappingProxyType(provider_tokens),
        external_auth_id=user_id,
    )
    if provider == ProviderType.GITHUB:
        installations = await client.get_github_installations()
    elif provider == ProviderType.BITBUCKET:
        installations = await client.get_bitbucket_workspaces()
    # ... 其他 provider
    items, next_page_id = paginate_results(installations, page_id, limit)
    return InstallationPage(items=items, next_page_id=next_page_id)

    page = 1
    decoded_page_id = decode_page_id(page_id)
    if decoded_page_id is not None:
        page = decoded_page_id

    if query:
        repos = await client.search_repositories(
            selected_provider=provider,
            query=query,
            per_page=limit + 1,  # 多取 1 条判断是否有下一页
            sort=search_sort,
            order=order,
            app_mode=get_global_config().app_mode,
        )
    else:
        repos = await client.get_repositories(
            sort='pushed',
            app_mode=get_global_config().app_mode,
            selected_provider=provider,
            page=page,
            per_page=limit + 1,
            installation_id=installation_id,
        )

    next_page_id = None
    if len(repos) > limit:
        repos = repos[:-1]              # 去掉多取的那条
        next_page_id = encode_page_id(page + 1)

    return RepositoryPage(items=repos, next_page_id=next_page_id)

    async def clone_or_init_git_repo(
        self,
        task: AppConversationStartTask,
        workspace: AsyncRemoteWorkspace,
        sandbox: SandboxInfo | None = None,
    ):
        request = task.request

        # 1. 创建项目目录
        result = await workspace.execute_command(
            f'mkdir -p {workspace.working_dir}', parent
        )

        # 2. 配置 git user settings
        await self._configure_git_user_settings(workspace)

        if not request.selected_repository:
            # 没有选仓库：初始化空 git repo
            if self.init_git_in_empty_workspace:
                cmd = (
                    'git init && git config --global '
                    f'--add safe.directory {workspace.working_dir}'
                )
                result = await workspace.execute_command(cmd, ...)
            return

        # 3. 获取认证过的 git URL
        user_info = await self.user_context.get_user_info()
        remote_repo_url = await self.user_context.get_authenticated_git_url(
            request.selected_repository
        )

        # 4. 浅克隆（默认）或全量克隆
        full_clone = bool(getattr(user_info, 'git_full_clone', False))
        clone_flags = ''
        if not full_clone:
            clone_flags = ' --depth 1'
            if request.selected_branch:
                clone_flags += f' --branch {shlex.quote(request.selected_branch)}'

        clone_command = (
            f'git clone{clone_flags} {quoted_remote_repo_url} {quoted_dir_name}'
        )
        result = await workspace.execute_command(
            clone_command, workspace.working_dir, 120  # 120s 超时
        )

        # 5. 分支处理
        if request.selected_branch:
            checkout_command = f'git checkout {shlex.quote(request.selected_branch)}'
        else:
            random_str = base62.encodebytes(os.urandom(16))
            openhands_workspace_branch = f'openhands-workspace-{random_str}'
            checkout_command = (
                f'git checkout -b {shlex.quote(openhands_workspace_branch)}'
            )

import subprocess

COMMON_BRANCH_EXAMPLES = "'main', 'feature/foo', or 'release/1.2.3'"

def is_valid_git_branch_name(branch_name: str) -> bool:
    """Return True when branch_name matches git branch naming rules."""
    if not branch_name:
        return False
    return (
        subprocess.run(
            ['git', 'check-ref-format', '--branch', branch_name],
            check=False,
            capture_output=True,
            text=True,
        ).returncode
        == 0
    )

def ensure_valid_git_branch_name(branch_name: str) -> None:
    """Raise ValueError when branch_name is not safe to pass to git checkout."""
    if is_valid_git_branch_name(branch_name):
        return
    raise ValueError(
        f'Invalid git branch name. Common GitHub/GitLab/Bitbucket '
        f'branch names look like {COMMON_BRANCH_EXAMPLES}.'
    )

#!/bin/bash
# This hook was installed by OpenHands
# It calls the pre-commit script in the .openhands directory

if [ -x ".openhands/pre-commit.sh" ]; then
    source ".openhands/pre-commit.sh"
    exit $?
else
    echo "Warning: .openhands/pre-commit.sh not found or not executable"
    exit 0
fi

suggested_task_pr_graphql_query = """
    query GetUserPRs($login: String!) {
        user(login: $login) {
        pullRequests(first: 50, states: [OPEN],
            orderBy: {field: UPDATED_AT, direction: DESC}) {
            nodes {
            number
            title
            repository { nameWithOwner }
            mergeable
            commits(last: 1) {
                nodes {
                commit {
                    statusCheckRollup { state }
                }
                }
            }
            reviews(first: 50,
                states: [CHANGES_REQUESTED, COMMENTED]) {
                nodes { state }
            }
            }
        }
        }
    }
"""

suggested_task_issue_graphql_query = """
    query GetUserIssues($login: String!) {
        user(login: $login) {
        issues(first: 50, states: [OPEN],
            filterBy: {assignee: $login},
            orderBy: {field: UPDATED_AT, direction: DESC}) {
            nodes {
            number
            title
            repository { nameWithOwner }
            }
        }
        }
    }
"""

class TaskType(str, Enum):
    MERGE_CONFLICTS = 'MERGE_CONFLICTS'
    FAILING_CHECKS = 'FAILING_CHECKS'
    UNRESOLVED_COMMENTS = 'UNRESOLVED_COMMENTS'
    OPEN_ISSUE = 'OPEN_ISSUE'
    OPEN_PR = 'OPEN_PR'

    def get_prompt_for_task(self) -> str:
        task_type = self.task_type
        issue_number = self.issue_number
        repo = self.repo

        env = Environment(
            loader=FileSystemLoader(
                'openhands/app_server/integrations/templates/suggested_task'
            )
        )

        if task_type == TaskType.MERGE_CONFLICTS:
            template = env.get_template('merge_conflict_prompt.j2')
        elif task_type == TaskType.FAILING_CHECKS:
            template = env.get_template('failing_checks_prompt.j2')
        elif task_type == TaskType.UNRESOLVED_COMMENTS:
            template = env.get_template('unresolved_comments_prompt.j2')
        elif task_type == TaskType.OPEN_ISSUE:
            template = env.get_template('open_issue_prompt.j2')

        terms = self.get_provider_terms()
        return template.render(issue_number=issue_number, repo=repo, **terms)

    async def get_authenticated_git_url(
        self,
        repo_name: str,
        is_optional: bool = False,
        specified_provider: ProviderType | None = None,
    ) -> str:
        """Get an authenticated git URL for a repository.

        Returns:
            Authenticated git URL if credentials are available,
            otherwise regular HTTPS URL
        """
        repository = await self.verify_repo_provider(repo_name)
        provider = repository.git_provider
        domain = self.PROVIDER_DOMAINS.get(provider, '')

        # 使用 token 中的 host 覆盖默认域名
        if self.provider_tokens and provider in self.provider_tokens:
            if provider != ProviderType.AZURE_DEVOPS:
                domain = self.provider_tokens[provider].host or domain

        # 检测 HTTP 协议安全性
        allow_insecure = os.environ.get(
            'ALLOW_INSECURE_GIT_ACCESS', 'false'
        ).lower() in ('true', '1', 'yes')
        if not allow_insecure:
            raise ValueError(
                'Attempting to connect to an insecure git repository over HTTP. '
                "Set ALLOW_INSECURE_GIT_ACCESS=true to allow this."
            )