From f5bfca76311c1d2441c4a3378805e8294428b6e7 Mon Sep 17 00:00:00 2001 From: xtaodada Date: Wed, 20 Mar 2024 23:10:14 +0800 Subject: [PATCH] support migrate org repo --- .gitignore | 3 +- defs/__init__.py | 0 defs/client.py | 94 ++++++++++++++++++++++++++++++++++++++++++++++++ defs/github.py | 12 +++++++ defs/models.py | 9 +++++ env.example.py | 2 ++ main.py | 25 +++++++++++++ requirements.txt | 2 ++ 8 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 defs/__init__.py create mode 100644 defs/client.py create mode 100644 defs/github.py create mode 100644 defs/models.py create mode 100644 env.example.py create mode 100644 main.py create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore index 68bc17f..aabac04 100644 --- a/.gitignore +++ b/.gitignore @@ -157,4 +157,5 @@ cython_debug/ # be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ +.idea/ +env.py diff --git a/defs/__init__.py b/defs/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/defs/client.py b/defs/client.py new file mode 100644 index 0000000..820beb8 --- /dev/null +++ b/defs/client.py @@ -0,0 +1,94 @@ +from env import URL, TOKEN + +from httpx import AsyncClient, TimeoutException + + +class GiteaError(Exception): + def __init__(self, status_code: int, message: str): + self.status_code = status_code + self.message = message + + def __str__(self): + return f"[{self.status_code}] {self.message}" + + +def retry(func): + async def wrapper(*args, **kwargs): + for _ in range(3): + try: + return await func(*args, **kwargs) + except TimeoutException: + continue + raise GiteaError(500, "Failed to get response after 3 retries") + + return wrapper + + +class Gitea: + def __init__(self, url: str, token: str): + self.url = url + self.token = token + self.client = AsyncClient(timeout=10) + self.headers = { + "Content-type": "application/json", + } + if token: + self.headers["Authorization"] = "token " + token + + def __get_url(self, endpoint): + url = self.url + "/api/v1" + endpoint + return url + + @staticmethod + def parse_result(result) -> dict: + """Parses the result-JSON to a dict.""" + if result.text and len(result.text) > 3: + return result.json() + return {} + + @retry + async def requests_get(self, endpoint: str, params: dict = None) -> dict: + combined_params = {} + if not params: + params = {} + combined_params.update(params) + request = await self.client.get( + url=self.__get_url(endpoint), headers=self.headers, params=combined_params + ) + if request.status_code not in [200, 201]: + message = f"Received status code: {request.status_code} ({request.url})" + if request.status_code in [403]: + message = f"Unauthorized: {request.url} - Check your permissions and try again! ({message})" + raise GiteaError(request.status_code, message) + return self.parse_result(request) + + @retry + async def requests_post(self, endpoint: str, data: dict): + request = await self.client.post( + url=self.__get_url(endpoint), headers=self.headers, json=data, + ) + if request.status_code not in [200, 201, 202]: + raise GiteaError( + request.status_code, + f"Received status code: {request.status_code} ({request.url}), {request.text}" + ) + return self.parse_result(request) + + async def get_version(self) -> str: + result = await self.requests_get("/version") + return result["version"] + + async def migrate_repo(self, owner: str, repo_name: str, git_url: str): + result = await self.requests_post( + "/repos/migrate", + data={ + "clone_addr": git_url, + "repo_name": repo_name, + "repo_owner": owner, + "mirror": True, + }, + ) + return result + + +gitea = Gitea(URL, TOKEN) diff --git a/defs/github.py b/defs/github.py new file mode 100644 index 0000000..9739aae --- /dev/null +++ b/defs/github.py @@ -0,0 +1,12 @@ +from typing import List + +from defs.models import GithubRepo + +from httpx import AsyncClient + + +async def get_github_repos(org_name: str) -> List[GithubRepo]: + async with AsyncClient() as client: + resp = await client.get(f"https://api.github.com/orgs/{org_name}/repos") + data = resp.json() + return [GithubRepo(**repo) for repo in data] diff --git a/defs/models.py b/defs/models.py new file mode 100644 index 0000000..cc552d9 --- /dev/null +++ b/defs/models.py @@ -0,0 +1,9 @@ +from pydantic import BaseModel + + +class GithubRepo(BaseModel): + name: str + full_name: str + private: bool + html_url: str + clone_url: str diff --git a/env.example.py b/env.example.py new file mode 100644 index 0000000..6494b32 --- /dev/null +++ b/env.example.py @@ -0,0 +1,2 @@ +URL = "http://" +TOKEN = "" diff --git a/main.py b/main.py new file mode 100644 index 0000000..3fb646b --- /dev/null +++ b/main.py @@ -0,0 +1,25 @@ +from defs.github import get_github_repos +from defs.client import gitea, GiteaError + + +async def main(): + print("gitea version:", await gitea.get_version()) + org_name = input("Please input the org name: ") + repos = await get_github_repos(org_name) + for repo in repos: + if repo.private: + continue + try: + await gitea.migrate_repo(org_name, repo.name, repo.clone_url) + print(f"Repo {repo.name} migrated successfully!") + except GiteaError as e: + if e.status_code == 409: + print(f"Repo {repo.name} already exists!") + else: + print(f"Failed to migrate repo {repo.name}: {e}") + + +if __name__ == "__main__": + import asyncio + + asyncio.run(main()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6b68903 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +httpx +pydantic