Produits Entr'ouvert » Corbo

        exclude = ('identifier',)

        fields = ('name', 'rss_feed_url')

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

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('corbo', '0006_auto_20160928_0833'),

    ]

    operations = [

        migrations.AddField(

            model_name='announce',

            name='identifier',

            field=models.CharField(max_length=256, null=True, blank=True),

        ),

        migrations.AddField(

            model_name='category',

            name='rss_feed_url',

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

        ),

    ]

import io

from time import mktime

from lxml import etree

import requests

import feedparser

    rss_feed_url = models.URLField(_('Feed URL'), 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_feed_url:

            return

        feed_response = requests.get(self.rss_feed_url)

        if feed_response.ok:

            content = feedparser.parse(feed_response.content)

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

                substitutions = []

                published = datetime.fromtimestamp(mktime(entry.published_parsed))

                html_tree = etree.HTML(entry['summary'])

                storage = DefaultStorage()

                for img in html_tree.xpath('//img'):

                    image_name = os.path.basename(img.attrib['src'])

                    r = requests.get(img.attrib['src'])

                    new_content = r.content

                    if storage.exists(image_name):

                        old_content = storage.open(image_name).read()

                        old_hash = hashlib.md5(old_content).hexdigest()

                        new_hash = hashlib.md5(new_content).hexdigest()

                        img.attrib['src'] = storage.url(image_name)

                        if new_hash == old_hash:

                            continue

                    new_image_name = storage.save(image_name, io.BytesIO(new_content))

                    img.attrib['src'] = storage.url(new_image_name)

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

                                    category=self)

                announce.title = entry['title']

                announce.text = etree.tostring(html_tree)

                announce.publication_time = published

                announce.save()

                if created:

                    Broadcast.objects.get_or_create(announce=announce)

    identifier = models.CharField(max_length=256, null=True, blank=True)

        html_tree = etree.HTML(self.announce.text)

<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 os

import pytest

import mock

import feedparser

from django.core.files.storage import DefaultStorage

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">

      <img src="http://example.com/logo.png">

      Feed entry content

    </content>

    <author>

      <name>Foo Bar</name>

    </author>

  </entry>

</feed>

"""

def mocked_request_get(*args, **kwargs):

    storage = DefaultStorage()

    class MockResponse:

        def __init__(self, content):

            self.ok = True

            self.content = content

    if args[0] == 'http://example.com/atom':

        return MockResponse(ATOM_FEED)

    else:

        logo_path = os.path.join(os.path.dirname(__file__), 'media', 'logo.png')

        return MockResponse(file(logo_path).read())

@mock.patch('corbo.models.requests.get', side_effect=mocked_request_get)

def test_announces_from_feed(mocked_get):

    storage = DefaultStorage()

    feed_content = feedparser.parse(ATOM_FEED)

    for category in CATEGORIES:

        c = Category.objects.create(name=category, rss_feed_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']]

            assert storage.url('logo.png') in announce.text

    # cleanup uploaded images

    os.unlink(storage.path('logo.png'))

  mock