Merge branch 'master' into 99-создать-методы-для-работы-get_assignee_story-в-обёртке

Author: cyberc001

75-CreatedDocker/docker-compose.yml

@@ -1,15 +1,41 @@
 version: '3'
 
 services:
+  db:
+    image: postgres:13
+    container_name: forgejo_db
+    environment:
+      POSTGRES_USER: forgejo
+      POSTGRES_PASSWORD: forgejo
+      POSTGRES_DB: forgejo
+    volumes:
+      - ./postgres-data:/var/lib/postgresql/data
+    restart: unless-stopped
+    networks:
+      - forgejo_network
+
   forgejo:
     image: codeberg.org/forgejo/forgejo:1.18
     container_name: forgejo
     environment:
       - USER_UID=1000
       - USER_GID=1000
+      - FORGEJO__database__DB_TYPE=postgres
+      - FORGEJO__database__HOST=db:5432
+      - FORGEJO__database__NAME=forgejo
+      - FORGEJO__database__USER=forgejo
+      - FORGEJO__database__PASSWD=forgejo
     volumes:
       - ./forgejo-data:/data
     ports:
       - "3000:3000"
       - "2222:22"
-    restart: unless-stopped
+    depends_on:
+      - db
+    restart: unless-stopped
+    networks:
+      - forgejo_network
+
+networks:
+  forgejo_network:
+    driver: bridge

ForgejoRepoAPI.py

@@ -299,8 +299,32 @@ def get_comments(self, repo, obj) -> list[Comment]:
 
         return result
 
-    def get_invites(self, repo: Repository) -> list[Invite]:
-        return []
+    def get_invites(self, repo: Repository, users: list[User] = None) -> list[Invite]:
+        if users is None:
+            return []
+
+        try:
+            collaborators = self.client.repository.repo_list_collaborators(
+                owner=repo.owner.login, repo=repo.name
+            )
+            collab_logins = {c.login for c in collaborators}
+
+            invites = []
+            for user in users:
+                if user.login not in collab_logins:
+                    invites.append(
+                        Invite(
+                            _id=0,
+                            invitee=user,
+                            created_at=None,
+                            html_url=user.html_url,
+                        )
+                    )
+            return invites
+
+        except Exception as e:
+            logging.error(f"Failed to simulate invites for Forgejo repo {repo.name}: {e}")
+            return []
 
     def get_rate_limiting(self) -> tuple[int, int]:
         return sys.maxsize, sys.maxsize
@@ -355,3 +379,17 @@ def get_base_url(self) -> str:
         print(
             f"Branch: {branch.name}, Last Commit: {branch.last_commit._id if branch.last_commit else 'None'}"
         )
+
+    # Получение приглашений
+    test_users = [
+        User(login="user1", username="User One", email="", html_url="", node_id="",
+             type="", bio="", site_admin=False, _id=""),
+        User(login="user2", username="User Two", email="", html_url="", node_id="",
+             type="", bio="", site_admin=False, _id=""),
+    ]
+
+    invites = api.get_invites(repo, users=test_users)
+    print(f"Total Invites: {len(invites)}")
+
+    for invite in invites:
+        print(f"Invitee: {invite.invitee.username}, URL: {invite.html_url}")

GitHubRepoAPI.py

@@ -30,7 +30,6 @@ def _client_validation(client: Github) -> Github:
             logging.error(
                 'Github: Connect: user could not be authenticated please try again.'
             )
-            exit(1)
         else:
             return client
 

git_logger.py

@@ -10,9 +10,12 @@
 TIMEZONE = 'Europe/Moscow'
 
 
-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)
+        return client
+    except Exception:
+        return None
 
 
 def get_tokens_from_file(tokens_path: str) -> list[str]:
@@ -30,50 +33,36 @@ 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):
-        self.clients = self._init_clients(source, tokens, base_url)
-        self.cur_client = None
+    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": login(source, token, base_url), "token": token} for token in tokens]
-        return clients
+        for token in tokens:
+            client = login(token, base_url)
+            if client:
+                self.clients.append(client)
+                self.token_map[client] = token
 
-    def get_next_client(self) -> IRepositoryAPI:
+        if not self.clients:
+            raise Exception("No valid tokens for either GitHub or Forgejo")
+
+    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, self.token_map[client]
 
-        self.cur_client = client
-        return client
-
-
-def get_next_repo(clients: Clients, repositories):
-    with open(repositories, 'r') as file:
-        list_repos = [x for x in file.read().split('\n') if x]
-    for repo_name in list_repos:
-        try:
-            cur_client = clients.get_next_client()
-            repo = cur_client['client'].get_repository(repo_name)
-        except Exception as err:
-            print(f'get_next_repo(): error {err}')
-            print(f'get_next_repo(): failed to load repository "{repo_name}"')
-        else:
-            yield cur_client['client'], repo, cur_client['token']
+    def get_next_client(self) -> tuple[IRepositoryAPI, str]:
+        return self._get_next_client()
 
 
 def get_next_binded_repo(clients: Clients, repositories: list[str]):

interface_wrapper.py

@@ -94,7 +94,7 @@ class PullRequest:
 class Invite:
     _id: int
     invitee: User
-    created_at: datetime
+    created_at: datetime | None
     html_url: str
 
 
@@ -198,25 +198,26 @@ def get_base_url(self) -> str:
         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))
 
 
 # Сервис для расчёта метрик