Merge pull request #123 from networktocode/develop

Release 2.0.7

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## v2.0.7 - 2021-12-01
+
+### Fixed
+
+- #120 - Improve handling of Zayo notifications.
+- #121 - Defer loading of `tzwhere` data until it's needed, to reduce memory overhead.
+
 ## v2.0.6 - 2021-11-30
 
 ### Added

circuit_maintenance_parser/parsers/zayo.py

@@ -1,5 +1,6 @@
 """Zayo parser."""
 import logging
+import re
 from typing import Dict
 
 import bs4  # type: ignore
@@ -22,15 +23,19 @@ class SubjectParserZayo1(EmailSubjectParser):
              END OF WINDOW NOTIFICATION***Customer Inc.***ZAYO TTN-0000123456 Planned***
              ***Customer Inc***ZAYO TTN-0001234567 Emergency MAINTENANCE NOTIFICATION***
              RESCHEDULE NOTIFICATION***Customer Inc***ZAYO TTN-0005423873 Planned***
+
+    Some degenerate examples have been seen as well:
+            [notices] CANCELLED NOTIFICATION***Customer,inc***ZAYO TTN-0005432100 Planned**
+            [notices] Rescheduled  Maintenance***ZAYO TTN-0005471719 MAINTENANCE NOTIFICATION***
     """
 
     def parse_subject(self, subject):
         """Parse subject of email message."""
         data = {}
-        tokens = subject.split("***")
+        tokens = re.split(r"\*+", subject)
         if len(tokens) == 4:
             data["account"] = tokens[1]
-            data["maintenance_id"] = tokens[2].split(" ")[1]
+        data["maintenance_id"] = tokens[-2].split(" ")[1]
         return [data]
 
 
@@ -48,7 +53,7 @@ def parse_html(self, soup):
                 text = soup.get_text()
                 if "will be commencing momentarily" in text:
                     data["status"] = Status("IN-PROCESS")
-                elif "has been completed" in text:
+                elif "has been completed" in text or "has closed" in text:
                     data["status"] = Status("COMPLETED")
 
         return [data]

circuit_maintenance_parser/provider.py

@@ -335,6 +335,11 @@ class Verizon(GenericProvider):
 class Zayo(GenericProvider):
     """Zayo provider custom class."""
 
+    _include_filter = {
+        "text/html": ["Maintenance Ticket #"],
+        "html": ["Maintenance Ticket #"],
+    }
+
     _processors: List[GenericProcessor] = [
         CombinedProcessor(data_parsers=[EmailDateParser, SubjectParserZayo1, HtmlParserZayo1]),
     ]

circuit_maintenance_parser/utils.py

@@ -16,36 +16,46 @@
 dirname = os.path.dirname(__file__)
 
 
+class classproperty:  # pylint: disable=invalid-name,too-few-public-methods
+    """Simple class-level equivalent of an @property."""
+
+    def __init__(self, method):
+        """Wrap a method."""
+        self.getter = method
+
+    def __get__(self, _, cls):
+        """Call the wrapped method."""
+        return self.getter(cls)
+
+
 class Geolocator:
     """Class to obtain Geo Location coordinates."""
 
     # Keeping caching of local DB and timezone in the class
-    db_location: Dict[Union[Tuple[str, str], str], Tuple[float, float]] = {}
-    timezone = None
+    _db_location: Dict[Union[Tuple[str, str], str], Tuple[float, float]] = {}
+    _timezone = None
 
-    def __init__(self):
-        """Initialize instance."""
-        self.load_db_location()
-        self.load_timezone()
-
-    @classmethod
-    def load_timezone(cls):
+    @classproperty
+    def timezone(cls):  # pylint: disable=no-self-argument
         """Load the timezone resolver."""
-        if cls.timezone is None:
-            cls.timezone = tzwhere.tzwhere()
+        if cls._timezone is None:
+            cls._timezone = tzwhere.tzwhere()
             logger.info("Loaded local timezone resolver.")
-
-    @classmethod
-    def load_db_location(cls):
-        """Load the localtions DB from CSV into a Dict."""
-        with open(os.path.join(dirname, "data", "worldcities.csv")) as csvfile:
-            reader = csv.DictReader(csvfile)
-            for row in reader:
-                # Index by city and country
-                cls.db_location[(row["city_ascii"], row["country"])] = (float(row["lat"]), float(row["lng"]))
-                # Index by city (first entry wins if duplicated names)
-                if row["city_ascii"] not in cls.db_location:
-                    cls.db_location[row["city_ascii"]] = (float(row["lat"]), float(row["lng"]))
+        return cls._timezone
+
+    @classproperty
+    def db_location(cls):  # pylint: disable=no-self-argument
+        """Load the locations DB from CSV into a Dict."""
+        if not cls._db_location:
+            with open(os.path.join(dirname, "data", "worldcities.csv")) as csvfile:
+                reader = csv.DictReader(csvfile)
+                for row in reader:
+                    # Index by city and country
+                    cls._db_location[(row["city_ascii"], row["country"])] = (float(row["lat"]), float(row["lng"]))
+                    # Index by city (first entry wins if duplicated names)
+                    if row["city_ascii"] not in cls._db_location:
+                        cls._db_location[row["city_ascii"]] = (float(row["lat"]), float(row["lng"]))
+        return cls._db_location
 
     def get_location(self, city: str) -> Tuple[float, float]:
         """Get location."""
@@ -64,7 +74,9 @@ def get_location_from_local_file(self, city: str) -> Tuple[float, float]:
         city_name = city.split(", ")[0]
         country = city.split(", ")[-1]
 
-        lat, lng = self.db_location.get((city_name, country), self.db_location.get(city_name, (None, None)))
+        lat, lng = self.db_location.get(  # pylint: disable=no-member
+            (city_name, country), self.db_location.get(city_name, (None, None))  # pylint: disable=no-member
+        )
         if lat and lng:
             logger.debug("Resolved %s to lat %s, lon %sfrom local locations DB.", city, lat, lng)
             return (lat, lng)
@@ -92,12 +104,12 @@ def city_timezone(self, city: str) -> str:
         if self.timezone is not None:
             try:
                 latitude, longitude = self.get_location(city)
-                timezone = self.timezone.tzNameAt(latitude, longitude)
+                timezone = self.timezone.tzNameAt(latitude, longitude)  # pylint: disable=no-member
                 if not timezone:
                     # In some cases, given a latitued and longitued, the tzwhere library returns
                     # an empty timezone, so we try with the coordinates from the API as an alternative
                     latitude, longitude = self.get_location_from_api(city)
-                    timezone = self.timezone.tzNameAt(latitude, longitude)
+                    timezone = self.timezone.tzNameAt(latitude, longitude)  # pylint: disable=no-member
 
                 if timezone:
                     logger.debug("Matched city %s to timezone %s", city, timezone)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "circuit-maintenance-parser"
-version = "2.0.6"
+version = "2.0.7"
 description = "Python library to parse Circuit Maintenance notifications and return a structured data back"
 authors = ["Network to Code <opensource@networktocode.com>"]
 license = "Apache-2.0"