#309 Add parsers for provider ATT (#313)

* Add dateutil and pandas[optional] as dependencies.

* Add openpyxl as optional dependency - required for open format spreadhseets in pandas.

* Add generic Xlsx Parser.

* Add parsers for ATT service provider.

* Skip 'Xlsx' tests if optional packages are not installed.

* Resolve issue with naive datetime.

* Remove trailing space requirement in regex.

* Update test emails.

* Add end to end tests for provider ATT.

* Catch exceptions in tests caused by missing, optional imports.

* Add coverage as a developer dependency.

* Specify exception as ProviderError rather than generic Exception.

* Raise 'ParserError' if fail to match record.

* Check for circuit ID key name when normalizing circuit IDs.

* Combine lines as requested.

* Raise 'ParserError' if spreadsheet is empty.

* Use poetry-install-options '--all-extras' to install optional packages in CI.

Author: grelleum

.github/workflows/ci.yml

@@ -1,9 +1,9 @@
 ---
 name: "CI"
-concurrency:  # Cancel any existing runs of this workflow for this same PR
+concurrency: # Cancel any existing runs of this workflow for this same PR
   group: "${{ github.workflow }}-${{ github.ref }}"
   cancel-in-progress: true
-on:  # yamllint disable
+on: # yamllint disable
   push:
     branches:
       - "main"
@@ -129,7 +129,7 @@ jobs:
         uses: "networktocode/gh-action-setup-poetry-environment@v6"
         with:
           python-version: "${{ matrix.python-version }}"
-          poetry-install-options: "--with dev"
+          poetry-install-options: "--all-extras"
       - name: "Run poetry Install"
         run: "poetry install"
       - name: "Run poetry Install"

README.md

@@ -69,6 +69,7 @@ By default, there is a `GenericProvider` that supports a `SimpleProcessor` using
 #### Supported providers based on other parsers
 
 - Apple
+- ATT
 - AWS
 - AquaComms
 - BSO
@@ -134,6 +135,12 @@ The library is available as a Python package in pypi and can be installed with p
 
 `pip install circuit-maintenance-parser[openai]`
 
+#### Xlsx Spreadsheets
+
+Some providers may attach a spreadsheet in their circuit maintenance notifications. Support for this is provided by installing the optional xlsx package.
+
+`pip install circuit-maintenance-parser[xlsx]`
+
 ## How to use it?
 
 The library requires two things:

circuit_maintenance_parser/__init__.py

@@ -14,6 +14,7 @@
     AquaComms,
     Apple,
     Arelion,
+    ATT,
     Cogent,
     Colt,
     CrownCastle,
@@ -44,6 +45,7 @@
     Apple,
     AquaComms,
     Arelion,
+    ATT,
     AWS,
     BSO,
     Cogent,

circuit_maintenance_parser/parser.py

@@ -1,4 +1,6 @@
 """Definition of Mainentance Notification base classes."""
+
+import io
 import logging
 import os
 import base64
@@ -12,10 +14,17 @@
 
 import bs4  # type: ignore
 from bs4.element import ResultSet  # type: ignore
-
 from pydantic import BaseModel, PrivateAttr
 from icalendar import Calendar  # type: ignore
 
+try:
+    from pandas import read_excel
+
+    READ_EXCEL_PRESENT = True
+except ImportError:
+    READ_EXCEL_PRESENT = False
+
+
 from circuit_maintenance_parser.errors import ParserError
 from circuit_maintenance_parser.output import Status, Impact, CircuitImpact
 from circuit_maintenance_parser.constants import EMAIL_HEADER_SUBJECT, EMAIL_HEADER_DATE
@@ -497,3 +506,35 @@ def parse_content(self, content):
         )
 
         return [data]
+
+
+class Xlsx(Parser):
+    """Xlsx parser."""
+
+    _data_types = PrivateAttr(
+        [
+            "application/octet-stream",
+            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        ]
+    )
+
+    def parser_hook(self, raw: bytes, content_type: str):
+        """Execute parsing."""
+        if not READ_EXCEL_PRESENT:
+            raise RuntimeError(
+                "Missing import 'pandas' required to read xlsx files. Install main package with option '[xlsx]'"
+            )
+
+        file_obj = io.BytesIO(raw)
+        xls = read_excel(file_obj)
+        xls = xls.drop_duplicates()
+        records = xls.to_dict(orient="records")
+        if not records:
+            raise ParserError("No rows found in attached spreadsheet.")
+        results = list(self.parse_xlsx(records))
+        return results
+
+    @staticmethod
+    def parse_xlsx(records: List[Dict]) -> List[Dict]:
+        """Provide placeholder method."""
+        raise NotImplementedError

circuit_maintenance_parser/parsers/att.py

@@ -0,0 +1,113 @@
+"""ATT Parser."""
+
+import logging
+import re
+import string
+from typing import Dict, List
+
+import dateutil
+from bs4.element import ResultSet  # type: ignore
+from circuit_maintenance_parser.errors import ParserError
+from circuit_maintenance_parser.parser import CircuitImpact, Html, Impact, Status, Xlsx
+
+logger = logging.getLogger(__name__)
+
+RE_EVENT = re.compile(
+    r"Event ID: (.*)[ \n]"
+    r"Customer Impact Description: (.*)[ \n]"
+    r"Summary: (.*)[ \n]"
+    r"Description: (.*)[ \n]"
+    r"Business Risk: (.*)"
+)
+RE_MAINTENANCE_WINDOW_GMT = re.compile(r"Start Time: (.* GMT).*End Time: (.* GMT)")
+RE_MAINTENANCE_WINDOW_NO_TIMEZONE = re.compile(r"Start Time: (.*)[ \n]End Time: (.*)")
+
+
+class XlsxParserATT1(Xlsx):
+    """Xlsx Parser for ATT file attachments."""
+
+    @staticmethod
+    def parse_xlsx(records: List[Dict]) -> List[Dict]:
+        """Parses ATT xlsx attachments."""
+        impact = Impact.OUTAGE
+        account_name, circuit_id_key = get_account_and_circuit_id_key(records[0])
+        circuit_ids = [r[circuit_id_key] for r in records]
+        if "Circuit/Asset" in records[0]:
+            circuit_ids = [normalize_lec_circuit_id(cid) for cid in circuit_ids]
+        circuits = [CircuitImpact(impact=impact, circuit_id=cid) for cid in circuit_ids]
+        data = [
+            {
+                "account": account_name,
+                "circuits": circuits,
+            }
+        ]
+        return data
+
+
+class HtmlParserATT1(Html):
+    """Notifications Parser for ATT notifications."""
+
+    def parse_html(self, soup):
+        """Parse ATT HTML notification."""
+        logger.debug("Parsing ATT HTML notification.")
+        data = self.parse_p_tags(soup)
+        data["start"] = self.dt2ts(data["start"])
+        data["end"] = self.dt2ts(data["end"])
+        data["status"] = Status.CONFIRMED
+        return [data]
+
+    @staticmethod
+    def parse_p_tags(soup: ResultSet) -> Dict:
+        """Parse <p> tags in HTML."""
+        data = {}
+        p_tags = soup.find_all("p")
+
+        for tag in p_tags:
+            text = remove_unprintable(tag.text.strip())
+
+            if match := RE_EVENT.search(text):
+                event_id, impact, summary, description, _ = match.groups()
+                data["maintenance_id"] = event_id
+                data["summary"] = f"{summary}: {impact} {description}"
+
+            elif match := RE_MAINTENANCE_WINDOW_GMT.search(text):
+                start_time_text, end_time_text = match.groups()
+                data["start"] = dateutil.parser.parse(start_time_text)
+                data["end"] = dateutil.parser.parse(end_time_text)
+
+            elif match := RE_MAINTENANCE_WINDOW_NO_TIMEZONE.search(text):
+                start_time_text, end_time_text = match.groups()
+                data["start"] = dateutil.parser.parse(start_time_text + " GMT")
+                data["end"] = dateutil.parser.parse(end_time_text + " GMT")
+
+        return data
+
+
+def get_account_and_circuit_id_key(record: Dict) -> tuple[str, str]:
+    """Return the account name and the key used to retrieve circuits IDs.
+
+    The key names may vary depending on the ATT business unit that initiated the notice.
+    """
+    if account := record.get("Customer"):
+        circuit_id_key = "Circuit/Asset"
+    elif account := record.get("Customer Name"):
+        circuit_id_key = "Circuit ID"
+    elif account := record.get("Customer Names"):
+        circuit_id_key = "Customer Circuit ID"
+    else:
+        raise ParserError("Could not parse 'Customer Name' and 'Circuit ID'.")
+
+    return str(account), circuit_id_key
+
+
+def normalize_lec_circuit_id(circuit_id: str) -> str:
+    """Standardize circuit IDs."""
+    circuit_id, *_ = circuit_id.split()
+    circuit_id = re.sub(r"^0+", "", circuit_id)  # Remove leading zeros.
+    circuit_id = re.sub(r"0+$", "ATI", circuit_id)  # Remove trailing zeros.
+    return circuit_id
+
+
+def remove_unprintable(text: str) -> str:
+    """Remove non-printing characters from text."""
+    return "".join(c for c in text if c in string.printable)

circuit_maintenance_parser/provider.py

@@ -17,6 +17,7 @@
 
 from circuit_maintenance_parser.parsers.apple import SubjectParserApple, TextParserApple
 from circuit_maintenance_parser.parsers.aquacomms import HtmlParserAquaComms1, SubjectParserAquaComms1
+from circuit_maintenance_parser.parsers.att import HtmlParserATT1, XlsxParserATT1
 from circuit_maintenance_parser.parsers.aws import SubjectParserAWS1, TextParserAWS1
 from circuit_maintenance_parser.parsers.bso import HtmlParserBSO1
 from circuit_maintenance_parser.parsers.cogent import HtmlParserCogent1, SubjectParserCogent1, TextParserCogent1
@@ -235,6 +236,17 @@ class Arelion(GenericProvider):
     _default_organizer = PrivateAttr("support@arelion.com")
 
 
+class ATT(GenericProvider):
+    """ATT provider custom class."""
+
+    _processors: List[GenericProcessor] = PrivateAttr(
+        [
+            CombinedProcessor(data_parsers=[EmailDateParser, HtmlParserATT1, XlsxParserATT1]),
+        ]
+    )
+    _default_organizer = PrivateAttr("g31654@att.com")
+
+
 class AWS(GenericProvider):
     """AWS provider custom class."""