invites and contributors fixes

Author: cyberc001

GitHubRepoAPI.py

@@ -9,7 +9,8 @@
     Branch,
     IRepositoryAPI,
     User,
-    Comment
+    Comment,
+    Invite
 )
 from github import Github
 
@@ -27,15 +28,18 @@ def get_repository(self, id: str) -> Repository | None:
             logging.error(f"Failed to get repository {id} from GitHub: {e}")
             return None
 
+    def get_collaborator_permission(self, repo: Repository, user: User) -> str:
+        return self.client.get_repo(repo._id).get_collaborator_permission(user.login)
+
     def get_commits(self, repo: Repository) -> list[Commit]:
         try:
             commits = self.client.get_repo(repo._id).get_commits()
             return [
                 Commit(
                     _id=c.sha,
                     message=c.commit.message,
-                    author=Contributor(
-                        c.author.login if c.author else "unknown", c.commit.author.email
+                    author=User(
+                        login=c.author.login,username=c.author.name,email=c.author.email
                     ),
                     date=c.commit.author.date,
                     files=[
@@ -187,6 +191,24 @@ def get_comments(self, repo, obj) -> list[Comment]:
 
         return result
 
+    def get_invites(self, repo: Repository) -> list[Invite]:
+        try:
+            invites = self.client.get_repo(repo._id).get_pending_invitations()
+            return [
+                Invite(
+                    _id=i._id,
+                    invitee=User(login=i.invitee.login, username=i.invitee.name, email=i.invitee.email),
+                    created_at=i.created_at,
+                    html_url=i.html_url
+                )
+                for i in invites
+            ]
+        except Exception as e:
+            logging.error(
+                f"Failed to get invites from GitHub for repo {repo.name}: {e}"
+            )
+            return []
+
 
 # Точка входа для тестирования
 if __name__ == "__main__":

contributors_parser.py

@@ -1,7 +1,7 @@
 from utils import logger
 from time import sleep
 from typing import Generator
-from interface_wrapper import IRepositoryAPI, Repository
+from interface_wrapper import IRepositoryAPI, Repository, User
 
 EMPTY_FIELD = 'Empty field'
 TIMEDELTA = 0.05
@@ -28,10 +28,10 @@ def log_repository_contributors(client: IRepositoryAPI, repository: Repository,
 
     for contributor_stat in contributors_stats.values():
         contributor = contributor_stat["contributor_object"]
-        contributor_permissons = repository.get_collaborator_permission(contributor)
+        contributor_permissions = client.get_collaborator_permission(repository, User(login=contributor,username="",email=""))
 
         info_tmp = {
-            'repository name': repository.full_name,
+            'repository name': repository.name,
             'login': contributor.login,
             'name': nvl(contributor.name),
             'email': nvl(contributor_stat['email']),
@@ -52,8 +52,6 @@ def log_repository_contributors(client: IRepositoryAPI, repository: Repository,
 
 def get_contributors_stats(client: IRepositoryAPI, repository: Repository) -> dict:
     contributors_stats = dict()
-
-    # Используем обёртку для получения коммитов
     commits = client.get_commits(repository)
 
     for commit in commits:
@@ -62,7 +60,7 @@ def get_contributors_stats(client: IRepositoryAPI, repository: Repository) -> di
         if not contributor.login in contributors_stats:
             contributors_stats[contributor.login] = {
                 'total_commits': 0,
-                'email': commit.commit.author.email,
+                'email': commit.author.email,
                 'contributor_object': contributor,
             }
 
@@ -79,15 +77,15 @@ def log_contributors(
 
     for repo in working_repos:
         try:
-            logger.log_title(repo.full_name)
-            log_repository_contributors(repo, csv_name)
+            logger.log_title(repo.name)
+            log_repository_contributors(client, repo, csv_name)
 
             if fork_flag:
                 for forked_repo in client.get_forks():
                     logger.log_title("FORKED:", forked_repo.name)
                     log_repository_contributors(client, forked_repo, csv_name)
                     sleep(TIMEDELTA)
 
-        except e:
+        except Exception as e:
             print(e)
             exit(1)

interface_wrapper.py

@@ -26,7 +26,7 @@ class User:
 class Commit:
     _id: str
     message: str
-    author: Contributor
+    author: User
     date: datetime
     files: list[str]
 
@@ -56,7 +56,6 @@ class Issue:
     closed_by: User
     labels: list[str]
     milestone: str
-    files: list[str]
 
 @dataclass
 class PullRequest:
@@ -75,6 +74,12 @@ class PullRequest:
     labels: list[str]
     milestone: str
 
+@dataclass
+class Invite:
+    _id: int
+    invitee: User
+    created_at: datetime
+    html_url: str
 
 @dataclass
 class Comment:
@@ -96,6 +101,10 @@ def get_repository(self, id: str) -> Repository | None:
         """Получить репозиторий по его идентификатору."""
         pass
 
+    @abstractmethod
+    def get_collaborator_permission(self, repo: Repository, user: User) -> str:
+        pass
+
     @abstractmethod
     def get_commits(self, repo: Repository) -> list[Commit]:
         """Получить список коммитов для репозитория."""
@@ -134,6 +143,10 @@ def get_wiki_pages(self, repo: Repository) -> list[WikiPage]:
     def get_comments(self, obj) -> list[Comment]:
         pass
 
+    @abstractmethod
+    def get_invites(self, repo: Repository) -> list[Invite]:
+        pass
+
 
 # Фабрика для создания API
 class RepositoryFactory:

invites_parser.py

@@ -1,52 +1,47 @@
+from utils import logger
 from time import sleep
-from interface_wrapper import IRepositoryAPI, PullRequest, Repository
+from interface_wrapper import IRepositoryAPI, Repository
 
+FIELDNAMES = (
+    'repository name',
+    'invited login',
+    'invite creation date',
+    'invitation url',
+)
 TIMEDELTA = 0.05
-TIMEZONE = 'Europe/Moscow'
-
-def login(token):
-    client = Github(login_or_token=token)
-
-    try:
-        # Проверяем аутентификацию через оригинальный метод
-        user_login = client.get_user().login
-    except GithubException as err:
-        print(f'Github: Connect: error {err.data}')
-        print('Github: Connect: user could not be authenticated please try again.')
-        exit(1)
-    else:
-        return client
-
-def get_next_repo(client: IRepositoryAPI, repositories):
-    api = GitHubRepoAPI.GitHubRepoAPI(client)  # Используем обёртку
-    with open(repositories, 'r') as file:
-        list_repos = [x for x in file.read().split('\n') if x]
-    print(list_repos)
-    for repo in list_repos:
-        try:
-            pass
-        except Exception as err:
-            print(f'Github: Connect: error {err.data}')
-            print(f'Github: Connect: failed to load repository "{repo.name}"')
-            exit(1)
-        else:
-            yield repo
-
-def get_assignee_story(github_object):
-    assignee_result = ""
-    events = (
-        github_object.get_issue_events()
-        if type(github_object) is PullRequest.PullRequest
-        else github_object.get_events()
-    )
-    for event in events:
-        if event.event in ["assigned", "unassigned"]:
-            date = event.created_at
-            assigner = github_object.user.login
-            assignee = event.assignee.login
-            assignee_result += f"{date}: {assigner} -"
-            if event.event == "unassigned":
-                assignee_result += "/"
-            assignee_result += f"> {assignee}; "
+
+
+def log_inviter(repo, invite, writer):
+    invite_info = [
+        repo.full_name,
+        invite.invitee.login,
+        invite.created_at.strftime("%d/%m/%Y, %H:%M:%S"),
+        invite.html_url,
+    ]
+    writer.writerow(invite_info)
+    print(invite_info)
+
+
+def log_repository_invitations(client: IRepositoryAPI, repository: Repository, csv_name):
+    invitations = client.get_invites(repository)
+    for invite in invitations:
+        invite_info = {
+            'repository name': repository.name,
+            'invited login': invite.invitee.login,
+            'invite creation date': invite.created_at.strftime("%d/%m/%Y, %H:%M:%S"),
+            'invitation url': invite.html_url,
+        }
+        logger.log_to_csv(csv_name, FIELDNAMES, invite_info)
+        logger.log_to_stdout(invite_info)
         sleep(TIMEDELTA)
-    return assignee_result
+
+
+def log_invitations(client: IRepositoryAPI, working_repos, csv_name):
+    logger.log_to_csv(csv_name, FIELDNAMES)
+
+    for repo in working_repos:
+        logger.log_title(repo.name)
+        try:
+            log_repository_invitations(client, repo, csv_name)
+        except Exception as e:
+            print(e)