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.merged_ids,
 203                 [podcast2.get_id()], podcast2.merged_ids)
 204
 205         podcast1.merged_slugs = set_filter(podcast1.merged_slugs,
 206                 [podcast2.slug], podcast2.merged_slugs)
 207
 208         podcast1.merged_oldids = set_filter(podcast1.merged_oldids,
 209                 [podcast2.oldid], podcast2.merged_oldids)
 210
 211         # the first URL in the list represents the podcast main URL
 212         main_url = podcast1.url
 213         podcast1.urls = set_filter(podcast1.urls, podcast2.urls)
 214         # so we insert it as the first again
 215         podcast1.urls.remove(main_url)
 216         podcast1.urls.insert(0, main_url)
 217
 218         # we ignore related_podcasts because
 219         # * the elements should be roughly the same
 220         # * element order is important but could not preserved exactly
 221
 222         podcast1.content_types = set_filter(podcast1.content_types,
 223                 podcast2.content_types)
 224
 225         key = lambda x: x.timestamp
 226         for a, b in utils.iterate_together(
 227                 [podcast1.subscribers, podcast2.subscribers], key):
 228
 229             if a is None or b is None: continue
 230
 231             # avoid increasing subscriber_count when merging
 232             # duplicate entries of a single podcast
 233             if a.subscriber_count == b.subscriber_count:
 234                 continue
 235
 236             a.subscriber_count += b.subscriber_count
 237
 238         for src, tags in podcast2.tags.items():
 239             podcast1.tags[src] = set_filter(podcast1.tags.get(src, []), tags)
 240
 241         podcast1.save()
 242
 243
 244     @repeat_on_conflict(['podcast2'])
 245     def _delete(self, podcast2):
 246         podcast2.delete()
 247
 248
 249     @repeat_on_conflict(['s'])
 250     def _save_state(self, s, podcast1):
 251         s.podcast = podcast1.get_id()
 252         s.save()
 253
 254
 255     @repeat_on_conflict(['e'])
 256     def _save_episode(self, e, podcast1):
 257         e.podcast = podcast1.get_id()
 258         e.save()
 259
 260
 261
 262     def reassign_episodes(self, podcast1, podcast2):
 263         # re-assign episodes to new podcast
 264         # if necessary, they will be merged later anyway
 265         for e in episodes_for_podcast(podcast2):
 266             self.actions['reassign-episode'] += 1
 267
 268             for s in all_episode_states(e):
 269                 self.actions['reassign-episode-state'] += 1
 270
 271                 self._save_state(s=s, podcast1=podcast1)
 272
 273             self._save_episode(e=e, podcast1=podcast1)
 274
 275
 276     def merge_states(self, podcast1, podcast2):
 277         """Merges the Podcast states that are associated with the two Podcasts.
 278
 279         This should be done after two podcasts are merged
 280         """
 281
 282         key = lambda x: x.user
 283         states1 = sorted(all_podcast_states(podcast1), key=key)
 284         states2 = sorted(all_podcast_states(podcast2), key=key)
 285
 286         for state, state2 in utils.iterate_together([states1, states2], key):
 287
 288             if state == state2:
 289                 continue
 290
 291             if state == None:
 292                 self.actions['move-podcast-state'] += 1
 293                 self._move_state(state2=state2, new_id=podcast1.get_id(),
 294                         new_url=podcast1.url)
 295
 296             elif state2 == None:
 297                 continue
 298
 299             else:
 300                 psm = PodcastStateMerger(state, state2, self.actions)
 301                 psm.merge()
 302
 303
 304     @repeat_on_conflict(['state2'])
 305     def _move_state(self, state2, new_id, new_url):
 306         state2.ref_url = new_url
 307         state2.podcast = new_id
 308         state2.save()
 309
 310     @repeat_on_conflict(['state2'])
 311     def _delete_state(state2):
 312         state2.delete()
 313
 314
 315
 316
 317 def similar_urls(a, b):
 318     """ Two Podcasts/Episodes are merged, if they have the same URLs"""
 319     return bool(utils.intersect(a.urls, b.urls))
 320
 321
 322
 323
 324
 325
 326 class EpisodeMerger(object):
 327
 328
 329     def __init__(self, episode1, episode2, actions, dry_run=False):
 330         if episode1 == episode2:
 331             raise IncorrectMergeException("can't merge episode into itself")
 332
 333         self.episode1 = episode1
 334         self.episode2 = episode2
 335         self.actions = actions
 336         self.dry_run = dry_run
 337
 338
 339     def merge(self):
 340         self._merge_objs(episode1=self.episode1, episode2=self.episode2)
 341         self.merge_states(self.episode1, self.episode2)
 342         self._delete(e=self.episode2)
 343         self.actions['merge-episode'] += 1
 344
 345
 346     @repeat_on_conflict(['episode1'])
 347     def _merge_objs(self, episode1, episode2):
 348
 349         episode1.urls = set_filter(episode1.urls, episode2.urls)
 350
 351         episode1.merged_ids = set_filter(episode1.merged_ids, [episode2._id],
 352                 episode2.merged_ids)
 353
 354         episode1.merged_slugs = set_filter(episode1.merged_slugs, [episode2.slug],
 355                 episode2.merged_slugs)
 356
 357         episode1.save()
 358
 359
 360     @repeat_on_conflict(['e'])
 361     def _delete(self, e):
 362         e.delete()
 363
 364
 365     def merge_states(self, episode, episode2):
 366
 367         key = lambda x: x.user
 368         states1 = sorted(all_episode_states(self.episode1), key=key)
 369         states2 = sorted(all_episode_states(self.episode2), key=key)
 370
 371         for state, state2 in utils.iterate_together([states1, states2], key):
 372
 373             if state == state2:
 374                 continue
 375
 376             if state == None:
 377                 self.actions['move-episode-state'] += 1
 378                 self._move(state2=state2, podcast_id=self.episode1.podcast,
 379                         episode_id=self.episode1._id)
 380
 381             elif state2 == None:
 382                 continue
 383
 384             else:
 385                 esm = EpisodeStateMerger(state, state2, self.actions)
 386                 esm.merge()
 387
 388
 389     @repeat_on_conflict(['state2'])
 390     def _move(self, state2, podcast_id, episode_id):
 391         state2.podcast = podcast_id
 392         state2.episode = episode_id
 393         state2.save()
 394
 395
 396
 397
 398
 399
 400 class PodcastStateMerger(object):
 401     """Merges the two given podcast states"""
 402
 403     def __init__(self, state, state2, actions, dry_run=False):
 404
 405         if state._id == state2._id:
 406             raise IncorrectMergeException("can't merge podcast state into itself")
 407
 408         if state.user != state2.user:
 409             raise IncorrectMergeException("states don't belong to the same user")
 410
 411         self.state = state
 412         self.state2 = state2
 413         self.actions = actions
 414         self.dry_run = dry_run
 415
 416
 417     def merge(self):
 418         self._do_merge(state=self.state, state2=self.state2)
 419         self._add_actions(state=self.state, actions=self.state2.actions)
 420         self._delete(state2=self.state2)
 421         self.actions['merged-podcast-state'] += 1
 422
 423
 424     @repeat_on_conflict(['state'])
 425     def _do_merge(self, state, state2):
 426
 427         # overwrite settings in state2 with state's settings
 428         settings = state2.settings
 429         settings.update(state.settings)
 430         state.settings = settings
 431
 432         state.disabled_devices = set_filter(state.disabled_devices,
 433                 state2.disabled_devices)
 434
 435         state.merged_ids = set_filter(state.merged_ids, [state2._id],
 436                 state2.merged_ids)
 437
 438         state.tags = set_filter(state.tags, state2.tags)
 439
 440         state.save()
 441
 442
 443     @repeat_on_conflict(['state'])
 444     def _add_actions(self, state, actions):
 445         try:
 446             state.add_actions(actions)
 447             state.save()
 448         except restkit.Unauthorized:
 449             # the merge could result in an invalid list of
 450             # subscribe/unsubscribe actions -- we ignore it and
 451             # just use the actions from state
 452             return
 453
 454     @repeat_on_conflict(['state2'])
 455     def _delete(self, state2):
 456         state2.delete()
 457
 458
 459
 460
 461
 462 class EpisodeStateMerger(object):
 463     """ Merges state2 in state """
 464
 465     def __init__(self, state, state2, actions, dry_run=False):
 466
 467         if state._id == state2._id:
 468             raise IncorrectMergeException("can't merge episode state into itself")
 469
 470         if state.user != state2.user:
 471             raise IncorrectMergeException("states don't belong to the same user")
 472
 473         self.state = state
 474         self.state2 = state2
 475         self.actions = actions
 476         self.dry_run = dry_run
 477
 478
 479     def merge(self):
 480         self._merge_obj(state=self.state, state2=self.state2)
 481         self._do_delete(state2=self.state2)
 482         self.actions['merge-episode-state'] += 1
 483
 484
 485     @repeat_on_conflict(['state'])
 486     def _merge_obj(self, state, state2):
 487         state.add_actions(state2.actions)
 488
 489         # overwrite settings in state2 with state's settings
 490         settings = state2.settings
 491         settings.update(state.settings)
 492         state.settings = settings
 493
 494         merged_ids = set(state.merged_ids + [state2._id] + state2.merged_ids)
 495         state.merged_ids = filter(None, merged_ids)
 496
 497         state.chapters = list(set(state.chapters + state2.chapters))
 498
 499         state.save()
 500
 501     @repeat_on_conflict(['state2'])
 502     def _do_delete(self, state2):
 503         state2.delete()
 504
 505
 506 def set_filter(*args):
 507     return filter(None, set(chain.from_iterable(args)))