small fixes

Author: cyberc001

ForgejoRepoAPI.py

@@ -305,6 +305,9 @@ def get_invites(self, repo: Repository) -> 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

@@ -268,6 +268,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

@@ -2,6 +2,7 @@
     RepositoryFactory,
     IRepositoryAPI
 )
+from GitHubRepoAPI import GitHubRepoAPI
 from time import sleep
 import requests
 
@@ -21,7 +22,14 @@ def get_tokens_from_file(tokens_path: str) -> list[str]:
     return tokens
 
 
-class GitClients:
+def get_repos_from_file(repos_path: str) -> list[str]:
+    with open(repos_path, 'r') as file:
+        list_repos = [x for x in file.read().split('\n') if x]
+
+    return list_repos
+
+
+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
@@ -54,38 +62,45 @@ def get_next_client(self) -> IRepositoryAPI:
         return client
 
 
-def get_next_repo(clients: GitClients, repositories):
+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]
-    print(list_repos)
     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}"')
-            exit(1)
         else:
-            print(cur_client['token'])
-            yield repo, cur_client['token']
+            yield cur_client['client'], repo, cur_client['token']
+
+
+def get_next_binded_repo(clients: Clients, repositories: list[str]):
+    for repo_name in repositories:
+        try:
+            cur_client = clients.get_next_client()
+            repo = cur_client['client'].get_repository(repo_name)
+        except Exception as err:
+            print(f'get_next_binded_repo(): error {err}')
+            print(f'get_next_binded_repo(): failed to load repository "{repo_name}"')
+        else:
+            yield cur_client['client'], repo, cur_client['token']
 
 
-def get_assignee_story(git_object):
+def get_assignee_story(git_object, client, token, repository):
     assignee_result = ""
 
     try:
-        repo_owner = git_object.repository.owner.username
-        repo_name = git_object.repository.name
-        issue_index = git_object.number  # Для pull request и issue одинаково
+        repo_owner = repository.owner.login
+        repo_name = repository.name
+        issue_index = git_object._id  # Для pull request и issue одинаково
 
-        client = git_object._client  # доступ к api-клиенту
-        token = client.token
-        base_url = client.base_url.rstrip('/')
+        base_url = client.get_base_url().rstrip('/')
 
         url = f"{base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_index}/timeline"
         headers = {
-            "Authorization": f"token {token}",
+            "Authorization": f"Bearer {token}" if client is GitHubRepoAPI else f"token {token}",
             "Accept": "application/json"
         }
 

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
+
 
 # Фабрика для создания API
 class RepositoryFactory:

issues_parser.py

@@ -152,7 +152,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(issue._id, repository.owner, repository.name, token)
                 if issue._id is not None
@@ -213,4 +213,3 @@ def log_issues(
             sleep(TIMEDELTA)
         except Exception as e:
             print("log_issues exception:", e)
-            exit(1)

utils.py

@@ -21,11 +21,11 @@ def log_title(title: str, title_len: int = TITLE_LEN):
     @staticmethod
     def log_to_csv(csv_name: str, field_names: tuple[str], row: dict | None = None):
         if isinstance(row, dict):
-            with open(csv_name, 'a', newline='') as file:
+            with open(csv_name, 'a', encoding='utf-8', newline='') as file:
                 writer = csv.DictWriter(file, fieldnames=field_names)
                 writer.writerow(row)
         elif row is None:
-            with open(csv_name, 'w', newline='') as file:
+            with open(csv_name, 'w', encoding='utf-8', newline='') as file:
                 writer = csv.writer(file)
                 writer.writerow(field_names)
         else: