"""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("&", "&")) 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