Update Telstra.py for new notification format (#190)

* Update Telstra.py for new notification format

* rename variables

* Update telstra.py

* add new telstra test files

* Update test_e2e for telstra

* Update test_parser for Telstra

* fix linting for telstra parser

* update regex condition

* fix useless import

* fix mypy lint

* keep new arelion instead of Telia

* fix indent

* fix pylint

* fix docstyle

* add good result.json

* fix json results

* fix test_e2e

* fix Telstra tests

* fix compatibility with old format

* fix parser organisation

* fix maintenance details

* fix black linting

* add summary to tests

* unit test fixing

* improve telsrtra parser and complete tests

* all test validated

---------

Co-authored-by: c.radet <[email protected]>

Author: unpapillon

circuit_maintenance_parser/parsers/telstra.py

@@ -1,12 +1,13 @@
 """Telstra parser."""
 import logging
 from typing import Dict, List
-
+import re
 from dateutil import parser
 from bs4.element import ResultSet  # type: ignore
 
 from circuit_maintenance_parser.parser import Html, Impact, CircuitImpact, Status
 
+
 # pylint: disable=too-many-branches
 
 
@@ -73,3 +74,91 @@ def parse_tables(self, tables: ResultSet, data: Dict):  # pylint: disable=too-ma
                         # First sentence containts 'Maintenance Details:' so we skip it
                         data["summary"] = ". ".join(sentences[1:])
             break
+
+
+class HtmlParserTelstra2(Html):
+    """Notifications Parser for Telstra notifications."""
+
+    def parse_html(self, soup):
+        """Execute parsing."""
+        data = {}
+        self.parse_tables(soup.find_all("table"), data)
+        return [data]
+
+    def add_maintenance_data(self, table: ResultSet, data: Dict):
+        """Populate data dict."""
+        for strong_element in table.find_all("strong"):
+            if not strong_element.string:
+                continue
+            strong_text = strong_element.string.strip()
+            strong_sibling = strong_element.next_sibling.next_sibling
+            if strong_text == "Reference number":
+                data["maintenance_id"] = strong_sibling.string.strip()
+            elif strong_text == "Start time":
+                text_start = strong_sibling.string
+                regex = re.search(r"\d{2}\s[a-zA-Z]{3}\s\d{4}\s\d{2}[:]\d{2}[:]\d{2}", text_start)
+                if regex is not None:
+                    start = parser.parse(regex.group())
+                    data["start"] = self.dt2ts(start)
+                else:
+                    data["start"] = "Not defined"
+            elif strong_text == "End time":
+                text_end = strong_sibling.string
+                regex = re.search(r"\d{2}\s[a-zA-Z]{3}\s\d{4}\s\d{2}[:]\d{2}[:]\d{2}", text_end)
+                if regex is not None:
+                    end = parser.parse(regex.group())
+                    data["end"] = self.dt2ts(end)
+                else:
+                    data["end"] = "is not defined"
+            elif strong_text == "Service/s under maintenance":
+                data["circuits"] = []
+                # TODO: This split is just an assumption of the multiple service, to be checked with more samples
+                impacted_circuits = strong_sibling.text.split(", ")
+                for circuit_id in impacted_circuits:
+                    data["circuits"].append(CircuitImpact(impact=Impact("OUTAGE"), circuit_id=circuit_id.strip()))
+            elif strong_text == "Maintenance details":
+                sentences: List[str] = []
+                for element in strong_element.next_elements:
+                    if element.string == "Reference number":
+                        break
+                    if element.string and element.string not in ["\n", "", "\xa0"] + sentences:
+                        sentences.append(element.string)
+                if sentences:
+                    # First sentence containts 'Maintenance Details' so we skip it
+                    data["summary"] = ". ".join(sentences[1:])
+
+    def parse_tables(self, tables: ResultSet, data: Dict):  # pylint: disable=too-many-locals
+        """Parse Table tag."""
+        for table in tables:
+            for p_element in table.find_all("p"):
+                # TODO: We should find a more consistent way to parse the status of a maintenance note
+                p_text = p_element.text.lower()
+                if "attention" in p_text:
+                    regex = re.search("[^attention ].*", p_text.strip())
+                    if regex is not None:
+                        data["account"] = regex.group()
+                    else:
+                        data["account"] = "not Found"
+            for span_element in table.find_all("span"):
+                span_text = span_element.text.lower()
+                if "planned maintenance to our network infrastructure" in span_text:
+                    data["status"] = Status("CONFIRMED")
+                elif "emergency maintenance to our network infrastructure" in span_text:
+                    data["status"] = Status("CONFIRMED")
+                elif "has been rescheduled" in span_text:
+                    data["status"] = Status("RE-SCHEDULED")
+                elif "has been completed successfully" in span_text:
+                    data["status"] = Status("COMPLETED")
+                elif (
+                    "did not proceed" in span_text
+                    or "has been withdrawn" in span_text
+                    or "has been cancelled" in span_text
+                ):
+                    data["status"] = Status("CANCELLED")
+                elif "was unsuccessful" in span_text:
+                    data["status"] = Status("CANCELLED")
+                else:
+                    continue
+                break
+            self.add_maintenance_data(table, data)
+            break

circuit_maintenance_parser/provider.py

@@ -35,7 +35,7 @@
     SubjectParserSeaborn2,
 )
 from circuit_maintenance_parser.parsers.sparkle import HtmlParserSparkle1
-from circuit_maintenance_parser.parsers.telstra import HtmlParserTelstra1
+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
 from circuit_maintenance_parser.parsers.zayo import HtmlParserZayo1, SubjectParserZayo1
@@ -330,6 +330,7 @@ class Telstra(GenericProvider):
 
     _processors: List[GenericProcessor] = [
         SimpleProcessor(data_parsers=[ICal]),
+        CombinedProcessor(data_parsers=[EmailDateParser, HtmlParserTelstra2]),
         CombinedProcessor(data_parsers=[EmailDateParser, HtmlParserTelstra1]),
     ]
     _default_organizer = "[email protected]"