Python源码示例：feedgen.feed.FeedGenerator()
示例1
def gen_feed():
fg = FeedGenerator()
fg.id(f"{ID}")
fg.title(f"{USERNAME} notes")
fg.author({"name": USERNAME})
fg.link(href=ID, rel="alternate")
fg.description(f"{USERNAME} notes")
fg.logo(ME.get("icon", {}).get("url"))
fg.language("en")
for item in DB.activities.find(
{
"box": Box.OUTBOX.value,
"type": "Create",
"meta.deleted": False,
"meta.public": True,
},
limit=10,
).sort("_id", -1):
fe = fg.add_entry()
fe.id(item["activity"]["object"].get("url"))
fe.link(href=item["activity"]["object"].get("url"))
fe.title(item["activity"]["object"]["content"])
fe.description(item["activity"]["object"]["content"])
return fg
示例2
def feed():
"""
Generate atom feed
"""
entries = parse_posts(0, C.feed_count)
fg = FeedGenerator()
fg.id(str(len(entries)))
fg.title(C.title)
fg.subtitle(C.subtitle)
fg.language(C.language)
fg.author(dict(name=C.author, email=C.email))
fg.link(href=C.root_url, rel='alternate')
fg.link(href=make_abs_url(C.root_url, 'feed'), rel='self')
for entry in entries:
fe = fg.add_entry()
fe.id(entry.get('url'))
fe.title(entry.get('title'))
fe.published(entry.get('date'))
fe.updated(entry.get('updated') or entry.get('date'))
fe.link(href=make_abs_url(C.root_url, entry.get('url')), rel='alternate')
fe.author(dict(name=entry.get('author'), email=entry.get('email')))
fe.content(entry.get('body'))
atom_feed = fg.atom_str(pretty=True)
return atom_feed
示例3
def generate_feed():
tz = pytz.timezone(config.timezone)
# Get latest X entries from database
entries = dbhelper.get_latest_entries()
fg = FeedGenerator()
# Feed id
fg.id(config.bot_link)
# Creator info (for Atom)
fg.author(name=config.author_name, email=config.author_email, replace=True )
# Self link to the feed
fg.link(href=config.feed_link, rel='self')
# Set description of your feed
fg.description(config.feed_description)
# Last time feed updated (use system time with timezone)
fg.lastBuildDate(formatdate(datetime.timestamp(datetime.now(tz)), localtime=True))
fg.title(config.feed_title)
# Set time-to-live (I really don't know why set this)
fg.ttl(5)
# Does this parameter mean anything?
fg.language(config.feed_language)
for entry in entries:
item = fg.add_entry()
# Use message id to form valid URL (new feature in Telegram since Feb 2016)
item.id("{!s}".format(entry["pid"]))
item.link(href="{!s}/{!s}".format(config.bot_link, entry["pid"]), rel="alternate")
# Set title and content from message text
item.title(entry["ptext"])
item.content(entry["ptext"])
# Set publish/update datetime
item.pubdate(entry["pdate"])
item.updated(entry["pdate"])
# Write RSS/Atom feed to file
# It's preferred to have only one type at a time (or just create two functions)
fg.atom_file('static/atom.xml')
# fg.rss_file('static/rss.xml')
示例4
def generate_feed():
tz = pytz.timezone(config.timezone)
# Get latest X entries from database
entries = dbhelper.get_latest_entries()
fg = FeedGenerator()
# Feed id
fg.id(config.bot_link)
# Creator info (for Atom)
fg.author(name=config.author_name, email=config.author_email, replace=True )
# Self link to the feed
fg.link(href=config.feed_link, rel='self')
# Set description of your feed
fg.description(config.feed_description)
# Last time feed updated (use system time with timezone)
fg.lastBuildDate(formatdate(datetime.timestamp(datetime.now(tz)), localtime=True))
fg.title(config.feed_title)
# Set time-to-live (I really don't know why set this)
fg.ttl(5)
# Does this parameter mean anything?
fg.language(config.feed_language)
for entry in entries:
item = fg.add_entry()
# Use message id to form valid URL (new feature in Telegram since Feb 2016)
item.id("{!s}".format(entry["pid"]))
item.link(href="{!s}/{!s}".format(config.bot_link, entry["pid"]), rel="alternate")
# Set title and content from message text
item.title(entry["ptext"])
item.content(entry["ptext"])
# Set publish/update datetime
item.pubdate(entry["pdate"])
item.updated(entry["pdate"])
# Write RSS/Atom feed to file
# It's preferred to have only one type at a time (or just create two functions)
fg.atom_file('static/atom.xml')
# fg.rss_file('static/rss.xml')
示例5
def podcast_feed():
logo_url = url_for("static", filename="wpclogo_big.png", _external=True)
fg = FeedGenerator()
fg.load_extension('podcast')
fg.podcast.itunes_category('Technology', 'Podcasting')
fg.podcast.itunes_image(logo_url)
fg.author({'name': 'Nathan Kellert', 'email': 'nathankellert@gmail.com'})
fg.link(href='http://watchpeoplecode.com/podcast_feed.xml', rel='self')
fg.title('WPC Coders Podcast')
fg.description('WPC Coders Podcast is a weekly peek into the lives of developers and the WatchPeopleCode community. Our goal is to keep our listeners entertained by giving them new and interesting insights into our industry as well as awesome things happening within our own community. Here, you can expect hear about some of the latest news, tools, and opportunities for developers in nearly every aread of our industry. Most importantly, we hope to have some fun and a few laughs in ways only other nerds know how.') # NOQA
episodes = [('ep1.mp3', 'Episode 1', datetime(2015, 02, 21, 23), 'Learn all about the WPC hosts, and where we came from in Episode 1!'),
('ep2.mp3', 'Episode 2', datetime(2015, 02, 28, 23), 'This week we cover your news, topics and questions in episode 2!'),
('ep3.mp3', 'Episode 3', datetime(2015, 03, 07, 23), "On todays podcast we talk to WatchPeopleCode's founder Alex Putilin. Hear about how the reddit search engine thousands watched him write. Also, hear the inside scoop of how WatchPeopleCode got started!"), # NOQA
('ep4.mp3', 'Episode 4', datetime(2015, 03, 14, 23), "This week we talk to FreeCodeCamps Quincy Larson(http://www.freecodecamp.com) about their project that combines teaching new developers how to code and completing projects for non-profits! Lets find out how this group of streamers code with a cause!")] # NOQA
for epfile, eptitle, epdate, epdescription in episodes[::-1]:
epurl = "https://s3.amazonaws.com/wpcpodcast/{}".format(epfile)
fe = fg.add_entry()
fe.id(epurl)
fe.title(eptitle)
fe.description(epdescription)
fe.podcast.itunes_image(logo_url)
fe.pubdate(epdate.replace(tzinfo=pytz.UTC))
fe.enclosure(epurl, 0, 'audio/mpeg')
return Response(response=fg.rss_str(pretty=True),
status=200,
mimetype='application/rss+xml')
示例6
def __init_feed(setup):
feed = FeedGenerator()
feed.id(setup.get("id"))
feed.title(setup.get("title"))
feed.subtitle(setup.get("subtitle"))
feed.author({"name": setup.get("name"), "email": setup.get("email")})
feed.link(href=setup.get("link"), rel="self")
feed.logo(setup.get("logo"))
return feed
示例7
def news_feed():
feed = FeedGenerator()
feed.id("https://jazzband.co/news/feed")
feed.link(href="https://jazzband.co/", rel="alternate")
feed.title("Jazzband News Feed")
feed.subtitle("We are all part of this.")
feed.link(href=full_url(request.url), rel="self")
# the list of updates of all news for setting the feed's updated value
updates = []
for page in news_pages:
if page.path == "index":
continue
# make the datetime timezone aware if needed
published = page.meta.get("published", None)
if published and published.tzinfo is None:
published = pytz.utc.localize(published)
updated = page.meta.get("updated", published)
if updated:
if updated.tzinfo is None:
updated = pytz.utc.localize(updated)
updates.append(updated)
summary = page.meta.get("summary", None)
author = page.meta.get("author", None)
author_link = page.meta.get("author_link", None)
url = full_url(url_for("content.news", path=page.path))
entry = feed.add_entry()
entry.id(url)
entry.title(page.meta["title"])
entry.summary(summary)
entry.content(content=str(page.html), type="html")
if author is not None:
author = {"name": author}
if author_link is not None:
author["uri"] = author_link
entry.author(author)
entry.link(href=url)
entry.updated(updated)
entry.published(published)
sorted_updates = sorted(updates)
feed.updated(sorted_updates and sorted_updates[-1] or datetime.utcnow())
return Response(feed.atom_str(pretty=True), mimetype="application/atom+xml")
示例8
def lambda_handler(event, context):
rss = event['rss']
bucket_name = event['bucket']
logging.info("Processing url: %s" % rss)
logging.info("Using bucket: %s" % bucket_name)
session = Session(region_name="us-west-2")
polly = session.client("polly")
s3 = resource('s3')
bucket = s3.Bucket(bucket_name)
logging.info("getting list of existing objects in the given bucket")
files = set(o.key for o in bucket.objects.all())
feed = feedparser.parse(rss)
title = feed['feed']['title']
fg = FeedGenerator()
fg.load_extension('podcast')
fg.title('Audio podcast based on: %s' % title)
fg.link(href=feed.feed.link, rel='alternate')
fg.subtitle(feed.feed.description)
ENTRY_URL = "http://{bucket}.s3-website.{region}.amazonaws.com/{filename}"
for entry in get_entries(feed):
filename = "%s.mp3" % entry['id']
fe = fg.add_entry()
fe.id(entry['id'])
fe.title(entry['title'])
fe.published(entry['published'])
entry_url = ENTRY_URL.format(bucket=bucket_name, filename=filename, region=os.environ["AWS_REGION"])
fe.enclosure(entry_url, 0, 'audio/mpeg')
if filename in files:
logging.info('Article "%s" with id %s already exist, skipping.'
% (entry['title'], entry['id']))
continue
try:
logging.info("Next entry, size: %d" % len(entry['content']))
logging.debug("Content: %s" % entry['content'])
response = polly.synthesize_speech(
Text=entry['content'],
OutputFormat="mp3",
VoiceId="Joanna")
with closing(response["AudioStream"]) as stream:
bucket.put_object(Key=filename, Body=stream.read())
except BotoCoreError as error:
logging.error(error)
bucket.put_object(Key='podcast.xml', Body=fg.rss_str(pretty=True))
