workflow runs parser

Author: PeeachPie

.github/workflows/smoke-tests.yml

@@ -17,6 +17,7 @@ jobs:
           - "--issues"
           - "--wikis"
           - "--contributors"
+          - "--workflow_runs"
 
     runs-on: ubuntu-latest
 

GitHubRepoAPI.py

@@ -13,6 +13,7 @@
     User,
     WikiPage,
     logging,
+    WorkflowRun,
 )
 
 
@@ -37,7 +38,7 @@ def get_user_data(self, user) -> User:
         return User(
             login=user.login,
             username=user.name,
-            email=user.email,
+            email=user.email,  # always None
             html_url=user.html_url,
             node_id=user.node_id,
             type=user.type,
@@ -46,6 +47,17 @@ def get_user_data(self, user) -> User:
             _id=user.id,
         )
 
+    def get_commit_data(self, commit, files=False) -> Commit:
+        return Commit(
+            _id=commit.sha,
+            message=commit.commit.message,
+            author=self.get_user_data(commit.author),
+            date=commit.commit.author.date,
+            files=[f.filename for f in commit.files] if files else None,
+            additions=commit.stats.additions,
+            deletions=commit.stats.deletions,
+        )
+
     def get_repository(self, id: str) -> Repository | None:
         try:
             repo = self.client.get_repo(id)
@@ -66,18 +78,7 @@ def get_collaborator_permission(self, repo: Repository, user: User) -> str:
     def get_commits(self, repo: Repository, files: bool = True) -> list[Commit]:
         try:
             commits = self.client.get_repo(repo._id).get_commits()
-            return [
-                Commit(
-                    _id=c.sha,
-                    message=c.commit.message,
-                    author=self.get_user_data(c.author),
-                    date=c.commit.author.date,
-                    files=[f.filename for f in c.files] if files else None,
-                    additions=c.stats.additions,
-                    deletions=c.stats.deletions,
-                )
-                for c in commits
-            ]
+            return [self.get_commit_data(c, files) for c in commits]
         except Exception as e:
             logging.error(
                 f"Failed to get commits from GitHub for repo {repo.name}: {e}"
@@ -153,18 +154,7 @@ def get_branches(self, repo: Repository) -> list[Branch]:
             for branch in branches:
                 commit = repo_client.get_commit(branch.commit.sha)
 
-                author = commit.author
-                contributor = Contributor(
-                    username=author.login if author else "unknown",
-                    email=commit.commit.author.email or "",
-                )
-
-                commit_obj = Commit(
-                    _id=commit.sha,
-                    message=commit.commit.message,
-                    author=contributor,
-                    date=commit.commit.author.date,
-                )
+                commit_obj = self.get_commit_data(commit)
 
                 result.append(Branch(name=branch.name, last_commit=commit_obj))
 
@@ -182,10 +172,21 @@ def get_wiki_pages(self, repo: Repository) -> list[WikiPage]:
     def get_forks(self, repo: Repository) -> list[Repository]:
         repo_client = self.client.get_repo(repo._id)
         result = []
+
         for r in repo_client.get_forks():
+            default_branch = (Branch(name=r.default_branch, last_commit=None),)
+            owner = (self.get_user_data(r.owner),)
+
             result.append(
-                Repository(_id=repo.full_name, name=repo.name, url=repo.html_url)
+                Repository(
+                    _id=r.full_name,
+                    name=r.name,
+                    url=r.html_url,
+                    default_branch=default_branch,
+                    owner=owner,
+                )
             )
+
         return result
 
     def get_comments(self, repo, obj) -> list[Comment]:
@@ -246,6 +247,34 @@ def get_invites(self, repo: Repository) -> list[Invite]:
     def get_rate_limiting(self) -> tuple[int, int]:
         return self.client.rate_limiting
 
+    def get_workflow_runs(self, repo) -> list[WorkflowRun]:
+        try:
+            runs = self.client.get_repo(repo._id).get_workflow_runs()
+
+            return [
+                WorkflowRun(
+                    display_title=r.display_title,
+                    event=r.event,
+                    head_branch=r.head_branch,
+                    head_sha=r.head_sha,
+                    name=r.name,
+                    path=r.path,
+                    created_at=r.created_at,
+                    run_started_at=r.run_started_at,
+                    updated_at=r.updated_at,
+                    conclusion=r.conclusion,
+                    status=r.status,
+                    url=r.url,
+                )
+                for r in runs
+            ]
+
+        except Exception as e:
+            logging.error(
+                f"Failed to get workflow runs from GitHub for repo {repo.name}: {e}"
+            )
+            return []
+
 
 # Точка входа для тестирования
 if __name__ == "__main__":

commits_parser.py

@@ -84,7 +84,7 @@ def log_commits(
             log_repository_commits(client, repo, csv_name, start, finish, branch)
             if fork_flag:
                 for forked_repo in client.get_forks(repo):
-                    logger.log_title("FORKED:", forked_repo.full_name)
+                    logger.log_title(f"FORKED: {forked_repo.name}")
                     log_repository_commits(
                         client, forked_repo, csv_name, start, finish, branch
                     )

contributors_parser.py

@@ -94,7 +94,7 @@ def log_contributors(
 
             if fork_flag:
                 for forked_repo in client.get_forks(repo):
-                    logger.log_title("FORKED:", forked_repo.name)
+                    logger.log_title(f"FORKED: {forked_repo.name}")
                     log_repository_contributors(client, forked_repo, csv_name)
                     sleep(TIMEDELTA)
 

interface_wrapper.py

@@ -111,6 +111,22 @@ class WikiPage:
     content: str
 
 
+@dataclass
+class WorkflowRun:
+    display_title: str
+    event: str
+    head_branch: str
+    head_sha: str
+    name: str
+    path: str
+    created_at: datetime
+    run_started_at: datetime
+    updated_at: datetime
+    conclusion: str
+    status: str
+    url: str
+
+
 # Интерфейс API
 class IRepositoryAPI(ABC):
 
@@ -173,6 +189,10 @@ def get_invites(self, repo: Repository) -> list[Invite]:
     def get_rate_limiting(self) -> tuple[int, int]:
         pass
 
+    @abstractmethod
+    def get_workflow_runs(self, repo: Repository) -> list[WorkflowRun]:
+        pass
+
 
 # Фабрика для создания API
 class RepositoryFactory:

issues_parser.py

@@ -205,7 +205,7 @@ def log_issues(
             if fork_flag:
                 forked_repos = client.get_forks(repo)
                 for forked_repo in forked_repos:
-                    logger.log_title("FORKED:", forked_repo.name)
+                    logger.log_title(f"FORKED: {forked_repo.name}")
                     log_repository_issues(
                         client, forked_repo, csv_name, token, start, finish
                     )

main.py

@@ -9,6 +9,7 @@
 import issues_parser
 import pull_requests_parser
 import wikipars
+import workflow_runs_parser
 from utils import parse_time
 
 
@@ -22,6 +23,9 @@ def parse_args():
     parser.add_argument("-i", "--issues", help="log issues", action="store_true")
     parser.add_argument("-w", "--wikis", help="log wikis", action="store_true")
     parser.add_argument("--contributors", help="log contributors", action="store_true")
+    parser.add_argument(
+        "--workflow_runs", help="log workflow runs", action="store_true"
+    )
     parser.add_argument(
         "--forks_include", help="logging data from forks", action="store_true"
     )
@@ -144,6 +148,10 @@ def run(args, binded_repos, repos_for_wiki=None):
         )
     if args.contributors:
         contributors_parser.log_contributors(binded_repos, args.out, args.forks_include)
+    if args.workflow_runs:
+        workflow_runs_parser.log_workflow_runs(
+            binded_repos, args.out, args.forks_include
+        )
     if args.wikis:
         wikipars.wikiparser(repos_for_wiki, args.download_repos, args.out)
     if args.export_google_sheets:

pull_requests_parser.py

@@ -208,7 +208,7 @@ def log_pull_requests(
             if fork_flag:
                 forked_repos = client.get_repo(repo._id).get_forks()
                 for forked_repo in forked_repos:
-                    logger.log_title("FORKED:", forked_repo.full_name)
+                    logger.log_title(f"FORKED: {forked_repo.name}")
                     log_repositories_pr(
                         client,
                         forked_repo,

test_token_usage.py

@@ -40,6 +40,7 @@ def setUp(self):
             wikis=False,
             contributors=False,
             invites=False,
+            workflow_runs=False,
             start="2000/01/01-00:00:00",
             finish="2400/01/01-00:00:00",
             branch="default",
@@ -88,6 +89,7 @@ def _get_usage(self, binded_repos, clients):
             param({'issues': True}, id='issues'),
             param({'invites': True}, id='invites'),
             param({'pull_requests': True}, id='pull_requests'),
+            param({'workflow_runs': True}, id='workflow_runs'),
         ],
     )
     def test_commits_parser(self, args: dict[str, bool]):

workflow_runs_parser.py

@@ -0,0 +1,78 @@
+from dataclasses import asdict, dataclass
+from time import sleep
+from typing import Generator
+
+from constants import TIMEDELTA
+from interface_wrapper import IRepositoryAPI, Repository
+from utils import logger
+
+
+@dataclass(kw_only=True, frozen=True)
+class WorkflowRunData:
+    repository_name: str = ''
+    display_title: str = ''
+    event: str = ''
+    head_branch: str = ''
+    head_sha: str = ''
+    name: str = ''
+    path: str = ''
+    started_at: str = ''
+    total_duration: float = 0.0
+    conclusion: str = ''
+    status: str = ''
+    url: str = ''
+
+
+def log_repository_workflow_runs(
+    client: IRepositoryAPI, repository: Repository, csv_name: str
+):
+    workflow_runs = client.get_workflow_runs(repository)
+
+    for run in workflow_runs:
+        total_duration = (run.updated_at - run.created_at).total_seconds()
+
+        workflow_run_data = WorkflowRunData(
+            repository_name=repository.name,
+            display_title=run.display_title,
+            event=run.event,
+            head_branch=run.head_branch,
+            head_sha=run.head_sha,
+            name=run.name,
+            path=run.path,
+            started_at=run.run_started_at,
+            total_duration=total_duration,
+            conclusion=run.conclusion,
+            status=run.status,
+            url=run.url,
+        )
+
+        info_dict = asdict(workflow_run_data)
+
+        logger.log_to_csv(csv_name, list(info_dict.keys()), info_dict)
+        logger.log_to_stdout(info_dict)
+
+        sleep(TIMEDELTA)
+
+
+def log_workflow_runs(
+    binded_repos: Generator[tuple[IRepositoryAPI, Repository, str], None, None],
+    csv_name: str,
+    fork_flag: bool,
+):
+    info = asdict(WorkflowRunData())
+    logger.log_to_csv(csv_name, list(info.keys()))
+
+    for client, repo, token in binded_repos:
+        try:
+            logger.log_title(repo.name)
+            log_repository_workflow_runs(client, repo, csv_name)
+
+            if fork_flag:
+                for forked_repo in client.get_forks(repo):
+                    logger.log_title(f"FORKED: {forked_repo.name}")
+                    log_repository_workflow_runs(client, forked_repo, csv_name)
+                    sleep(TIMEDELTA)
+
+        except Exception as e:
+            print(e)
+            exit(1)