/usr/lib/python2.7/dist-packages/pelican/writers.py is in python-pelican 3.4.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | # -*- coding: utf-8 -*-
from __future__ import with_statement, unicode_literals, print_function
import six
import os
import locale
import logging
if not six.PY3:
from codecs import open
from feedgenerator import Atom1Feed, Rss201rev2Feed
from jinja2 import Markup
from six.moves.urllib.parse import urlparse
from pelican.paginator import Paginator
from pelican.utils import (get_relative_path, path_to_url, set_date_tzinfo,
is_selected_for_writing)
from pelican import signals
logger = logging.getLogger(__name__)
class Writer(object):
def __init__(self, output_path, settings=None):
self.output_path = output_path
self.reminder = dict()
self.settings = settings or {}
self._written_files = set()
self._overridden_files = set()
def _create_new_feed(self, feed_type, context):
feed_class = Rss201rev2Feed if feed_type == 'rss' else Atom1Feed
sitename = Markup(context['SITENAME']).striptags()
feed = feed_class(
title=sitename,
link=(self.site_url + '/'),
feed_url=self.feed_url,
description=context.get('SITESUBTITLE', ''))
return feed
def _add_item_to_the_feed(self, feed, item):
title = Markup(item.title).striptags()
link = '%s/%s' % (self.site_url, item.url)
feed.add_item(
title=title,
link=link,
unique_id='tag:%s,%s:%s' % (urlparse(link).netloc,
item.date.date(),
urlparse(link).path.lstrip('/')),
description=item.get_content(self.site_url),
categories=item.tags if hasattr(item, 'tags') else None,
author_name=getattr(item, 'author', ''),
pubdate=set_date_tzinfo(
item.modified if hasattr(item, 'modified') else item.date,
self.settings.get('TIMEZONE', None)))
def _open_w(self, filename, encoding, override=False):
"""Open a file to write some content to it.
Exit if we have already written to that file, unless one (and no more
than one) of the writes has the override parameter set to True.
"""
if filename in self._overridden_files:
if override:
raise RuntimeError('File %s is set to be overridden twice'
% filename)
else:
logger.info('skipping %s' % filename)
filename = os.devnull
elif filename in self._written_files:
if override:
logger.info('overwriting %s' % filename)
else:
raise RuntimeError('File %s is to be overwritten' % filename)
if override:
self._overridden_files.add(filename)
self._written_files.add(filename)
return open(filename, 'w', encoding=encoding)
def write_feed(self, elements, context, path=None, feed_type='atom'):
"""Generate a feed with the list of articles provided
Return the feed. If no path or output_path is specified, just
return the feed object.
:param elements: the articles to put on the feed.
:param context: the context to get the feed metadata.
:param path: the path to output.
:param feed_type: the feed type to use (atom or rss)
"""
if not is_selected_for_writing(self.settings, path):
return
old_locale = locale.setlocale(locale.LC_ALL)
locale.setlocale(locale.LC_ALL, str('C'))
try:
self.site_url = context.get(
'SITEURL', path_to_url(get_relative_path(path)))
self.feed_domain = context.get('FEED_DOMAIN')
self.feed_url = '{}/{}'.format(self.feed_domain, path)
feed = self._create_new_feed(feed_type, context)
max_items = len(elements)
if self.settings['FEED_MAX_ITEMS']:
max_items = min(self.settings['FEED_MAX_ITEMS'], max_items)
for i in range(max_items):
self._add_item_to_the_feed(feed, elements[i])
if path:
complete_path = os.path.join(self.output_path, path)
try:
os.makedirs(os.path.dirname(complete_path))
except Exception:
pass
encoding = 'utf-8' if six.PY3 else None
with self._open_w(complete_path, encoding) as fp:
feed.write(fp, 'utf-8')
logger.info('writing %s' % complete_path)
return feed
finally:
locale.setlocale(locale.LC_ALL, old_locale)
def write_file(self, name, template, context, relative_urls=False,
paginated=None, override_output=False, **kwargs):
"""Render the template and write the file.
:param name: name of the file to output
:param template: template to use to generate the content
:param context: dict to pass to the templates.
:param relative_urls: use relative urls or absolutes ones
:param paginated: dict of article list to paginate - must have the
same length (same list in different orders)
:param override_output: boolean telling if we can override previous
output with the same name (and if next files written with the same
name should be skipped to keep that one)
:param **kwargs: additional variables to pass to the templates
"""
if name is False or name == "" or\
not is_selected_for_writing(self.settings,\
os.path.join(self.output_path, name)):
return
elif not name:
# other stuff, just return for now
return
def _write_file(template, localcontext, output_path, name, override):
"""Render the template write the file."""
output = template.render(localcontext)
path = os.path.join(output_path, name)
try:
os.makedirs(os.path.dirname(path))
except Exception:
pass
with self._open_w(path, 'utf-8', override=override) as f:
f.write(output)
logger.info('writing {}'.format(path))
# Send a signal to say we're writing a file with some specific
# local context.
signals.content_written.send(path, context=localcontext)
localcontext = context.copy()
if relative_urls:
relative_url = path_to_url(get_relative_path(name))
context['localsiteurl'] = relative_url
localcontext['SITEURL'] = relative_url
localcontext['output_file'] = name
localcontext.update(kwargs)
# pagination
if paginated:
# pagination needed, init paginators
paginators = {key: Paginator(name, val, self.settings)
for key, val in paginated.items()}
# generated pages, and write
for page_num in range(list(paginators.values())[0].num_pages):
paginated_localcontext = localcontext.copy()
for key in paginators.keys():
paginator = paginators[key]
previous_page = paginator.page(page_num) \
if page_num > 0 else None
page = paginator.page(page_num + 1)
next_page = paginator.page(page_num + 2) \
if page_num + 1 < paginator.num_pages else None
paginated_localcontext.update(
{'%s_paginator' % key: paginator,
'%s_page' % key: page,
'%s_previous_page' % key: previous_page,
'%s_next_page' % key: next_page})
_write_file(template, paginated_localcontext, self.output_path,
page.save_as, override_output)
else:
# no pagination
_write_file(template, localcontext, self.output_path, name,
override_output)
|