Add amazon exception sensor (#420)

custom_components/mail_and_packages/const.py

@@ -69,6 +69,11 @@
 AMAZON_HUB_EMAIL = "[email protected]"
 AMAZON_HUB_SUBJECT = "(You have a package to pick up)(.*)- (\\d{6})"
 AMAZON_TIME_PATTERN = "will arrive:,estimated delivery date is:,guaranteed delivery date is:,Arriving:,Arriver:"
+AMAZON_EXCEPTION_SUBJECT = "Delivery update:"
+AMAZON_EXCEPTION_BODY = "running late"
+AMAZON_EXCEPTION = "amazon_exception"
+AMAZON_EXCEPTION_ORDER = "amazon_exception_order"
+AMAZON_PATTERN = "[0-9]{3}-[0-9]{7}-[0-9]{7}"
 
 # Sensor Data
 SENSOR_DATA = {
@@ -221,6 +226,7 @@
         "package(s)",
         "mdi:package-variant-closed",
     ],
+    "amazon_exception": ["Mail Amazon Exception", "package(s)", "mdi:archive-alert"],
     "amazon_hub": ["Mail Amazon Hub Packages", "package(s)", "mdi:amazon"],
     "capost_delivered": [
         "Mail Canada Post Delivered",

custom_components/mail_and_packages/helpers.py

@@ -304,6 +304,10 @@ def fetch(
         value = amazon_hub(account, amazon_fwds)
         count[sensor] = value[const.ATTR_COUNT]
         count[const.AMAZON_HUB_CODE] = value[const.ATTR_CODE]
+    elif sensor == const.AMAZON_EXCEPTION:
+        info = amazon_exception(account, amazon_fwds)
+        count[sensor] = info[const.ATTR_COUNT]
+        count[const.AMAZON_EXCEPTION_ORDER] = info[const.ATTR_ORDER]
     elif "_packages" in sensor:
         prefix = sensor.split("_")[0]
         delivering = fetch(hass, config, account, data, f"{prefix}_delivering")
@@ -1003,6 +1007,54 @@ def amazon_hub(account: Type[imaplib.IMAP4_SSL], fwds: Optional[str] = None) ->
     return info
 
 
+def amazon_exception(
+    account: Type[imaplib.IMAP4_SSL], fwds: Optional[str] = None
+) -> dict:
+    """Find Amazon exception emails.
+
+    Returns dict of sensor data
+    """
+    orderNum = []
+    tfmt = get_formatted_date()
+    subject = const.AMAZON_EXCEPTION_SUBJECT
+    pattern = const.AMAZON_PATTERN
+    count = 0
+    info = {}
+    domains = const.Amazon_Domains.split(",")
+    if isinstance(fwds, list):
+        for fwd in fwds:
+            if fwd != '""':
+                domains.append(fwd)
+                _LOGGER.debug("Amazon email adding %s to list", str(fwd))
+
+    for domain in domains:
+        if "@" in domain:
+            email_address = domain.strip('"')
+            _LOGGER.debug("Amazon email search address: %s", str(email_address))
+        else:
+            email_address = []
+            addresses = const.AMAZON_SHIPMENT_TRACKING
+            for address in addresses:
+                email_address.append(f"{address}@{domain}")
+            _LOGGER.debug("Amazon email search address: %s", str(email_address))
+
+        (rv, sdata) = email_search(account, email_address, tfmt, subject)
+
+        if rv != "OK":
+            continue
+
+        count += len(sdata[0].split())
+        _LOGGER.debug("Found %s Amazon exceptions", count)
+        ordernums = get_tracking(sdata[0], account, pattern)
+        for order in ordernums:
+            orderNum.append(order)
+
+    info[const.ATTR_COUNT] = count
+    info[const.ATTR_ORDER] = orderNum
+
+    return info
+
+
 def get_items(
     account: Type[imaplib.IMAP4_SSL],
     param: str = None,

custom_components/mail_and_packages/sensor.py

@@ -92,7 +92,10 @@ def device_state_attributes(self) -> Optional[str]:
         data = self.coordinator.data
 
         if "Amazon" in self._name:
-            attr[const.ATTR_ORDER] = data[const.AMAZON_ORDER]
+            if self._name == "amazon_exception":
+                attr[const.ATTR_ORDER] = data[const.AMAZON_EXCEPTION_ORDER]
+            else:
+                attr[const.ATTR_ORDER] = data[const.AMAZON_ORDER]
         elif "Mail USPS Mail" == self._name:
             attr[const.ATTR_IMAGE] = data[const.ATTR_IMAGE_NAME]
         elif "_delivering" in self.type and tracking in self.data.keys():

tests/conftest.py

@@ -978,3 +978,28 @@ def mock_copytree():
     with patch("custom_components.mail_and_packages.helpers.copytree") as mock_copytree:
         mock_copytree.return_value = True
         yield mock_copytree
+
+
+@pytest.fixture()
+def mock_imap_amazon_exception():
+    """ Mock imap class values. """
+    with patch(
+        "custom_components.mail_and_packages.helpers.imaplib"
+    ) as mock_imap_amazon_exception:
+        mock_conn = mock.Mock(spec=imaplib.IMAP4_SSL)
+        mock_imap_amazon_exception.IMAP4_SSL.return_value = mock_conn
+
+        mock_conn.login.return_value = (
+            "OK",
+            [b"[email protected] authenticated (Success)"],
+        )
+        mock_conn.list.return_value = (
+            "OK",
+            [b'(\\HasNoChildren) "/" "INBOX"'],
+        )
+        mock_conn.search.return_value = ("OK", [b"1"])
+        f = open("tests/test_emails/amazon_exception.eml", "r")
+        email_file = f.read()
+        mock_conn.fetch.return_value = ("OK", [(b"", email_file.encode("utf-8"))])
+        mock_conn.select.return_value = ("OK", [])
+        yield mock_conn

tests/const.py

@@ -51,6 +51,7 @@
         "zpackages_delivered",
         "zpackages_transit",
         "amazon_delivered",
+        "amazon_exception",
         "amazon_hub",
         "amazon_packages",
         "capost_delivered",
@@ -184,6 +185,7 @@
     "port": 993,
     "resources": [
         "amazon_delivered",
+        "amazon_exception",
         "amazon_hub",
         "amazon_packages",
         "capost_delivered",