"""Helper functions for Mail and Packages."""

from __future__ import annotations

import base64
import datetime
import email
import hashlib
import imaplib
import logging
import os
import quopri
import re
import subprocess  # nosec
import uuid
from datetime import timezone
from email.header import decode_header
from shutil import copyfile, copytree, which
from typing import Any, List, Optional, Type, Union

import aiohttp
import dateparser
import homeassistant.helpers.config_validation as cv
from bs4 import BeautifulSoup
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONF_HOST,
    CONF_PASSWORD,
    CONF_PORT,
    CONF_RESOURCES,
    CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.util import ssl
from PIL import Image, ImageOps

from .const import (
    AMAZON_DELIVERED,
    AMAZON_DELIVERED_SUBJECT,
    AMAZON_EXCEPTION,
    AMAZON_EXCEPTION_ORDER,
    AMAZON_EXCEPTION_SUBJECT,
    AMAZON_HUB,
    AMAZON_HUB_BODY,
    AMAZON_HUB_CODE,
    AMAZON_HUB_EMAIL,
    AMAZON_HUB_SUBJECT,
    AMAZON_HUB_SUBJECT_SEARCH,
    AMAZON_IMG_PATTERN,
    AMAZON_ORDER,
    AMAZON_PACKAGES,
    AMAZON_PATTERN,
    AMAZON_SHIPMENT_TRACKING,
    AMAZON_TIME_PATTERN,
    AMAZON_TIME_PATTERN_END,
    ATTR_AMAZON_IMAGE,
    ATTR_BODY,
    ATTR_CODE,
    ATTR_COUNT,
    ATTR_EMAIL,
    ATTR_IMAGE_NAME,
    ATTR_IMAGE_PATH,
    ATTR_ORDER,
    ATTR_PATTERN,
    ATTR_SUBJECT,
    ATTR_TRACKING,
    ATTR_USPS_MAIL,
    CONF_ALLOW_EXTERNAL,
    CONF_AMAZON_DAYS,
    CONF_AMAZON_DOMAIN,
    CONF_AMAZON_FWDS,
    CONF_CUSTOM_IMG,
    CONF_CUSTOM_IMG_FILE,
    CONF_DURATION,
    CONF_FOLDER,
    CONF_GENERATE_MP4,
    CONF_IMAP_SECURITY,
    CONF_STORAGE,
    CONF_VERIFY_SSL,
    DEFAULT_AMAZON_DAYS,
    OVERLAY,
    SENSOR_DATA,
    SENSOR_TYPES,
    SHIPPERS,
)

NO_SSL = "Email will be accessed without encryption using this method and is not recommended."
_LOGGER = logging.getLogger(__name__)

# Config Flow Helpers


def get_resources() -> dict:
    """Resource selection schema.

    Returns dict of user selected sensors
    """
    known_available_resources = {
        sensor_id: sensor.name for sensor_id, sensor in SENSOR_TYPES.items()
    }

    return known_available_resources


async def _check_ffmpeg() -> bool:
    """Check if ffmpeg is installed.

    Returns boolean
    """
    return which("ffmpeg")


async def _test_login(
    host: str, port: int, user: str, pwd: str, security: str, verify: bool
) -> bool:
    """Test IMAP login to specified server.

    Returns success boolean
    """
    # Catch invalid mail server / host names
    try:
        if security == "SSL":
            if not verify:
                context = ssl.client_context_no_verify()
            else:
                context = ssl.client_context()
            account = imaplib.IMAP4_SSL(host=host, port=port, ssl_context=context)
        elif security == "startTLS":
            if not verify:
                context = ssl.client_context_no_verify()
            else:
                context = ssl.client_context()
            account = imaplib.IMAP4(host=host, port=port)
            account.starttls(context)
        else:
            _LOGGER.warning(NO_SSL)
            account = imaplib.IMAP4(host=host, port=port)
    except Exception as err:
        _LOGGER.error("Error connecting into IMAP Server: %s", str(err))
        return False
    # Validate we can login to mail server
    try:
        account.login(user, pwd)
        return True
    except Exception as err:
        _LOGGER.error("Error logging into IMAP Server: %s", str(err))
        return False


# Email Data helpers


def default_image_path(
    hass: HomeAssistant, config_entry: ConfigEntry  # pylint: disable=unused-argument
) -> str:
    """Return value of the default image path.

    Returns the default path based on logic
    """
    storage = None
    try:
        storage = config_entry.get(CONF_STORAGE)
    except AttributeError:
        storage = config_entry.data[CONF_STORAGE]

    if storage:
        return storage
    return "custom_components/mail_and_packages/images/"


def process_emails(hass: HomeAssistant, config: ConfigEntry) -> dict:
    """Process emails and return value.

    Returns dict containing sensor data
    """
    host = config.get(CONF_HOST)
    port = config.get(CONF_PORT)
    user = config.get(CONF_USERNAME)
    pwd = config.get(CONF_PASSWORD)
    folder = config.get(CONF_FOLDER)
    resources = config.get(CONF_RESOURCES)
    imap_security = config.get(CONF_IMAP_SECURITY)
    verify_ssl = config.get(CONF_VERIFY_SSL)

    # Create the dict container
    data = {}

    # Login to email server and select the folder
    account = login(host, port, user, pwd, imap_security, verify_ssl)

    # Do not process if account returns false
    if not account:
        return data

    if not selectfolder(account, folder):
        # Bail out on error
        return data

    # Create image file name dict container
    _image = {}

    # USPS Mail Image name
    image_name = image_file_name(hass, config)
    _LOGGER.debug("Image name: %s", image_name)
    _image[ATTR_IMAGE_NAME] = image_name

    # Amazon delivery image name
    image_name = image_file_name(hass, config, True)
    _LOGGER.debug("Amazon Image Name: %s", image_name)
    _image[ATTR_AMAZON_IMAGE] = image_name

    image_path = default_image_path(hass, config)
    _LOGGER.debug("Image path: %s", image_path)
    _image[ATTR_IMAGE_PATH] = image_path
    data.update(_image)

    # Only update sensors we're intrested in
    for sensor in resources:
        try:
            fetch(hass, config, account, data, sensor)
        except Exception as err:
            _LOGGER.error("Error updating sensor: %s reason: %s", sensor, err)

    # Copy image file to www directory if enabled
    if config.get(CONF_ALLOW_EXTERNAL):
        copy_images(hass, config)

    return data


def copy_images(hass: HomeAssistant, config: ConfigEntry) -> None:
    """Copy images to www directory if enabled."""
    paths = []
    src = f"{hass.config.path()}/{default_image_path(hass, config)}"
    dst = f"{hass.config.path()}/www/mail_and_packages/"

    # Setup paths list
    paths.append(dst)
    paths.append(dst + "amazon/")

    # Clean up the destination directory
    for path in paths:
        # Path check
        path_check = os.path.exists(path)
        if not path_check:
            try:
                os.makedirs(path)
            except OSError as err:
                _LOGGER.error("Problem creating: %s, error returned: %s", path, err)
                return
        cleanup_images(path)

    try:
        copytree(src, dst, dirs_exist_ok=True)
    except Exception as err:
        _LOGGER.error(
            "Problem copying files from %s to %s error returned: %s", src, dst, err
        )
        return


def image_file_name(
    hass: HomeAssistant, config: ConfigEntry, amazon: bool = False
) -> str:
    """Determine if filename is to be changed or not.

    Returns filename
    """
    mail_none = None
    path = None
    image_name = None

    if amazon:
        mail_none = f"{os.path.dirname(__file__)}/no_deliveries.jpg"
        image_name = "no_deliveries.jpg"
        path = f"{hass.config.path()}/{default_image_path(hass, config)}amazon"
    else:
        path = f"{hass.config.path()}/{default_image_path(hass, config)}"
        if config.get(CONF_CUSTOM_IMG):
            mail_none = config.get(CONF_CUSTOM_IMG_FILE)
        else:
            mail_none = f"{os.path.dirname(__file__)}/mail_none.gif"
        image_name = os.path.split(mail_none)[1]

    # Path check
    path_check = os.path.exists(path)
    if not path_check:
        try:
            os.makedirs(path)
        except OSError as err:
            _LOGGER.error("Problem creating: %s, error returned: %s", path, err)
            return image_name

    # SHA1 file hash check
    try:
        sha1 = hash_file(mail_none)
    except OSError as err:
        _LOGGER.error("Problem accessing file: %s, error returned: %s", mail_none, err)
        return image_name

    ext = None
    ext = ".jpg" if amazon else ".gif"

    for file in os.listdir(path):
        if file.endswith(".gif") or (file.endswith(".jpg") and amazon):
            try:
                created = datetime.datetime.fromtimestamp(
                    os.path.getctime(os.path.join(path, file))
                ).strftime("%d-%b-%Y")
            except OSError as err:
                _LOGGER.error(
                    "Problem accessing file: %s, error returned: %s", file, err
                )
                return image_name
            today = get_formatted_date()
            _LOGGER.debug("Created: %s, Today: %s", created, today)
            # If image isn't mail_none and not created today,
            # return a new filename
            if sha1 != hash_file(os.path.join(path, file)) and today != created:
                image_name = f"{str(uuid.uuid4())}{ext}"
            else:
                image_name = file

    # If we find no images in the image directory generate a new filename
    if image_name in mail_none:
        image_name = f"{str(uuid.uuid4())}{ext}"
    _LOGGER.debug("Image Name: %s", image_name)

    # Insert place holder image
    _LOGGER.debug("Copying %s to %s", mail_none, os.path.join(path, image_name))

    copyfile(mail_none, os.path.join(path, image_name))

    return image_name


def hash_file(filename: str) -> str:
    """Return the SHA-1 hash of the file passed into it.

    Returns hash of file as string
    """
    # make a hash object
    the_hash = hashlib.sha1()  # nosec

    # open file for reading in binary mode
    with open(filename, "rb") as file:
        # loop till the end of the file
        chunk = 0
        while chunk != b"":
            # read only 1024 bytes at a time
            chunk = file.read(1024)
            the_hash.update(chunk)

    # return the hex representation of digest
    return the_hash.hexdigest()


def fetch(
    hass: HomeAssistant, config: ConfigEntry, account: Any, data: dict, sensor: str
) -> int:
    """Fetch data for a single sensor, including any sensors it depends on.

    Returns integer of sensor passed to it
    """
    if sensor in data:
        return data[sensor]

    img_out_path = f"{hass.config.path()}/{default_image_path(hass, config)}"
    gif_duration = config.get(CONF_DURATION)
    generate_mp4 = config.get(CONF_GENERATE_MP4)
    amazon_fwds = cv.ensure_list_csv(config.get(CONF_AMAZON_FWDS))
    image_name = data[ATTR_IMAGE_NAME]
    amazon_image_name = data[ATTR_AMAZON_IMAGE]
    amazon_days = config.get(CONF_AMAZON_DAYS)

    # Conditional variables
    nomail = (
        config.get(CONF_CUSTOM_IMG_FILE) if config.get(CONF_CUSTOM_IMG_FILE) else None
    )
    amazon_domain = (
        config.get(CONF_AMAZON_DOMAIN) if config.get(CONF_AMAZON_DOMAIN) else None
    )

    count = {}

    if sensor == "usps_mail":
        count[sensor] = get_mails(
            account,
            img_out_path,
            gif_duration,
            image_name,
            generate_mp4,
            nomail,
        )
    elif sensor == AMAZON_PACKAGES:
        count[sensor] = get_items(
            account,
            ATTR_COUNT,
            amazon_fwds,
            amazon_days,
            amazon_domain,
        )
        count[AMAZON_ORDER] = get_items(
            account,
            ATTR_ORDER,
            amazon_fwds,
            amazon_days,
            amazon_domain,
        )
    elif sensor == AMAZON_HUB:
        value = amazon_hub(account, amazon_fwds)
        count[sensor] = value[ATTR_COUNT]
        count[AMAZON_HUB_CODE] = value[ATTR_CODE]
    elif sensor == AMAZON_EXCEPTION:
        info = amazon_exception(account, amazon_fwds, amazon_domain)
        count[sensor] = info[ATTR_COUNT]
        count[AMAZON_EXCEPTION_ORDER] = info[ATTR_ORDER]
    elif "_packages" in sensor:
        prefix = sensor.replace("_packages", "")
        delivering = fetch(hass, config, account, data, f"{prefix}_delivering")
        delivered = fetch(hass, config, account, data, f"{prefix}_delivered")
        count[sensor] = delivering + delivered
    elif "_delivering" in sensor:
        prefix = sensor.replace("_delivering", "")
        delivered = fetch(hass, config, account, data, f"{prefix}_delivered")
        info = get_count(account, sensor, True, amazon_domain=amazon_domain)
        count[sensor] = max(0, info[ATTR_COUNT] - delivered)
        count[f"{prefix}_tracking"] = info[ATTR_TRACKING]
    elif sensor == "zpackages_delivered":
        count[sensor] = 0  # initialize the variable
        for shipper in SHIPPERS:
            delivered = f"{shipper}_delivered"
            if delivered in data and delivered != sensor:
                count[sensor] += fetch(hass, config, account, data, delivered)
    elif sensor == "zpackages_transit":
        total = 0
        for shipper in SHIPPERS:
            delivering = f"{shipper}_delivering"
            if delivering in data and delivering != sensor:
                total += fetch(hass, config, account, data, delivering)
        count[sensor] = max(0, total)
    elif sensor == "mail_updated":
        count[sensor] = update_time()
    else:
        count[sensor] = get_count(
            account,
            sensor,
            False,
            img_out_path,
            hass,
            amazon_image_name,
            amazon_domain,
            amazon_fwds,
        )[ATTR_COUNT]

    data.update(count)
    _LOGGER.debug("Sensor: %s Count: %s", sensor, str(count[sensor]))
    return count[sensor]


def login(
    host: str, port: int, user: str, pwd: str, security: str, verify: bool = True
) -> Union[bool, Type[imaplib.IMAP4_SSL]]:
    """Login to IMAP server.

    Returns account object
    """
    try:
        if security == "SSL":
            if not verify:
                context = ssl.client_context_no_verify()
            else:
                context = ssl.client_context()
            account = imaplib.IMAP4_SSL(host=host, port=port, ssl_context=context)
        elif security == "startTLS":
            if not verify:
                context = ssl.client_context_no_verify()
            else:
                context = ssl.client_context()
            account = imaplib.IMAP4(host=host, port=port)
            account.starttls(context)
        else:
            account = imaplib.IMAP4(host=host, port=port)

    except Exception as err:
        _LOGGER.error("Network error while connecting to server: %s", str(err))
        return False

    # If login fails give error message
    try:
        account.login(user, pwd)
    except Exception as err:
        _LOGGER.error("Error logging into IMAP Server: %s", str(err))
        return False

    return account


def selectfolder(account: Type[imaplib.IMAP4_SSL], folder: str) -> bool:
    """Select folder inside the mailbox."""
    try:
        account.list()
    except Exception as err:
        _LOGGER.error("Error listing folders: %s", str(err))
        return False
    try:
        account.select(folder, readonly=True)
    except Exception as err:
        _LOGGER.error("Error selecting folder: %s", str(err))
        return False
    return True


def get_formatted_date() -> str:
    """Return today in specific format.

    Returns current timestamp as string
    """
    today = datetime.datetime.today().strftime("%d-%b-%Y")
    #
    # for testing
    # today = "11-Jan-2021"
    #
    return today


def update_time() -> Any:
    """Get update time.

    Returns current timestamp as string
    """
    # updated = datetime.datetime.now().strftime("%b-%d-%Y %I:%M %p")
    # updated = datetime.datetime.now(timezone.utc).isoformat(timespec="minutes")
    updated = datetime.datetime.now(timezone.utc)

    return updated


def build_search(address: list, date: str, subject: str = None) -> tuple:
    """Build IMAP search query.

    Return tuple of utf8 flag and search query.
    """
    the_date = f"SINCE {date}"
    imap_search = None
    utf8_flag = False
    prefix_list = None
    email_list = None

    if isinstance(address, list):
        if len(address) == 1:
            email_list = address[0]
        else:
            email_list = '" FROM "'.join(address)
            prefix_list = " ".join(["OR"] * (len(address) - 1))
    else:
        email_list = address

    _LOGGER.debug("DEBUG subject: %s", subject)

    if subject is not None:
        if not subject.isascii():
            utf8_flag = True
            if prefix_list is not None:
                imap_search = f'{prefix_list} FROM "{email_list}" {the_date} SUBJECT'
            else:
                imap_search = f'FROM "{email_list}" {the_date} SUBJECT'
            # imap_search = f"{the_date} SUBJECT"
        else:
            if prefix_list is not None:
                imap_search = f'({prefix_list} FROM "{email_list}" SUBJECT "{subject}" {the_date})'
            else:
                imap_search = f'(FROM "{email_list}" SUBJECT "{subject}" {the_date})'
    else:
        if prefix_list is not None:
            imap_search = f'({prefix_list} FROM "{email_list}" {the_date})'
        else:
            imap_search = f'(FROM "{email_list}" {the_date})'

    _LOGGER.debug("DEBUG imap_search: %s", imap_search)

    return (utf8_flag, imap_search)


def email_search(
    account: Type[imaplib.IMAP4_SSL], address: list, date: str, subject: str = None
) -> tuple:
    """Search emails with from, subject, senton date.

    Returns a tuple
    """
    utf8_flag, search = build_search(address, date, subject)
    value = ("", [""])

    if utf8_flag:
        subject = subject.encode("utf-8")
        account.literal = subject
        try:
            value = account.search("utf-8", search)
        except Exception as err:
            _LOGGER.debug(
                "Error searching emails with unicode characters: %s", str(err)
            )
            value = "BAD", err.args[0]
    else:
        try:
            value = account.search(None, search)
        except Exception as err:
            _LOGGER.error("Error searching emails: %s", str(err))
            value = "BAD", err.args[0]

    _LOGGER.debug("email_search value: %s", value)

    (check, new_value) = value
    if new_value[0] is None:
        _LOGGER.debug("email_search value was invalid: None")
        value = (check, [b""])

    return value


def email_fetch(
    account: Type[imaplib.IMAP4_SSL], num: int, parts: str = "(RFC822)"
) -> tuple:
    """Download specified email for parsing.

    Returns tuple
    """
    # iCloud doesn't support RFC822 so override the 'message parts'
    if account.host == "imap.mail.me.com":
        parts = "BODY[]"

    try:
        value = account.fetch(num, parts)
    except Exception as err:
        _LOGGER.error("Error fetching emails: %s", str(err))
        value = "BAD", err.args[0]

    return value


def get_mails(
    account: Type[imaplib.IMAP4_SSL],
    image_output_path: str,
    gif_duration: int,
    image_name: str,
    gen_mp4: bool = False,
    custom_img: str = None,
) -> int:
    """Create GIF image based on the attachments in the inbox."""
    image_count = 0
    images = []
    images_delete = []
    msg = ""

    _LOGGER.debug("Attempting to find Informed Delivery mail")
    _LOGGER.debug("Informed delivery search date: %s", get_formatted_date())

    (server_response, data) = email_search(
        account,
        SENSOR_DATA[ATTR_USPS_MAIL][ATTR_EMAIL],
        get_formatted_date(),
        SENSOR_DATA[ATTR_USPS_MAIL][ATTR_SUBJECT][0],
    )

    # Bail out on error
    if server_response != "OK" or data[0] is None:
        return image_count

    # Check to see if the path exists, if not make it
    if not os.path.isdir(image_output_path):
        try:
            os.makedirs(image_output_path)
        except Exception as err:
            _LOGGER.critical("Error creating directory: %s", str(err))

    # Clean up image directory
    _LOGGER.debug("Cleaning up image directory: %s", str(image_output_path))
    cleanup_images(image_output_path)

    # Copy overlays to image directory
    _LOGGER.debug("Checking for overlay files in: %s", str(image_output_path))
    copy_overlays(image_output_path)

    if server_response == "OK":
        _LOGGER.debug("Informed Delivery email found processing...")
        for num in data[0].split():
            msg = email_fetch(account, num, "(RFC822)")[1]
            for response_part in msg:
                if isinstance(response_part, tuple):
                    msg = email.message_from_bytes(response_part[1])
                    _LOGGER.debug("msg: %s", msg)

                    # walking through the email parts to find images
                    for part in msg.walk():
                        if part.get_content_type() == "text/html":
                            _LOGGER.debug("Found html email processing...")
                            part = part.get_payload(decode=True)
                            part = part.decode("utf-8", "ignore")
                            soup = BeautifulSoup(part, "html.parser")
                            found_images = soup.find_all(id="mailpiece-image-src-id")
                            if not found_images:
                                continue
                            if "data:image/jpeg;base64" not in part:
                                _LOGGER.debug("Unexpected html format found.")
                                continue
                            _LOGGER.debug("Found images: %s", bool(found_images))

                            # Convert all the images to binary data
                            for image in found_images:
                                filename = random_filename()
                                data = str(image["src"]).split(",")[1]
                                try:
                                    with open(
                                        image_output_path + filename, "wb"
                                    ) as the_file:
                                        the_file.write(base64.b64decode(data))
                                        images.append(image_output_path + filename)
                                        image_count = image_count + 1
                                except Exception as err:
                                    _LOGGER.critical(
                                        "Error opening filepath: %s", str(err)
                                    )
                                    return image_count

                        # Log error message if we are unable to open the filepath for
                        # some reason
                        elif part.get_content_type() == "image/jpeg":
                            _LOGGER.debug("Extracting image from email")
                            filename = part.get_filename()
                            junkmail = ["mailer", "content"]
                            if any(junk in filename for junk in junkmail):
                                _LOGGER.debug("Discarding junk mail.")
                                continue
                            try:
                                with open(
                                    image_output_path + filename, "wb"
                                ) as the_file:
                                    the_file.write(part.get_payload(decode=True))
                                    images.append(image_output_path + filename)
                                    image_count = image_count + 1
                            except Exception as err:
                                _LOGGER.critical("Error opening filepath: %s", str(err))
                                return image_count

                        elif part.get_content_type() == "multipart":
                            continue

        # Remove duplicate images
        _LOGGER.debug("Removing duplicate images.")
        images = list(dict.fromkeys(images))

        # Create copy of image list for deleting temporary images
        images_delete = images[:]

        # Look for mail pieces without images image
        if re.compile(r"\bimage-no-mailpieces?700\.jpg\b").search(str(msg)) is not None:
            images.append(os.path.dirname(__file__) + "/image-no-mailpieces700.jpg")
            image_count = image_count + 1
            _LOGGER.debug("Placeholder image found using: image-no-mailpieces700.jpg.")

        # Remove USPS announcement images
        _LOGGER.debug("Removing USPS announcement images.")
        images = [
            el
            for el in images
            if not any(
                ignore in el
                for ignore in ["mailerProvidedImage", "ra_0", "Mail Attachment.txt"]
            )
        ]
        image_count = len(images)
        _LOGGER.debug("Image Count: %s", str(image_count))

        if image_count > 0:
            all_images = []

            _LOGGER.debug("Resizing images to 724x320...")
            # Resize images to 724x320
            all_images = resize_images(images, 724, 320)

            # Create copy of image list for deleting temporary images
            for image in all_images:
                images_delete.append(image)

            # Create numpy array of images
            _LOGGER.debug("Creating array of image files...")
            img, *imgs = [Image.open(file) for file in all_images]

            try:
                _LOGGER.debug("Generating animated GIF")
                # Use Pillow to create mail images
                img.save(
                    fp=os.path.join(image_output_path, image_name),
                    format="GIF",
                    append_images=imgs,
                    save_all=True,
                    duration=gif_duration * 1000,
                    loop=0,
                )
                _LOGGER.debug("Mail image generated.")
            except Exception as err:
                _LOGGER.error("Error attempting to generate image: %s", str(err))
            for image in images_delete:
                cleanup_images(f"{os.path.split(image)[0]}/", os.path.split(image)[1])

        elif image_count == 0:
            _LOGGER.debug("No mail found.")
            if os.path.isfile(image_output_path + image_name):
                _LOGGER.debug("Removing " + image_output_path + image_name)
                cleanup_images(image_output_path, image_name)

            try:
                _LOGGER.debug("Copying nomail gif")
                if custom_img is not None:
                    nomail = custom_img
                else:
                    nomail = os.path.dirname(__file__) + "/mail_none.gif"
                copyfile(nomail, image_output_path + image_name)
            except Exception as err:
                _LOGGER.error("Error attempting to copy image: %s", str(err))

        if gen_mp4:
            _generate_mp4(image_output_path, image_name)

    return image_count


def random_filename(ext: str = ".jpg") -> str:
    """Generate random filename."""
    return f"{str(uuid.uuid4())}{ext}"


def _generate_mp4(path: str, image_file: str) -> None:
    """Generate mp4 from gif.

    use a subprocess so we don't lock up the thread
    comamnd: ffmpeg -f gif -i infile.gif outfile.mp4
    """
    gif_image = os.path.join(path, image_file)
    mp4_file = os.path.join(path, image_file.replace(".gif", ".mp4"))
    filecheck = os.path.isfile(mp4_file)
    _LOGGER.debug("Generating mp4: %s", mp4_file)
    if filecheck:
        cleanup_images(os.path.split(mp4_file))
        _LOGGER.debug("Removing old mp4: %s", mp4_file)

    # TODO: find a way to call ffmpeg the right way from HA
    subprocess.call(
        [
            "ffmpeg",
            "-i",
            gif_image,
            "-pix_fmt",
            "yuv420p",
            mp4_file,
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )


def resize_images(images: list, width: int, height: int) -> list:
    """Resize images.

    This should keep the aspect ratio of the images
    Returns list of images
    """
    all_images = []
    for image in images:
        try:
            with open(image, "rb") as fd_img:
                try:
                    img = Image.open(fd_img)
                    img.thumbnail((width, height), resample=Image.Resampling.LANCZOS)

                    # Add padding as needed
                    img = ImageOps.pad(
                        img, (width, height), method=Image.Resampling.LANCZOS
                    )
                    # Crop to size
                    img = img.crop((0, 0, width, height))

                    pre = os.path.splitext(image)[0]
                    image = pre + ".gif"
                    img.save(image, img.format)
                    fd_img.close()
                    all_images.append(image)
                except Exception as err:
                    _LOGGER.error(
                        "Error attempting to read image %s: %s", str(image), str(err)
                    )
                    continue
        except Exception as err:
            _LOGGER.error("Error attempting to open image %s: %s", str(image), str(err))
            continue

    return all_images


def copy_overlays(path: str) -> None:
    """Copy overlay images to image output path."""
    overlays = OVERLAY
    check = all(item in overlays for item in os.listdir(path))

    # Copy files if they are missing
    if not check:
        for file in overlays:
            _LOGGER.debug("Copying file to: %s", str(path + file))
            copyfile(
                os.path.dirname(__file__) + "/" + file,
                path + file,
            )


def cleanup_images(path: str, image: Optional[str] = None) -> None:
    """Clean up image storage directory.

    Only supose to delete .gif, .mp4, and .jpg files
    """
    if image is not None:
        try:
            os.remove(path + image)
        except Exception as err:
            _LOGGER.error("Error attempting to remove image: %s", str(err))
        return

    for file in os.listdir(path):
        if file.endswith(".gif") or file.endswith(".mp4") or file.endswith(".jpg"):
            try:
                os.remove(path + file)
            except Exception as err:
                _LOGGER.error("Error attempting to remove found image: %s", str(err))


def get_count(
    account: Type[imaplib.IMAP4_SSL],
    sensor_type: str,
    get_tracking_num: bool = False,
    image_path: Optional[str] = None,
    hass: Optional[HomeAssistant] = None,
    amazon_image_name: Optional[str] = None,
    amazon_domain: Optional[str] = None,
    amazon_fwds: Optional[str] = None,
) -> dict:
    """Get Package Count.

    Returns dict of sensor data
    """
    count = 0
    tracking = []
    result = {}
    today = get_formatted_date()
    track = None
    found = []

    # Return Amazon delivered info
    if sensor_type == AMAZON_DELIVERED:
        result[ATTR_COUNT] = amazon_search(
            account, image_path, hass, amazon_image_name, amazon_domain, amazon_fwds
        )
        result[ATTR_TRACKING] = ""
        return result

    # Bail out if unknown sensor type
    if ATTR_EMAIL not in SENSOR_DATA[sensor_type]:
        _LOGGER.debug("Unknown sensor type: %s", str(sensor_type))
        result[ATTR_COUNT] = count
        result[ATTR_TRACKING] = ""
        return result

    subjects = SENSOR_DATA[sensor_type][ATTR_SUBJECT]
    for subject in subjects:
        _LOGGER.debug(
            "Attempting to find mail from (%s) with subject (%s)",
            SENSOR_DATA[sensor_type][ATTR_EMAIL],
            subject,
        )

        (server_response, data) = email_search(
            account, SENSOR_DATA[sensor_type][ATTR_EMAIL], today, subject
        )
        if server_response == "OK" and data[0] is not None:
            if ATTR_BODY in SENSOR_DATA[sensor_type].keys():
                count += find_text(data, account, SENSOR_DATA[sensor_type][ATTR_BODY])
            else:
                count += len(data[0].split())

            _LOGGER.debug(
                "Search for (%s) with subject (%s) results: %s count: %s",
                SENSOR_DATA[sensor_type][ATTR_EMAIL],
                subject,
                data[0],
                count,
            )
            found.append(data[0])

    if (
        f"{'_'.join(sensor_type.split('_')[:-1])}_tracking" in SENSOR_DATA
        and ATTR_PATTERN
        in SENSOR_DATA[f"{'_'.join(sensor_type.split('_')[:-1])}_tracking"].keys()
    ):
        track = SENSOR_DATA[f"{'_'.join(sensor_type.split('_')[:-1])}_tracking"][
            ATTR_PATTERN
        ][0]

    if track is not None and get_tracking_num and count > 0:
        for sdata in found:
            tracking.extend(get_tracking(sdata, account, track))
        tracking = list(dict.fromkeys(tracking))

    if len(tracking) > 0:
        # Use tracking numbers found for count (more accurate)
        count = len(tracking)

    result[ATTR_TRACKING] = tracking

    result[ATTR_COUNT] = count
    return result


def get_tracking(
    sdata: Any, account: Type[imaplib.IMAP4_SSL], the_format: Optional[str] = None
) -> list:
    """Parse tracking numbers from email.

    Returns list of tracking numbers
    """
    tracking = []
    pattern = None
    mail_list = sdata.split()
    _LOGGER.debug("Searching for tracking numbers in %s messages...", len(mail_list))

    pattern = re.compile(rf"{the_format}")
    for i in mail_list:
        data = email_fetch(account, i, "(RFC822)")[1]
        for response_part in data:
            if isinstance(response_part, tuple):
                msg = email.message_from_bytes(response_part[1])
                _LOGGER.debug("Checking message subject...")

                # Search subject for a tracking number
                email_subject = msg["subject"]
                if (found := pattern.findall(email_subject)) and len(found) > 0:
                    _LOGGER.debug(
                        "Found tracking number in email subject: (%s)",
                        found[0],
                    )
                    if found[0] not in tracking:
                        tracking.append(found[0])
                    continue

                # Search in email body for tracking number
                _LOGGER.debug("Checking message body using %s ...", the_format)
                for part in msg.walk():
                    _LOGGER.debug("Content type: %s", part.get_content_type())
                    if part.get_content_type() not in ["text/html", "text/plain"]:
                        continue
                    email_msg = part.get_payload(decode=True)
                    email_msg = email_msg.decode("utf-8", "ignore")
                    if (found := pattern.findall(email_msg)) and len(found) > 0:
                        # DHL is special
                        if " " in the_format:
                            found[0] = found[0].split(" ")[1]

                        _LOGGER.debug(
                            "Found tracking number in email body: %s", found[0]
                        )
                        if found[0] not in tracking:
                            tracking.append(found[0])
                        continue

    if len(tracking) == 0:
        _LOGGER.debug("No tracking numbers found")

    return tracking


def find_text(sdata: Any, account: Type[imaplib.IMAP4_SSL], search_terms: list) -> int:
    """Filter for specific words in email.

    Return count of items found as integer
    """
    _LOGGER.debug("Searching for (%s) in (%s) emails", search_terms, len(sdata))
    mail_list = sdata[0].split()
    count = 0
    found = None

    for i in mail_list:
        data = email_fetch(account, i, "(RFC822)")[1]
        for response_part in data:
            if isinstance(response_part, tuple):
                msg = email.message_from_bytes(response_part[1])

                for part in msg.walk():
                    for search in search_terms:
                        _LOGGER.debug("Content type: %s", part.get_content_type())
                        if part.get_content_type() not in ["text/html", "text/plain"]:
                            continue
                        email_msg = part.get_payload(decode=True)
                        email_msg = email_msg.decode("utf-8", "ignore")
                        pattern = re.compile(rf"{search}")
                        if (found := pattern.findall(email_msg)) and len(found) > 0:
                            _LOGGER.debug(
                                "Found (%s) in email %s times.", search, str(len(found))
                            )
                            count += len(found)

    _LOGGER.debug("Search for (%s) count results: %s", search_terms, count)
    return count


def amazon_search(
    account: Type[imaplib.IMAP4_SSL],
    image_path: str,
    hass: HomeAssistant,
    amazon_image_name: str,
    amazon_domain: str,
    fwds: str = None,
) -> int:
    """Find Amazon Delivered email.

    Returns email found count as integer
    """
    _LOGGER.debug("Searching for Amazon delivered email(s)...")

    subjects = AMAZON_DELIVERED_SUBJECT
    today = get_formatted_date()
    count = 0

    _LOGGER.debug("Cleaning up amazon images...")
    cleanup_images(f"{image_path}amazon/")

    address_list = amazon_email_addresses(fwds, amazon_domain)
    _LOGGER.debug("Amazon email list: %s", str(address_list))

    for subject in subjects:
        (server_response, data) = email_search(account, address_list, today, subject)

        if server_response == "OK" and data[0] is not None:
            count += len(data[0].split())
            _LOGGER.debug("Amazon delivered email(s) found: %s", count)
            get_amazon_image(
                data[0],
                account,
                image_path,
                hass,
                amazon_image_name,
            )

    if count == 0:
        _LOGGER.debug("No Amazon deliveries found.")
        nomail = f"{os.path.dirname(__file__)}/no_deliveries.jpg"
        try:
            copyfile(nomail, f"{image_path}amazon/" + amazon_image_name)
        except Exception as err:
            _LOGGER.error("Error attempting to copy image: %s", str(err))

    return count


def get_amazon_image(
    sdata: Any,
    account: Type[imaplib.IMAP4_SSL],
    image_path: str,
    hass: HomeAssistant,
    image_name: str,
) -> None:
    """Find Amazon delivery image."""
    _LOGGER.debug("Searching for Amazon image in emails...")

    img_url = None
    mail_list = sdata.split()
    _LOGGER.debug("HTML Amazon emails found: %s", len(mail_list))

    for i in mail_list:
        data = email_fetch(account, i, "(RFC822)")[1]
        for response_part in data:
            if isinstance(response_part, tuple):
                msg = email.message_from_bytes(response_part[1])
                _LOGGER.debug("Email Multipart: %s", str(msg.is_multipart()))
                _LOGGER.debug("Content Type: %s", str(msg.get_content_type()))

                for part in msg.walk():
                    if part.get_content_type() != "text/html":
                        continue
                    _LOGGER.debug("Processing HTML email...")
                    part = part.get_payload(decode=True)
                    part = part.decode("utf-8", "ignore")
                    pattern = re.compile(rf"{AMAZON_IMG_PATTERN}")
                    found = pattern.findall(part)
                    for url in found:
                        if url[1] != "us-prod-temp.s3.amazonaws.com":
                            continue
                        img_url = url[0] + url[1] + url[2]
                        _LOGGER.debug("Amazon img URL: %s", img_url)
                        break

    if img_url is not None:
        _LOGGER.debug("Attempting to download Amazon image.")
        # Download the image we found
        hass.add_job(download_img(hass, img_url, image_path, image_name))


async def download_img(
    hass: HomeAssistant, img_url: str, img_path: str, img_name: str
) -> None:
    """Download image from url."""
    img_path = f"{img_path}amazon/"
    filepath = f"{img_path}{img_name}"

    async with aiohttp.ClientSession() as session:
        async with session.get(img_url.replace("&amp;", "&")) as resp:
            if resp.status != 200:
                _LOGGER.error("Problem downloading file http error: %s", resp.status)
                return
            content_type = resp.headers["content-type"]
            _LOGGER.debug("URL content-type: %s", content_type)
            if "image" in content_type:
                data = await resp.read()
                _LOGGER.debug("Downloading image to: %s", filepath)
                the_file = await hass.async_add_executor_job(open, filepath, "wb")
                the_file.write(data)
                _LOGGER.debug("Amazon image downloaded")


def _process_amazon_forwards(email_list: str | list | None) -> list:
    """Process amazon forward emails.

    Returns list of email addresses
    """
    result = []
    if email_list is not None:
        if not isinstance(email_list, list):
            email_list = email_list.split()
        for fwd in email_list:
            if fwd and fwd != '""' and fwd not in result:
                result.append(fwd)

    _LOGGER.debug("Processed forwards: %s", result)
    return result


def amazon_hub(account: Type[imaplib.IMAP4_SSL], fwds: Optional[str] = None) -> dict:
    """Find Amazon Hub info emails.

    Returns dict of sensor data
    """
    email_addresses = []
    email_addresses.extend(_process_amazon_forwards(fwds))
    body_regex = AMAZON_HUB_BODY
    subject_regex = AMAZON_HUB_SUBJECT_SEARCH
    info = {}
    today = get_formatted_date()

    email_addresses.extend(AMAZON_HUB_EMAIL)
    _LOGGER.debug("[Hub] Amazon email list: %s", str(email_addresses))

    for address in email_addresses:
        (server_response, sdata) = email_search(
            account, address, today, subject=AMAZON_HUB_SUBJECT
        )

        # Bail out on error
        if server_response != "OK" or sdata[0] is None:
            return info

        if len(sdata) == 0:
            info[ATTR_COUNT] = 0
            info[ATTR_CODE] = []
            return info

        found = []
        id_list = sdata[0].split()
        _LOGGER.debug("Amazon hub emails found: %s", str(len(id_list)))
        for i in id_list:
            data = email_fetch(account, i, "(RFC822)")[1]
            for response_part in data:
                if isinstance(response_part, tuple):
                    msg = email.message_from_bytes(response_part[1])

                    # Get combo number from subject line
                    email_subject = msg["subject"]
                    pattern = re.compile(rf"{subject_regex}")
                    search = pattern.search(email_subject)
                    if search is not None:
                        if len(search.groups()) > 1:
                            found.append(search.group(3))
                            continue

                    # Get combo number from message body
                    try:
                        if msg.is_multipart():
                            email_msg = quopri.decodestring(str(msg.get_payload(0)))
                        else:
                            email_msg = quopri.decodestring(str(msg.get_payload()))
                    except Exception as err:
                        _LOGGER.debug("Problem decoding email message: %s", str(err))
                        continue
                    email_msg = email_msg.decode("utf-8", "ignore")
                    pattern = re.compile(rf"{body_regex}")
                    search = pattern.search(email_msg)
                    if search is not None:
                        if len(search.groups()) > 1:
                            found.append(search.group(2))

    info[ATTR_COUNT] = len(found)
    info[ATTR_CODE] = found

    return info


def amazon_exception(
    account: Type[imaplib.IMAP4_SSL],
    fwds: Optional[list] = None,
    the_domain: str = None,
) -> dict:
    """Find Amazon exception emails.

    Returns dict of sensor data
    """
    order_number = []
    tfmt = get_formatted_date()
    count = 0
    info = {}

    address_list = amazon_email_addresses(fwds, the_domain)
    _LOGGER.debug("Amazon email list: %s", str(address_list))

    (server_response, sdata) = email_search(
        account, address_list, tfmt, AMAZON_EXCEPTION_SUBJECT
    )

    if server_response == "OK":
        count += len(sdata[0].split())
        _LOGGER.debug("Found %s Amazon exceptions", count)
        order_numbers = get_tracking(sdata[0], account, AMAZON_PATTERN)
        for order in order_numbers:
            order_number.append(order)

    info[ATTR_COUNT] = count
    info[ATTR_ORDER] = order_number

    return info


def amazon_date_search(email_msg: str) -> int:
    """Search for amazon date strings in email message."""
    for pattern in AMAZON_TIME_PATTERN_END:
        if (result := email_msg.find(pattern)) != -1:
            return result
    return -1


def amazon_date_format(arrive_date: str, lang: str) -> tuple:
    """Return the date format."""
    if "de_" in lang:
        return (arrive_date.split(",", 1)[1].strip(), "%d %B")

    if "today" in arrive_date or "tomorrow" in arrive_date:
        return (arrive_date.split(",")[1].strip(), "%B %d")

    if arrive_date.endswith(","):
        return (arrive_date.rstrip(","), "%A, %B %d")

    if "," not in arrive_date:
        return (arrive_date, "%A %d %B")

    return (arrive_date, "%A, %B %d")


def amazon_email_addresses(
    fwds: Optional[str] = None, the_domain: str = None
) -> list | None:
    """Return Amazon email addresses in list format."""
    domains = []
    domains.extend(_process_amazon_forwards(fwds))
    the_domain = the_domain.split()
    domains.extend(the_domain)
    value = []

    for domain in domains:
        if "@" in domain:
            email_address = domain.strip('"')
            value.append(email_address)
        else:
            email_address = []
            addresses = AMAZON_SHIPMENT_TRACKING
            for address in addresses:
                email_address.append(f"{address}@{domain}")
            value.extend(email_address)

    _LOGGER.debug("Amazon email search addresses: %s", str(value))
    return value


def get_items(
    account: Type[imaplib.IMAP4_SSL],
    param: str = None,
    fwds: Optional[str] = None,
    days: int = DEFAULT_AMAZON_DAYS,
    the_domain: str = None,
) -> Union[List[str], int]:
    """Parse Amazon emails for delivery date and order number.

    Returns list of order numbers or email count as integer
    """
    _LOGGER.debug("Attempting to find Amazon email with item list ...")

    # Limit to past X days
    past_date = datetime.date.today() - datetime.timedelta(days=days)
    tfmt = past_date.strftime("%d-%b-%Y")
    deliveries_today = []
    order_number = []

    address_list = amazon_email_addresses(fwds, the_domain)
    _LOGGER.debug("Amazon email list: %s", str(address_list))

    (server_response, sdata) = email_search(account, address_list, tfmt)

    if server_response == "OK":
        mail_ids = sdata[0]
        id_list = mail_ids.split()
        _LOGGER.debug("Amazon emails found: %s", str(len(id_list)))
        for i in id_list:
            data = email_fetch(account, i, "(RFC822)")[1]
            for response_part in data:
                if isinstance(response_part, tuple):
                    msg = email.message_from_bytes(response_part[1])

                    _LOGGER.debug("Email Multipart: %s", str(msg.is_multipart()))
                    _LOGGER.debug("Content Type: %s", str(msg.get_content_type()))

                    # Get order number from subject line
                    encoding = decode_header(msg["subject"])[0][1]
                    if encoding is not None:
                        email_subject = decode_header(msg["subject"])[0][0].decode(
                            encoding, "ignore"
                        )
                    else:
                        email_subject = decode_header(msg["subject"])[0][0]

                    if not isinstance(email_subject, str):
                        _LOGGER.debug("Converting subject to string.")
                        email_subject = email_subject.decode("utf-8", "ignore")

                    _LOGGER.debug("Amazon Subject: %s", str(email_subject))
                    pattern = re.compile(r"[0-9]{3}-[0-9]{7}-[0-9]{7}")

                    # Don't add the same order number twice
                    if (
                        (found := pattern.findall(email_subject))
                        and len(found) > 0
                        and found[0] not in order_number
                    ):
                        order_number.append(found[0])

                    try:
                        if msg.is_multipart():
                            email_msg = quopri.decodestring(str(msg.get_payload(0)))
                        else:
                            email_msg = quopri.decodestring(str(msg.get_payload()))
                    except Exception as err:
                        _LOGGER.debug("Problem decoding email message: %s", str(err))
                        _LOGGER.error("Unable to process this email. Skipping.")
                        continue
                    email_msg = email_msg.decode("utf-8", "ignore")

                    # Check message body for order number
                    if (
                        (found := pattern.findall(email_msg))
                        and len(found) > 0
                        and found[0] not in order_number
                    ):
                        order_number.append(found[0])

                    for search in AMAZON_TIME_PATTERN:
                        _LOGGER.debug("Looking for: %s", search)
                        if search not in email_msg:
                            continue

                        start = email_msg.find(search) + len(search)
                        end = amazon_date_search(email_msg)

                        arrive_date = email_msg[start:end].replace(">", "").strip()
                        _LOGGER.debug("First pass: %s", arrive_date)
                        arrive_date = arrive_date.split(" ")
                        arrive_date = arrive_date[0:3]
                        arrive_date = " ".join(arrive_date).strip()

                        # Get the date object
                        dateobj = dateparser.parse(arrive_date)

                        if (
                            dateobj is not None
                            and dateobj.day == datetime.date.today().day
                            and dateobj.month == datetime.date.today().month
                        ):
                            deliveries_today.append("Amazon Order")

    value = None
    if param == "count":
        _LOGGER.debug("Amazon Count: %s", str(len(deliveries_today)))
        if len(deliveries_today) > len(order_number):
            value = len(order_number)
        else:
            value = len(deliveries_today)
    else:
        _LOGGER.debug("Amazon order: %s", str(order_number))
        value = order_number

    return value
