12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485 |
- import argparse
- import collections
- import re
- import sys
- import typing
- import symuid
- from symuid._uuid import generate_uuid4_bytes
- def _log_path(track_path, msg, stream=None):
- if not stream: # pytest capsys
- stream = sys.stdout
- stream.write("{!r}: {}\n".format(track_path, msg))
- def _log_path_error(track_path, msg):
- _log_path(track_path, msg, stream=sys.stderr)
- class _SyncPosition:
- def __init__(self):
- self._tracks = set() # type: typing.Set[symuid.Track]
- self._play_counts = set() # type: typing.Set[symuid.PlayCount]
- def add_track(self, track: symuid.Track) -> None:
- self._tracks.add(track)
- self._play_counts.update(track.get_play_counts())
- def sync(self, play_count_added_cb=None) -> None:
- for track in self._tracks:
- track_play_counts = set(track.get_play_counts())
- for play_count in self._play_counts:
- if play_count not in track_play_counts:
- track.register_play_count(
- play_count, tag_set_cb=play_count_added_cb)
- def __repr__(self) -> str:
- return repr(vars(self))
- def sync(path, path_ignore_regex, show_ignored=False,
- play_count_added_cb=None):
- sync_positions = collections.defaultdict(_SyncPosition)
- for track in symuid.Track.walk(
- root_path=path,
- path_ignore_regex=path_ignore_regex,
- ignored_cb=lambda p: show_ignored and _log_path(p, 'ignored'),
- unsupported_cb=lambda p, e:
- _log_path_error(p, 'unsupported type, skipped'),
- ):
- if track.get_uuid() is None:
- track.assign_uuid(generate_uuid4_bytes())
- _log_path(track.path, 'assigned uuid {!r}'.format(
- track.get_uuid()))
- sync_positions[track.get_uuid()].add_track(track)
- for sync_position in sync_positions.values():
- sync_position.sync(play_count_added_cb=play_count_added_cb)
- def _init_argparser():
- argparser = argparse.ArgumentParser(description=None)
- argparser.add_argument('path')
- argparser.add_argument(
- '--path-ignore-regex',
- default=symuid.Track.PATH_DEFAULT_IGNORE_REGEX,
- nargs=1,
- metavar='pattern',
- dest='path_ignore_regex',
- type=re.compile,
- help='(default: %(default)s)',
- )
- argparser.add_argument(
- '--show-ignored',
- action='store_true',
- )
- return argparser
- def _main():
- argparser = _init_argparser()
- args = argparser.parse_args()
- sync(**vars(args),
- play_count_added_cb=lambda track, tag:
- _log_path(track.path, 'added play count tag {!r}'.format(tag)))
|