Forgejo Automatic Selection/Github

Author: Wave0004

git_logger.py

@@ -4,9 +4,13 @@
 from interface_wrapper import IRepositoryAPI, RepositoryFactory
 
 
-def login(source, token, base_url):
-    client = RepositoryFactory.create_api(source, token, base_url)
-    return client
+def login(token, base_url):
+    try:
+        client = RepositoryFactory.create_api(token, base_url)
+        client.get_user_data("test")  
+        return client
+    except Exception:
+        return None
 
 
 def get_tokens_from_file(tokens_path: str) -> list[str]:
@@ -24,61 +28,38 @@ def get_repos_from_file(repos_path: str) -> list[str]:
 
 
 class Clients:
-    def __init__(self, source: str, tokens: list[str], base_url: str | None = None):
-        # Возможно это можно переписать покрасивее
-        if source == 'github':
-            self.clients = self._init_clients(source, tokens, base_url)
-        elif base_url == 'forgejo':
-            self.client = RepositoryFactory.create_api(source, tokens[0], base_url)
-            self.token = tokens[0]
-        else:
-            print(f"Unavailable source {source}, use [ 'github' | 'forgejo' ] instead")
-
-        self.source = source
+    def __init__(self, tokens: list[str], base_url: str | None = None):
+        self.clients = []
+        self.token_map = {}
 
-    def _init_clients(
-        self, source: str, tokens: list[str], base_url: str | None
-    ) -> list[dict]:
-        clients = [
-            {
-                "client": RepositoryFactory.create_api(source, token, base_url),
-                "token": token,
-            }
-            for token in tokens
-        ]
+        for token in tokens:
+            client = login(token, base_url)
+            if client:
+                self.clients.append(client)
+                self.token_map[client] = token
 
-        return clients
+        if not self.clients:
+            raise Exception("No valid tokens for either GitHub or Forgejo")
 
-    def _get_next_git_client(self) -> tuple[IRepositoryAPI, str]:
+    def _get_next_client(self) -> tuple[IRepositoryAPI, str]:
         client = None
         max_remaining_limit = -1
 
-        for client_tmp in self.clients:
-            remaining_limit, limit = client_tmp["client"].get_rate_limiting()
-
-            # можно добавить вывод износа токена
-            # можно дополнительно проверять на 403 и временно пропускать эти токены,
-            # либо завести константу "минимальный коэффициент износа" и не трогать "изношенные" токены
-
-            if remaining_limit > max_remaining_limit:
-                client = client_tmp
-                max_remaining_limit = remaining_limit
-
+        for c in self.clients:
+            remaining, _ = c.get_rate_limiting()
+            if remaining > max_remaining_limit:
+                client = c
+                max_remaining_limit = remaining
             sleep(TIMEDELTA)
 
         if client is None:
             raise Exception("No git clients available")
 
-        return client['client'], client['token']
-
-    def _get_next_forgejo_client(self) -> tuple[IRepositoryAPI, str]:
-        return self.client, self.token
+        return client, self.token_map[client]
 
     def get_next_client(self) -> tuple[IRepositoryAPI, str]:
-        if self.source == 'github':
-            return self._get_next_git_client()
-        elif self.source == 'forgejo':
-            return self._get_next_forgejo_client
+        return self._get_next_client()
+  
 
 
 def get_next_binded_repo(clients: Clients, repositories: list[str]):

interface_wrapper.py

@@ -194,25 +194,26 @@ def get_workflow_runs(self, repo: Repository) -> list[WorkflowRun]:
         pass
 
 
-# Фабрика для создания API
 class RepositoryFactory:
     @staticmethod
-    def create_api(
-        source: str, token: str, base_url: str | None = None
-    ) -> IRepositoryAPI:
+    def create_api(token: str, base_url: str | None = None) -> IRepositoryAPI:
         from ForgejoRepoAPI import ForgejoRepoAPI
         from GitHubRepoAPI import GitHubRepoAPI
 
-        if source == 'github':
+        errors = []
+
+        try:
             return GitHubRepoAPI(Github(auth=Auth.Token(token)))
-        elif source == 'forgejo':
-            if not isinstance(base_url, str):
-                raise ValueError(
-                    f"base_url for PyforgejoApi should be str, got {type(base_url)}"
-                )
-            return ForgejoRepoAPI(PyforgejoApi(api_key=token, base_url=base_url))
-        else:
-            raise ValueError(f"Unsupported source: {source}")
+        except Exception as e:
+            errors.append(f"GitHub login failed: {e}")
+
+        if base_url:
+            try:
+                return ForgejoRepoAPI(PyforgejoApi(api_key=token, base_url=base_url))
+            except Exception as e:
+                errors.append(f"Forgejo login failed: {e}")
+
+        raise Exception(" / ".join(errors))
 
 
 # Сервис для расчёта метрик

main.py

@@ -36,6 +36,13 @@ def parse_args():
         action="store_true",
     )
 
+    parser.add_argument(
+    '--base_url',
+    type=str,
+    required=False,
+    help='Base URL for Forgejo instance (if using Forgejo)',
+    )
+
     token = parser.add_mutually_exclusive_group(required=True)
     token.add_argument('-t', '--token', type=str, help='account access token')
     token.add_argument('--tokens', type=str, help='path to your tokens')
@@ -170,16 +177,15 @@ def main():
 
     repositories = git_logger.get_repos_from_file(args.list)
 
-    print(repositories)
-
     try:
-        clients = git_logger.Clients("github", tokens)
+        clients = git_logger.Clients(tokens, args.base_url)
         binded_repos = git_logger.get_next_binded_repo(clients, repositories)
     except Exception as e:
-        print(e)
-        print(traceback.print_exc())
-    else:
-        run(args, binded_repos, repositories)
+        print(f"Failed to initialize any clients: {e}")
+        print(traceback.format_exc())
+        return
+
+    run(args, binded_repos, repositories)
 
 
 if __name__ == '__main__':