-
Notifications
You must be signed in to change notification settings - Fork 16
/
atom.py
177 lines (134 loc) · 5.92 KB
/
atom.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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
from flask import Blueprint, request, make_response
from portality import models as models
from portality.core import app
from portality.crosswalks.atom import AtomCrosswalk
from lxml import etree
from datetime import datetime, timedelta
from portality.lib import plausible, dates
from portality.lib.dates import FMT_DATETIME_STD
blueprint = Blueprint('atom', __name__)
@blueprint.route('/feed')
@plausible.pa_event(app.config.get('ANALYTICS_CATEGORY_ATOM', 'Atom'),
action=app.config.get('ANALYTICS_ACTION_ACTION', 'Feed Request'))
def feed():
# get the feed for this base_url (which is just used to set the metadata of
# the feed, but we want to do this outside of a request context so it
# is testable)
f = get_feed(request.base_url)
# serialise and respond with the atom xml
resp = make_response(f.serialise())
resp.mimetype = "application/atom+xml"
return resp
def get_feed(base_url=None):
"""
Main method for generating the feed. Gets all of the settings
out of config and returns the feed object, which can then
be serialised and delivered by the web layer
:param base_url: The base url to include in the feed metadata
:return: AtomFeed object
"""
max_size = app.config.get("MAX_FEED_ENTRIES", 20)
max_age = app.config.get("MAX_FEED_ENTRY_AGE", 2592000)
from_date = (dates.now() - timedelta(0, max_age)).strftime(FMT_DATETIME_STD)
dao = models.AtomRecord()
records = dao.list_records(from_date, max_size)
title = app.config.get("FEED_TITLE", "untitled")
url = base_url
generator = app.config.get('FEED_GENERATOR', "")
icon = app.config.get("FEED_LOGO", "")
logo = app.config.get("FEED_LOGO", "")
link = app.config.get('BASE_URL', "")
rights = app.config.get('FEED_LICENCE', "")
xwalk = AtomCrosswalk()
f = AtomFeed(title, url, generator, icon, logo, link, rights)
for record in records:
entry = xwalk.crosswalk(record)
f.add_entry(entry)
return f
class AtomFeed(object):
ATOM_NAMESPACE = "http://www.w3.org/2005/Atom"
ATOM = "{%s}" % ATOM_NAMESPACE
NSMAP = {None: ATOM_NAMESPACE}
def __init__(self, title, url, generator, icon, logo, link, rights):
self.title = title
self.url = url
self.generator = generator
self.icon = icon
self.logo = logo
self.link = link
self.rights = rights
self.last_updated = None
self.entries = {}
def add_entry(self, entry):
# update the "last_updated" property if necessary
lu = entry.get("updated")
dr = dates.parse(lu)
if self.last_updated is None or dr > self.last_updated:
self.last_updated = dr
# record the entries by date
if lu in self.entries:
self.entries[lu].append(entry)
else:
self.entries[lu] = [entry]
def serialise(self):
if self.last_updated is None:
self.last_updated = dates.now()
feed = etree.Element(self.ATOM + "feed", nsmap=self.NSMAP)
title = etree.SubElement(feed, self.ATOM + "title")
title.text = self.title
if self.generator is not None:
generator = etree.SubElement(feed, self.ATOM + "generator")
generator.text = self.generator
icon = etree.SubElement(feed, self.ATOM + "icon")
icon.text = self.icon
if self.logo is not None:
logo = etree.SubElement(feed, self.ATOM + "logo")
logo.text = self.logo
self_link = etree.SubElement(feed, self.ATOM + "link")
self_link.set("rel", "self")
self_link.set("href", self.url)
link = etree.SubElement(feed, self.ATOM + "link")
link.set("rel", "related")
link.set("href", self.link)
rights = etree.SubElement(feed, self.ATOM + "rights")
rights.text = self.rights
updated = etree.SubElement(feed, self.ATOM + "updated")
dr = datetime.strftime(self.last_updated, FMT_DATETIME_STD)
updated.text = dr
entry_dates = list(self.entries.keys())
entry_dates.sort(reverse=True)
for ed in entry_dates:
es = self.entries.get(ed)
for e in es:
self._serialise_entry(feed, e)
tree = etree.ElementTree(feed)
return etree.tostring(tree, pretty_print=True, xml_declaration=True, encoding="utf-8")
def _serialise_entry(self, feed, e):
entry = etree.SubElement(feed, self.ATOM + "entry")
author = etree.SubElement(entry, self.ATOM + "author")
name = etree.SubElement(author, self.ATOM + "name")
name.text = e['author']
for cat in e.get("categories", []):
c = etree.SubElement(entry, self.ATOM + "category")
c.set("term", cat)
cont = etree.SubElement(entry, self.ATOM + "content")
cont.set("src", e['content_src'])
id = etree.SubElement(entry, self.ATOM + "id")
id.text = e['id']
# this is not strictly necessary, as we have an atom:content element, but it can't harm
alt = etree.SubElement(entry, self.ATOM + "link")
alt.set("rel", "alternate")
alt.set("href", e['alternate'])
if "related" in e:
rel = etree.SubElement(entry, self.ATOM + "link")
rel.set("rel", "related")
rel.set("href", e['related'])
rights = etree.SubElement(entry, self.ATOM + "rights")
rights.text = e['rights']
summary = etree.SubElement(entry, self.ATOM + "summary")
summary.set("type", "text")
summary.text = e['summary']
title = etree.SubElement(entry, self.ATOM + "title")
title.text = e['title']
updated = etree.SubElement(entry, self.ATOM + "updated")
updated.text = e['updated']