/
feed_entry.py
160 lines (138 loc) · 5.44 KB
/
feed_entry.py
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
"""Feed Entry."""
import logging
import re
from abc import ABC, abstractmethod
from datetime import datetime
from typing import List, Optional, Tuple, Type
from .consts import CUSTOM_ATTRIBUTE
from .geo_rss_distance_helper import GeoRssDistanceHelper
from .xml_parser.feed_item import FeedItem
from .xml_parser.geometry import Geometry, Point, Polygon, BoundingBox
_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 = home_coordinates
self._rss_entry = rss_entry
def __repr__(self):
"""Return string representation of this entry."""
return "<{}(id={})>".format(self.__class__.__name__, 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) -> Optional[List[Geometry]]:
"""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) -> Optional[Tuple[float, float]]:
"""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) -> Optional[str]:
"""Return the external id of this entry."""
if self._rss_entry:
external_id = 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 = hash(self.coordinates)
return external_id
return None
def _search_in_external_id(self, regexp) -> Optional[str]:
"""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) -> Optional[str]:
"""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) -> Optional[str]:
"""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) -> Optional[str]:
"""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("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) -> Optional[str]:
"""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) -> Optional[datetime]:
"""Return the published date of this entry."""
if self._rss_entry:
return self._rss_entry.published_date
return None
@property
def updated(self) -> Optional[datetime]:
"""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"}