mygpo/api/legacy.py

   1 from django.http import HttpResponse
   2 from django.utils.datastructures import MultiValueDictKeyError
   3 from django.views.decorators.csrf import csrf_exempt
   4 from django.views.decorators.cache import never_cache
   5 from django.contrib.auth import get_user_model
   6
   7 from mygpo.podcasts.models import Podcast
   8 from mygpo.api.opml import Importer, Exporter
   9 from mygpo.api.backend import get_device
  10 from mygpo.utils import normalize_feed_url
  11 from mygpo.subscriptions.tasks import subscribe, unsubscribe
  12
  13 import logging
  14
  15 logger = logging.getLogger(__name__)
  16
  17
  18 LEGACY_DEVICE_NAME = "Legacy Device"
  19 LEGACY_DEVICE_UID = "legacy"
  20
  21
  22 @never_cache
  23 @csrf_exempt
  24 def upload(request):
  25     try:
  26         emailaddr = request.POST["username"]
  27         password = request.POST["password"]
  28         action = request.POST["action"]
  29         protocol = request.POST["protocol"]
  30         opml = request.FILES["opml"].read()
  31     except MultiValueDictKeyError:
  32         return HttpResponse("@PROTOERROR", content_type="text/plain")
  33
  34     user = auth(emailaddr, password)
  35     if not user:
  36         return HttpResponse("@AUTHFAIL", content_type="text/plain")
  37
  38     dev = get_device(user, LEGACY_DEVICE_UID, request.META.get("HTTP_USER_AGENT", ""))
  39
  40     existing_urls = [x.url for x in dev.get_subscribed_podcasts()]
  41
  42     i = Importer(opml)
  43
  44     podcast_urls = [p["url"] for p in i.items]
  45     podcast_urls = map(normalize_feed_url, podcast_urls)
  46     podcast_urls = list(filter(None, podcast_urls))
  47
  48     new = [u for u in podcast_urls if u not in existing_urls]
  49     rem = [u for u in existing_urls if u not in podcast_urls]
  50
  51     # remove duplicates
  52     new = list(set(new))
  53     rem = list(set(rem))
  54
  55     for n in new:
  56         p = Podcast.objects.get_or_create_for_url(n).object
  57         subscribe(p.pk, user.pk, dev.uid)
  58
  59     for r in rem:
  60         p = Podcast.objects.get_or_create_for_url(r).object
  61         unsubscribe(p.pk, user.pk, dev.uid)
  62
  63     return HttpResponse("@SUCCESS", content_type="text/plain")
  64
  65
  66 @never_cache
  67 @csrf_exempt
  68 def getlist(request):
  69     emailaddr = request.GET.get("username", None)
  70     password = request.GET.get("password", None)
  71
  72     user = auth(emailaddr, password)
  73     if user is None:
  74         return HttpResponse("@AUTHFAIL", content_type="text/plain")
  75
  76     dev = get_device(
  77         user, LEGACY_DEVICE_UID, request.META.get("HTTP_USER_AGENT", ""), undelete=True
  78     )
  79     podcasts = dev.get_subscribed_podcasts()
  80
  81     title = "{username}'s subscriptions".format(username=user.username)
  82     exporter = Exporter(title)
  83
  84     opml = exporter.generate(podcasts)
  85
  86     return HttpResponse(opml, content_type="text/xml")
  87
  88
  89 def auth(emailaddr, password):
  90     if emailaddr is None or password is None:
  91         return None
  92
  93     User = get_user_model()
  94     try:
  95         user = User.objects.get(email=emailaddr)
  96     except User.DoesNotExist:
  97         return None
  98
  99     if not user.check_password(password):
 100         return None
 101
 102     if not user.is_active:
 103         return None
 104
 105     return user