методы для работы get_assignee_story() в обёртке

ForgejoRepoAPI.py

@@ -329,6 +329,9 @@ def get_invites(self, repo: Repository, users: list[User] = None) -> list[Invite
     def get_rate_limiting(self) -> tuple[int, int]:
         return sys.maxsize, sys.maxsize
 
+    def get_base_url(self) -> str:
+        return self.client.base_url
+
 
 # Точка входа для тестирования
 if __name__ == "__main__":

GitHubRepoAPI.py

@@ -267,6 +267,9 @@ def get_workflow_runs(self, repo) -> list[WorkflowRun]:
             )
             return []
 
+    def get_base_url(self) -> str:
+        return 'https://api.github.com'
+
 
 # Точка входа для тестирования
 if __name__ == "__main__":

git_logger.py

@@ -1,7 +1,13 @@
+from interface_wrapper import (
+    RepositoryFactory,
+    IRepositoryAPI
+)
+from GitHubRepoAPI import GitHubRepoAPI
 from time import sleep
+import requests
 
-from constants import TIMEDELTA
-from interface_wrapper import IRepositoryAPI, RepositoryFactory
+TIMEDELTA = 0.05
+TIMEZONE = 'Europe/Moscow'
 
 
 def login(token, base_url):
@@ -53,7 +59,6 @@ def _get_next_client(self) -> tuple[IRepositoryAPI, str]:
 
         if client is None:
             raise Exception("No git clients available")
-
         return client, self.token_map[client]
 
     def get_next_client(self) -> tuple[IRepositoryAPI, str]:
@@ -66,31 +71,48 @@ def get_next_binded_repo(clients: Clients, repositories: list[str]):
             client, token = clients.get_next_client()
             repo = 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}"')
-            exit(1)
+            print(f'get_next_binded_repo(): error {err}')
+            print(f'get_next_binded_repo(): failed to load repository "{repo_name}"')
         else:
             yield client, repo, token
 
 
-def get_assignee_story(git_object):
-    # TODO
-    return ""
-
-    '''assignee_result = ""
-    events = (
-        git_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}; "
-        sleep(TIMEDELTA)
-    return assignee_result'''
+def get_assignee_story(git_object, client, token, repository):
+    assignee_result = ""
+
+    try:
+        repo_owner = repository.owner.login
+        repo_name = repository.name
+        issue_index = git_object._id  # Для pull request и issue одинаково
+
+        base_url = client.get_base_url().rstrip('/')
+
+        url = f"{base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_index}/timeline"
+        headers = {
+            "Authorization": f"Bearer {token}" if client is GitHubRepoAPI else f"token {token}",
+            "Accept": "application/json"
+        }
+
+        response = requests.get(url, headers=headers)
+        if response.status_code != 200:
+            raise Exception(f"Failed to fetch issue timeline: {response.status_code}, {response.text}")
+
+        events = response.json()
+
+        for event in events:
+            if event.get('event') in ["assigned", "unassigned"]:
+                date = event.get('created_at')
+                assigner = event.get('actor', {}).get('login', 'unknown')
+                assignee = event.get('assignee', {}).get('login', 'unknown')
+
+                assignee_result += f"{date}: {assigner} -"
+                if event['event'] == "unassigned":
+                    assignee_result += "/"
+                assignee_result += f"> {assignee}; "
+
+                sleep(TIMEDELTA)
+
+    except Exception as e:
+        print(f"get_assignee_story(): error {e}")
+
+    return assignee_result

interface_wrapper.py

@@ -60,7 +60,7 @@ class Repository:
 
 @dataclass
 class Issue:
-    _id: str
+    _id: int
     title: str
     state: str
     created_at: datetime
@@ -193,6 +193,10 @@ def get_rate_limiting(self) -> tuple[int, int]:
     def get_workflow_runs(self, repo: Repository) -> list[WorkflowRun]:
         pass
 
+    @abstractmethod
+    def get_base_url(self) -> str:
+        pass
+
 
 class RepositoryFactory:
     @staticmethod

issues_parser.py

@@ -205,7 +205,7 @@ def get_info(obj, attr):
             closer_name=issue.closed_by.username if issue.closed_by else None,
             closer_login=issue.closed_by.login if issue.closed_by else None,
             closer_email=issue.closed_by.email if issue.closed_by else None,
-            assignee_story=get_assignee_story(issue),
+            assignee_story=get_assignee_story(issue, client, token, repository),
             connected_pull_requests=(
                 get_connected_pulls(token, issue._id, repository.owner, repository.name)
                 if issue._id is not None

pull_requests_parser.py

@@ -150,7 +150,7 @@ def get_info(obj, attr):
             merger_email=pull.merged_by.email if pull.merged_by else None,
             source_branch=pull.head_ref,
             target_branch=pull.base_ref,
-            assignee_story=get_assignee_story(pull),
+            assignee_story=get_assignee_story(pull, client, token, repository),
             related_issues=(
                 get_related_issues(pull._id, repository.owner, repository.name, token)
                 if pull.issue_url is not None