Produits Entr'ouvert » Corbo

        fields = ('name', 'rss_stream_url')

# corbo - Announces Manager

# Copyright (C) 2016 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from django.core.management.base import BaseCommand, CommandError

from corbo.models import Category

class Command(BaseCommand):

    def handle(self, *args, **kwargs):

        for c in Category.objects.filter(rss_stream_url__isnull=False):

            c.save()

# -*- coding: utf-8 -*-

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('corbo', '0006_sms'),

    ]

    operations = [

        migrations.AddField(

            model_name='category',

            name='rss_stream_url',

            field=models.URLField(help_text='if defined, announces will be automatically created from rss items', null=True, blank=True),

            preserve_default=True,

        ),

    ]

import requests

import feedparser

    rss_stream_url = models.URLField(blank=True, null=True,

                help_text=_('if defined, announces will be automatically created from rss items'))

    def save(self, *args, **kwargs):

        super(Category, self).save(*args, **kwargs)

        if not self.rss_stream_url:

            return

        feed_response = requests.get(self.rss_stream_url)

        if feed_response.ok:

            content = feedparser.parse(feed_response.content)

            for entry in content.get('entries', []):

                announce, created = Announce.objects.get_or_create(title=entry['title'],

                                                            category=self)

                if created:

                    announce.publication_time = entry['updated']

                    announce.text = entry['summary']

                    announce.save()

<a href="{% url 'edit_category' object.id %}" rel="popup">{% trans 'Edit' %}</a>

  <h2>{% trans "Edit Category" %}</h2>

    python-emails,

    python-requests,

    python-feedparser

0 * * * * corbo corbo-manage tenant_command sync_external_feeds --all-tenants

feedparser

requests

        'feedparser',

        'requests'

import pytest

import mock

import feedparser

from django.core.urlresolvers import reverse

from django.core.urlresolvers import reverse

from corbo.models import Category

pytestmark = pytest.mark.django_db

CATEGORIES = ('Alerts',)

ATOM_FEED = """

<?xml version="1.0" encoding="UTF-8"?>

<feed xml:lang="en-US" xmlns="http://www.w3.org/2005/Atom" xmlns:wfw="http://wellformedweb.org/CommentAPI/">

  <id>tag:linuxfr.org,2005:/news</id>

  <title>Sample RSS Feeds</title>

  <updated>2016-09-16T10:29:46+02:00</updated>

  <entry>

    <id>tag:linuxfr.org,2005:News/37537</id>

    <published>2016-09-16T10:29:46+02:00</published>

    <updated>2016-09-16T11:27:00+02:00</updated>

    <title>Feed entry sample</title>

    <content type="html">

      Feed entry content

    </content>

    <author>

      <name>Foo Bar</name>

    </author>

  </entry>

</feed>

"""

@mock.patch('corbo.models.requests.get')

def test_announces_from_feed(mocked_get):

    feed_content = feedparser.parse(ATOM_FEED)

    mocked_get.return_value = mock.Mock(ok=True, content=ATOM_FEED)

    for category in CATEGORIES:

        c = Category.objects.create(name=category, rss_stream_url='http://example.com/atom')

        assert c.announce_set.count() == len(feed_content['entries'])

        for announce in c.announce_set.all():

            assert announce.title in [feed['title'] for feed in feed_content['entries']]

  mock