1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161
|
"""Feed Entry."""
from __future__ import annotations
import logging
import re
from abc import ABC, abstractmethod
from datetime import datetime
from .consts import CUSTOM_ATTRIBUTE
from .geo_rss_distance_helper import GeoRssDistanceHelper
from .xml_parser.feed_item import FeedItem
from .xml_parser.geometry import BoundingBox, Geometry, Point, Polygon
_LOGGER = logging.getLogger(__name__)
DEFAULT_FEATURES = [Point, Polygon, BoundingBox]
class FeedEntry(ABC):
"""Feed entry base class."""
def __init__(self, home_coordinates: tuple[float, float], rss_entry: FeedItem):
"""Initialise this feed entry."""
self._home_coordinates: tuple[float, float] = home_coordinates
self._rss_entry: FeedItem = rss_entry
def __repr__(self):
"""Return string representation of this entry."""
return f"<{self.__class__.__name__}(id={self.external_id})>"
@property
def features(self) -> list[type[Geometry]]:
"""Return the list of geometry types that this feed entry supports."""
return DEFAULT_FEATURES
@property
def geometries(self) -> list[Geometry] | None:
"""Return all geometries of this entry."""
if self._rss_entry:
# Return all geometries that are of type defined in features.
return list(
filter(lambda x: type(x) in self.features, self._rss_entry.geometries)
)
return None
@property
def coordinates(self) -> tuple[float, float] | None:
"""Return the best coordinates (latitude, longitude) of this entry."""
# This looks for the first point in the list of geometries. If there
# is no point then return the first entry.
if self.geometries and len(self.geometries) >= 1:
for entry in self.geometries:
if isinstance(entry, Point):
return GeoRssDistanceHelper.extract_coordinates(entry)
# No point found.
return GeoRssDistanceHelper.extract_coordinates(self.geometries[0])
return None
@property
def external_id(self) -> str | None:
"""Return the external id of this entry."""
if self._rss_entry:
external_id: str | None = self._rss_entry.guid
if not external_id:
external_id = self.title
if not external_id:
# Use geometry as ID as a fallback.
external_id = str(hash(self.coordinates))
return external_id
return None
def _search_in_external_id(self, regexp) -> str | None:
"""Find a sub-string in the entry's external id."""
if self.external_id:
match = re.search(regexp, self.external_id)
if match:
return match.group(CUSTOM_ATTRIBUTE)
return None
@property
def title(self) -> str | None:
"""Return the title of this entry."""
if self._rss_entry:
return self._rss_entry.title
return None
def _search_in_title(self, regexp):
"""Find a sub-string in the entry's title."""
if self.title:
match = re.search(regexp, self.title)
if match:
return match.group(CUSTOM_ATTRIBUTE)
return None
@property
def category(self) -> str | None:
"""Return the category of this entry."""
if (
self._rss_entry
and self._rss_entry.category
and isinstance(self._rss_entry.category, list)
):
# To keep this simple, just return the first category.
return self._rss_entry.category[0]
return None
@property
@abstractmethod
def attribution(self) -> str | None:
"""Return the attribution of this entry."""
return None
@property
def distance_to_home(self) -> float:
"""Return the distance in km of this entry to the home coordinates."""
# This goes through all geometries and reports back the closest
# distance to any of them.
distance: float = float("inf")
if self.geometries and len(self.geometries) >= 1:
for geometry in self.geometries:
distance = min(
distance,
GeoRssDistanceHelper.distance_to_geometry(
self._home_coordinates, geometry
),
)
return distance
@property
def description(self) -> str | None:
"""Return the description of this entry."""
if self._rss_entry and self._rss_entry.description:
return self._rss_entry.description
return None
@property
def published(self) -> datetime | None:
"""Return the published date of this entry."""
if self._rss_entry:
return self._rss_entry.published_date
return None
@property
def updated(self) -> datetime | None:
"""Return the updated date of this entry."""
if self._rss_entry:
return self._rss_entry.updated_date
return None
def _search_in_description(self, regexp):
"""Find a sub-string in the entry's description."""
if self.description:
match = re.search(regexp, self.description)
if match:
return match.group(CUSTOM_ATTRIBUTE)
return None
@staticmethod
def _string2boolean(value: str) -> bool:
"""Convert value to boolean."""
return isinstance(value, str) and value.strip().lower() in {"true", "yes", "1"}
|