# -*- coding: utf-8 -*-
'''
feedgen.entry
~~~~~~~~~~~~~
:copyright: 2013-2020, Lars Kiesow <lkiesow@uos.de>
:license: FreeBSD and LGPL, see license.* for more details.
'''
from datetime import datetime
from ..dateutil import parser as dateutil_parser
from ..dateutil import tz as dateutil_tz
import warnings
from lxml.etree import CDATA # nosec - adding CDATA entry is safe
from .compat import string_types
from .util import ensure_format, formatRFC2822, xml_fromstring, xml_elem
def _add_text_elm(entry, data, name):
"""Add a text subelement to an entry"""
if not data:
return
elm = xml_elem(name, entry)
type_ = data.get('type')
if data.get('src'):
if name != 'content':
raise ValueError("Only the 'content' element of an entry can "
"contain a 'src' attribute")
elm.attrib['src'] = data['src']
elif data.get(name):
# Surround xhtml with a div tag, parse it and embed it
if type_ == 'xhtml':
xhtml = '<div xmlns="http://www.w3.org/1999/xhtml">' \
+ data.get(name) + '</div>'
elm.append(xml_fromstring(xhtml))
elif type_ == 'CDATA':
elm.text = CDATA(data.get(name))
# Parse XML and embed it
elif type_ and (type_.endswith('/xml') or type_.endswith('+xml')):
elm.append(xml_fromstring(data[name]))
# Embed the text in escaped form
elif not type_ or type_.startswith('text') or type_ == 'html':
elm.text = data.get(name)
# Everything else should be included base64 encoded
else:
raise NotImplementedError(
'base64 encoded {} is not supported at the moment. '
'Pull requests adding support are welcome.'.format(name)
)
# Add type description of the content
if type_:
elm.attrib['type'] = type_
class FeedEntry(object):
'''FeedEntry call representing an ATOM feeds entry node or an RSS feeds item
node.
'''
def __init__(self):
# ATOM
# required
self.__atom_id = None
self.__atom_title = None
self.__atom_updated = datetime.now(dateutil_tz.tzutc())
# recommended
self.__atom_author = None
self.__atom_content = None
self.__atom_link = None
self.__atom_summary = None
# optional
self.__atom_category = None
self.__atom_contributor = None
self.__atom_published = None
self.__atom_source = None
self.__atom_rights = None
# RSS
self.__rss_author = None
self.__rss_category = None
self.__rss_comments = None
self.__rss_description = None
self.__rss_content = None
self.__rss_enclosure = None
self.__rss_guid = {}
self.__rss_link = None
self.__rss_pubDate = None
self.__rss_source = None
self.__rss_title = None
# Extension list:
self.__extensions = {}
self.__extensions_register = {}
def atom_entry(self, extensions=True):
'''Create an ATOM entry and return it.'''
entry = xml_elem('entry')
if not (self.__atom_id and self.__atom_title and self.__atom_updated):
raise ValueError('Required fields not set')
id = xml_elem('id', entry)
id.text = self.__atom_id
title = xml_elem('title', entry)
title.text = self.__atom_title
updated = xml_elem('updated', entry)
updated.text = self.__atom_updated.isoformat()
# An entry must contain an alternate link if there is no content
# element.
if not self.__atom_content:
links = self.__atom_link or []
if not [l for l in links if l.get('rel') == 'alternate']:
raise ValueError('Entry must contain an alternate link or ' +
'a content element.')
# Add author elements
for a in self.__atom_author or []:
# Atom requires a name. Skip elements without.
if not a.get('name'):
continue
author = xml_elem('author', entry)
name = xml_elem('name', author)
name.text = a.get('name')
if a.get('email'):
email = xml_elem('email', author)
email.text = a.get('email')
if a.get('uri'):
uri = xml_elem('uri', author)
uri.text = a.get('uri')
_add_text_elm(entry, self.__atom_content, 'content')
for l in self.__atom_link or []:
link = xml_elem('link', entry, href=l['href'])
if l.get('rel'):
link.attrib['rel'] = l['rel']
if l.get('type'):
link.attrib['type'] = l['type']
if l.get('hreflang'):
link.attrib['hreflang'] = l['hreflang']
if l.get('title'):
link.attrib['title'] = l['title']
if l.get('length'):
link.attrib['length'] = l['length']
_add_text_elm(entry, self.__atom_summary, 'summary')
for c in self.__atom_category or []:
cat = xml_elem('category', entry, term=c['term'])
if c.get('scheme'):
cat.attrib['scheme'] = c['scheme']
if c.get('label'):
cat.attrib['label'] = c['label']
# Add author elements
for c in self.__atom_contributor or []:
# Atom requires a name. Skip elements without.
if not c.get('name'):
continue
contrib = xml_elem('contributor', entry)
name = xml_elem('name', contrib)
name.text = c.get('name')
if c.get('email'):
email = xml_elem('email', contrib)
email.text = c.get('email')
if c.get('uri'):
uri = xml_elem('uri', contrib)
uri.text = c.get('uri')
if self.__atom_published:
published = xml_elem('published', entry)
published.text = self.__atom_published.isoformat()
if self.__atom_rights:
rights = xml_elem('rights', entry)
rights.text = self.__atom_rights
if self.__atom_source:
source = xml_elem('source', entry)
if self.__atom_source.get('title'):
source_title = xml_elem('title', source)
source_title.text = self.__atom_source['title']
if self.__atom_source.get('link'):
xml_elem('link', source, href=self.__atom_source['link'])
if extensions:
for ext in self.__extensions.values() or []:
if ext.get('atom'):
ext['inst'].extend_atom(entry)
return entry
def rss_entry(self, extensions=True):
'''Create a RSS item and return it.'''
entry = xml_elem('item')
if not (self.__rss_title or
self.__rss_description or
self.__rss_content):
raise ValueError('Required fields not set')
if self.__rss_title:
title = xml_elem('title', entry)
title.text = self.__rss_title
if self.__rss_link:
link = xml_elem('link', entry)
link.text = self.__rss_link
if self.__rss_description and self.__rss_content:
description = xml_elem('description', entry)
description.text = self.__rss_description
XMLNS_CONTENT = 'http://purl.org/rss/1.0/modules/content/'
content = xml_elem('{%s}encoded' % XMLNS_CONTENT, entry)
content.text = CDATA(self.__rss_content['content']) \
if self.__rss_content.get('type', '') == 'CDATA' \
else self.__rss_content['content']
elif self.__rss_description:
description = xml_elem('description', entry)
description.text = self.__rss_description
elif self.__rss_content:
description = xml_elem('description', entry)
description.text = CDATA(self.__rss_content['content']) \
if self.__rss_content.get('type', '') == 'CDATA' \
else self.__rss_content['content']
for a in self.__rss_author or []:
author = xml_elem('author', entry)
author.text = a
if self.__rss_guid.get('guid'):
guid = xml_elem('guid', entry)
guid.text = self.__rss_guid['guid']
permaLink = str(self.__rss_guid.get('permalink', False)).lower()
guid.attrib['isPermaLink'] = permaLink
for cat in self.__rss_category or []:
category = xml_elem('category', entry)
category.text = cat['value']
if cat.get('domain'):
category.attrib['domain'] = cat['domain']
if self.__rss_comments:
comments = xml_elem('comments', entry)
comments.text = self.__rss_comments
if self.__rss_enclosure:
enclosure = xml_elem('enclosure', entry)
enclosure.attrib['url'] = self.__rss_enclosure['url']
enclosure.attrib['length'] = self.__rss_enclosure['length']
enclosure.attrib['type'] = self.__rss_enclosure['type']
if self.__rss_pubDate:
pubDate = xml_elem('pubDate', entry)
pubDate.text = formatRFC2822(self.__rss_pubDate)
if self.__rss_source:
source = xml_elem('source', entry, url=self.__rss_source['url'])
source.text = self.__rss_source['title']
if extensions:
for ext in self.__extensions.values() or []:
if ext.get('rss'):
ext['inst'].extend_rss(entry)
return entry
def title(self, title=None):
'''Get or set the title value of the entry. It should contain a human
readable title for the entry. Title is mandatory for both ATOM and RSS
and should not be blank.
:param title: The new title of the entry.
:returns: The entriess title.
'''
if title is not None:
self.__atom_title = title
self.__rss_title = title
return self.__atom_title
def id(self, id=None):
'''Get or set the entry id which identifies the entry using a
universally unique and permanent URI. Two entries in a feed can have
the same value for id if they represent the same entry at different
points in time. This method will also set rss:guid with permalink set
to False. Id is mandatory for an ATOM entry.
:param id: New Id of the entry.
:returns: Id of the entry.
'''
if id is not None:
self.__atom_id = id
self.__rss_guid = {'guid': id, 'permalink': False}
return self.__atom_id
def guid(self, guid=None, permalink=False):
'''Get or set the entries guid which is a string that uniquely
identifies the item. This will also set atom:id.
:param guid: Id of the entry.
:param permalink: If this is a permanent identifier for this item
:returns: Id and permalink setting of the entry.
'''
if guid is not None:
self.__atom_id = guid
self.__rss_guid = {'guid': guid, 'permalink': permalink}
return self.__rss_guid
def updated(self, updated=None):
'''Set or get the updated value which indicates the last time the entry
was modified in a significant way.
The value can either be a string which will automatically be parsed or
a datetime.datetime object. In any case it is necessary that the value
include timezone information.
:param updated: The modification date.
:returns: Modification date as datetime.datetime
'''
if updated is not None:
if isinstance(updated, string_types):
updated = dateutil_parser.parse(updated)
if not isinstance(updated, datetime):
raise ValueError('Invalid datetime format')
if updated.tzinfo is None:
raise ValueError('Datetime object has no timezone info')
self.__atom_updated = updated
self.__rss_lastBuildDate = updated
return self.__atom_updated
def author(self, author=None, replace=False, **kwargs):
'''Get or set author data. An author element is a dict containing a
name, an email address and a uri. Name is mandatory for ATOM, email is
mandatory for RSS.
This method can be called with:
- the fields of an author as keyword arguments
- the fields of an author as a dictionary
- a list of dictionaries containing the author fields
An author has the following fields:
- *name* conveys a human-readable name for the person.
- *uri* contains a home page for the person.
- *email* contains an email address for the person.
:param author: Dict or list of dicts with author data.
:param replace: Add or replace old data.
Example::
>>> author({'name':'John Doe', 'email':'jdoe@example.com'})
[{'name':'John Doe','email':'jdoe@example.com'}]
>>> author([{'name': 'Mr. X'}, {'name': 'Max'}])
[{'name':'John Doe','email':'jdoe@example.com'},
{'name':'John Doe'}, {'name':'Max'}]
>>> author(name='John Doe', email='jdoe@example.com', replace=True)
[{'name':'John Doe','email':'jdoe@example.com'}]
'''
if author is None and kwargs:
author = kwargs
if author is not None:
if replace or self.__atom_author is None:
self.__atom_author = []
self.__atom_author += ensure_format(author,
set(['name', 'email', 'uri']),
set())
self.__rss_author = []
for a in self.__atom_author:
if a.get('email'):
if a.get('name'):
self.__rss_author.append('%(email)s (%(name)s)' % a)
else:
self.__rss_author.append('%(email)s' % a)
return self.__atom_author
def content(self, content=None, src=None, type=None):
'''Get or set the content of the entry which contains or links to the
complete content of the entry. Content must be provided for ATOM
entries if there is no alternate link, and should be provided if there
is no summary. If the content is set (not linked) it will also set
rss:description.
:param content: The content of the feed entry.
:param src: Link to the entries content.
:param type: If type is CDATA content would not be escaped.
:returns: Content element of the entry.
'''
if src is not None:
self.__atom_content = {'src': src}
elif content is not None:
self.__atom_content = {'content': content}
self.__rss_content = {'content': content}
if type is not None:
self.__atom_content['type'] = type
self.__rss_content['type'] = type
return self.__atom_content
def link(self, link=None, replace=False, **kwargs):
'''Get or set link data. An link element is a dict with the fields
href, rel, type, hreflang, title, and length. Href is mandatory for
ATOM.
This method can be called with:
- the fields of a link as keyword arguments
- the fields of a link as a dictionary
- a list of dictionaries containing the link fields
A link has the following fields:
- *href* is the URI of the referenced resource (typically a Web page)
- *rel* contains a single link relationship type. It can be a full URI,
or one of the following predefined values (default=alternate):
- *alternate* an alternate representation of the entry or feed, for
example a permalink to the html version of the entry, or the
front page of the weblog.
- *enclosure* a related resource which is potentially large in size
and might require special handling, for example an audio or video
recording.
- *related* an document related to the entry or feed.
- *self* the feed itself.
- *via* the source of the information provided in the entry.
- *type* indicates the media type of the resource.
- *hreflang* indicates the language of the referenced resource.
- *title* human readable information about the link, typically for
display purposes.
- *length* the length of the resource, in bytes.
RSS only supports one link with nothing but a URL. So for the RSS link
element the last link with rel=alternate is used.
RSS also supports one enclusure element per entry which is covered by
the link element in ATOM feed entries. So for the RSS enclusure element
the last link with rel=enclosure is used.
:param link: Dict or list of dicts with data.
:param replace: Add or replace old data.
:returns: List of link data.
'''
if link is None and kwargs:
link = kwargs
if link is not None:
if replace or self.__atom_link is None:
self.__atom_link = []
self.__atom_link += ensure_format(
link,
set(['href', 'rel', 'type', 'hreflang', 'title', 'length']),
set(['href']),
{'rel': ['alternate', 'enclosure', 'related', 'self', 'via']},
{'rel': 'alternate'})
# RSS only needs one URL. We use the first link for RSS:
for l in self.__atom_link:
if l.get('rel') == 'alternate':
self.__rss_link = l['href']
elif l.get('rel') == 'enclosure':
self.__rss_enclosure = {'url': l['href']}
self.__rss_enclosure['type'] = l.get('type')
self.__rss_enclosure['length'] = l.get('length') or '0'
# return the set with more information (atom)
return self.__atom_link
def summary(self, summary=None, type=None):
'''Get or set the summary element of an entry which conveys a short
summary, abstract, or excerpt of the entry. Summary is an ATOM only
element and should be provided if there either is no content provided
for the entry, or that content is not inline (i.e., contains a src
attribute), or if the content is encoded in base64. This method will
also set the rss:description field if it wasn't previously set or
contains the old value of summary.
:param summary: Summary of the entries contents.
:returns: Summary of the entries contents.
'''
if summary is not None:
# Replace the RSS description with the summary if it was the
# summary before. Not if it is the description.
if not self.__rss_description or (
self.__atom_summary and
self.__rss_description == self.__atom_summary.get("summary")
):
self.__rss_description = summary
self.__atom_summary = {'summary': summary}
if type is not None:
self.__atom_summary['type'] = type
return self.__atom_summary
def description(self, description=None, isSummary=False):
'''Get or set the description value which is the item synopsis.
Description is an RSS only element. For ATOM feeds it is split in
summary and content. The isSummary parameter can be used to control
which ATOM value is set when setting description.
:param description: Description of the entry.
:param isSummary: If the description should be used as content or
summary.
:returns: The entries description.
'''
if description is not None:
self.__rss_description = description
if isSummary:
self.__atom_summary = description
else:
self.__atom_content = {'content': description}
return self.__rss_description
def category(self, category=None, replace=False, **kwargs):
'''Get or set categories that the entry belongs to.
This method can be called with:
- the fields of a category as keyword arguments
- the fields of a category as a dictionary
- a list of dictionaries containing the category fields
A categories has the following fields:
- *term* identifies the category
- *scheme* identifies the categorization scheme via a URI.
- *label* provides a human-readable label for display
If a label is present it is used for the RSS feeds. Otherwise the term
is used. The scheme is used for the domain attribute in RSS.
:param category: Dict or list of dicts with data.
:param replace: Add or replace old data.
:returns: List of category data.
'''
if category is None and kwargs:
category = kwargs
if category is not None:
if replace or self.__atom_category is None:
self.__atom_category = []
self.__atom_category += ensure_format(
category,
set(['term', 'scheme', 'label']),
set(['term']))
# Map the ATOM categories to RSS categories. Use the atom:label as
# name or if not present the atom:term. The atom:scheme is the
# rss:domain.
self.__rss_category = []
for cat in self.__atom_category:
rss_cat = {}
rss_cat['value'] = cat.get('label', cat['term'])
if cat.get('scheme'):
rss_cat['domain'] = cat['scheme']
self.__rss_category.append(rss_cat)
return self.__atom_category
def contributor(self, contributor=None, replace=False, **kwargs):
'''Get or set the contributor data of the feed. This is an ATOM only
value.
This method can be called with:
- the fields of an contributor as keyword arguments
- the fields of an contributor as a dictionary
- a list of dictionaries containing the contributor fields
An contributor has the following fields:
- *name* conveys a human-readable name for the person.
- *uri* contains a home page for the person.
- *email* contains an email address for the person.
:param contributor: Dictionary or list of dictionaries with contributor
data.
:param replace: Add or replace old data.
:returns: List of contributors as dictionaries.
'''
if contributor is None and kwargs:
contributor = kwargs
if contributor is not None:
if replace or self.__atom_contributor is None:
self.__atom_contributor = []
self.__atom_contributor += ensure_format(
contributor, set(['name', 'email', 'uri']), set(['name']))
return self.__atom_contributor
def published(self, published=None):
'''Set or get the published value which contains the time of the initial
creation or first availability of the entry.
The value can either be a string which will automatically be parsed or
a datetime.datetime object. In any case it is necessary that the value
include timezone information.
:param published: The creation date.
:returns: Creation date as datetime.datetime
'''
if published is not None:
if isinstance(published, string_types):
published = dateutil_parser.parse(published)
if not isinstance(published, datetime):
raise ValueError('Invalid datetime format')
if published.tzinfo is None:
raise ValueError('Datetime object has no timezone info')
self.__atom_published = published
self.__rss_pubDate = published
return self.__atom_published
def pubDate(self, pubDate=None):
'''Get or set the pubDate of the entry which indicates when the entry
was published. This method is just another name for the published(...)
method.
'''
return self.published(pubDate)
def pubdate(self, pubDate=None):
'''Get or set the pubDate of the entry which indicates when the entry
was published. This method is just another name for the published(...)
method.
pubdate(…) is deprecated and may be removed in feedgen ≥ 0.8. Use
pubDate(…) instead.
'''
warnings.warn('pubdate(…) is deprecated and may be removed in feedgen '
'≥ 0.8. Use pubDate(…) instead.')
return self.published(pubDate)
def rights(self, rights=None):
'''Get or set the rights value of the entry which conveys information
about rights, e.g. copyrights, held in and over the entry. This ATOM
value will also set rss:copyright.
:param rights: Rights information of the feed.
:returns: Rights information of the feed.
'''
if rights is not None:
self.__atom_rights = rights
return self.__atom_rights
def comments(self, comments=None):
'''Get or set the value of comments which is the URL of the comments
page for the item. This is a RSS only value.
:param comments: URL to the comments page.
:returns: URL to the comments page.
'''
if comments is not None:
self.__rss_comments = comments
return self.__rss_comments
def source(self, url=None, title=None):
'''Get or set the source for the current feed entry.
Note that ATOM feeds support a lot more sub elements than title and URL
(which is what RSS supports) but these are currently not supported.
Patches are welcome.
:param url: Link to the source.
:param title: Title of the linked resource
:returns: Source element as dictionaries.
'''
if url is not None and title is not None:
self.__rss_source = {'url': url, 'title': title}
self.__atom_source = {'link': url, 'title': title}
return self.__rss_source
def enclosure(self, url=None, length=None, type=None):
'''Get or set the value of enclosure which describes a media object
that is attached to the item. This is a RSS only value which is
represented by link(rel=enclosure) in ATOM. ATOM feeds can furthermore
contain several enclosures while RSS may contain only one. That is why
this method, if repeatedly called, will add more than one enclosures to
the feed. However, only the last one is used for RSS.
:param url: URL of the media object.
:param length: Size of the media in bytes.
:param type: Mimetype of the linked media.
:returns: Data of the enclosure element.
'''
if url is not None:
self.link(href=url, rel='enclosure', type=type, length=length)
return self.__rss_enclosure
def ttl(self, ttl=None):
'''Get or set the ttl value. It is an RSS only element. ttl stands for
time to live. It's a number of minutes that indicates how long a
channel can be cached before refreshing from the source.
:param ttl: Integer value representing the time to live.
:returns: Time to live of of the entry.
'''
if ttl is not None:
self.__rss_ttl = int(ttl)
return self.__rss_ttl
def load_extension(self, name, atom=True, rss=True):
'''Load a specific extension by name.
:param name: Name of the extension to load.
:param atom: If the extension should be used for ATOM feeds.
:param rss: If the extension should be used for RSS feeds.
'''
# Check loaded extensions
if not isinstance(self.__extensions, dict):
self.__extensions = {}
if name in self.__extensions.keys():
raise ImportError('Extension already loaded')
# Load extension
extname = name[0].upper() + name[1:] + 'EntryExtension'
try:
supmod = __import__('feedgen.ext.%s_entry' % name)
extmod = getattr(supmod.ext, name + '_entry')
except ImportError:
# Use FeedExtension module instead
supmod = __import__('feedgen.ext.%s' % name)
extmod = getattr(supmod.ext, name)
ext = getattr(extmod, extname)
self.register_extension(name, ext, atom, rss)
def register_extension(self, namespace, extension_class_entry=None,
atom=True, rss=True):
'''Register a specific extension by classes to a namespace.
:param namespace: namespace for the extension
:param extension_class_entry: Class of the entry extension to load.
:param atom: If the extension should be used for ATOM feeds.
:param rss: If the extension should be used for RSS feeds.
'''
# Check loaded extensions
# `load_extension` ignores the "Extension" suffix.
if not isinstance(self.__extensions, dict):
self.__extensions = {}
if namespace in self.__extensions.keys():
raise ImportError('Extension already loaded')
if not extension_class_entry:
raise ImportError('No extension class')
extinst = extension_class_entry()
setattr(self, namespace, extinst)
# `load_extension` registry
self.__extensions[namespace] = {
'inst': extinst,
'extension_class_entry': extension_class_entry,
'atom': atom,
'rss': rss
}