transformers

1
import os
2
import zipfile
3

4
import requests
5
from get_ci_error_statistics import download_artifact, get_artifacts_links
6

7

8
def get_daily_ci_runs(token, num_runs=7):
9
    """Get the workflow runs of the scheduled (daily) CI.
10

11
    This only selects the runs triggered by the `schedule` event on the `main` branch.
12
    """
13
    headers = None
14
    if token is not None:
15
        headers = {"Accept": "application/vnd.github+json", "Authorization": f"Bearer {token}"}
16

17
    # The id of a workflow (not of a workflow run)
18
    workflow_id = "636036"
19

20
    url = f"https://api.github.com/repos/huggingface/transformers/actions/workflows/{workflow_id}/runs"
21
    # On `main` branch + event being `schedule` + not returning PRs + only `num_runs` results
22
    url += f"?branch=main&event=schedule&exclude_pull_requests=true&per_page={num_runs}"
23

24
    result = requests.get(url, headers=headers).json()
25

26
    return result["workflow_runs"]
27

28

29
def get_last_daily_ci_runs(token):
30
    """Get the last completed workflow run id of the scheduled (daily) CI."""
31
    workflow_runs = get_daily_ci_runs(token)
32
    workflow_run_id = None
33
    for workflow_run in workflow_runs:
34
        if workflow_run["status"] == "completed":
35
            workflow_run_id = workflow_run["id"]
36
            break
37

38
    return workflow_run_id
39

40

41
def get_last_daily_ci_artifacts(artifact_names, output_dir, token):
42
    """Get the artifacts of last completed workflow run id of the scheduled (daily) CI."""
43
    workflow_run_id = get_last_daily_ci_runs(token)
44
    if workflow_run_id is not None:
45
        artifacts_links = get_artifacts_links(worflow_run_id=workflow_run_id, token=token)
46
        for artifact_name in artifact_names:
47
            if artifact_name in artifacts_links:
48
                artifact_url = artifacts_links[artifact_name]
49
                download_artifact(
50
                    artifact_name=artifact_name, artifact_url=artifact_url, output_dir=output_dir, token=token
51
                )
52

53

54
def get_last_daily_ci_reports(artifact_names, output_dir, token):
55
    """Get the artifacts' content of the last completed workflow run id of the scheduled (daily) CI."""
56
    get_last_daily_ci_artifacts(artifact_names, output_dir, token)
57

58
    results = {}
59
    for artifact_name in artifact_names:
60
        artifact_zip_path = os.path.join(output_dir, f"{artifact_name}.zip")
61
        if os.path.isfile(artifact_zip_path):
62
            results[artifact_name] = {}
63
            with zipfile.ZipFile(artifact_zip_path) as z:
64
                for filename in z.namelist():
65
                    if not os.path.isdir(filename):
66
                        # read the file
67
                        with z.open(filename) as f:
68
                            results[artifact_name][filename] = f.read().decode("UTF-8")
69

70
    return results
71