Add parser for FLAG (fka Global Cloud Xchange)

changes/419.added

@@ -0,0 +1 @@
+Added parser for FLAG (fka Globalcloudexchange)

circuit_maintenance_parser/__init__.py

@@ -10,6 +10,7 @@
     ATT,
     AWS,
     BSO,
+    FLAG,
     GTT,
     HGC,
     NTT,
@@ -58,6 +59,7 @@
     CrownCastle,
     Equinix,
     EUNetworks,
+    FLAG,
     GlobalCloudXchange,
     Google,
     GTT,

circuit_maintenance_parser/parsers/flag.py

@@ -0,0 +1,111 @@
+"""Circuit Maintenance Parser for FLAG Notifications.
+
+Note: this is a fork of Globalcloudexchange parser.
+"""
+
+import re
+from datetime import datetime
+from typing import Any, Dict, List
+
+from bs4 import BeautifulSoup
+from bs4.element import ResultSet  # type: ignore
+
+from circuit_maintenance_parser.output import Impact
+from circuit_maintenance_parser.parser import EmailSubjectParser, Html, Status
+
+
+class HtmlParserFlag1(Html):
+    """Custom Parser for HTML portion of FLAG circuit maintenance notifications."""
+
+    def parse_html(self, soup: BeautifulSoup) -> List[Dict]:
+        """Parse an FLAG circuit maintenance email.
+
+        Args:
+            soup (BeautifulSoup): beautiful soup object containing the html portion of an email.
+
+        Returns:
+            Dict: The data dict containing circuit maintenance data.
+        """
+        data: Dict[str, Any] = {"circuits": []}
+        self.parse_tables(soup.find_all("table", attrs={"border-collapse": "collapse"}), data)
+        self.parse_paragraphs(soup.find_all("p"), data)
+
+        return [data]
+
+    def parse_tables(self, tables: ResultSet, data: Dict):
+        """Parse table elements to find maintenance windows (start/end) and circuit ID's."""
+        date_format = "%d-%b-%Y %H:%M"
+        for table in tables:
+            table_type = ""
+            for row in table.find_all("tr"):
+                cols = row.find_all("td")
+                if cols[0].text.strip() == "Service ID":
+                    table_type = "circuits"
+                    continue
+                if cols[0].text.strip() == "Window":
+                    table_type = "windows"
+                    continue
+
+                # this table is listing all circuits
+                if table_type == "circuits":
+                    impact = Impact.OUTAGE
+                    if "at risk" in cols[1].text.lower():
+                        impact = Impact.REDUCED_REDUNDANCY
+
+                    data["circuits"].append({"circuit_id": cols[0].text.strip(), "impact": impact})
+                # this table is listing windows (note: for now, we will only use the last listed window)
+                elif table_type == "windows":
+                    data["start"] = self.dt2ts(datetime.strptime(cols[1].text.strip(), date_format))
+                    data["end"] = self.dt2ts(datetime.strptime(cols[2].text.strip(), date_format))
+
+    def parse_paragraphs(self, paragraphs: ResultSet, data: Dict):
+        """Parse paragraph elements to find account and summary."""
+        for p in paragraphs:
+            for pstring in p.strings:
+                search = re.search("Dear (.*),", pstring)
+                if search:
+                    data["account"] = search.group(1).strip()
+                    continue
+
+                # after account has been set, next paragraph is the summary
+                if "account" in data and "summary" not in data:
+                    data["summary"] = pstring.strip()
+                    continue
+
+
+class SubjectParserFlag1(EmailSubjectParser):
+    """Parse the subject of a FLAG circuit maintenance email. The subject contains the maintenance ID and status."""
+
+    def parse_subject(self, subject: str) -> List[Dict]:
+        """Parse the FLAG Email subject for maintenance ID and status.
+
+        Args:
+            subject (str): subject of email
+            e.g. 'FLAG | PE2025102750538 | Planned Event | Rescheduled'.
+
+
+        Returns:
+            List[Dict]: Returns the data object with maintenance_id and status fields.
+        """
+        data = {}
+        search = re.search(
+            r"^FLAG \| ([A-Z0-9]+)\b",
+            subject,
+        )
+        if search:
+            data["maintenance_id"] = search.group(1)
+
+        if "completed" in subject.lower():
+            data["status"] = Status.COMPLETED
+        elif "rescheduled" in subject.lower():
+            data["status"] = Status.RE_SCHEDULED
+        elif "scheduled" in subject.lower() or "reminder" in subject.lower() or "notice" in subject.lower():
+            data["status"] = Status.CONFIRMED
+        elif "cancelled" in subject.lower():
+            data["status"] = Status.CANCELLED
+        else:
+            # Some FLAG notifications don't clearly state a status in their subject.
+            # From inspection of examples, it looks like "Confirmed" would be the most appropriate in this case.
+            data["status"] = Status.CONFIRMED
+
+        return [data]

circuit_maintenance_parser/provider.py

@@ -23,6 +23,7 @@
 from circuit_maintenance_parser.parsers.colt import CsvParserColt1, SubjectParserColt1, SubjectParserColt2
 from circuit_maintenance_parser.parsers.crowncastle import HtmlParserCrownCastle1
 from circuit_maintenance_parser.parsers.equinix import HtmlParserEquinix, SubjectParserEquinix
+from circuit_maintenance_parser.parsers.flag import HtmlParserFlag1, SubjectParserFlag1
 from circuit_maintenance_parser.parsers.globalcloudxchange import HtmlParserGcx1, SubjectParserGcx1
 from circuit_maintenance_parser.parsers.google import HtmlParserGoogle1, SubjectParserGoogle1
 from circuit_maintenance_parser.parsers.gtt import HtmlParserGTT1
@@ -362,6 +363,17 @@ class EUNetworks(GenericProvider):
     _default_organizer = "noc@eunetworks.com"
 
 
+class FLAG(GenericProvider):
+    """FLAG provider custom class."""
+
+    _processors: List[GenericProcessor] = PrivateAttr(
+        [
+            CombinedProcessor(data_parsers=[EmailDateParser, SubjectParserFlag1, HtmlParserFlag1]),
+        ]
+    )
+    _default_organizer = PrivateAttr("change@flagtel.com")
+
+
 class GlobalCloudXchange(GenericProvider):
     """Global Cloud Xchange provider custom class."""