Release v2.7.0

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## v2.7.0 - 2024-12-02
+
+### Added
+
+- [#297](https://github.com/networktocode/circuit-maintenance-parser/pull/297) - Added new parser for Tata Communications.
+
+### Dependencies
+
+- [#295](https://github.com/networktocode/circuit-maintenance-parser/pull/295) - Remove pydantic dotenv extra
+
+### Changed
+
+- [#291](https://github.com/networktocode/circuit-maintenance-parser/pull/291) - Update Windstream Parser for new emails
+
 ## v2.6.1 - 2024-06-04
 
 ### Fixed

README.md

@@ -84,6 +84,7 @@ By default, there is a `GenericProvider` that supports a `SimpleProcessor` using
 - Netflix (AS2906 only)
 - Seaborn
 - Sparkle
+- Tata
 - Telstra (\*)
 - Turkcell
 - Verizon

circuit_maintenance_parser/__init__.py

@@ -28,6 +28,7 @@
     PacketFabric,
     Seaborn,
     Sparkle,
+    Tata,
     Telia,
     Telstra,
     Turkcell,
@@ -59,6 +60,7 @@
     PacketFabric,
     Seaborn,
     Sparkle,
+    Tata,
     Telia,
     Telstra,
     Turkcell,

circuit_maintenance_parser/parsers/tata.py

@@ -0,0 +1,71 @@
+# pylint: disable=disallowed-name
+"""Circuit maintenance parser for Tata Email notifications."""
+from typing import List, Dict, Any
+from datetime import datetime
+
+from bs4.element import ResultSet  # type: ignore
+from circuit_maintenance_parser.output import Impact, Status
+from circuit_maintenance_parser.parser import Html, EmailSubjectParser
+
+
+class HtmlParserTata(Html):
+    """Custom Parser for HTML portion of Tata circuit maintenance notifications."""
+
+    def parse_html(self, soup: ResultSet) -> List[Dict]:
+        """Parse Tata circuit maintenance email."""
+        prev: str = ""
+        data: Dict[str, Any] = {
+            "account": "N/A",
+            "circuits": [],
+            "organizer": soup.select("a[href^=mailto]")[0].text.strip(),
+        }
+        for span in soup.find_all("span"):
+            curr = span.text.strip()
+            if curr != prev:
+                prev_lower = prev.lower()
+                if prev_lower == "ticket reference - tcl":
+                    data["maintenance_id"] = curr
+                elif prev_lower == "service id":
+                    for circuit in curr.split(","):
+                        data["circuits"].append(
+                            {
+                                "circuit_id": circuit.strip(),
+                                "impact": Impact.OUTAGE,
+                            }
+                        )
+                elif prev_lower in ("activity window (gmt)", "revised activity window (gmt)"):
+                    start_end = curr.split("to")
+                    data["start"] = self._parse_time(start_end[0])
+                    data["end"] = self._parse_time(start_end[1])
+                elif "extended up to time window" in prev_lower:
+                    if "gmt" in curr.lower():
+                        data["end"] = self._parse_time(curr)
+            prev = span.text.strip()
+
+        return [data]
+
+    @staticmethod
+    def _parse_time(string: str) -> int:
+        """Convert YYYY-MM-DD HH:MM:SS GMT to epoch."""
+        return int((datetime.strptime(string.strip(), "%Y-%m-%d %H:%M:%S GMT") - datetime(1970, 1, 1)).total_seconds())
+
+
+class SubjectParserTata(EmailSubjectParser):
+    """Custom Parser for Email subject of Tata circuit maintenance notifications."""
+
+    def parse_subject(self, subject: str) -> List[Dict]:
+        """Parse Tata Email subject for summary and status."""
+        data: Dict[str, Any] = {"summary": subject.strip().replace("\n", "")}
+        subject_lower = subject.lower()
+        if "completion" in subject_lower:
+            data["status"] = Status.COMPLETED
+        elif "reschedule" in subject_lower or "extension" in subject_lower:
+            data["status"] = Status.RE_SCHEDULED
+        elif "reminder" in subject_lower:
+            data["status"] = Status.CONFIRMED
+        elif "cancellation" in subject_lower:
+            data["status"] = Status.CANCELLED
+        else:
+            data["status"] = Status.CONFIRMED
+
+        return [data]

circuit_maintenance_parser/parsers/windstream.py

@@ -41,28 +41,25 @@ def parse_html(self, soup):
 
         data["summary"] = summary_text
 
-        table = soup.find("table")
-        for row in table.find_all("tr"):
-            if len(row) < 2:
-                continue
-            cols = row.find_all("td")
-            header_tag = cols[0].string
-            if header_tag is None or header_tag == "Maintenance Address:":
-                continue
-            header_tag = header_tag.string.strip()
-            value_tag = cols[1].string.strip()
-            if header_tag == "WMT:":
-                data["maintenance_id"] = value_tag
-            elif "Date & Time:" in header_tag:
-                dt_time = convert_timezone(value_tag)
-                if "Event Start" in header_tag:
-                    data["start"] = int(dt_time.replace(tzinfo=timezone.utc).timestamp())
-                elif "Event End" in header_tag:
-                    data["end"] = int(dt_time.replace(tzinfo=timezone.utc).timestamp())
-            elif header_tag == "Outage":
-                impact = Impact("OUTAGE")
-            else:
-                continue
+        impact = soup.find("td", string="Outage").find_next_sibling("td").string
+        if impact:
+            impact = Impact("OUTAGE")
+
+        maint_id = soup.find("td", string="WMT:").find_next_sibling("td").string
+        if maint_id:
+            data["maintenance_id"] = maint_id
+
+        event = soup.find("td", string="Event Start Date & Time:").find_next_sibling("td").string
+        if event:
+            dt_time = convert_timezone(event)
+            data["start"] = int(dt_time.replace(tzinfo=timezone.utc).timestamp())
+            event = ""
+
+        event = soup.find("td", string="Event End Date & Time:").find_next_sibling("td").string
+        if event:
+            dt_time = convert_timezone(event)
+            data["end"] = int(dt_time.replace(tzinfo=timezone.utc).timestamp())
+            event = ""
 
         table = soup.find("table", "circuitTable")
         for row in table.find_all("tr"):

circuit_maintenance_parser/provider.py

@@ -37,6 +37,7 @@
     SubjectParserSeaborn2,
 )
 from circuit_maintenance_parser.parsers.sparkle import HtmlParserSparkle1
+from circuit_maintenance_parser.parsers.tata import HtmlParserTata, SubjectParserTata
 from circuit_maintenance_parser.parsers.telstra import HtmlParserTelstra1, HtmlParserTelstra2
 from circuit_maintenance_parser.parsers.turkcell import HtmlParserTurkcell1
 from circuit_maintenance_parser.parsers.verizon import HtmlParserVerizon1
@@ -428,6 +429,19 @@ class Sparkle(GenericProvider):
     _default_organizer = PrivateAttr("[email protected]")
 
 
+class Tata(GenericProvider):
+    """Tata provider custom class."""
+
+    _include_filter = PrivateAttr({EMAIL_HEADER_SUBJECT: ["Planned Work Notification"]})
+
+    _processors: List[GenericProcessor] = PrivateAttr(
+        [
+            CombinedProcessor(data_parsers=[HtmlParserTata, SubjectParserTata, EmailDateParser]),
+        ]
+    )
+    _default_organizer = PrivateAttr("[email protected]")
+
+
 class Telia(Arelion):
     """Telia provider custom class."""