Revert "bundle queries for episode states by URLs (fixes bug 1588)"
[mygpo.git] / mygpo / api / advanced / __init__.py
blobdaff083fc31b77942aff82bb13ad77619526febe
2 # This file is part of my.gpodder.org.
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.
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.
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/>.
18 from functools import partial
19 from itertools import imap, chain
20 from collections import defaultdict, namedtuple
21 from datetime import datetime
23 import dateutil.parser
25 try:
26 import gevent
27 except ImportError:
28 gevent = None
30 from django.http import HttpResponse, HttpResponseBadRequest, Http404, HttpResponseNotFound
31 from django.contrib.sites.models import RequestSite
32 from django.db import IntegrityError
33 from django.views.decorators.csrf import csrf_exempt
34 from django.views.decorators.cache import never_cache
36 from mygpo.api.constants import EPISODE_ACTION_TYPES, DEVICE_TYPES
37 from mygpo.api.httpresponse import JsonResponse
38 from mygpo.api.sanitizing import sanitize_url, sanitize_urls
39 from mygpo.api.advanced.directory import episode_data, podcast_data
40 from mygpo.api.backend import get_device, BulkSubscribe
41 from mygpo.couch import BulkException, bulk_save_retry
42 from mygpo.log import log
43 from mygpo.utils import parse_time, format_time, parse_bool, get_timestamp
44 from mygpo.decorators import allowed_methods, repeat_on_conflict
45 from mygpo.core import models
46 from mygpo.core.models import SanitizingRule, Podcast
47 from mygpo.core.tasks import auto_flattr_episode
48 from mygpo.users.models import PodcastUserState, EpisodeAction, \
49 EpisodeUserState, DeviceDoesNotExist, DeviceUIDException, \
50 InvalidEpisodeActionAttributes
51 from mygpo.users.settings import FLATTR_AUTO
52 from mygpo.json import json, JSONDecodeError
53 from mygpo.api.basic_auth import require_valid_user, check_username
54 from mygpo.db.couchdb.episode import episode_by_id, \
55 favorite_episodes_for_user, episodes_for_podcast
56 from mygpo.db.couchdb.podcast import podcast_for_url
57 from mygpo.db.couchdb.podcast_state import subscribed_podcast_ids_by_device
58 from mygpo.db.couchdb.episode_state import get_podcasts_episode_states, \
59 episode_state_for_ref_urls, get_episode_actions
62 # keys that are allowed in episode actions
63 EPISODE_ACTION_KEYS = ('position', 'episode', 'action', 'device', 'timestamp',
64 'started', 'total', 'podcast')
67 @csrf_exempt
68 @require_valid_user
69 @check_username
70 @never_cache
71 @allowed_methods(['GET', 'POST'])
72 def subscriptions(request, username, device_uid):
74 now = datetime.now()
75 now_ = get_timestamp(now)
77 if request.method == 'GET':
79 try:
80 device = request.user.get_device_by_uid(device_uid)
81 except DeviceDoesNotExist as e:
82 return HttpResponseNotFound(str(e))
84 since_ = request.GET.get('since', None)
85 if since_ == None:
86 return HttpResponseBadRequest('parameter since missing')
87 try:
88 since = datetime.fromtimestamp(float(since_))
89 except ValueError:
90 return HttpResponseBadRequest('since-value is not a valid timestamp')
92 changes = get_subscription_changes(request.user, device, since, now)
94 return JsonResponse(changes)
96 elif request.method == 'POST':
97 d = get_device(request.user, device_uid,
98 request.META.get('HTTP_USER_AGENT', ''))
100 if not request.raw_post_data:
101 return HttpResponseBadRequest('POST data must not be empty')
103 actions = json.loads(request.raw_post_data)
104 add = actions['add'] if 'add' in actions else []
105 rem = actions['remove'] if 'remove' in actions else []
107 add = filter(None, add)
108 rem = filter(None, rem)
110 try:
111 update_urls = update_subscriptions(request.user, d, add, rem)
112 except IntegrityError, e:
113 return HttpResponseBadRequest(e)
115 return JsonResponse({
116 'timestamp': now_,
117 'update_urls': update_urls,
121 def update_subscriptions(user, device, add, remove):
123 for a in add:
124 if a in remove:
125 raise IntegrityError('can not add and remove %s at the same time' % a)
127 add_s = list(sanitize_urls(add, 'podcast'))
128 rem_s = list(sanitize_urls(remove, 'podcast'))
130 assert len(add) == len(add_s) and len(remove) == len(rem_s)
132 updated_urls = filter(lambda (a, b): a != b, zip(add + remove, add_s + rem_s))
134 add_s = filter(None, add_s)
135 rem_s = filter(None, rem_s)
137 # If two different URLs (in add and remove) have
138 # been sanitized to the same, we ignore the removal
139 rem_s = filter(lambda x: x not in add_s, rem_s)
141 subscriber = BulkSubscribe(user, device)
143 for a in add_s:
144 subscriber.add_action(a, 'subscribe')
146 for r in rem_s:
147 subscriber.add_action(r, 'unsubscribe')
149 try:
150 subscriber.execute()
151 except BulkException as be:
152 for err in be.errors:
153 log('Advanced API: %(username)s: Updating subscription for '
154 '%(podcast_url)s on %(device_uid)s failed: '
155 '%(rerror)s (%(reason)s)'.format(username=user.username,
156 podcast_url=err.doc, device_uid=device.uid,
157 error=err.error, reason=err.reason)
160 return updated_urls
163 def get_subscription_changes(user, device, since, until):
164 add_urls, rem_urls = device.get_subscription_changes(since, until)
165 until_ = get_timestamp(until)
166 return {'add': add_urls, 'remove': rem_urls, 'timestamp': until_}
169 @csrf_exempt
170 @require_valid_user
171 @check_username
172 @never_cache
173 @allowed_methods(['GET', 'POST'])
174 def episodes(request, username, version=1):
176 version = int(version)
177 now = datetime.now()
178 now_ = get_timestamp(now)
179 ua_string = request.META.get('HTTP_USER_AGENT', '')
181 if request.method == 'POST':
182 try:
183 actions = json.loads(request.raw_post_data)
184 except (JSONDecodeError, UnicodeDecodeError) as e:
185 msg = 'Advanced API: could not decode episode update POST data for user %s: %s' % (username, e)
186 log(msg)
187 return HttpResponseBadRequest(msg)
189 try:
190 update_urls = update_episodes(request.user, actions, now, ua_string)
191 except DeviceUIDException as e:
192 import traceback
193 log('could not update episodes for user %s: %s %s: %s' % (username, e, traceback.format_exc(), actions))
194 return HttpResponseBadRequest(str(e))
195 except InvalidEpisodeActionAttributes as e:
196 import traceback
197 log('could not update episodes for user %s: %s %s: %s' % (username, e, traceback.format_exc(), actions))
198 return HttpResponseBadRequest(str(e))
200 return JsonResponse({'timestamp': now_, 'update_urls': update_urls})
202 elif request.method == 'GET':
203 podcast_url= request.GET.get('podcast', None)
204 device_uid = request.GET.get('device', None)
205 since_ = request.GET.get('since', None)
206 aggregated = parse_bool(request.GET.get('aggregated', False))
208 try:
209 since = int(since_) if since_ else None
210 except ValueError:
211 return HttpResponseBadRequest('since-value is not a valid timestamp')
213 if podcast_url:
214 podcast = podcast_for_url(podcast_url)
215 if not podcast:
216 raise Http404
217 else:
218 podcast = None
220 if device_uid:
222 try:
223 device = request.user.get_device_by_uid(device_uid)
224 except DeviceDoesNotExist as e:
225 return HttpResponseNotFound(str(e))
227 else:
228 device = None
230 changes = get_episode_changes(request.user, podcast, device, since,
231 now_, aggregated, version)
233 return JsonResponse(changes)
237 def convert_position(action):
238 """ convert position parameter for API 1 compatibility """
239 pos = getattr(action, 'position', None)
240 if pos is not None:
241 action.position = format_time(pos)
242 return action
246 def get_episode_changes(user, podcast, device, since, until, aggregated, version):
248 devices = dict( (dev.id, dev.uid) for dev in user.devices )
250 args = {}
251 if podcast is not None:
252 args['podcast_id'] = podcast.get_id()
254 if device is not None:
255 args['device_id'] = device.id
257 actions = get_episode_actions(user._id, since, until, **args)
259 if version == 1:
260 actions = imap(convert_position, actions)
262 clean_data = partial(clean_episode_action_data,
263 user=user, devices=devices)
265 actions = map(clean_data, actions)
266 actions = filter(None, actions)
268 if aggregated:
269 actions = dict( (a['episode'], a) for a in actions ).values()
271 return {'actions': actions, 'timestamp': until}
276 def clean_episode_action_data(action, user, devices):
278 if None in (action.get('podcast', None), action.get('episode', None)):
279 return None
281 if 'device_id' in action:
282 device_id = action['device_id']
283 device_uid = devices.get(device_id)
284 if device_uid:
285 action['device'] = device_uid
287 del action['device_id']
289 # remove superfluous keys
290 for x in action.keys():
291 if x not in EPISODE_ACTION_KEYS:
292 del action[x]
294 # set missing keys to None
295 for x in EPISODE_ACTION_KEYS:
296 if x not in action:
297 action[x] = None
299 if action['action'] != 'play':
300 if 'position' in action:
301 del action['position']
303 if 'total' in action:
304 del action['total']
306 if 'started' in action:
307 del action['started']
309 if 'playmark' in action:
310 del action['playmark']
312 else:
313 action['position'] = action.get('position', False) or 0
315 return action
321 def update_episodes(user, actions, now, ua_string):
322 update_urls = []
324 grouped_actions = defaultdict(list)
326 # group all actions by their episode
327 for action in actions:
329 podcast_url = action['podcast']
330 podcast_url = sanitize_append(podcast_url, 'podcast', update_urls)
331 if podcast_url == '': continue
333 episode_url = action['episode']
334 episode_url = sanitize_append(episode_url, 'episode', update_urls)
335 if episode_url == '': continue
337 act = parse_episode_action(action, user, update_urls, now, ua_string)
338 grouped_actions[ (podcast_url, episode_url) ].append(act)
341 auto_flattr_episodes = []
343 # Prepare the updates for each episode state
344 obj_funs = []
346 for (p_url, e_url), action_list in grouped_actions.iteritems():
347 episode_state = episode_state_for_ref_urls(user, p_url, e_url)
349 if any(a['action'] == 'play' for a in actions):
350 auto_flattr_episodes.append(episode_state.episode)
352 fun = partial(update_episode_actions, action_list=action_list)
353 obj_funs.append( (episode_state, fun) )
355 bulk_save_retry(obj_funs)
357 if user.get_wksetting(FLATTR_AUTO):
358 for episode_id in auto_flattr_episodes:
359 auto_flattr_episode.delay(user, episode_id)
361 return update_urls
364 def update_episode_actions(episode_state, action_list):
365 """ Adds actions to the episode state and saves if necessary """
367 len1 = len(episode_state.actions)
368 episode_state.add_actions(action_list)
370 if len(episode_state.actions) == len1:
371 return None
373 return episode_state
377 def parse_episode_action(action, user, update_urls, now, ua_string):
378 action_str = action.get('action', None)
379 if not valid_episodeaction(action_str):
380 raise Exception('invalid action %s' % action_str)
382 new_action = EpisodeAction()
384 new_action.action = action['action']
386 if action.get('device', False):
387 device = get_device(user, action['device'], ua_string)
388 new_action.device = device.id
390 if action.get('timestamp', False):
391 new_action.timestamp = dateutil.parser.parse(action['timestamp'])
392 else:
393 new_action.timestamp = now
394 new_action.timestamp = new_action.timestamp.replace(microsecond=0)
396 new_action.upload_timestamp = get_timestamp(now)
398 new_action.started = action.get('started', None)
399 new_action.playmark = action.get('position', None)
400 new_action.total = action.get('total', None)
402 return new_action
405 @csrf_exempt
406 @require_valid_user
407 @check_username
408 @never_cache
409 # Workaround for mygpoclient 1.0: It uses "PUT" requests
410 # instead of "POST" requests for uploading device settings
411 @allowed_methods(['POST', 'PUT'])
412 def device(request, username, device_uid):
413 d = get_device(request.user, device_uid,
414 request.META.get('HTTP_USER_AGENT', ''))
416 data = json.loads(request.raw_post_data)
418 if 'caption' in data:
419 if not data['caption']:
420 return HttpResponseBadRequest('caption must not be empty')
421 d.name = data['caption']
423 if 'type' in data:
424 if not valid_devicetype(data['type']):
425 return HttpResponseBadRequest('invalid device type %s' % data['type'])
426 d.type = data['type']
429 request.user.update_device(d)
431 return HttpResponse()
434 def valid_devicetype(type):
435 for t in DEVICE_TYPES:
436 if t[0] == type:
437 return True
438 return False
440 def valid_episodeaction(type):
441 for t in EPISODE_ACTION_TYPES:
442 if t[0] == type:
443 return True
444 return False
447 @csrf_exempt
448 @require_valid_user
449 @check_username
450 @never_cache
451 @allowed_methods(['GET'])
452 def devices(request, username):
453 devices = filter(lambda d: not d.deleted, request.user.devices)
454 devices = map(device_data, devices)
455 return JsonResponse(devices)
458 def device_data(device):
459 return dict(
460 id = device.uid,
461 caption = device.name,
462 type = device.type,
463 subscriptions= len(subscribed_podcast_ids_by_device(device)),
468 def get_podcast_data(podcasts, domain, url):
469 """ Gets podcast data for a URL from a dict of podcasts """
470 podcast = podcasts.get(url)
471 return podcast_data(podcast, domain)
474 def get_episode_data(podcasts, domain, clean_action_data, include_actions, episode_status):
475 """ Get episode data for an episode status object """
476 podcast_id = episode_status.episode.podcast
477 podcast = podcasts.get(podcast_id, None)
478 t = episode_data(episode_status.episode, domain, podcast)
479 t['status'] = episode_status.status
481 # include latest action (bug 1419)
482 if include_actions and episode_status.action:
483 t['action'] = clean_action_data(episode_status.action)
485 return t
488 @csrf_exempt
489 @require_valid_user
490 @check_username
491 @never_cache
492 def updates(request, username, device_uid):
493 now = datetime.now()
494 now_ = get_timestamp(now)
496 try:
497 device = request.user.get_device_by_uid(device_uid)
498 except DeviceDoesNotExist as e:
499 return HttpResponseNotFound(str(e))
501 since_ = request.GET.get('since', None)
502 if since_ == None:
503 return HttpResponseBadRequest('parameter since missing')
504 try:
505 since = datetime.fromtimestamp(float(since_))
506 except ValueError:
507 return HttpResponseBadRequest('since-value is not a valid timestamp')
509 include_actions = parse_bool(request.GET.get('include_actions', False))
511 ret = get_subscription_changes(request.user, device, since, now)
512 domain = RequestSite(request).domain
514 subscriptions = list(device.get_subscribed_podcasts())
516 podcasts = dict( (p.url, p) for p in subscriptions )
517 prepare_podcast_data = partial(get_podcast_data, podcasts, domain)
519 ret['add'] = map(prepare_podcast_data, ret['add'])
521 devices = dict( (dev.id, dev.uid) for dev in request.user.devices )
522 clean_action_data = partial(clean_episode_action_data,
523 user=request.user, devices=devices)
525 # index subscribed podcasts by their Id for fast access
526 podcasts = dict( (p.get_id(), p) for p in subscriptions )
527 prepare_episode_data = partial(get_episode_data, podcasts, domain,
528 clean_action_data, include_actions)
530 episode_updates = get_episode_updates(request.user, subscriptions, since)
531 ret['updates'] = map(prepare_episode_data, episode_updates)
533 return JsonResponse(ret)
536 def get_episode_updates(user, subscribed_podcasts, since):
537 """ Returns the episode updates since the timestamp """
539 EpisodeStatus = namedtuple('EpisodeStatus', 'episode status action')
541 episode_status = {}
543 # get episodes
544 if gevent:
545 episode_jobs = [gevent.spawn(episodes_for_podcast, p, since) for p in
546 subscribed_podcasts]
547 gevent.joinall(episode_jobs)
548 episodes = chain.from_iterable(job.get() for job in episode_jobs)
550 else:
551 episodes = chain.from_iterable(episodes_for_podcast(p, since) for p
552 in subscribed_podcasts)
555 for episode in episodes:
556 episode_status[episode._id] = EpisodeStatus(episode, 'new', None)
559 # get episode states
560 if gevent:
561 e_action_jobs = [gevent.spawn(get_podcasts_episode_states, p, user._id)
562 for p in subscribed_podcasts]
563 gevent.joinall(e_action_jobs)
564 e_actions = chain.from_iterable(job.get() for job in e_action_jobs)
566 else:
567 e_actions = [get_podcasts_episode_states(p, user._id) for p
568 in subscribed_podcasts]
571 for action in e_actions:
572 e_id = action['episode_id']
574 if e_id in episode_status:
575 episode = episode_status[e_id].episode
576 else:
577 episode = episode_by_id(e_id)
579 episode_status[e_id] = EpisodeStatus(episode, action['action'], action)
581 return episode_status.itervalues()
584 @require_valid_user
585 @check_username
586 @never_cache
587 def favorites(request, username):
588 favorites = favorite_episodes_for_user(request.user)
589 domain = RequestSite(request).domain
590 e_data = lambda e: episode_data(e, domain)
591 ret = map(e_data, favorites)
592 return JsonResponse(ret)
595 def sanitize_append(url, obj_type, sanitized_list):
596 urls = sanitize_url(url, obj_type)
597 if url != urls:
598 sanitized_list.append( (url, urls) )
599 return urls