support migrate org repo

.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

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)

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]

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

env.example.py

@@ -0,0 +1,2 @@
+URL = "http://"
+TOKEN = ""

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())

requirements.txt

@@ -0,0 +1,2 @@
+httpx
+pydantic