# What does this PR do?

From the offline discussion + the comments from @patrickvonplaten in https://github.com/huggingface/transformers/pull/24827#issuecomment-1641750464 I propose a new design for tightly integrating PEFT into transformers. 
This integration enables loading any PEFT adapter that is saved locally or on the Hub directly into PEFT without dispatching the entire model creation process to PEFT as introduced in #24827. 

This would also enable an easier pipeline integration (a one-liner to load adapter weights) | EDIT: pipeline should work out of the box

Let's constraint this integration to few PEFT methods only, for simplicity and redirect users to use PEFT for advanced features (e.g. merge and unload) and advanced PEFT methods (adaptation prompt, prompt learning).

Current API:

## Load a model with an adapter locally or from the Hub:

```python
import torch
from transformers import AutoModelForCausalLM, OPTForCausalLM

model_id = "facebook/opt-350m"
adapter_model_id = "ybelkada/opt-350m-lora"

# directly on from_pretrained
model = AutoModelForCausalLM.from_pretrained(adapter_model_id)
print(model)

# directly on from_pretrained
model = OPTForCausalLM.from_pretrained(adapter_model_id)
print(model)
```

## Load and attach adapter to an existing model

```python
from transformers import AutoModelForCausalLM

# with load_adapter
model = AutoModelForCausalLM.from_pretrained(model_id)
model.load_adapter(adapter_model_id)

print(model)

# 8-bit + multiGPU compatiblity
model = AutoModelForCausalLM.from_pretrained(model_id, load_in_8bit=True, device_map="balanced")
model.load_adapter(adapter_model_id)

print(model)
print(set(model.hf_device_map.values()))

_ = model(torch.LongTensor([[0, 1, 2, 3]]).to(0))
```

## Attach an adapter, iteratively enable / disable adapters

```python
from transformers import AutoModelForCausalLM, OPTForCausalLM, AutoTokenizer
from peft import PeftConfig

model_id = "facebook/opt-350m"
adapter_model_id = "ybelkada/opt-350m-lora"
tokenizer = AutoTokenizer.from_pretrained(model_id)
text = "Hello"
inputs = tokenizer(text, return_tensors="pt")

model = AutoModelForCausalLM.from_pretrained(model_id)
peft_config = PeftConfig.from_pretrained(adapter_model_id)

# To get random weights
peft_config.init_lora_weights = False

model.add_adapter(peft_config)
print(model)

model.disable_adapters()
output_disabled = model.generate(**inputs)
print(tokenizer.decode(output_disabled[0], skip_special_tokens=True))
>>> Hello, I'm a newbie to this sub. I'm looking for a good place to

model.enable_adapters()
output_enabled = model.generate(**inputs)
print(tokenizer.decode(output_enabled[0], skip_special_tokens=True))
>>> Hello, MMMMMMMM
```

## Add multiple adapters

```python
from transformers import AutoModelForCausalLM, OPTForCausalLM, AutoTokenizer
from peft import PeftConfig, LoraConfig

model_id = "facebook/opt-350m"

# directly on from_pretrained
model = AutoModelForCausalLM.from_pretrained(model_id)

lora_config = LoraConfig(
    target_modules=["q_proj", "k_proj"],
    init_lora_weights=False
)

model.add_adapter(lora_config, adapter_name="adapter_1")

# attach new adapter with same config
model.add_adapter(lora_config, adapter_name="adapter_2")

model.set_adapter("adapter_1")
output_disabled = model.generate(**inputs)
print(tokenizer.decode(output_disabled[0], skip_special_tokens=True))
>>> Hello, I'm a newbie to this sub. I'm looking for a good place to

model.set_adapter("adapter_2")
output_enabled = model.generate(**inputs)
print(tokenizer.decode(output_enabled[0], skip_special_tokens=True))
>>> Hello, I'm a newbie to the game. I'm looking for a good way to
```

## Save adapters

```python
from transformers import AutoModelForCausalLM, OPTForCausalLM, AutoTokenizer
from peft import PeftConfig, LoraConfig

model_id = "facebook/opt-350m"

# directly on from_pretrained
model = AutoModelForCausalLM.from_pretrained(model_id)

lora_config = LoraConfig(
    target_modules=["q_proj", "k_proj"],
)

model.add_adapter(lora_config)

... # train here

model.save_pretrained(save_dir) 

# you can either load it back with transformers or PEFT

from peft import AutoPeftModelForCausalLM

model = AutoPeftModelForCausalLM.from_pretrained(save_dir)

# or

model = AutoModelForCausalLM.from_pretrained(save_dir)
```

## Train adapters using Trainer

Check this gist: https://gist.github.com/younesbelkada/cdda6e4abcb09e58f6324d75e0d88862 

This PR is on par with: https://github.com/huggingface/peft/pull/749

Features to support:

- [x] loading PEFT adapters
- [x] using multiple adapters
- [x] deal with models loaded with accelerate
- [x] Loading directly from `from_pretrained`
- [ ] Merging adapter weights - to not support
- [ ] Unload adapter weights - to not support
- [x] Training with BC with expected PEFT checkpoints format (do we really want to support training? Shall we just redirect users to load a classic `PeftModel` if they want to train a model?)
- [x] What about `save_pretrained` ? 
 
Features to **not** support:

- [x] disabling adapters
- [x] prompt tuning / prompt learning methods

TODOs:

- [ ] docs
- [x] tests

cc @sgugger @patrickvonplaten @BenjaminBossan @pacman100 
[`PEFT`] Peft integration alternative design . # What does this PR do?

From the offline discussion + the comments from @patrickvonplaten in https://github.com/huggingface/transformers/pull/24827#issuecomment-1641750464 I propose a new design for tightly integrating PEFT into transformers. 
This integration enables loading any PEFT adapter that is saved locally or on the Hub directly into PEFT without dispatching the entire model creation process to PEFT as introduced in #24827. 

This would also enable an easier pipeline integration (a one-liner to load adapter weights) | EDIT: pipeline should work out of the box

Let's constraint this integration to few PEFT methods only, for simplicity and redirect users to use PEFT for advanced features (e.g. merge and unload) and advanced PEFT methods (adaptation prompt, prompt learning).

Current API:

## Load a model with an adapter locally or from the Hub:

```python
import torch
from transformers import AutoModelForCausalLM, OPTForCausalLM

model_id = "facebook/opt-350m"
adapter_model_id = "ybelkada/opt-350m-lora"

# directly on from_pretrained
model = AutoModelForCausalLM.from_pretrained(adapter_model_id)
print(model)

# directly on from_pretrained
model = OPTForCausalLM.from_pretrained(adapter_model_id)
print(model)
```

## Load and attach adapter to an existing model

```python
from transformers import AutoModelForCausalLM

# with load_adapter
model = AutoModelForCausalLM.from_pretrained(model_id)
model.load_adapter(adapter_model_id)

print(model)

# 8-bit + multiGPU compatiblity
model = AutoModelForCausalLM.from_pretrained(model_id, load_in_8bit=True, device_map="balanced")
model.load_adapter(adapter_model_id)

print(model)
print(set(model.hf_device_map.values()))

_ = model(torch.LongTensor([[0, 1, 2, 3]]).to(0))
```

## Attach an adapter, iteratively enable / disable adapters

```python
from transformers import AutoModelForCausalLM, OPTForCausalLM, AutoTokenizer
from peft import PeftConfig

model_id = "facebook/opt-350m"
adapter_model_id = "ybelkada/opt-350m-lora"
tokenizer = AutoTokenizer.from_pretrained(model_id)
text = "Hello"
inputs = tokenizer(text, return_tensors="pt")

model = AutoModelForCausalLM.from_pretrained(model_id)
peft_config = PeftConfig.from_pretrained(adapter_model_id)

# To get random weights
peft_config.init_lora_weights = False

model.add_adapter(peft_config)
print(model)

model.disable_adapters()
output_disabled = model.generate(**inputs)
print(tokenizer.decode(output_disabled[0], skip_special_tokens=True))
>>> Hello, I'm a newbie to this sub. I'm looking for a good place to

model.enable_adapters()
output_enabled = model.generate(**inputs)
print(tokenizer.decode(output_enabled[0], skip_special_tokens=True))
>>> Hello, MMMMMMMM
```

## Add multiple adapters

```python
from transformers import AutoModelForCausalLM, OPTForCausalLM, AutoTokenizer
from peft import PeftConfig, LoraConfig

model_id = "facebook/opt-350m"

# directly on from_pretrained
model = AutoModelForCausalLM.from_pretrained(model_id)

lora_config = LoraConfig(
    target_modules=["q_proj", "k_proj"],
    init_lora_weights=False
)

model.add_adapter(lora_config, adapter_name="adapter_1")

# attach new adapter with same config
model.add_adapter(lora_config, adapter_name="adapter_2")

model.set_adapter("adapter_1")
output_disabled = model.generate(**inputs)
print(tokenizer.decode(output_disabled[0], skip_special_tokens=True))
>>> Hello, I'm a newbie to this sub. I'm looking for a good place to

model.set_adapter("adapter_2")
output_enabled = model.generate(**inputs)
print(tokenizer.decode(output_enabled[0], skip_special_tokens=True))
>>> Hello, I'm a newbie to the game. I'm looking for a good way to
```

## Save adapters

```python
from transformers import AutoModelForCausalLM, OPTForCausalLM, AutoTokenizer
from peft import PeftConfig, LoraConfig

model_id = "facebook/opt-350m"

# directly on from_pretrained
model = AutoModelForCausalLM.from_pretrained(model_id)

lora_config = LoraConfig(
    target_modules=["q_proj", "k_proj"],
)

model.add_adapter(lora_config)

... # train here

model.save_pretrained(save_dir) 

# you can either load it back with transformers or PEFT

from peft import AutoPeftModelForCausalLM

model = AutoPeftModelForCausalLM.from_pretrained(save_dir)

# or

model = AutoModelForCausalLM.from_pretrained(save_dir)
```

## Train adapters using Trainer

Check this gist: https://gist.github.com/younesbelkada/cdda6e4abcb09e58f6324d75e0d88862 

This PR is on par with: https://github.com/huggingface/peft/pull/749

Features to support:

- [x] loading PEFT adapters
- [x] using multiple adapters
- [x] deal with models loaded with accelerate
- [x] Loading directly from `from_pretrained`
- [ ] Merging adapter weights - to not support
- [ ] Unload adapter weights - to not support
- [x] Training with BC with expected PEFT checkpoints format (do we really want to support training? Shall we just redirect users to load a classic `PeftModel` if they want to train a model?)
- [x] What about `save_pretrained` ? 
 
Features to **not** support:

- [x] disabling adapters
- [x] prompt tuning / prompt learning methods

TODOs:

- [ ] docs
- [x] tests

cc @sgugger @patrickvonplaten @BenjaminBossan @pacman100 
# Copyright 2022 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


client = WebClient(token=os.environ["CI_SLACK_BOT_TOKEN"])


def handle_test_results(test_results):
    expressions = test_results.split(" ")

    failed = 0
    success = 0

    # When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
    # When it is too long, those signs are not present.
    time_spent = expressions[-2] if "=" in expressions[-1] else expressions[-1]

    for i, expression in enumerate(expressions):
        if "failed" in expression:
            failed += int(expressions[i - 1])
        if "passed" in expression:
            success += int(expressions[i - 1])

    return failed, success, time_spent


def extract_first_line_failure(failures_short_lines):
    failures = {}
    file = None
    in_error = False
    for line in failures_short_lines.split("\n"):
        if re.search(r"_ \[doctest\]", line):
            in_error = True
            file = line.split(" ")[2]
        elif in_error and not line.split(" ")[0].isdigit():
            failures[file] = line
            in_error = False

    return failures


class Message:
    def __init__(self, title: str, doc_test_results: Dict):
        self.title = title

        self._time_spent = doc_test_results["time_spent"].split(",")[0]
        self.n_success = doc_test_results["success"]
        self.n_failures = doc_test_results["failures"]
        self.n_tests = self.n_success + self.n_failures

        # Failures and success of the modeling tests
        self.doc_test_results = doc_test_results

    @property
    def time(self) -> str:
        time_spent = [self._time_spent]
        total_secs = 0

        for time in time_spent:
            time_parts = time.split(":")

            # Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
            if len(time_parts) == 1:
                time_parts = [0, 0, time_parts[0]]

            hours, minutes, seconds = int(time_parts[0]), int(time_parts[1]), float(time_parts[2])
            total_secs += hours * 3600 + minutes * 60 + seconds

        hours, minutes, seconds = total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
        return f"{int(hours)}h{int(minutes)}m{int(seconds)}s"

    @property
    def header(self) -> Dict:
        return {"type": "header", "text": {"type": "plain_text", "text": self.title}}

    @property
    def no_failures(self) -> Dict:
        return {
            "type": "section",
            "text": {
                "type": "plain_text",
                "text": f"🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.",
                "emoji": True,
            },
            "accessory": {
                "type": "button",
                "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                "url": f"https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}",
            },
        }

    @property
    def failures(self) -> Dict:
        return {
            "type": "section",
            "text": {
                "type": "plain_text",
                "text": (
                    f"There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in"
                    f" {self.time}."
                ),
                "emoji": True,
            },
            "accessory": {
                "type": "button",
                "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                "url": f"https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}",
            },
        }

    @property
    def category_failures(self) -> Dict:
        line_length = 40
        category_failures = {k: v["failed"] for k, v in doc_test_results.items() if isinstance(v, dict)}

        report = ""
        for category, failures in category_failures.items():
            if len(failures) == 0:
                continue

            if report != "":
                report += "\n\n"

            report += f"*{category} failures*:".ljust(line_length // 2).rjust(line_length // 2) + "\n"
            report += "`"
            report += "`\n`".join(failures)
            report += "`"

        return {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"The following examples had failures:\n\n\n{report}\n",
            },
        }

    @property
    def payload(self) -> str:
        blocks = [self.header]

        if self.n_failures > 0:
            blocks.append(self.failures)

        if self.n_failures > 0:
            blocks.extend([self.category_failures])

        if self.n_failures == 0:
            blocks.append(self.no_failures)

        return json.dumps(blocks)

    @staticmethod
    def error_out():
        payload = [
            {
                "type": "section",
                "text": {
                    "type": "plain_text",
                    "text": "There was an issue running the tests.",
                },
                "accessory": {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                    "url": f"https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}",
                },
            }
        ]

        print("Sending the following payload")
        print(json.dumps({"blocks": json.loads(payload)}))

        client.chat_postMessage(
            channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"],
            text="There was an issue running the tests.",
            blocks=payload,
        )

    def post(self):
        print("Sending the following payload")
        print(json.dumps({"blocks": json.loads(self.payload)}))

        text = f"{self.n_failures} failures out of {self.n_tests} tests," if self.n_failures else "All tests passed."

        self.thread_ts = client.chat_postMessage(
            channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"],
            blocks=self.payload,
            text=text,
        )

    def get_reply_blocks(self, job_name, job_link, failures, text):
        failures_text = ""
        for key, value in failures.items():
            value = value[:200] + " [Truncated]" if len(value) > 250 else value
            failures_text += f"*{key}*\n_{value}_\n\n"

        title = job_name
        content = {"type": "section", "text": {"type": "mrkdwn", "text": text}}

        if job_link is not None:
            content["accessory"] = {
                "type": "button",
                "text": {"type": "plain_text", "text": "GitHub Action job", "emoji": True},
                "url": job_link,
            }

        return [
            {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
            content,
            {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
        ]

    def post_reply(self):
        if self.thread_ts is None:
            raise ValueError("Can only post reply if a post has been made.")

        job_link = self.doc_test_results.pop("job_link")
        self.doc_test_results.pop("failures")
        self.doc_test_results.pop("success")
        self.doc_test_results.pop("time_spent")

        sorted_dict = sorted(self.doc_test_results.items(), key=lambda t: t[0])
        for job, job_result in sorted_dict:
            if len(job_result["failures"]):
                text = f"*Num failures* :{len(job_result['failed'])} \n"
                failures = job_result["failures"]
                blocks = self.get_reply_blocks(job, job_link, failures, text=text)

                print("Sending the following reply")
                print(json.dumps({"blocks": blocks}))

                client.chat_postMessage(
                    channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"],
                    text=f"Results for {job}",
                    blocks=blocks,
                    thread_ts=self.thread_ts["ts"],
                )

                time.sleep(1)


def get_job_links():
    run_id = os.environ["GITHUB_RUN_ID"]
    url = f"https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"
    result = requests.get(url).json()
    jobs = {}

    try:
        jobs.update({job["name"]: job["html_url"] for job in result["jobs"]})
        pages_to_iterate_over = math.ceil((result["total_count"] - 100) / 100)

        for i in range(pages_to_iterate_over):
            result = requests.get(url + f"&page={i + 2}").json()
            jobs.update({job["name"]: job["html_url"] for job in result["jobs"]})

        return jobs
    except Exception as e:
        print("Unknown error, could not fetch links.", e)

    return {}


def retrieve_artifact(name: str):
    _artifact = {}

    if os.path.exists(name):
        files = os.listdir(name)
        for file in files:
            try:
                with open(os.path.join(name, file), encoding="utf-8") as f:
                    _artifact[file.split(".")[0]] = f.read()
            except UnicodeDecodeError as e:
                raise ValueError(f"Could not open {os.path.join(name, file)}.") from e

    return _artifact


def retrieve_available_artifacts():
    class Artifact:
        def __init__(self, name: str):
            self.name = name
            self.paths = []

        def __str__(self):
            return self.name

        def add_path(self, path: str):
            self.paths.append({"name": self.name, "path": path})

    _available_artifacts: Dict[str, Artifact] = {}

    directories = filter(os.path.isdir, os.listdir())
    for directory in directories:
        artifact_name = directory
        if artifact_name not in _available_artifacts:
            _available_artifacts[artifact_name] = Artifact(artifact_name)

            _available_artifacts[artifact_name].add_path(directory)

    return _available_artifacts


if __name__ == "__main__":
    github_actions_job_links = get_job_links()
    available_artifacts = retrieve_available_artifacts()

    docs = collections.OrderedDict(
        [
            ("*.py", "API Examples"),
            ("*.md", "MD Examples"),
        ]
    )

    # This dict will contain all the information relative to each doc test category:
    # - failed: list of failed tests
    # - failures: dict in the format 'test': 'error_message'
    doc_test_results = {
        v: {
            "failed": [],
            "failures": {},
        }
        for v in docs.values()
    }

    # Link to the GitHub Action job
    doc_test_results["job_link"] = github_actions_job_links.get("run_doctests")

    artifact_path = available_artifacts["doc_tests_gpu_test_reports"].paths[0]
    artifact = retrieve_artifact(artifact_path["name"])
    if "stats" in artifact:
        failed, success, time_spent = handle_test_results(artifact["stats"])
        doc_test_results["failures"] = failed
        doc_test_results["success"] = success
        doc_test_results["time_spent"] = time_spent[1:-1] + ", "

        all_failures = extract_first_line_failure(artifact["failures_short"])
        for line in artifact["summary_short"].split("\n"):
            if re.search("FAILED", line):
                line = line.replace("FAILED ", "")
                line = line.split()[0].replace("\n", "")

                if "::" in line:
                    file_path, test = line.split("::")
                else:
                    file_path, test = line, line

                for file_regex in docs.keys():
                    if fnmatch(file_path, file_regex):
                        category = docs[file_regex]
                        doc_test_results[category]["failed"].append(test)

                        failure = all_failures[test] if test in all_failures else "N/A"
                        doc_test_results[category]["failures"][test] = failure
                        break

    message = Message("🤗 Results of the doc tests.", doc_test_results)
    message.post()
    message.post_reply()

# Copyright 2022 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


client = WebClient(token=os.environ["CI_SLACK_BOT_TOKEN"])


def handle_test_results(test_results):
    expressions = test_results.split(" ")

    failed = 0
    success = 0

    # When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
    # When it is too long, those signs are not present.
    time_spent = expressions[-2] if "=" in expressions[-1] else expressions[-1]

    for i, expression in enumerate(expressions):
        if "failed" in expression:
            failed += int(expressions[i - 1])
        if "passed" in expression:
            success += int(expressions[i - 1])

    return failed, success, time_spent


def extract_first_line_failure(failures_short_lines):
    failures = {}
    file = None
    in_error = False
    for line in failures_short_lines.split("\n"):
        if re.search(r"_ \[doctest\]", line):
            in_error = True
            file = line.split(" ")[2]
        elif in_error and not line.split(" ")[0].isdigit():
            failures[file] = line
            in_error = False

    return failures


class Message:
    def __init__(self, title: str, doc_test_results: Dict):
        self.title = title

        self._time_spent = doc_test_results["time_spent"].split(",")[0]
        self.n_success = doc_test_results["success"]
        self.n_failures = doc_test_results["failures"]
        self.n_tests = self.n_success + self.n_failures

        # Failures and success of the modeling tests
        self.doc_test_results = doc_test_results

    @property
    def time(self) -> str:
        time_spent = [self._time_spent]
        total_secs = 0

        for time in time_spent:
            time_parts = time.split(":")

            # Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
            if len(time_parts) == 1:
                time_parts = [0, 0, time_parts[0]]

            hours, minutes, seconds = int(time_parts[0]), int(time_parts[1]), float(time_parts[2])
            total_secs += hours * 3600 + minutes * 60 + seconds

        hours, minutes, seconds = total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
        return f"{int(hours)}h{int(minutes)}m{int(seconds)}s"

    @property
    def header(self) -> Dict:
        return {"type": "header", "text": {"type": "plain_text", "text": self.title}}

    @property
    def no_failures(self) -> Dict:
        return {
            "type": "section",
            "text": {
                "type": "plain_text",
                "text": f"🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.",
                "emoji": True,
            },
            "accessory": {
                "type": "button",
                "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                "url": f"https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}",
            },
        }

    @property
    def failures(self) -> Dict:
        return {
            "type": "section",
            "text": {
                "type": "plain_text",
                "text": (
                    f"There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in"
                    f" {self.time}."
                ),
                "emoji": True,
            },
            "accessory": {
                "type": "button",
                "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                "url": f"https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}",
            },
        }

    @property
    def category_failures(self) -> Dict:
        line_length = 40
        category_failures = {k: v["failed"] for k, v in doc_test_results.items() if isinstance(v, dict)}

        report = ""
        for category, failures in category_failures.items():
            if len(failures) == 0:
                continue

            if report != "":
                report += "\n\n"

            report += f"*{category} failures*:".ljust(line_length // 2).rjust(line_length // 2) + "\n"
            report += "`"
            report += "`\n`".join(failures)
            report += "`"

        return {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"The following examples had failures:\n\n\n{report}\n",
            },
        }

    @property
    def payload(self) -> str:
        blocks = [self.header]

        if self.n_failures > 0:
            blocks.append(self.failures)

        if self.n_failures > 0:
            blocks.extend([self.category_failures])

        if self.n_failures == 0:
            blocks.append(self.no_failures)

        return json.dumps(blocks)

    @staticmethod
    def error_out():
        payload = [
            {
                "type": "section",
                "text": {
                    "type": "plain_text",
                    "text": "There was an issue running the tests.",
                },
                "accessory": {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                    "url": f"https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}",
                },
            }
        ]

        print("Sending the following payload")
        print(json.dumps({"blocks": json.loads(payload)}))

        client.chat_postMessage(
            channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"],
            text="There was an issue running the tests.",
            blocks=payload,
        )

    def post(self):
        print("Sending the following payload")
        print(json.dumps({"blocks": json.loads(self.payload)}))

        text = f"{self.n_failures} failures out of {self.n_tests} tests," if self.n_failures else "All tests passed."

        self.thread_ts = client.chat_postMessage(
            channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"],
            blocks=self.payload,
            text=text,
        )

    def get_reply_blocks(self, job_name, job_link, failures, text):
        failures_text = ""
        for key, value in failures.items():
            value = value[:200] + " [Truncated]" if len(value) > 250 else value
            failures_text += f"*{key}*\n_{value}_\n\n"

        title = job_name
        content = {"type": "section", "text": {"type": "mrkdwn", "text": text}}

        if job_link is not None:
            content["accessory"] = {
                "type": "button",
                "text": {"type": "plain_text", "text": "GitHub Action job", "emoji": True},
                "url": job_link,
            }

        return [
            {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
            content,
            {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
        ]

    def post_reply(self):
        if self.thread_ts is None:
            raise ValueError("Can only post reply if a post has been made.")

        job_link = self.doc_test_results.pop("job_link")
        self.doc_test_results.pop("failures")
        self.doc_test_results.pop("success")
        self.doc_test_results.pop("time_spent")

        sorted_dict = sorted(self.doc_test_results.items(), key=lambda t: t[0])
        for job, job_result in sorted_dict:
            if len(job_result["failures"]):
                text = f"*Num failures* :{len(job_result['failed'])} \n"
                failures = job_result["failures"]
                blocks = self.get_reply_blocks(job, job_link, failures, text=text)

                print("Sending the following reply")
                print(json.dumps({"blocks": blocks}))

                client.chat_postMessage(
                    channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"],
                    text=f"Results for {job}",
                    blocks=blocks,
                    thread_ts=self.thread_ts["ts"],
                )

                time.sleep(1)


def get_job_links():
    run_id = os.environ["GITHUB_RUN_ID"]
    url = f"https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"
    result = requests.get(url).json()
    jobs = {}

    try:
        jobs.update({job["name"]: job["html_url"] for job in result["jobs"]})
        pages_to_iterate_over = math.ceil((result["total_count"] - 100) / 100)

        for i in range(pages_to_iterate_over):
            result = requests.get(url + f"&page={i + 2}").json()
            jobs.update({job["name"]: job["html_url"] for job in result["jobs"]})

        return jobs
    except Exception as e:
        print("Unknown error, could not fetch links.", e)

    return {}


def retrieve_artifact(name: str):
    _artifact = {}

    if os.path.exists(name):
        files = os.listdir(name)
        for file in files:
            try:
                with open(os.path.join(name, file), encoding="utf-8") as f:
                    _artifact[file.split(".")[0]] = f.read()
            except UnicodeDecodeError as e:
                raise ValueError(f"Could not open {os.path.join(name, file)}.") from e

    return _artifact


def retrieve_available_artifacts():
    class Artifact:
        def __init__(self, name: str):
            self.name = name
            self.paths = []

        def __str__(self):
            return self.name

        def add_path(self, path: str):
            self.paths.append({"name": self.name, "path": path})

    _available_artifacts: Dict[str, Artifact] = {}

    directories = filter(os.path.isdir, os.listdir())
    for directory in directories:
        artifact_name = directory
        if artifact_name not in _available_artifacts:
            _available_artifacts[artifact_name] = Artifact(artifact_name)

            _available_artifacts[artifact_name].add_path(directory)

    return _available_artifacts


if __name__ == "__main__":
    github_actions_job_links = get_job_links()
    available_artifacts = retrieve_available_artifacts()

    docs = collections.OrderedDict(
        [
            ("*.py", "API Examples"),
            ("*.md", "MD Examples"),
        ]
    )

    # This dict will contain all the information relative to each doc test category:
    # - failed: list of failed tests
    # - failures: dict in the format 'test': 'error_message'
    doc_test_results = {
        v: {
            "failed": [],
            "failures": {},
        }
        for v in docs.values()
    }

    # Link to the GitHub Action job
    doc_test_results["job_link"] = github_actions_job_links.get("run_doctests")

    artifact_path = available_artifacts["doc_tests_gpu_test_reports"].paths[0]
    artifact = retrieve_artifact(artifact_path["name"])
    if "stats" in artifact:
        failed, success, time_spent = handle_test_results(artifact["stats"])
        doc_test_results["failures"] = failed
        doc_test_results["success"] = success
        doc_test_results["time_spent"] = time_spent[1:-1] + ", "

        all_failures = extract_first_line_failure(artifact["failures_short"])
        for line in artifact["summary_short"].split("\n"):
            if re.search("FAILED", line):
                line = line.replace("FAILED ", "")
                line = line.split()[0].replace("\n", "")

                if "::" in line:
                    file_path, test = line.split("::")
                else:
                    file_path, test = line, line

                for file_regex in docs.keys():
                    if fnmatch(file_path, file_regex):
                        category = docs[file_regex]
                        doc_test_results[category]["failed"].append(test)

                        failure = all_failures[test] if test in all_failures else "N/A"
                        doc_test_results[category]["failures"][test] = failure
                        break

    message = Message("🤗 Results of the doc tests.", doc_test_results)
    message.post()
    message.post_reply()
