mygpo/maintenance/merge.py

   1 from itertools import chain, imap as map
   2 import logging
   3 from functools import partial
   4
   5 import restkit
   6
   7 from mygpo.core.models import Podcast, Episode, PodcastGroup
   8 from mygpo.users.models import PodcastUserState, EpisodeUserState
   9 from mygpo import utils
  10 from mygpo.decorators import repeat_on_conflict
  11 from mygpo.db.couchdb.episode import episodes_for_podcast
  12 from mygpo.db.couchdb.podcast_state import all_podcast_states
  13 from mygpo.db.couchdb.episode_state import all_episode_states
  14 from mygpo.db.couchdb.utils import multi_request_view
  15
  16
  17 class IncorrectMergeException(Exception):
  18     pass
  19
  20
  21 def podcast_url_wrapper(r):
  22     url = r['key']
  23     doc = r['doc']
  24     if doc['doc_type'] == 'Podcast':
  25         obj = Podcast.wrap(doc)
  26     elif doc['doc_type'] == 'PodcastGroup':
  27         obj = PodcastGroup.wrap(doc)
  28
  29     return obj.get_podcast_by_url(url)
  30
  31
  32 def merge_objects(podcasts=True, podcast_states=False, episodes=False,
  33         episode_states=False, dry_run=False):
  34     """
  35     Merges objects (podcasts, episodes, states) based on different criteria
  36     """
  37
  38     # The "smaller" podcast is merged into the "greater"
  39     podcast_merge_order = lambda a, b: cmp(a.subscriber_count(), b.subscriber_count())
  40     no_merge_order = lambda a, b: 0
  41
  42     merger = partial(merge_from_iterator, dry_run=dry_run,
  43             progress_callback=utils.progress)
  44
  45
  46     if podcasts:
  47
  48         print 'Merging Podcasts by URL'
  49         podcasts, total = get_view_count_iter(Podcast,
  50                 'podcasts/by_url',
  51                 wrap = False,
  52                 include_docs=True)
  53         podcasts = map(podcast_url_wrapper, podcasts)
  54         merger(podcasts, similar_urls, podcast_merge_order, total,
  55                 merge_podcasts)
  56         print
  57
  58
  59         print 'Merging Podcasts by Old-Id'
  60         podcasts, total = get_view_count_iter(Podcast,
  61                 'podcasts/by_oldid',
  62                 wrap = False,
  63                 include_docs=True)
  64         podcasts = imap(podcast_oldid_wrapper, podcasts)
  65         merger(podcasts, similar_oldid, podcast_merge_order, total,
  66                 merge_podcasts)
  67         print
  68
  69
  70     if podcast_states:
  71         print 'Merging Duplicate Podcast States'
  72         states, total = get_view_count_iter(PodcastUserState,
  73                 'podcast_states/by_user',
  74                 include_docs=True)
  75         should_merge = lambda a, b: a == b
  76         merger(states, should_merge, no_merge_order, total,
  77                 merge_podcast_states)
  78         print
  79
  80
  81     if episodes:
  82         print 'Merging Episodes by URL'
  83         episodes, total = get_view_count_iter(Episode,
  84                 'episodes/by_podcast_url',
  85                 include_docs=True)
  86         should_merge = lambda a, b: a.podcast == b.podcast and \
  87                 similar_urls(a, b)
  88         merger(episodes, should_merge, no_merge_order, total, merge_episodes)
  89         print
  90
  91
  92         print 'Merging Episodes by Old-Id'
  93         episodes, total = get_view_count_iter(Episode,
  94                 'episodes/by_oldid',
  95                 include_docs=True)
  96         should_merge = lambda a, b: a.podcast == b.podcast and \
  97                 similar_oldid(a, b)
  98         merger(episodes, should_merge, no_merge_order, total, merge_episodes)
  99         print
 100
 101
 102     if episode_states:
 103         print 'Merging Duplicate Episode States'
 104         states, total = get_view_count_iter(EpisodeUserState,
 105                 'episode_states/by_user_episode',
 106                 include_docs=True)
 107         should_merge = lambda a, b: (a.user, a.episode) == \
 108                                     (b.user, b.episode)
 109         merger(states, should_merge, no_merge_order, total,
 110                 merge_episode_states)
 111         print
 112
 113
 114
 115 def get_view_count_iter(cls, view, *args, **kwargs):
 116     iterator = multi_request_view(cls, view, *args, **kwargs)
 117     total = cls.view(view, limit=0).total_rows
 118     return iterator, total
 119
 120
 121 def merge_from_iterator(obj_it, should_merge, cmp, total, merge_func,
 122         dry_run=False, progress_callback=lambda *args, **kwargs: None):
 123     """
 124     Iterates over the objects in obj_it and calls should_merge for each pair of
 125     objects. This implies that the objects returned by obj_it should be sorted
 126     such that potential merge-candiates appear after each other.
 127
 128     If should_merge returns True, the pair of objects is going to be merged.
 129     The smaller object (according to cmp) is merged into the larger one.
 130     merge_func performs the actual merge. It is passed the two objects to be
 131     merged (first the larger, then the smaller one).
 132     """
 133
 134     obj_it = iter(obj_it)
 135
 136     try:
 137         prev = obj_it.next()
 138     except StopIteration:
 139         return
 140
 141     for n, p in enumerate(obj_it):
 142         if should_merge(p, prev):
 143             items = sorted([p, prev], cmp=cmp)
 144             logging.info('merging {old} into {new}'.
 145                     format(old=items[1], new=items[0]))
 146
 147             merge_func(*items, dry_run=dry_run)
 148
 149         prev = p
 150         progress_callback(n, total)
 151
 152
 153
 154 class PodcastMerger(object):
 155     """ Merges podcast2 into podcast
 156
 157     Also merges the related podcast states, and re-assignes podcast2's episodes
 158     to podcast, but does neither merge their episodes nor their episode states
 159     """
 160
 161
 162     def __init__(self, podcasts, actions, groups, dry_run=False):
 163
 164         for n, podcast1 in enumerate(podcasts):
 165             for m, podcast2 in enumerate(podcasts):
 166                 if podcast1 == podcast2 and n != m:
 167                     raise IncorrectMergeException("can't merge podcast into itself")
 168
 169         self.podcasts = podcasts
 170         self.actions = actions
 171         self.groups = groups
 172         self.dry_run = dry_run
 173
 174
 175     def merge(self):
 176         podcast1 = self.podcasts.pop(0)
 177
 178         for podcast2 in self.podcasts:
 179             self._merge_objs(podcast1=podcast1, podcast2=podcast2)
 180             self.merge_states(podcast1, podcast2)
 181             self.merge_episodes()
 182             self.reassign_episodes(podcast1, podcast2)
 183             self._delete(podcast2=podcast2)
 184
 185         self.actions['merge-podcast'] += 1
 186
 187
 188     def merge_episodes(self):
 189         for n, episodes in self.groups:
 190
 191             episode = episodes.pop(0)
 192
 193             for ep in episodes:
 194
 195                 em = EpisodeMerger(episode, ep, self.actions)
 196                 em.merge()
 197
 198
 199     @repeat_on_conflict(['podcast1', 'podcast2'])
 200     def _merge_objs(self, podcast1, podcast2):
 201
 202         podcast1.merged_ids = set_filter(podcast1.get_id(),
 203                 podcast1.merged_ids, [podcast2.get_id()], podcast2.merged_ids)
 204
 205         podcast1.merged_slugs = set_filter(podcast1.slug,
 206                 podcast1.merged_slugs, [podcast2.slug], podcast2.merged_slugs)
 207
 208         podcast1.merged_oldids = set_filter(podcast1.oldid,
 209                 podcast1.merged_oldids, [podcast2.oldid],
 210                 podcast2.merged_oldids)
 211
 212         # the first URL in the list represents the podcast main URL
 213         main_url = podcast1.url
 214         podcast1.urls = set_filter(None, podcast1.urls, podcast2.urls)
 215         # so we insert it as the first again
 216         podcast1.urls.remove(main_url)
 217         podcast1.urls.insert(0, main_url)
 218
 219         # we ignore related_podcasts because
 220         # * the elements should be roughly the same
 221         # * element order is important but could not preserved exactly
 222
 223         podcast1.content_types = set_filter(None, podcast1.content_types,
 224                 podcast2.content_types)
 225
 226         key = lambda x: x.timestamp
 227         for a, b in utils.iterate_together(
 228                 [podcast1.subscribers, podcast2.subscribers], key):
 229
 230             if a is None or b is None: continue
 231
 232             # avoid increasing subscriber_count when merging
 233             # duplicate entries of a single podcast
 234             if a.subscriber_count == b.subscriber_count:
 235                 continue
 236
 237             a.subscriber_count += b.subscriber_count
 238
 239         for src, tags in podcast2.tags.items():
 240             podcast1.tags[src] = set_filter(None, podcast1.tags.get(src, []),
 241                     tags)
 242
 243         podcast1.save()
 244
 245
 246     @repeat_on_conflict(['podcast2'])
 247     def _delete(self, podcast2):
 248         podcast2.delete()
 249
 250
 251     @repeat_on_conflict(['s'])
 252     def _save_state(self, s, podcast1):
 253         s.podcast = podcast1.get_id()
 254         s.save()
 255
 256
 257     @repeat_on_conflict(['e'])
 258     def _save_episode(self, e, podcast1):
 259         e.podcast = podcast1.get_id()
 260         e.save()
 261
 262
 263
 264     def reassign_episodes(self, podcast1, podcast2):
 265         # re-assign episodes to new podcast
 266         # if necessary, they will be merged later anyway
 267         for e in episodes_for_podcast(podcast2):
 268             self.actions['reassign-episode'] += 1
 269
 270             for s in all_episode_states(e):
 271                 self.actions['reassign-episode-state'] += 1
 272
 273                 self._save_state(s=s, podcast1=podcast1)
 274
 275             self._save_episode(e=e, podcast1=podcast1)
 276
 277
 278     def merge_states(self, podcast1, podcast2):
 279         """Merges the Podcast states that are associated with the two Podcasts.
 280
 281         This should be done after two podcasts are merged
 282         """
 283
 284         key = lambda x: x.user
 285         states1 = sorted(all_podcast_states(podcast1), key=key)
 286         states2 = sorted(all_podcast_states(podcast2), key=key)
 287
 288         for state, state2 in utils.iterate_together([states1, states2], key):
 289
 290             if state == state2:
 291                 continue
 292
 293             if state == None:
 294                 self.actions['move-podcast-state'] += 1
 295                 self._move_state(state2=state2, new_id=podcast1.get_id(),
 296                         new_url=podcast1.url)
 297
 298             elif state2 == None:
 299                 continue
 300
 301             else:
 302                 psm = PodcastStateMerger(state, state2, self.actions)
 303                 psm.merge()
 304
 305
 306     @repeat_on_conflict(['state2'])
 307     def _move_state(self, state2, new_id, new_url):
 308         state2.ref_url = new_url
 309         state2.podcast = new_id
 310         state2.save()
 311
 312     @repeat_on_conflict(['state2'])
 313     def _delete_state(state2):
 314         state2.delete()
 315
 316
 317
 318
 319 def similar_urls(a, b):
 320     """ Two Podcasts/Episodes are merged, if they have the same URLs"""
 321     return bool(utils.intersect(a.urls, b.urls))
 322
 323
 324
 325
 326
 327
 328 class EpisodeMerger(object):
 329
 330
 331     def __init__(self, episode1, episode2, actions, dry_run=False):
 332         if episode1 == episode2:
 333             raise IncorrectMergeException("can't merge episode into itself")
 334
 335         self.episode1 = episode1
 336         self.episode2 = episode2
 337         self.actions = actions
 338         self.dry_run = dry_run
 339
 340
 341     def merge(self):
 342         self._merge_objs(episode1=self.episode1, episode2=self.episode2)
 343         self.merge_states(self.episode1, self.episode2)
 344         self._delete(e=self.episode2)
 345         self.actions['merge-episode'] += 1
 346
 347
 348     @repeat_on_conflict(['episode1'])
 349     def _merge_objs(self, episode1, episode2):
 350
 351         episode1.urls = set_filter(None, episode1.urls, episode2.urls)
 352
 353         episode1.merged_ids = set_filter(episode1._id, episode1.merged_ids,
 354                 [episode2._id], episode2.merged_ids)
 355
 356         episode1.merged_slugs = set_filter(episode1.slug,
 357                 episode1.merged_slugs, [episode2.slug], episode2.merged_slugs)
 358
 359         episode1.save()
 360
 361
 362     @repeat_on_conflict(['e'])
 363     def _delete(self, e):
 364         e.delete()
 365
 366
 367     def merge_states(self, episode, episode2):
 368
 369         key = lambda x: x.user
 370         states1 = sorted(all_episode_states(self.episode1), key=key)
 371         states2 = sorted(all_episode_states(self.episode2), key=key)
 372
 373         for state, state2 in utils.iterate_together([states1, states2], key):
 374
 375             if state == state2:
 376                 continue
 377
 378             if state == None:
 379                 self.actions['move-episode-state'] += 1
 380                 self._move(state2=state2, podcast_id=self.episode1.podcast,
 381                         episode_id=self.episode1._id)
 382
 383             elif state2 == None:
 384                 continue
 385
 386             else:
 387                 esm = EpisodeStateMerger(state, state2, self.actions)
 388                 esm.merge()
 389
 390
 391     @repeat_on_conflict(['state2'])
 392     def _move(self, state2, podcast_id, episode_id):
 393         state2.podcast = podcast_id
 394         state2.episode = episode_id
 395         state2.save()
 396
 397
 398
 399
 400
 401
 402 class PodcastStateMerger(object):
 403     """Merges the two given podcast states"""
 404
 405     def __init__(self, state, state2, actions, dry_run=False):
 406
 407         if state._id == state2._id:
 408             raise IncorrectMergeException("can't merge podcast state into itself")
 409
 410         if state.user != state2.user:
 411             raise IncorrectMergeException("states don't belong to the same user")
 412
 413         self.state = state
 414         self.state2 = state2
 415         self.actions = actions
 416         self.dry_run = dry_run
 417
 418
 419     def merge(self):
 420         self._do_merge(state=self.state, state2=self.state2)
 421         self._add_actions(state=self.state, actions=self.state2.actions)
 422         self._delete(state2=self.state2)
 423         self.actions['merged-podcast-state'] += 1
 424
 425
 426     @repeat_on_conflict(['state'])
 427     def _do_merge(self, state, state2):
 428
 429         # overwrite settings in state2 with state's settings
 430         settings = state2.settings
 431         settings.update(state.settings)
 432         state.settings = settings
 433
 434         state.disabled_devices = set_filter(None, state.disabled_devices,
 435                 state2.disabled_devices)
 436
 437         state.merged_ids = set_filter(state._id, state.merged_ids,
 438                 [state2._id], state2.merged_ids)
 439
 440         state.tags = set_filter(None, state.tags, state2.tags)
 441
 442         state.save()
 443
 444
 445     @repeat_on_conflict(['state'])
 446     def _add_actions(self, state, actions):
 447         try:
 448             state.add_actions(actions)
 449             state.save()
 450         except restkit.Unauthorized:
 451             # the merge could result in an invalid list of
 452             # subscribe/unsubscribe actions -- we ignore it and
 453             # just use the actions from state
 454             return
 455
 456     @repeat_on_conflict(['state2'])
 457     def _delete(self, state2):
 458         state2.delete()
 459
 460
 461
 462
 463
 464 class EpisodeStateMerger(object):
 465     """ Merges state2 in state """
 466
 467     def __init__(self, state, state2, actions, dry_run=False):
 468
 469         if state._id == state2._id:
 470             raise IncorrectMergeException("can't merge episode state into itself")
 471
 472         if state.user != state2.user:
 473             raise IncorrectMergeException("states don't belong to the same user")
 474
 475         self.state = state
 476         self.state2 = state2
 477         self.actions = actions
 478         self.dry_run = dry_run
 479
 480
 481     def merge(self):
 482         self._merge_obj(state=self.state, state2=self.state2)
 483         self._do_delete(state2=self.state2)
 484         self.actions['merge-episode-state'] += 1
 485
 486
 487     @repeat_on_conflict(['state'])
 488     def _merge_obj(self, state, state2):
 489         state.add_actions(state2.actions)
 490
 491         # overwrite settings in state2 with state's settings
 492         settings = state2.settings
 493         settings.update(state.settings)
 494         state.settings = settings
 495
 496         merged_ids = set(state.merged_ids + [state2._id] + state2.merged_ids)
 497         state.merged_ids = filter(None, merged_ids)
 498
 499         state.chapters = list(set(state.chapters + state2.chapters))
 500
 501         state.save()
 502
 503     @repeat_on_conflict(['state2'])
 504     def _do_delete(self, state2):
 505         state2.delete()
 506
 507
 508 def set_filter(orig, *args):
 509     """ chain args, and remove falsy values and orig """
 510     s = set(chain.from_iterable(args))
 511     s = s - set([orig])
 512     s = filter(None, s)
 513     return s