SeaCogs/bible/bible.py

#   _____                         _
#  / ____|                       (_)
# | (___   ___  __ _ _____      ___ _ __ ___  _ __ ___   ___ _ __
#  \___ \ / _ \/ _` / __\ \ /\ / / | '_ ` _ \| '_ ` _ \ / _ \ '__|
#  ____) |  __/ (_| \__ \\ V  V /| | | | | | | | | | | |  __/ |
# |_____/ \___|\__,_|___/ \_/\_/ |_|_| |_| |_|_| |_| |_|\___|_|

import random
from asyncio import create_task
from io import BytesIO

import aiohttp
import numpy as np
from discord import Colour, Embed, File
from PIL import Image
from red_commons.logging import getLogger
from redbot.core import Config, commands, data_manager
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import bold, error, humanize_list

import bible.errors
from bible.models import Version


class Bible(commands.Cog):
    """Retrieve Bible verses from the API.bible API."""

    __author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
    __git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
    __version__ = "1.1.2"
    __documentation__ = "https://seacogs.coastalcommits.com/pterodactyl/"

    def __init__(self, bot: Red):
        super().__init__()
        self.bot = bot
        self.session = aiohttp.ClientSession()
        self.config = Config.get_conf(self, identifier=481923957134912, force_registration=True)
        self.logger = getLogger("red.SeaCogs.Bible")
        self.config.register_global(bible="de4e12af7f28f599-02")
        self.config.register_user(bible=None)

    async def cog_unload(self):
        create_task(self.session.close())

    def format_help_for_context(self, ctx: commands.Context) -> str:
        pre_processed = super().format_help_for_context(ctx) or ""
        n = "\n" if "\n\n" not in pre_processed else ""
        text = [
            f"{pre_processed}{n}",
            f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
            f"{bold('Author:')} {humanize_list(self.__author__)}",
            f"{bold('Documentation:')} {self.__documentation__}",
        ]
        return "\n".join(text)

    def get_icon(self, color: Colour) -> File:
        """Get the docs.api.bible favicon with a given color."""
        image_path = data_manager.bundled_data_path(self) / "api.bible-logo.png"
        image = Image.open(image_path)
        image = image.convert("RGBA")
        data = np.array(image)
        red, green, blue, alpha = data.T  # pylint: disable=unused-variable
        white_areas = (red == 255) & (blue == 255) & (green == 255)
        data[..., :-1][white_areas.T] = color.to_rgb()
        image = Image.fromarray(data)

        with BytesIO() as image_binary:
            image.save(image_binary, "PNG")
            image_binary.seek(0)
            return File(image_binary, filename="icon.png", description="API.Bible Icon")

    async def translate_book_name(self, bible_id: str, book_name: str) -> str:
        """Translate a book name to a book ID."""
        book_name_list = [w.lower() if w.lower() == "of" else w.title() for w in book_name.split()]
        book_name = " ".join(book_name_list)
        books = await self._get_books(bible_id)
        for book in books:
            if book_name in (book["abbreviation"], book["name"]):
                return book["id"]
        raise ValueError(error(f"Book {book_name} not found."))

    async def get_version(self, bible_id: str) -> Version:
        """Retrieve the version of the Bible being used."""
        url = f"https://api.scripture.api.bible/v1/bibles/{bible_id}"
        headers = await self.bot.get_shared_api_tokens("api.bible")

        async with self.session.get(url, headers=headers) as response:
            data = await response.json()
            self.logger.debug(
                "get_version executed with a response code of: %s",
                response.status,
            )
            if response.status == 401:
                raise bible.errors.Unauthorized()
            if response.status == 403:
                raise bible.errors.BibleAccessError()
            if response.status == 503:
                raise bible.errors.ServiceUnavailable()
            return Version(
                bible_id,
                data["data"]["abbreviation"],
                data["data"]["language"]["name"],
                data["data"]["abbreviationLocal"],
                data["data"]["language"]["nameLocal"],
                data["data"]["description"],
                data["data"]["descriptionLocal"],
                data["data"]["copyright"],
            )

    async def _get_passage(
        self,
        ctx: commands.Context,
        bible_id: str,
        passage_id: str,
        include_verse_numbers: bool,
    ) -> dict:
        """Get a Bible passage from the API.bible API."""
        url = f"https://api.scripture.api.bible/v1/bibles/{bible_id}/passages/{passage_id}"
        headers = await self.bot.get_shared_api_tokens("api.bible")
        params = {
            "fums-version": "3",
            "content-type": "text",
            "include-notes": "false",
            "include-titles": "false",
            "include-chapter-numbers": "false",
            "include-verse-numbers": str(include_verse_numbers).lower(),
            "include-verse-spans": "false",
            "use-org-id": "false",
        }

        async with self.session.get(url, headers=headers, params=params) as response:
            data = await response.json()
            self.logger.debug(
                "_get_passage executed with a response code of: %s",
                response.status,
            )
            if response.status == 400:
                raise bible.errors.InexplicableError()
            if response.status == 401:
                raise bible.errors.Unauthorized()
            if response.status == 403:
                raise bible.errors.BibleAccessError()
            if response.status == 404:
                raise bible.errors.NotFound()
            if response.status == 503:
                raise bible.errors.ServiceUnavailable()

        fums_url = "https://fums.api.bible/f3"
        fums_params = {
            "t": data["meta"]["fumsToken"],
            "dId": self.bot.user.id,
            "sId": ctx.message.created_at.timestamp(),
            "uId": hash(str(ctx.author.id)),
        }

        async with self.session.get(fums_url, params=fums_params) as response:
            self.logger.debug(
                "_get_passage FUMS executed with a response code of: %s\nDevice ID: %s\nSession ID: %s\nUser ID: %s (%s)",
                response.status,
                self.bot.user.id,
                ctx.message.created_at.timestamp(),
                hash(str(ctx.author.id)),
                ctx.author.id,
            )
        return data["data"]

    async def _get_books(self, bible_id: str) -> dict:
        """Get the books of the Bible from the API.bible API."""
        url = f"https://api.scripture.api.bible/v1/bibles/{bible_id}/books"
        headers = await self.bot.get_shared_api_tokens("api.bible")

        async with self.session.get(url, headers=headers) as response:
            data = await response.json()
            self.logger.debug(
                "_get_books executed with a response code of: %s",
                response.status,
            )
            if response.status == 401:
                raise bible.errors.Unauthorized()
            if response.status == 403:
                raise bible.errors.BibleAccessError()
            if response.status == 503:
                raise bible.errors.ServiceUnavailable()
            return data["data"]

    async def _get_chapters(self, bible_id: str, book_id: str) -> dict:
        """Get the chapters of a book from the API.bible API."""
        url = f"https://api.scripture.api.bible/v1/bibles/{bible_id}/books/{book_id}/chapters"
        headers = await self.bot.get_shared_api_tokens("api.bible")

        async with self.session.get(url, headers=headers) as response:
            data = await response.json()
            self.logger.debug(
                "_get_chapters executed with a response code of: %s",
                response.status,
            )
            if response.status == 401:
                raise bible.errors.Unauthorized()
            if response.status == 403:
                raise bible.errors.BibleAccessError()
            if response.status == 503:
                raise bible.errors.ServiceUnavailable()
            return data["data"]

    async def _get_verses(self, bible_id: str, book_id: str, chapter: int) -> dict:
        """Get the verses of a chapter from the API.bible API."""
        url = f"https://api.scripture.api.bible/v1/bibles/{bible_id}/chapters/{book_id}.{chapter}/verses"
        headers = await self.bot.get_shared_api_tokens("api.bible")

        async with self.session.get(url, headers=headers) as response:
            data = await response.json()
            self.logger.debug(
                "_get_verses executed with a response code of: %s",
                response.status,
            )
            if response.status == 401:
                raise bible.errors.Unauthorized()
            if response.status == 403:
                raise bible.errors.BibleAccessError()
            if response.status == 503:
                raise bible.errors.ServiceUnavailable()
            return data["data"]

    @commands.group(autohelp=True)
    async def bible(self, ctx: commands.Context):
        """Retrieve Bible verses from the API.bible API."""

    @bible.command(name="passage", aliases=["verse"])
    async def bible_passage(self, ctx: commands.Context, book: str, passage: str):
        """Get a Bible passage.

        Example usage:
            `[p]bible passage Genesis 1:1`
            `[p]bible passage John 3:16-3:17`"""
        bible_id = await self.config.bible()

        try:
            book_id = await self.translate_book_name(bible_id, book)
        except ValueError as e:
            await ctx.send(str(e))
            return

        try:
            version = await self.get_version(bible_id)
            if len(passage.split("-")) == 2:
                from_verse, to_verse = passage.replace(":", ".").split("-")
                if "." not in to_verse:
                    to_verse = f"{from_verse.split('.')[0]}.{to_verse}"
                passage = await self._get_passage(ctx, bible_id, f"{book_id}.{from_verse}-{book_id}.{to_verse}", True)
            else:
                passage = await self._get_passage(ctx, bible_id, f"{book_id}.{passage.replace(':', '.')}", False)
        except (
            bible.errors.BibleAccessError,
            bible.errors.NotFound,
            bible.errors.InexplicableError,
            bible.errors.ServiceUnavailable,
            bible.errors.Unauthorized,
        ) as e:
            await ctx.send(e.message)
            return

        if len(passage["content"]) > 4096:
            await ctx.send("The passage is too long to send.")
            return

        if await ctx.embed_requested():
            icon = self.get_icon(await ctx.embed_color())
            embed = Embed(
                title=f"{passage['reference']}",
                description=passage["content"].replace("¶ ", ""),
                color=await ctx.embed_color(),
            )
            embed.set_footer(text=f"{ctx.prefix}bible passage - Powered by API.Bible - {version.abbreviationLocal} ({version.languageLocal}, {version.descriptionLocal})", icon_url="attachment://icon.png")
            await ctx.send(embed=embed, file=icon)
        else:
            await ctx.send(f"## {passage['reference']}\n{passage['content']}")

    @bible.command(name="random")
    async def bible_random(self, ctx: commands.Context):
        """Get a random Bible verse."""
        bible_id = await self.config.bible()

        try:
            version = await self.get_version(bible_id)
            books = await self._get_books(bible_id)
            book = random.choice(books)

            chapters = await self._get_chapters(bible_id, book["id"])
            chapter = random.choice(chapters)

            verses = await self._get_verses(bible_id, book["id"], chapter["number"])
            verse = random.choice(verses)["id"]

            passage = await self._get_passage(ctx, bible_id, verse, False)
        except (
            bible.errors.BibleAccessError,
            bible.errors.NotFound,
            bible.errors.InexplicableError,
            bible.errors.ServiceUnavailable,
            bible.errors.Unauthorized,
        ) as e:
            await ctx.send(e.message)
            return

        if await ctx.embed_requested():
            icon = self.get_icon(await ctx.embed_color())
            embed = Embed(
                title=f"{passage['reference']}",
                description=passage["content"].replace("¶ ", ""),
                color=await ctx.embed_color(),
            )
            embed.set_footer(text=f"{ctx.prefix}bible random - Powered by API.Bible - {version.abbreviationLocal} ({version.languageLocal}, {version.descriptionLocal})", icon_url="attachment://icon.png")
            await ctx.send(embed=embed, file=icon)
        else:
            await ctx.send(f"## {passage['reference']}\n{passage['content']}")