Merge pull request #282 from networktocode/release-v2.6.0

Release v2.6.0

.github/workflows/ci.yml

@@ -114,6 +114,10 @@ jobs:
       fail-fast: true
       matrix:
         python-version: ["3.8", "3.9", "3.10", "3.11"]
+        pydantic: ["2.x"]
+        include:
+          - python-version: "3.11"
+            pydantic: "1.x"
     runs-on: "ubuntu-20.04"
     env:
       INVOKE_LOCAL: "True"
@@ -128,6 +132,9 @@ jobs:
           poetry-install-options: "--with dev"
       - name: "Run poetry Install"
         run: "poetry install"
+      - name: "Run poetry Install"
+        run: "pip install pydantic==1.10.13"
+        if: matrix.pydantic == '1.x'
       - name: "Run Tests"
         run: "poetry run invoke pytest --local"
     needs:

CHANGELOG.md

@@ -1,6 +1,23 @@
 # Changelog
 
-## v.2.5.0 - 2024-03-13
+## v2.6.0 - 2024-04-04
+
+### Added
+
+- [#273](https://github.com/networktocode/circuit-maintenance-parser/pull/273) - Add iCal parsing to GTT/EXA
+- [#280](https://github.com/networktocode/circuit-maintenance-parser/pull/280) - Add new Windstream Parser
+
+### Changed
+
+- [#277](https://github.com/networktocode/circuit-maintenance-parser/pull/277) - Refactor the output validator `validate_empty_circuit`
+- [#281](https://github.com/networktocode/circuit-maintenance-parser/pull/281) Add the ability to support pydantic 1 and 2
+
+### Fixed
+
+- [#272](https://github.com/networktocode/circuit-maintenance-parser/pull/272) - Fix the logic in the output validator `validate_empty_circuit`
+- [#278](https://github.com/networktocode/circuit-maintenance-parser/pull/278) - Increase robustness of Crown Castle parsing
+
+## v2.5.0 - 2024-03-13
 
 ### Added
 

README.md

@@ -60,9 +60,10 @@ By default, there is a `GenericProvider` that supports a `SimpleProcessor` using
 
 - Arelion (previously Telia)
 - EuNetworks
+- EXA (formerly GTT) (\*)
 - NTT
 - PacketFabric
-- Telstra
+- Telstra (\*)
 
 #### Supported providers based on other parsers
 
@@ -73,7 +74,7 @@ By default, there is a `GenericProvider` that supports a `SimpleProcessor` using
 - Colt
 - Crown Castle Fiber
 - Equinix
-- EXA (formerly GTT)
+- EXA (formerly GTT) (\*)
 - HGC
 - Global Cloud Xchange
 - Google
@@ -83,11 +84,14 @@ By default, there is a `GenericProvider` that supports a `SimpleProcessor` using
 - Netflix (AS2906 only)
 - Seaborn
 - Sparkle
-- Telstra
+- Telstra (\*)
 - Turkcell
 - Verizon
+- Windstream
 - Zayo
 
+(\*) Providers in both lists, with BCOP standard and nonstandard parsers.
+
 > Note: Because these providers do not support the BCOP standard natively, maybe there are some gaps on the implemented parser that will be refined with new test cases. We encourage you to report related **issues**!
 
 #### LLM-powered Parsers

circuit_maintenance_parser/__init__.py

@@ -32,6 +32,7 @@
     Telstra,
     Turkcell,
     Verizon,
+    Windstream,
     Zayo,
 )
 
@@ -62,6 +63,7 @@
     Telstra,
     Turkcell,
     Verizon,
+    Windstream,
     Zayo,
 )
 

circuit_maintenance_parser/output.py

@@ -8,7 +8,14 @@
 
 from typing import List
 
-from pydantic import field_validator, BaseModel, StrictStr, StrictInt, PrivateAttr
+try:
+    from pydantic import field_validator
+except ImportError:
+    # TODO: This exception handling is required for Pydantic 1.x compatibility. To be removed when the dependency is deprecated.
+    from pydantic import validator as field_validator  # type: ignore
+
+
+from pydantic import BaseModel, StrictStr, StrictInt, PrivateAttr
 
 
 class Impact(str, Enum):
@@ -197,7 +204,7 @@ def validate_empty_strings(cls, value):
     def validate_empty_circuits(cls, value, values):
         """Validate non-cancel notifications have a populated circuit list."""
         values = values.data
-        if len(value) < 1 and str(values["status"]) in ("CANCELLED", "COMPLETED"):
+        if len(value) < 1 and values["status"] not in (Status.CANCELLED, Status.COMPLETED):
             raise ValueError("At least one circuit has to be included in the maintenance")
         return value
 

circuit_maintenance_parser/parser.py

@@ -42,7 +42,11 @@ class Parser(BaseModel):
     @classmethod
     def get_data_types(cls) -> List[str]:
         """Return the expected data type."""
-        return cls._data_types.get_default()
+        try:
+            return cls._data_types.get_default()
+        except AttributeError:
+            # TODO: This exception handling is required for Pydantic 1.x compatibility. To be removed when the dependency is deprecated.
+            return cls._data_types
 
     @classmethod
     def get_name(cls) -> str:

circuit_maintenance_parser/parsers/crowncastle.py

@@ -56,11 +56,8 @@ def parse_strong(self, soup, data):
         for strong in soup.find_all("strong"):
             if strong.string.strip() == "Ticket Number:":
                 data["maintenance_id"] = strong.next_sibling.strip()
-            if strong.string.strip() == "Description:":
-                summary = strong.parent.next_sibling.next_sibling.contents[0].string.strip()
-                summary = re.sub(r"[\n\r]", "", summary)
-                data["summary"] = summary
-            if strong.string.strip().startswith("Work Description:"):
+            val = strong.string.strip()
+            if val == "Description:" or val.startswith("Work Description:"):
                 for sibling in strong.parent.next_siblings:
                     summary = "".join(sibling.strings)
                     summary = re.sub(r"[\n\r]", "", summary)

circuit_maintenance_parser/parsers/windstream.py

@@ -0,0 +1,76 @@
+"""Windstream parser."""
+import logging
+from datetime import timezone
+
+from circuit_maintenance_parser.parser import Html, Impact, CircuitImpact, Status
+from circuit_maintenance_parser.utils import convert_timezone
+
+# pylint: disable=too-many-nested-blocks, too-many-branches
+
+logger = logging.getLogger(__name__)
+
+
+class HtmlParserWindstream1(Html):
+    """Notifications Parser for Windstream notifications."""
+
+    def parse_html(self, soup):
+        """Execute parsing."""
+        data = {}
+        data["circuits"] = []
+        impact = Impact("NO-IMPACT")
+        confirmation_words = [
+            "Demand Maintenance Notification",
+            "Planned Maintenance Notification",
+            "Emergency Maintenance Notification",
+        ]
+        cancellation_words = ["Postponed Maintenance Notification", "Cancelled Maintenance Notification"]
+
+        h1_tag = soup.find("h1")
+        if h1_tag.string.strip() == "Completed Maintenance Notification":
+            data["status"] = Status("COMPLETED")
+        elif any(keyword in h1_tag.string.strip() for keyword in confirmation_words):
+            data["status"] = Status("CONFIRMED")
+        elif h1_tag.string.strip() == "Updated Maintenance Notification":
+            data["status"] = Status("RE-SCHEDULED")
+        elif any(keyword in h1_tag.string.strip() for keyword in cancellation_words):
+            data["status"] = Status("CANCELLED")
+
+        div_tag = h1_tag.find_next_sibling("div")
+        summary_text = div_tag.get_text(separator="\n", strip=True)
+        summary_text = summary_text.split("\nDESCRIPTION OF MAINTENANCE")[0]
+
+        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
+
+        table = soup.find("table", "circuitTable")
+        for row in table.find_all("tr"):
+            cols = row.find_all("td")
+            if len(cols) == 9:
+                if cols[0].string.strip() == "Name":
+                    continue
+                data["account"] = cols[0].string.strip()
+                data["circuits"].append(CircuitImpact(impact=impact, circuit_id=cols[2].string.strip()))
+
+        return [data]

circuit_maintenance_parser/provider.py

@@ -40,6 +40,7 @@
 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.windstream import HtmlParserWindstream1
 from circuit_maintenance_parser.parsers.zayo import HtmlParserZayo1, SubjectParserZayo1
 from circuit_maintenance_parser.processor import CombinedProcessor, GenericProcessor, SimpleProcessor
 from circuit_maintenance_parser.utils import rgetattr
@@ -150,22 +151,38 @@ def get_maintenances(self, data: NotificationData) -> Iterable[Maintenance]:
     @classmethod
     def get_default_organizer(cls) -> str:
         """Expose default_organizer as class attribute."""
-        return cls._default_organizer.get_default()  # type: ignore
+        try:
+            return cls._default_organizer.get_default()  # type: ignore
+        except AttributeError:
+            # TODO: This exception handling is required for Pydantic 1.x compatibility. To be removed when the dependency is deprecated.
+            return cls._default_organizer
 
     @classmethod
     def get_default_processors(cls) -> List[GenericProcessor]:
         """Expose default_processors as class attribute."""
-        return cls._processors.get_default()  # type: ignore
+        try:
+            return cls._processors.get_default()  # type: ignore
+        except AttributeError:
+            # TODO: This exception handling is required for Pydantic 1.x compatibility. To be removed when the dependency is deprecated.
+            return cls._processors
 
     @classmethod
     def get_default_include_filters(cls) -> Dict[str, List[str]]:
         """Expose include_filter as class attribute."""
-        return cls._include_filter.get_default()  # type: ignore
+        try:
+            return cls._include_filter.get_default()  # type: ignore
+        except AttributeError:
+            # TODO: This exception handling is required for Pydantic 1.x compatibility. To be removed when the dependency is deprecated.
+            return cls._include_filter
 
     @classmethod
     def get_default_exclude_filters(cls) -> Dict[str, List[str]]:
         """Expose exclude_filter as class attribute."""
-        return cls._exclude_filter.get_default()  # type: ignore
+        try:
+            return cls._exclude_filter.get_default()  # type: ignore
+        except AttributeError:
+            # TODO: This exception handling is required for Pydantic 1.x compatibility. To be removed when the dependency is deprecated.
+            return cls._exclude_filter
 
     @classmethod
     def get_extended_data(cls):
@@ -307,10 +324,13 @@ class GTT(GenericProvider):
     """EXA (formerly GTT) provider custom class."""
 
     # "Planned Work Notification", "Emergency Work Notification"
-    _include_filter = PrivateAttr({EMAIL_HEADER_SUBJECT: ["Work Notification"]})
+    _include_filter = PrivateAttr(
+        {"Icalendar": ["BEGIN"], "ical": ["BEGIN"], EMAIL_HEADER_SUBJECT: ["Work Notification"]}
+    )
 
     _processors: List[GenericProcessor] = PrivateAttr(
         [
+            SimpleProcessor(data_parsers=[ICal]),
             CombinedProcessor(data_parsers=[EmailDateParser, HtmlParserGTT1]),
         ]
     )
@@ -449,6 +469,17 @@ class Verizon(GenericProvider):
     _default_organizer = PrivateAttr("[email protected]")
 
 
+class Windstream(GenericProvider):
+    """Windstream provider custom class."""
+
+    _processors: List[GenericProcessor] = PrivateAttr(
+        [
+            CombinedProcessor(data_parsers=[EmailDateParser, HtmlParserWindstream1]),
+        ]
+    )
+    _default_organizer = PrivateAttr("[email protected]")
+
+
 class Zayo(GenericProvider):
     """Zayo provider custom class."""
 

circuit_maintenance_parser/utils.py

@@ -3,6 +3,8 @@
 import logging
 from typing import Tuple, Dict, Union
 import csv
+import datetime
+import pytz
 
 from geopy.exc import GeocoderUnavailable, GeocoderTimedOut, GeocoderServiceError  # type: ignore
 from geopy.geocoders import Nominatim  # type: ignore
@@ -128,6 +130,52 @@ def city_timezone(self, city: str) -> str:
         raise ParserError("Timezone resolution not properly initalized.")
 
 
+def convert_timezone(time_str):
+    """
+    Converts a string representing a date/time in the format 'MM/DD/YY HH:MM Timezone' to a datetime object in UTC.
+
+    Args:
+        time_str (str): A string representing a date/time followed by a timezone abbreviation.
+
+    Returns:
+        datetime: A datetime object representing the converted date/time in UTC.
+
+    Example:
+        convert_timezone("01/20/24 06:00 ET")
+    """
+    # Convert timezone abbreviation to timezone string for pytz.
+    timezone_mapping = {
+        "ET": "US/Eastern",
+        "CT": "US/Central",
+        "MT": "US/Mountain",
+        "PT": "US/Pacific"
+        # Add more mappings as needed
+    }
+
+    datetime_str, tz_abbr = time_str.rsplit(maxsplit=1)
+    # Parse the datetime string
+    dt_time = datetime.datetime.strptime(datetime_str, "%m/%d/%y %H:%M")
+
+    timezone = timezone_mapping.get(tz_abbr)
+    if timezone is None:
+        try:
+            # Get the timezone object
+            tz_zone = pytz.timezone(tz_abbr)
+        except ValueError as exc:
+            raise ValueError("Timezone not found: " + str(exc))  # pylint: disable=raise-missing-from
+    else:
+        # Get the timezone object
+        tz_zone = pytz.timezone(timezone)
+
+    # Convert to the specified timezone
+    dt_time = tz_zone.localize(dt_time)
+
+    # Convert to UTC
+    dt_utc = dt_time.astimezone(pytz.utc)
+
+    return dt_utc
+
+
 def rgetattr(obj, attr):
     """Recursive GetAttr to look for nested attributes."""
     nested_value = getattr(obj, attr)