mygpo/api/legacy.py

   1 #
   2 # This file is part of my.gpodder.org.
   3 #
   4 # my.gpodder.org is free software: you can redistribute it and/or modify it
   5 # under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or (at your
   7 # option) any later version.
   8 #
   9 # my.gpodder.org is distributed in the hope that it will be useful, but
  10 # WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  11 # or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
  12 # License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with my.gpodder.org. If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 from datetime import datetime
  19
  20 from django.http import HttpResponse
  21 from django.utils.datastructures import MultiValueDictKeyError
  22 from django.views.decorators.csrf import csrf_exempt
  23 from django.views.decorators.cache import never_cache
  24 from django.contrib.auth import get_user_model
  25
  26 from mygpo.podcasts.models import Podcast
  27 from mygpo.api.opml import Importer, Exporter
  28 from mygpo.users.models import SubscriptionException
  29 from mygpo.api.backend import get_device
  30 from mygpo.utils import normalize_feed_url
  31 from mygpo.subscriptions import subscribe, unsubscribe
  32
  33 import logging
  34 logger = logging.getLogger(__name__)
  35
  36
  37 LEGACY_DEVICE_NAME = 'Legacy Device'
  38 LEGACY_DEVICE_UID  = 'legacy'
  39
  40 @never_cache
  41 @csrf_exempt
  42 def upload(request):
  43     try:
  44         emailaddr = request.POST['username']
  45         password  = request.POST['password']
  46         action    = request.POST['action']
  47         protocol  = request.POST['protocol']
  48         opml      = request.FILES['opml'].read()
  49     except MultiValueDictKeyError:
  50         return HttpResponse("@PROTOERROR", content_type='text/plain')
  51
  52     user = auth(emailaddr, password)
  53     if (not user):
  54         return HttpResponse('@AUTHFAIL', content_type='text/plain')
  55
  56     dev = get_device(user, LEGACY_DEVICE_UID,
  57             request.META.get('HTTP_USER_AGENT', ''))
  58
  59     existing_urls = [x.url for x in dev.get_subscribed_podcasts()]
  60
  61     i = Importer(opml)
  62
  63     podcast_urls = [p['url'] for p in i.items]
  64     podcast_urls = list(map(normalize_feed_url, podcast_urls))
  65     podcast_urls = [_f for _f in podcast_urls if _f]
  66
  67     new = [u for u in podcast_urls if u not in existing_urls]
  68     rem = [u for e in existing_urls if u not in podcast_urls]
  69
  70     #remove duplicates
  71     new = list(set(new))
  72     rem = list(set(rem))
  73
  74     for n in new:
  75         p = Podcast.objects.get_or_create_for_url(n)
  76         subscribe(p, user, dev)
  77
  78     for r in rem:
  79         p = Podcast.objects.get_or_create_for_url(r)
  80         unsubscribe(p, user, dev)
  81
  82     return HttpResponse('@SUCCESS', content_type='text/plain')
  83
  84 @never_cache
  85 @csrf_exempt
  86 def getlist(request):
  87     emailaddr = request.GET.get('username', None)
  88     password = request.GET.get('password', None)
  89
  90     user = auth(emailaddr, password)
  91     if user is None:
  92         return HttpResponse('@AUTHFAIL', content_type='text/plain')
  93
  94     dev = get_device(user, LEGACY_DEVICE_UID,
  95             request.META.get('HTTP_USER_AGENT', ''),
  96             undelete=True)
  97     podcasts = dev.get_subscribed_podcasts()
  98
  99     title = "{username}'s subscriptions".format(username=user.username)
 100     exporter = Exporter(title)
 101
 102     opml = exporter.generate(podcasts)
 103
 104     return HttpResponse(opml, content_type='text/xml')
 105
 106
 107 def auth(emailaddr, password):
 108     if emailaddr is None or password is None:
 109         return None
 110
 111     User = get_user_model()
 112     try:
 113         user = User.objects.get(email=emailaddr)
 114     except User.DoesNotExist:
 115         return None
 116
 117     if not user.check_password(password):
 118         return None
 119
 120     if not user.is_active:
 121         return None
 122
 123     return user