From 0f1bf5f2ea0d368e48531348a2ee89481d5808fb Mon Sep 17 00:00:00 2001 From: Aleteoryx Date: Wed, 1 Oct 2025 18:36:50 -0400 Subject: [PATCH] simple scrobbler --- playback2scrob.py | 233 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 233 insertions(+) create mode 100755 playback2scrob.py diff --git a/playback2scrob.py b/playback2scrob.py new file mode 100755 index 0000000000000000000000000000000000000000..7672047877955fe09be5577a538cff154f23e0d3 --- /dev/null +++ b/playback2scrob.py @@ -0,0 +1,233 @@ +#!/bin/env python + +doc = ''' +scrobble .rockbox/playback.log, handling exclusions and such. +this script assumes filenames of the form '$root//? (Disc <#>)?/<#> - .' + +config is INI-format. it supports the following parameters: + [pb2s] + roots - comma-separated list of valid directories for music + offset - time-zone offset. e.g. -5 for EST + +''' + +API_KEY = '67c56bce3d1ae7d5574b7d51028e8db0' +SECRET = '8c19d08c8ad2a4c7b490dea2b53876a0' # i guess i have to make this public? WTF. + +from configparser import ConfigParser +from sys import stderr, argv, exit +from dataclasses import dataclass +from typing import Optional, Iterator, List, Set +import re +import json +from hashlib import md5 +from urllib.request import urlopen +from urllib.parse import urlencode +from urllib.error import HTTPError +from time import sleep, time + +argv0 = 'playback2scrob' + +def die(why): + print('fatal:', why, file=stderr) + exit(-1) + +def usage(): + global argv0, doc + print(f'usage: {argv0} ') + print() + print(doc) + exit(-1) + + +### CONFIG ### + +@dataclass(init=False) +class Config: + roots: Set[str] + offset: int + + def __init__(self, config: ConfigParser): + try: + self.roots = config.get('pb2s', 'roots', fallback='/') + self.roots = set([x.strip() for x in self.roots.split(',')]) + + self.offset = config.getint('pb2s', 'offset', fallback=0)*3600 + + except Exception as e: + die(f'config invalid: {e}') + + +### PARSING ### + +@dataclass +class RawLogEntry: + when: int + what: str + elapsed: int + duration: int + +@dataclass +class Scrobble: + when: int + track: str + artist: str + album: Optional[str] + +def split_logfile(config: Config, lines: Iterator[str]) -> Iterator[RawLogEntry]: + for line in lines: + line = line.strip() + if line[0] in '#': + continue + + when,elapsed,duration,what = line.split(':') + yield RawLogEntry(int(when)-config.offset, what, int(elapsed), int(duration)) + +def filter_logfile(config: Config, entries: Iterator[RawLogEntry]) -> Iterator[RawLogEntry]: + for entry in entries: + if entry.elapsed < 30_000: + continue + if entry.elapsed < entry.duration // 2: + continue + if (entry.when + (14 * 86_400)) < time(): # 2 weeks + continue + for root in config.roots: + if entry.what.startswith(root): + entry.what = entry.what[len(root):] + if entry.what[0] == '/': + entry.what = entry.what[1:] + break + else: + continue + + yield entry + +def scrobblify(entries: Iterator[RawLogEntry]) -> Iterator[Scrobble]: + for entry in entries: + segs = entry.what.split('/') + if len(segs) < 3: + continue + + artist, album, track = segs[-3:] + + if (m := re.search(' ?\\(Disc \\d+\\)$', album)) is not None: + album = album[:m.start()] + + if (m := re.match('\\d+ - ', track)) is not None: + track = track[m.end():] + track = track.rsplit('.', 1)[0] + + yield Scrobble(entry.when, track, artist, album) + + +### FUCKING LAST.FM ### + +class APIException(Exception): + def __init__(self, method, code, msg): + super().__init__(f'got error {code} in {method}: {msg}') + +def api_call(method: str, *, post=False, **params): + global API_KEY, SECRET + params['api_key'] = API_KEY + params['method'] = method + + keys = sorted(params.keys(), key=lambda k: k.encode('utf-8')) + sig = md5() + for k in keys: + sig.update(k.encode('utf-8')) + sig.update(params[k].encode('utf-8')) + sig.update(SECRET.encode('utf-8')) + + params['api_sig'] = sig.hexdigest() + params['format'] = 'json' + + params = urlencode(params) + + url = 'https://ws.audioscrobbler.com/2.0/' + data = None + if not post: + url += '?'+params + else: + data = params.encode('utf-8') + + try: + req = urlopen(url, data) + except HTTPError as e: + req = e + res = json.loads(req.read()) + + if 'error' in res: + raise APIException(method, res['error'], res['message']) + + print(res) + return res + +def put_scrobbles(sk: str, scrobbles: List[Scrobble]): + params = {'sk':sk} + for i,scrobble in enumerate(scrobbles): + params[f'timestamp[{i}]'] = str(scrobble.when) + params[f'artist[{i}]'] = scrobble.artist + params[f'albumArtist[{i}]'] = scrobble.artist + params[f'albumArtist[{i}]'] = scrobble.artist + params[f'album[{i}]'] = scrobble.album + params[f'track[{i}]'] = scrobble.track + + api_call('track.scrobble', post=True, **params) + + +### BOOT ### + +if len(argv) < 1: + usage() +argv0 = argv[0] +argv = argv[1:] +if len(argv) != 2: + usage() + +config = ConfigParser(strict=False, interpolation=None) +try: + config.read_file(open(argv[0])) +except Exception as e: + die(f"can't read {argv[0]}: {e}") +config = Config(config) + +try: + logfile = open(argv[1]).readlines() +except Exception as e: + die(f"can't read {argv[1]}: {e}") + +print(config) + + +logfile = split_logfile(config, logfile) +logfile = filter_logfile(config, logfile) +scrobbles = scrobblify(logfile) + +token = api_call('auth.getToken')['token'] +while True: + print("open the link and log in, then press enter") + print(f'https://www.last.fm/api/auth/?api_key={API_KEY}&token={token}') + input() + try: + session = api_call('auth.getSession', token=token) + break + except APIException as e: + print('the server sent back an error. try again?\n') + +sk = session['session']['key'] + +acc = [] +for scrobble in scrobbles: + acc.append(scrobble) + if len(acc) < 50: + continue + put_scrobbles(sk, acc) + acc = [] + print("sent 50 scrobbles...") + sleep(5) + +if len(acc) > 0: + put_scrobbles(sk, acc) + print(f"sent {len(acc)} scrobbes...") + +print("done!")