123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384 |
- 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=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())
- @property
- def play_counts(self) -> typing.Set[symuid.PlayCount]:
- return self._play_counts
- @property
- def tracks(self) -> typing.Set[symuid.Track]:
- return self._tracks
- def __repr__(self) -> str:
- return repr(vars(self))
- def sync(path, path_ignore_regex, show_ignored=False):
- 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():
- for track in sync_position.tracks:
- track_play_counts = set(track.get_play_counts())
- for play_count in sync_position.play_counts:
- if play_count not in track_play_counts:
- track.register_play_count(play_count)
- 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))
|