#!/usr/bin/env python3 """ Simple ID3 tag manipulation tool with special functions such as - filling in missing ID3v2 tags from ID3v1 and vice-versa (--fill-missing) - interactive synchronization between ID3v1 tags, ID3v2 tags and file name (--audit) - removal of any metadata from embedded artwork images (--clean-images) - reduction of embedded artwork file size by transcoding PNG to JPEG or re-encoding JPEG to maximal 700x700 pixels, 100 kB (--reduce-images) - automatic import of embedded artwork from PNG or JPEG files with appropriate names (e.g. "Folder.jpg"; --auto-image) - semi-automatic interactive BPM analysis by "tapping" with the beat (--tap-bpm; requires FFmpeg's ffplay application to be installed) - special "albumize" mode that fills in track and disc numbers and album artists (--albumize) - automatic purging of uncommon or proprietary tags (--clean-tags) - reordering ID3v2 tags into a canonical order (--sort) Processing a file is done in three phases: 1. Albumization (if --albumize is selected) For each input directory (--albumize only accepts whole directories as input), the following tags will be set in all files if not already set: - track number from filename ("04 title.mp3", "track04.mp3") - total track count from maximum track number in album - disc number from filename ("cd 2 - 04 - title.mp3", "trackb04.mp3") - total disc count from maximum disc number in album - album artist from most frequent artist in album 2. Setting individual tags, i.e. what the --title/--artist/--album/etc., --set, --remove and --image/--add-image options do. 3. Application of the global actions in the order specified on the command line. If no action and no individual tag setting is specified, the default action "--show" (just display the tags) is performed. Limitations: - ID3v2.2 and ID3v2.4 tags will always be converted to ID3v2.3 - doesn't read ID3v2.4 at file end - discards ID3v2.4 recording dates and times (but not the year) - --delete-backups only deletes backups of files that still exist Examples: - remove comment tag and embedded artwork of all files in the current directory: $ kjid3 --comment "" --remove-images *.mp3 - interactively resolve all discrepancies between filenames, ID3v1 and ID3v2 tags, show the results, but keep backups of the original files: $ kjid3 --backup --audit --show my_music/ - full treatment of all albums, including stripping unneccesary tags, importing embedded artwork, and generating BPM values: $ kjid3 --albumize --tap-bpm --auto albums/* """ __version__ = "1.0.2" __author__ = "Martin J. Fiedler " __copyright__ = """ This software is published under the terms of KeyJ's Research License, version 0.2. Usage of this software is subject to the following conditions: 0. There's no warranty whatsoever. The author(s) of this software can not be held liable for any damages that occur when using this software. 1. This software may be used freely for both non-commercial and commercial purposes. 2. This software may be redistributed freely as long as no fees are charged for the distribution and this license information is included. 3. This software may be modified freely except for this license information, which must not be changed in any way. 4. If anything other than configuration, indentation or comments have been altered in the code, the original author(s) must receive a copy of the modified code. """ __history__ = """ 1.0 [2017-08-29] - initial public version 1.0.1 [2019-01-25] - fixed bug that updated files would not be written if only --image was set 1.0.2 [2024-02-??] - Python 3 port """ import sys, os, re, struct, io, zlib, collections, argparse, glob import subprocess, time try: from PIL import Image have_pil = True except ImportError: have_pil = False ID3V1_GENRES = ["Blues","Classic Rock","Country","Dance","Disco","Funk", "Grunge","Hip-Hop","Jazz","Metal","New Age","Oldies","Other","Pop","R&B", "Rap","Reggae","Rock","Techno","Industrial","Alternative","Ska", "Death Metal","Pranks","Soundtrack","Euro-Techno","Ambient","Trip-Hop", "Vocal","Jazz+Funk","Fusion","Trance","Classical","Instrumental","Acid", "House","Game","Sound Clip","Gospel","Noise","AlternRock","Bass","Soul", "Punk","Space","Meditative","Instrumental Pop","Instrumental Rock","Ethnic", "Gothic","Darkwave","Techno-Industrial","Electronic","Pop-Folk","Eurodance", "Dream","Southern Rock","Comedy","Cult","Gangsta Rap","Top 40", "Christian Rap","Pop / Funk","Jungle","Native American","Cabaret","New Wave", "Psychedelic","Rave","Showtunes","Trailer","Lo-Fi","Tribal","Acid Punk", "Acid Jazz","Polka","Retro","Musical","Rock & Roll","Hard Rock","Folk", "Folk-Rock","National Folk","Swing","Fast Fusion","Bebob","Latin","Revival", "Celtic","Bluegrass","Avantgarde","Gothic Rock","Progressive Rock", "Psychedelic Rock","Symphonic Rock","Slow Rock","Big Band","Chorus", "Easy Listening","Acoustic","Humour","Speech","Chanson","Opera", "Chamber Music","Sonata","Symphony","Booty Bass","Primus","Porn Groove", "Satire","Slow Jam","Club","Tango","Samba","Folklore","Ballad", "Power Ballad","Rhythmic Soul","Freestyle","Duet","Punk Rock", "Drum Solo","A Cappella","Euro-House","Dance Hall","Goa","Drum & Bass", "Club-House","Hardcore","Terror","Indie","BritPop","Negerpunk","Polsk Punk", "Beat","Christian Gangsta Rap","Heavy Metal","Black Metal","Crossover", "Contemporary Christian","Christian Rock","Merengue","Salsa","Thrash Metal", "Anime","JPop","Synthpop","Abstract","Art Rock","Baroque","Bhangra", "Big Beat","Breakbeat","Chillout","Downtempo","Dub","EBM","Eclectic", "Electro","Electroclash","Emo","Experimental","Garage","Global","IDM", "Illbient","Industro-Goth","Jam Band","Krautrock","Leftfield","Lounge", "Math Rock","New Romantic","Nu-Breakz","Post-Punk","Post-Rock","Psytrance", "Shoegaze","Space Rock","Trop Rock","World Music","Neoclassical","Audiobook", "Audio Theatre","Neue Deutsche Welle","Podcast","Indie Rock","G-Funk", "Dubstep","Garage Rock","Psybient","Glam Rock","Dream Pop","Merseybeat", "K-Pop","Chiptune","Grime","Grindcore","Indietronic","Indietronica", "Jazz Rock","Jazz Fusion","Post-Punk Revival","Electronica","Psychill", "Ethnotronic","Americana","Ambient Dub","Digital Dub","Chillwave", "Stoner Rock"] ID3V1_GENRES_INV = dict((n.lower().replace(' ', '').replace('-', ''), i) for i,n in enumerate(ID3V1_GENRES)) ID3V2_PICTYPES = ["other","32x32-pixel file icon","other file icon", "front cover","back cover","leaflet page","label","lead artist","artist", "conductor","band/Orchestra","composer","lyricist","recording location", "recording","performance","video screen capture","a bright coloured fish", "illustration","artist logotype","publisher logotype"] ID3V2_PICTYPE_MAP = { "cover":3, "front":3, "back":4, "leaflet":5, "disc":6, "media":6, "label":6, "artist":8, "conductor":9, "band":10, "orchestra":10, "composer":11, "lyricist":12, "studio":13, "recording":14, "performance":15, "gig":15, "video":16, "demo":16, "illustration":18, "logo":19, "bandlogo":19, "labellogo":20, "studiologo":21, "icon":2, "other":0 } ID3V2_PICTYPE_PRIORITY = dict((i,p) for p,i in enumerate([3, 16, 4, 6, 7, 8, 19, 10, 11, 12, 14, 15, 13])) ID3V2_KNOWN_FRAMES = set(['AENC','APIC','ASPI','COMM','COMR','ENCR','EQUA', 'EQU2','ETCO','GEOB','GRID','IPLS','LINK','MCDI','MLLT','OWNE','PRIV','PCNT', 'POPM','POSS','RBUF','RVAD','RVA2','RVRB','SEEK','SIGN','SYLT','SYTC','TALB', 'TBPM','TCOM','TCON','TCOP','TDAT','TDEN','TDLY','TDOR','TDRC','TDRL','TDTG', 'TENC','TEXT','TFLT','TIPL','TIME','TIT1','TIT2','TIT3','TKEY','TLAN','TLEN', 'TMCL','TMED','TMOO','TOAL','TOFN','TOLY','TOPE','TORY','TOWN','TPE1','TPE2', 'TPE3','TPE4','TPOS','TPRO','TPUB','TRCK','TRDA','TRSN','TRSO','TSOA','TSOP', 'TSIZ','TSOT','TSRC','TSSE','TSST','TYER','TXXX','UFID','USER','USLT','WCOM', 'WCOP','WOAF','WOAR','WOAS','WORS','WPAY','WPUB','WXXX']) ID3V2_USUAL_FRAMES = set(['APIC','COMM','SYLT','SYTC','TALB','TBPM','TCOM', 'TCON','TCOP','TDOR','TDRC','TDRL','TEXT','TIPL','TIT1','TIT2','TIT3','TKEY', 'TLAN','TLEN','TMCL','TMOO','TOAL','TOLY','TOPE','TORY','TPE1','TPE2','TPE3', 'TPE4','TPOS','TPUB','TRCK','TRSN','TRSO','TSOA','TSOP','TSOT','TSRC','TYER', 'UFID','USLT','WCOM','WCOP','WOAF','WOAR','WOAS','WORS','WPUB']) ID3V2_MULTI_FRAMES = set(['APIC','COMM', 'PRIV', 'TXXX', 'WXXX']) ID3V22_FRAME_MAP = { 'BUF':'RBUF', 'CNT':'PCNT', 'COM':'COMM', 'CRA':'AENC', 'ETC':'ETCO', 'EQU':'EQUA', 'GEO':'GEOB', 'IPL':'IPLS', 'LNK':'LINK', 'MCI':'MCDI', 'MLL':'MLLT', 'PIC':'APIC', 'POP':'POPM', 'REV':'RVRB', 'RVA':'RVAD', 'SLT':'SYLT', 'STC':'SYTC', 'TAL':'TALB', 'TBP':'TBPM', 'TCM':'TCOM', 'TCO':'TCON', 'TCR':'TCOP', 'TDA':'TDAT', 'TDY':'TDLY', 'TEN':'TENC', 'TFT':'TFLT', 'TIM':'TIME', 'TKE':'TKEY', 'TLA':'TLAN', 'TLE':'TLEN', 'TMT':'TMED', 'TOA':'TOPE', 'TOF':'TOFN', 'TOL':'TOLY', 'TOR':'TORY', 'TOT':'TOAL', 'TP1':'TPE1', 'TP2':'TPE2', 'TP3':'TPE3', 'TP4':'TPE4', 'TPA':'TPOS', 'TPB':'TPUB', 'TRC':'TSRC', 'TRD':'TRDA', 'TRK':'TRCK', 'TSI':'TSIZ', 'TSS':'TSSE', 'TT1':'TIT1', 'TT2':'TIT2', 'TT3':'TIT3', 'TXT':'TEXT', 'TXX':'TXXX', 'TYE':'TYER', 'UFI':'UFID', 'ULT':'USLT', 'WAF':'WOAF', 'WAR':'WOAR', 'WAS':'WOAS', 'WCM':'WCOM', 'WCP':'WCOP', 'WPB':'WPUB', 'WXX':'WXXX' } def IsJPEG(img): return img[:2] == b'\xff\xd8' def IsPNG(img): return img[:8] == b'\x89PNG\r\n\x1a\n' def _encode_ss32(size): assert size < (1 << 28) return ((size << 3) & 0x7F000000) \ | ((size << 2) & 0x7F0000) \ | ((size << 1) & 0x7F00) \ | ( size & 0x7F) def atoi(x, default=0): try: return int(x) except ValueError: return default def is_only_char(s, c): for sc in s: if sc != c: return False return True class ID3FrameBase(object): def __init__(self, frame, major=0, group=-1, discardable=False, specific=False, read_only=False): self.frame = frame self.group = group self.flags = 0 if discardable: self.flags |= 0x8000 if specific: self.flags |= 0x4000 if read_only: self.flags |= 0x2000 if group >= 0: self.flags |= 0x0020 @property def discardable(self): return bool(self.flags & 0x4000) @property def read_only(self): return bool(self.flags & 0x1000) @staticmethod def _decstring(enc, data, offset=0, need_terminator=True, use_terminator=True): if isinstance(enc, bytes): enc = enc[0] try: align, enc = { 0x00: (1, 'windows-1252'), 0x01: (2, 'utf-16'), 0x02: (2, 'utf-16-be'), 0x03: (1, 'utf-8-sig'), }[enc] except KeyError: raise IOError(f"uses unknown text encoding 0x{enc:02X}") if use_terminator: end = offset while True: end = data.find(b'\0' * align, end) if end < 0: if need_terminator: raise IOError("contains an unterminated text field") end = len(data) break if (end - offset) % align: end += 1 else: end += align break else: end = len(data) try: return data[offset:end].decode(enc).rstrip('\0'), end except UnicodeDecodeError as e: raise IOError(f"contains encoding error in text - {e}") @staticmethod def _getenc(text): try: text.encode('ascii') return b'\x00', 'iso8859-1', b'\x00' except UnicodeEncodeError: return b'\x01', 'utf-16', b'\x00\x00' @staticmethod def raw_drepr(data): hexdump = ' '.join(f"{c:02X}" for c in data[:4]) ellipsis = " ..." if (len(data) > 4) else "" return f"<{len(data)} raw bytes: {hexdump}{ellipsis}>" def dump(self, verbosity): if verbosity >= 3: print(f"ID3v2 {self.frame!r} frame: {self.drepr()} (flags: {bin(self.flags)}, group: {self.group})") elif verbosity >= 2: print(f"ID3v2 {self.frame!r} frame: {self.drepr()}") def genhdr(self, payload): if self.group >= 0: payload = bytes([self.group]) + payload return struct.pack('>4sIH', self.frame.encode('ascii', 'replace'), len(payload), self.flags) + payload def generate(self): return b'' class ID3RawFrame(ID3FrameBase): def __init__(self, frame, data, major, group, discardable, specific, read_only, compressed, encrypted): ID3FrameBase.__init__(self, frame, major, group, discardable, specific, read_only) if compressed: self.flags |= 0x0080 if encrypted: self.flags |= 0x0040 self.data = data def drepr(self): return self.raw_drepr(self.data) def generate(self): return self.genhdr(self.data) class ID3TextFrame(ID3FrameBase): def __init__(self, frame, data, major, group, discardable, specific, read_only): ID3FrameBase.__init__(self, frame, major, group, discardable, specific, read_only) self.text = self._decstring(data, data, 1, need_terminator=False, use_terminator=(frame in ID3V2_USUAL_FRAMES))[0] def drepr(self): return repr(self.text) def generate(self): if not self.text: return b'' prefix, enc, sep = self._getenc(self.text) return self.genhdr(prefix + self.text.encode(enc, 'ignore')) class ID3GenreFrame(ID3TextFrame): def __init__(self, frame, data, major, group, discardable, specific, read_only): ID3FrameBase.__init__(self, frame, major, group, discardable, specific, read_only) # genre is kept in ID3v2.3 format internally -> convert null-separated ID3v2.4 # to parenthesis-separated ID3v2.3 here self.text = "" for part in self._decstring(data, data, 1, False, False)[0].split('\0'): if part.startswith('('): self.text += part break if re.match(r'\d+|RX|CR', part): self.text += f"({part})" else: self.text += part break class ID3NewTextFrame(ID3TextFrame): def __init__(self, frame, text, discardable=False, specific=False, read_only=False): ID3FrameBase.__init__(self, frame, discardable=discardable, specific=specific, read_only=read_only) self.text = text def generate(self): if self.frame.startswith('1') or not(self.text): return b'' else: return ID3TextFrame.generate(self) class ID3CommentFrame(ID3TextFrame): def __init__(self, frame, data, major, group, discardable, specific, read_only): ID3FrameBase.__init__(self, frame, major, group, discardable, specific, read_only) self.lang = data[1:4].decode('ascii', 'replace') self.desc, pos = self._decstring(data, data, 4) self.text = self._decstring(data, data, pos, False)[0] def drepr(self): return f"Comment(lang={self.lang!r}, desc={self.desc!r}, text={self.text!r})" def generate(self): prefix, enc, sep = self._getenc(self.desc + self.text) return self.genhdr(prefix + self.lang[:3].encode('ascii', 'replace').ljust(3, b"\x00") + self.desc.encode(enc, 'ignore') + sep + self.text.encode(enc, 'ignore')) class ID3NewCommentFrame(ID3CommentFrame): def __init__(self, text): ID3FrameBase.__init__(self, 'COMM') self.lang = 'und' self.desc = "" self.text = text class ID3ImageFrame(ID3FrameBase): def __init__(self, frame, data, major, group, discardable, specific, read_only): ID3FrameBase.__init__(self, frame, major, group, discardable, specific, read_only) if major < 3: mtype = data[1:4].lower().strip() self.mimetype = "image/" + ("jpeg" if (mtype == "jpg") else mtype) pos = 4 else: self.mimetype, pos = self._decstring(0, data, 1) if not('/' in self.mimetype): self.mimetype = "image/" + self.mimetype self.pictype = data[pos] pos += 1 self.desc, pos = self._decstring(data, data, pos) self.data = data[pos:] def drepr(self): pictype_str = "unknown" if (self.pictype >= len(ID3V2_PICTYPES)) else ID3V2_PICTYPES[self.pictype] return f"Image({self.mimetype}, type {self.pictype} ({pictype_str}), {self.desc!r}, {self.raw_drepr(self.data)})" def generate(self): prefix, enc, sep = self._getenc(self.desc) return self.genhdr(prefix + self.mimetype.encode('iso8859-1', 'ignore') + bytes([0, self.pictype]) + self.desc.encode(enc, errors='replace') + sep + self.data) class ID3NewImageFrame(ID3ImageFrame): def __init__(self, filename, pictype=None, desc=None): ID3FrameBase.__init__(self, 'APIC') if pictype in (None, ''): self.pictype = 3 else: if isinstance(pictype, str): pictype = pictype.strip().lower() try: self.pictype = int(ID3V2_PICTYPE_MAP.get(pictype, pictype)) except ValueError: self.pictype = -1 if (self.pictype < 0) or (self.pictype > 255): raise ValueError(f"invalid picture type {pictype!r}") self.desc = (desc or '').strip() with open(filename.strip(), 'rb') as f: self.data = f.read() if IsJPEG(self.data): self.mimetype = "image/jpeg" elif IsPNG(self.data): self.mimetype = "image/png" else: raise ValueError("unsupported image format (neither JPEG nor PNG)") class ID3Info(object): def __init__(self, filename, verbosity=0): self.verbosity = verbosity self.filename = os.path.normpath(os.path.abspath(filename)) self.has_versions = set() self.ignore_read_only = False self.payload_start = 0 self.payload_end = 0 self.modified = False self.good = False self.data = [] self.index = {} if self.verbosity >= 1: print if self.verbosity >= 2: print("File: ", self.filename) try: f = open(self.filename, "rb") except EnvironmentError as e: return self.fail(f"could not open the input file - {e}") self.good = True if f.read(3) == b'ID3': version = list(f.read(1) or b'\0')[0] try: if version in (2, 3, 4): self._read_id3v2(f, version) else: self.fail(f"unrecognized version ID3v2.{version}") except EnvironmentError as e: self.fail(e) try: f.seek(-128, 2) except EnvironmentError: pass if f.read(3) == b'TAG': try: self._read_id3v1(f) except EnvironmentError as e: self.fail(e) f.close() # convert ID3v2.4 into ID3v2.3 for old, new in (("TDRC", "TYER"), ("TDOR", "TORY")): if not(old in self.index): continue if (new in self.index): if self.gettext(old) == self.gettext(new): self.log(3, f"ID3v2.4 {old!r} and ID3v2.3 {new!r} have same contents, removing {old!r}") self.remove(old) continue frame = self.index[old] m = re.match(r'(\d+)\s*', frame.text) if not m: pass # not a valid date, can't convert elif m.end(0) == len(frame.text): self.log(3, f"converting ID3v2.4 {old!r} into ID3v2.3 {new!r}") frame.frame = new del self.index[old] self.index[new] = frame else: self.log(3, f"creating ID3v2.4 {old!r} from ID3v2.3 {new!r}") self.settext(new, m.group(1)) self.modified = False # we didn't change anything deliberately yet for item in self.data: item.dump(self.verbosity) if self.verbosity >= 1: if self.has_versions: print(f"{self.filename}:", ', '.join(f"ID3v{ver}" for ver in sorted(self.has_versions))) else: print(f"{self.filename}: no ID3 tags found") def copyfrom(self, id3): if not(id3) or not(id3.good): return if self.data or id3.data: self.data = id3.data[:] self.index = dict(id3.index.items()) self.modified = True def add(self, frame): self.data.append(frame) multi = (frame.frame in ID3V2_MULTI_FRAMES) if frame.frame in self.index: existing = self.index[frame.frame] if multi: self.index[frame.frame].append(frame) elif existing.discardable and not(frame.discardable): self.index[frame.frame] = frame elif not(existing.discardable) and frame.discardable: pass else: self.fail(f"multiple ID3v2 {frame.frame} frames found") self.index[frame.frame] = frame elif multi: self.index[frame.frame] = [frame] else: self.index[frame.frame] = frame self.modified = True def log(self, v, msg): if self.verbosity >= v: print(msg) def fail(self, msg): print(f"{self.filename}:", msg, file=sys.stderr) def _readstruct(self, f, code, bytes, what): data = f.read(bytes) if len(data) != bytes: raise IOError("unexpected EOF while reading " + what) return struct.unpack(code, data) def _read_id3v1(self, f): title, artist, album, year, comment, genre = self._readstruct(f, '30s30s30s4s30sB', 125, "ID3v1 header") self.payload_end = f.tell() - 128 if (comment[-2] == 0) and (comment[-1] != 0): track = comment[-1] comment = comment[:-2] self.log(3, f"ID3v1.1 track number: {track!r}") self.settext('1track', str(track)) self.has_versions.add("1.1") else: self.has_versions.add("1.0") for name, value in (("title", title), ("artist", artist), ("album", album), ("year", year), ("comment", comment)): value = value.rstrip(b' \0').decode('windows-1252', errors='replace') if not value: continue self.log(3, f"ID3v1 {name}: {value!r}") self.settext('1' + name, value) if genre < len(ID3V1_GENRES): genre = f"({genre}){ID3V1_GENRES[genre]}" elif genre < 255: genre = f"({genre})" else: genre = None self.log(3, f"ID3v1 genre: {genre!r}") if genre: self.settext('1genre', genre) return True def _decode_ss32(self, size, where): if not(size) or (size & 0x80808080): raise IOError(f"invalid syncsafe size {size:08X} in {where}") return ((size & 0x7F000000) >> 3) \ | ((size & 0x7F0000) >> 2) \ | ((size & 0x7F00) >> 1) \ | (size & 0x7F) def _read_id3v2(self, f, major=4): minor, flags, size = self._readstruct(f, '>BBI', 6, "ID3v2 header") self.has_versions.add(f"2.{major}.{minor}") self.log(2, f"parsing ID3v2.{major}.{minor}") size = self._decode_ss32(size, "ID3v2 header") self.log(3, f"ID3v2 size: {size} bytes, flags: {bin(flags)}") if flags & 0x10: size += 10 # read 10 additional bytes if footer is present unsync_all = bool(flags & 0x80) ignore_tag = (major == 2) and (flags & 0x40) if flags & 0x40: extsize = self._readstruct(f, '>I', 4, "ID3v2 extended header")[0] size -= 4 # we already read 4 bytes into the tag by now extsize = self._decode_ss32(extsize, "ID3v2 extended header") self.log(3, f"ID3v2 extended header size: {extsize}") if extsize > size: raise IOError("ID3v2 extended header size is greater than total ID3v2 tag size") else: extsize = 0 tag = f.read(size) if len(tag) != size: raise IOError("unexpected EOF while reading ID3v2 tag") self.payload_start = f.tell() if self.verbosity >= 3: print("first few audio data bytes:", ' '.join(f"{c:02X}" for c in f.read(4))) if ignore_tag: return True if (extsize > 4) and (tag[extsize-4:extsize] in ID3V2_KNOWN_FRAMES): self.fail("ID3v2 extended header size is invalid") extsize -= 4 tag = tag[extsize:] if (major < 4) and unsync_all: tag = tag.replace(b'\xff\x00', b'\xff') unsync_all = False tag = io.BytesIO(tag) while True: hsize = (10 if (major > 2) else 6) header = tag.read(hsize) if len(header) < hsize: break if major > 2: frame, size, flags = struct.unpack('>4sIH', header) else: frame, sizeh, sizel = struct.unpack('>3sHB', header) size = (sizeh << 8) | sizel flags = 0 frame = frame.decode('ascii', 'replace') if not frame.strip('\0'): break if major == 4: size = self._decode_ss32(size, f"ID3v2 {frame!r} frame") self.log(3, f"ID3v2 frame: {frame!r}, {size} bytes, flags: {bin(flags)}") data = tag.read(size) if len(data) != size: self.fail(f"ID3v2 {frame!r} frame extends past end of tag") continue if major == 3: discardable = bool(flags & 0x8000) specific = bool(flags & 0x4000) read_only = bool(flags & 0x2000) compressed = bool(flags & 0x0080) encrypted = bool(flags & 0x0040) grouped = bool(flags & 0x0020) unsync = False dli = False else: discardable = bool(flags & 0x4000) specific = bool(flags & 0x2000) read_only = bool(flags & 0x1000) compressed = bool(flags & 0x0008) encrypted = bool(flags & 0x0004) grouped = bool(flags & 0x0040) unsync = bool(flags & 0x0002) dli = bool(flags & 0x0001) # the order of unsynchronization, compression, group byte # and DLI is unspecified ... I'm guessing here :( if dli: data = data[4:] # not checking DLI, just ignoring it if unsync: data = data.replace(b'\xff\x00', b'\xff') if grouped: group = data[0] data = data[1:] else: group = -1 if compressed and not(encrypted): try: data = zlib.decompress(data) except: self.fail(f"ID3v2 {frame!r} frame has errors in compressed data") continue if major == 2: frame = ID3V22_FRAME_MAP.get(frame, frame.ljust(4)) try: if encrypted: frame_class = None elif frame.startswith('T'): frame_class = ID3TextFrame else: frame_class = { 'TCON': ID3GenreFrame, 'COMM': ID3CommentFrame, 'APIC': ID3ImageFrame, }.get(frame) if frame_class: frame = frame_class(frame, data, major, group, discardable, specific, read_only) else: frame = ID3RawFrame(frame, data, major, group, discardable, specific, read_only, compressed, encrypted) self.add(frame) except EnvironmentError as e: self.fail(f"ID3v2 {frame!r} frame {e}") del tag return True def get(self, frame, default=None): return self.index.get(frame, default) def gettext(self, frame): frame = self.index.get(frame) if isinstance(frame, ID3TextFrame): return frame.text elif isinstance(frame, list) and frame and isinstance(frame[0], ID3CommentFrame): return frame[0].text return "" def settext(self, frame_id, text): frame = self.index.get(frame_id) if isinstance(frame, ID3TextFrame): if frame.read_only and not(self.ignore_read_only): raise ValueError(f"ID3v2 {frame_id!r} frame is read-only") if text != frame.text: self.modified = True frame.text = text elif isinstance(frame, list) and frame and isinstance(frame[0], ID3CommentFrame): eligible = [sf for sf in frame if not sf.read_only] if not eligible: if self.ignore_read_only: eligible = frame else: raise ValueError(f"all ID3v2 {frame_id!r} frames are read-only") frame = eligible[0] if text != frame.text: self.modified = True frame.text = text elif frame: raise TypeError("invalid frame type for settext()") else: if frame_id == 'COMM': frame = ID3NewCommentFrame(text) else: frame = ID3NewTextFrame(frame_id, text) # try to add new text frames before album artwork if frame_id.startswith('1'): self.data.append(frame) else: self.data.insert(min([len(self.data)] + [i for i,f in enumerate(self.data) if f.frame=='APIC']), frame) self.index[frame_id] = frame if text: self.modified = True def remove(self, *frame_ids): for f in frame_ids: try: del self.index[f] self.modified = True except KeyError: pass frame_ids = set(frame_ids) self.data = [f for f in self.data if not(f.frame in frame_ids)] def write(self, dest=None, backup=False): root, base = os.path.split(self.filename) base, ext = os.path.splitext(base) outfile = dest or self.filename self.log(1, f"writing modified file '{outfile}'") if dest: infile = self.filename remove = False else: n = 0 while True: infile = os.path.join(root, f".kjid3-backup.{base}.{n}{ext}") if not os.path.exists(infile): break n += 1 remove = not(backup) try: os.rename(self.filename, infile) except EnvironmentError as e: raise IOError(f"cannot rename - {e}") fin = None fout = None try: id3v2 = b''.join(f.generate() for f in self.data) if id3v2: id3v2 = b'ID3\x03\x00\x00' + struct.pack('>I', _encode_ss32(len(id3v2))) + id3v2 id3v1 = self.shorten(self.gettext('1title'), 30).encode('windows-1252', 'ignore')[:30].ljust(30, b'\0') \ + self.shorten(self.gettext('1artist'), 30).encode('windows-1252', 'ignore')[:30].ljust(30, b'\0') \ + self.shorten(self.gettext('1album'), 30).encode('windows-1252', 'ignore')[:30].ljust(30, b'\0') \ + self.shorten(self.gettext('1year'), 30).encode('windows-1252', 'ignore').split(b'T', 1)[0].split(b'/', 1)[0].split(b'-', 1)[0][:4].ljust(4, b'\0') \ + self.gettext('1comment' ).encode('windows-1252', 'ignore')[:30].ljust(30, b'\0') try: id3v1 = id3v1[:-2] + b'\0' + bytes([int(self.gettext('1track').split('/', 1)[0])]) except ValueError: pass genre_str = self.gettext('1genre').strip() genre = 255 m = re.match(r'\((\d+)\)', genre_str) if m: i = int(m.group(1)) if (i >= 0) and (i < 255): genre = i if genre == 255: genre = ID3V1_GENRES_INV.get(genre_str.rsplit(')', 1)[-1].lower().replace(' ', '').replace('-', ''), 255) if id3v1.strip(b'\0') or (genre != 255): id3v1 = b'TAG' + id3v1 + bytes([genre]) assert len(id3v1) == 128 else: id3v1 = b'' with open(outfile, 'wb') as fout: if id3v2: fout.write(id3v2) with open(infile, 'rb') as fin: if not self.payload_end: fin.seek(0, 2) self.payload_end = fin.tell() fin.seek(self.payload_start) nbytes = self.payload_end - self.payload_start while nbytes > 0: block = fin.read(min(nbytes, 512 * 1024)) if not block: break fout.write(block) nbytes -= len(block) new_pl_end = fout.tell() if id3v1: fout.write(id3v1) except Exception as e: raise print(f"{self.filename}: error while writing tag -", e, file=sys.stderr) if infile != self.filename: try: os.unlink(outfile) except EnvironmentError: pass try: os.rename(infile, self.filename) except EnvironmentError: pass return if dest == self.filename: self.payload_start = len(id3v2) self.payload_end = new_pl_end if remove: try: os.unlink(infile) except EnvironmentError as e: print(f"{infile}: can not delete temporary file -", e, file=sys.stderr) def shorten(self, text, length=30): text = text.lstrip().split('\n', 1)[0].rstrip() orig = text if len(text) <= length: return text while len(text) > length: # try to remove parenthesized parts at the end if text.endswith(')') and ('(' in text): text = text.rsplit('(', 1)[0].strip() elif text.endswith(']') and ('[' in text): text = text.rsplit(']', 1)[0].strip() elif text.endswith('>') and ('<' in text): text = text.rsplit('>', 1)[0].strip() elif text.endswith('}') and ('{' in text): text = text.rsplit('}', 1)[0].strip() else: sep = None for s in ("--", " -", " feat.", " feat ", " ft.", " ft ", " vs.", " vs ", " pres.", " pres ", " & "): if s in text: sep = s break if sep: text = text.rsplit(sep, 1)[0].strip() else: break # nothing found to remove :( if len(text) > length: # still too long -> remove leading articles in a few common languages ltext = text.lower() for prefix in ( "the ", "a ", "an ", "le ", "la ", "l'", "el ", "un ", "une ", "una ", "il ", "der ", "das ", "ein ", "eine ", # not "die" -- may be a different english word ): if ltext.startswith(prefix): text = text[len(prefix):].strip() break if len(text) > length: # still too long -> generate ellipsis last_space = text.rfind(' ', 0, length-3) if (last_space >= 0) and ((length - last_space) < 8): text = text[:last_space].strip() + " ..." else: text = text[:length-3] + "..." self.log(1, f"text {orig!r} shortened to {text!r} for ID3v1 tag") return text def settext_ex(self, frame_id, text): if frame_id in ('TRCK', '1track'): self.settext(frame_id, self.modify_track_no(text, 'TRCK', '1track', True)) elif frame_id == 'TPOS': self.settext(frame_id, self.modify_track_no(text, 'TPOS')) else: self.settext(frame_id, text) def modify_track_no(self, pattern, tag, alt_tag=None, use_filename=False): if not pattern: return pattern # split pattern into current (n) and total (t) try: c, t = pattern.split('/', 1) except ValueError: c = pattern t = None if c and(not(atoi(c))) and use_filename: # if current number is not a valid number, get it from the filename m = re.search(r'\d+', os.path.basename(self.filename)) c = m.group(0).lstrip('0') if m else '' if not c: # if no current number specified at all, use the existing value c = (self.gettext(tag) or self.gettext(alt_tag)).split('/', 1)[0] # assemble the final value if t: return c + '/' + t else: return c ################################################################################ def CleanJPEG(img): if not IsJPEG(img): return img offset = 2 while True: if (offset > (len(img) - 4)) or (img[offset] != b'\xff'): break # not a JPEG marker at all, or too close to EOF marker = img[offset+1] size = (img[offset+2] << 8) + img[offset+3] + 2 if (offset + size) >= len(img): break # invalid size if ((marker & 0xF0) == 0xE0) or (marker == 0xFE): img = img[:offset] + img[offset+size:] # APPn, COM: remove elif ((marker & 0xF0) == 0xC0) or (marker in (0xDB, 0xDC, 0xDD, 0xDE, 0xDF)): offset += size # SOSn, DHT, DAC, DQT, DNL, DRI, DHP, EXP: ignore else: break # all others (in particular SOS): stop processing eoi_idx = img.find(b'\xff\xd9') if eoi_idx > 40: img = img[:eoi_idx+2] return img def CleanPNG(img): if not IsPNG(img): return img offset = 8 while True: if (offset > (len(img) - 12)): break # EOF size, chunk = struct.unpack('>I4s', img[offset:offset+8]) size += 12 if chunk[0] != chunk[0].upper(): img = img[:offset] + img[offset+size:] # remove non-essential chunk else: offset += size if (chunk == b'IEND') and (offset < len(img)): img = img[:offset] return img def ReduceImage(data, mimetype, max_pixels=0, max_bytes=0, default_quality=90, min_quality=75, oversize_tolerance=20, logger=None): in_max_pix = ((max_pixels * (100 + oversize_tolerance) + 50) // 100) if max_pixels else 32768 try: if isinstance(data, bytes): img = Image.open(io.BytesIO(data)) in_format = img.format else: in_format = None if (in_format == 'JPEG') \ and (len(data) < max_bytes) \ and (img.size[0] <= in_max_pix) \ and (img.size[1] <= in_max_pix): return data, mimetype # image is OK as it is if in_format: img.load() except EnvironmentError as e: if logger: logger.fail(f"failed to load and re-encode image - {e}") return data, mimetype oldfmt = f"{in_format if in_format else 'new image'}, {img.size[0]}x{img.size[1]}, {len(data)*0.001:.1f}kB" img = img.convert('L' if (img.mode in 'L1') else 'RGB') for k in list(img.info): del img.info[k] if (img.size[0] > in_max_pix) or (img.size[1] > in_max_pix): img = img.resize(min(img.size, (max_pixels, (max_pixels * img.size[1] + img.size[0] // 2) // img.size[0]), ((max_pixels * img.size[0] + img.size[1] // 2) // img.size[1], max_pixels) ), Image.BICUBIC) q = default_quality while True: f = io.BytesIO() img.save(f, 'JPEG', quality=q, optimize=True) data = f.getvalue() logger.log(2, f"re-encoding with quality {q} -> {len(data)*0.001:.1f}kB") if (len(data) < max_bytes) or (q <= min_quality): break q = max(min_quality, q * 98 // 100) if logger: logger.log(1, f"re-encoding image from {oldfmt} to JPEG, {img.size[0]}x{img.size[1]}, {len(data)*0.001:.1f}kB") return data, "image/jpeg" ################################################################################ AvailableActions = {} def action(param): def decorator(func): AvailableActions[param] = func return func return decorator @action("--remove-v1") def RemoveID3v1(id3): """ remove all ID3v1 tags """ for f in [f.frame for f in id3.data]: if f.startswith('1'): id3.remove(f) @action("--remove-v2") def RemoveID3v2(id3): """ remove all ID3v2 tags """ for f in [f.frame for f in id3.data]: if not f.startswith('1'): id3.remove(f) #@action("--remove-all") # implicitly done by --new def RemoveAll(id3): """ remove all ID3 tags """ if id3.data: id3.modified = True id3.data = [] id3.index = {} @action("--clean-tags") def RemoveExotic(id3): """ remove exotic and non-standard ID3v2 tags """ frames = set(f.frame for f in id3.data if not f.frame.startswith('1')) - ID3V2_USUAL_FRAMES if frames: fcounts = [(fid, sum(1 for f in id3.data if f.frame == fid)) for fid in frames] id3.log(1, "removing exotic ID3v2 frames: " + ", ".join((repr(fid) if (cnt < 2) else f"{fid!r}({cnt}x)") for fid, cnt in fcounts)) id3.remove(*frames) @action("--fill-missing") def FillMissingTags(id3): """ re-create missing ID3v1 tag(s) from ID3v2 and vice-versa """ from1to2 = [] from2to1 = [] for tag1, tag2 in ( ('1title', 'TIT2'), ('1artist', 'TPE1'), ('1album', 'TALB'), ('1genre', 'TCON'), ('1track', 'TRCK'), ('1year', 'TYER'), ): value1 = id3.gettext(tag1) value2 = id3.gettext(tag2) if (value1 and value2) or (not(value1) and not(value2)): continue # both already present, or none -> ignore try: if value1: # create ID3v2 from ID3v1 from1to2.append(tag1[1:]) id3.settext(tag2, value1) elif value2: # create ID3v1 from ID3v2 from2to1.append(tag1[1:]) id3.settext(tag1, id3.shorten(value2, 30)) except Exception as e: print(f"{id3.filename}: failed to synchronize {tag1!r}/{tag2!r} tags - [{e.__class__.__name__}]", e, file=sys.stderr) raise value1 = id3.gettext('1comment') value2 = id3.gettext('COMM') if value1 and not(value2): from1to2.append("comment") id3.settext('COMM', value1) elif value2 and not(value1): from2to1.append("comment") id3.settext('1comment', id3.shorten(value2, 28 if id3.gettext('1track') else 30)) if from1to2: id3.log(1, f"copying ID3v1 tag(s) to ID3v2: " + ", ".join(from1to2)) if from2to1: id3.log(1, f"copying ID3v2 tag(s) to ID3v1: " + ", ".join(from2to1)) @action("--sort") def ReorderTags(id3): """ bring the ID3v2 frames into a standard order; move album art to the end """ def category(frame): return min((-len(prefix), i) for i, prefix in enumerate( ["TIT", "TPE1", "TALB", "TPE", "T", "COMM", "USLT", "SYLT", "", "APIC", "1"] ) if frame.frame.startswith(prefix))[1] old_order = list(map(id, id3.data)) id3.data.sort(key=category) if list(map(id, id3.data)) != old_order: id3.modified = True @action("--clean-images") def CleanImages(id3): """ remove all metadata from embedded artwork images """ for f in id3.get('APIC', []): old_data = f.data f.data = CleanPNG(CleanJPEG(old_data)) if f.data != old_data: diff = len(old_data) - len(f.data) if diff < 10000: id3.log(1, f"removing metadata from artwork, saving {diff} bytes") else: id3.log(1, f"removing metadata from artwork, saving {diff*0.001:.1f}kB") id3.modified = True @action("--reduce-images") def ReduceImages(id3, max_pixels=700, max_kb=100, default_quality=90, min_quality=75, oversize_tolerance=20): """ re-encode PNG images or large JPEG images to smaller JPEG images (max. 700x700 pixels in 100 kB) """ if not have_pil: print("Error: can not reduce image sizes: PIL/Pillow is not installed", file=sys.stderr) return for f in id3.get('APIC', []): old_data = f.data f.data, f.mimetype = ReduceImage(old_data, f.mimetype, logger=id3, max_pixels=max_pixels, max_bytes=max_kb*1000, default_quality=default_quality, min_quality=min_quality, oversize_tolerance=oversize_tolerance) if f.data != old_data: id3.modified = True @action("--single-image") def RemoveAdditionalImages(id3, keep=1): """ discard all embedded artwork except for one image """ imgs = id3.get('APIC', []) if len(imgs) <= keep: return # sort by priority, keep order inside category imgs.sort(key = lambda img: (-1) if isinstance(img, ID3NewImageFrame) else ID3V2_PICTYPE_PRIORITY.get(img.pictype, 99)) # increase "keep" to ensure we don't delete newly added images while (keep < len(imgs)) and isinstance(imgs[keep], ID3NewImageFrame): keep += 1 if len(imgs) <= keep: return id3.log(1, f"removing {len(imgs) - keep} additional artwork image(s)") for img in imgs[keep:]: id3.data.remove(img) id3.modified = True id3.index['APIC'] = imgs @action("--remove-images") def RemoveImages(id3): """ remove all embedded artwork """ id3.remove('APIC') @action("--auto-image") def AutoAddImage(id3): """ if no embedded artwork is present, import it from external file with same name, or common names like Folder.jpg """ if id3.get('APIC'): return try: imgs_in_folder = [(os.path.splitext(f)[0].lower().replace('-', ' ').split(), f) \ for f in os.listdir(os.path.dirname(id3.filename)) \ if os.path.splitext(f)[-1].strip('.').lower() in ("jpg", "jpeg", "png")] except EnvironmentError: pass search_wordlists = list(enumerate([ [], ["album"], ["albumart"], ["cover"], ["front"], ["folder"], os.path.splitext(os.path.basename(id3.filename))[0].lower().replace('-', ' ').split()])) # enumerate all files, rank them by the highest-scored wordlist that matches root = os.path.dirname(id3.filename) try: match_imgs = sorted( (max(score for score, wordlist in search_wordlists if all((w in words) for w in wordlist)), filename) for words, filename in ((base.lower().replace('-', ' ').split(), base + ext) for base, ext in map(os.path.splitext, os.listdir(root)) if ext.strip('.').lower() in set(["jpg", "jpeg", "png"]))) except EnvironmentError: pass if not match_imgs: return # no images found at all # only keep the files from the highest-scored tier highscore = match_imgs[-1][0] match_imgs = [filename for score, filename in match_imgs if score >= highscore] if not(highscore) and ((len(match_imgs) > 1) \ or os.path.exists(os.path.join(os.path.dirname(id3.filename), os.path.splitext(match_imgs[0])[0] + ".mp3"))): # no match for a "preferred" filename -> only accept if there is a single # image *and* this is not directly associated with another .mp3 file return if (len(match_imgs) > 1) and not(highscore): return # no decent hit for any of the standard file names, and more than one file in the directory img = os.path.join(root, match_imgs[0]) id3.log(1, f"auto-adding artwork image '{img}'") try: id3.add(ID3NewImageFrame(img)) except EnvironmentError as e: id3.fail(f"could not import image file '{img}' - {e}") except ValueError as e: id3.fail(f"image file '{img}' has unrecognized format") @action("--extract-images") def ExtractImages(id3): """ extract embedded artwork into separate files """ imgs = id3.get('APIC', []) if not imgs: return # get index of image with highest priority prio = [(ID3V2_PICTYPE_PRIORITY.get(img.pictype, 99), i) for i, img in enumerate(imgs)] maxprio = min(prio)[0] prio = [(p, i) for p, i in prio if p == maxprio] default_idx = prio[0][1] if (len(prio) == 1) else -1 basename = os.path.splitext(id3.filename)[0] for i, img in enumerate(imgs): add_number = (i != default_idx) if IsJPEG(img.data): ext = ".jpg" elif IsPNG (img.data): ext = ".png" elif img.mimetype: ext = '.' + img.mimetype.split('/')[-1] else: ext = ".???" name = basename if i != default_idx: name += f"-{i+1}" name += ext if name.endswith('?'): id3.fail(f"unknown file type for image '{name}'") continue try: f = open(name, "wb") f.write(img.data) if (ext == ".jpg") and not(img.data.endswith("\xff\xd9")): f.write("\xff\xd9") f.close() except EnvironmentError as e: id3.fail(f"could not write image file '{name}' - {e}") @action("--clean") def CleanAll(id3, add_stuff=False): """ equivalent to --single-image --reduce-images --clean-tags --clean-images --fill-missing --sort """ RemoveAdditionalImages(id3) if add_stuff: AutoAddImage(id3) ReduceImages(id3) RemoveExotic(id3) CleanImages(id3) if add_stuff: FillMissingTags(id3) ReorderTags(id3) @action("--auto") def AutoProcess(id3): """ equivalent to --auto-image --fill-missing --clean """ return CleanAll(id3, True) @action("--comment2title") def CommentToTitle(id3): """ append bracketed ID3v1 comment into ID3v2 title (useful if comment tag has been used to denote remixes) """ comm1 = id3.gettext('1comment').strip() comm2 = id3.gettext('COMM').strip() if not(comm1) or (comm2 and (comm1 != comm2)): return comm = comm2 or comm1 if not(comm.startswith('(') and comm.endswith(')')): return title1 = id3.gettext('1title') title2 = id3.gettext('TIT2') if not(title1) or (title2 and (title1 != title2)): return title = title2 or title1 id3.log(1, f"appending comment {comm!r} to title {title!r}") id3.settext('TIT2', title + ' ' + comm) id3.remove('1comment') id3.remove('COMM') ################################################################################ _ffplay = None def EstimateBPM(taps): if len(taps) < 2: return None, None hist = collections.Counter() for step in range(1, len(taps)): hist.update([round(600 * step / max(b-a, 1E-3)) for a,b in zip(taps, taps[step:])] * step) bpm, count = hist.most_common(1)[0] halfmax = sorted(b for b,c in hist.items() if 2 * c >= count) return bpm / 10, (2 * max(bpm - halfmax[0], halfmax[-1] - bpm) + 1) / 20 @action("--tap-bpm") def TapBPM(id3): """ interactively set BPM by playing audio and tapping keys """ global _ffplay if not _ffplay: exes = ["ffplay", "avplay"] if sys.platform == "win32": exes = [x + ".exe" for x in exes] for d in [".", os.path.dirname(sys.argv[0])] + os.getenv('PATH').split(os.path.pathsep): for exe in exes: path = os.path.join(d, exe) if os.access(path, os.X_OK): _ffplay = path break if _ffplay: break if not _ffplay: print("Error: FFplay or AVplay need to be installed and in the PATH", file=sys.stderr) print(" in order to use the --tap-bpm function.", file=sys.stderr) sys.exit(3) if id3.get('TBPM'): return print(f"{id3.filename}: no BPM information available") print("Tap the Enter key with the beat to estimate the BPM value.") print("Enter the desired BPM number when ready.") print("Special values: '-' / '+' = seek in audio file, '/' = reset, '.' = cancel") player = None play_t0 = 0 seek = 90 replay = True while True: if replay: if player: player.terminate() player.wait() try: player = subprocess.Popen([_ffplay, "-loglevel", "quiet", "-nodisp", "-ss", str(seek), id3.filename], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except EnvironmentError as e: print("Error: failed to run FFplay -", e, file=sys.stderr) return play_t0 = time.time() - seek replay = False taps = [] if not taps: current = "???.? +/- ?.?" try: value = input(f"TapBPM [current: {current} BPM] ==> ").strip() except (EnvironmentError, EOFError, KeyboardInterrupt): print("Aborted.") player.terminate() player.wait() return if not value: taps.append(time.perf_counter()) bpm, var = EstimateBPM(taps) if bpm: current = f"{bpm:5.1f} +/- {min(var or 10, 9.9):.1f}" elif value == '*': seek = 0 replay = True elif is_only_char(value, '-') or is_only_char(value, '<'): seek = max(0, int(time.time() - play_t0 - 10 * len(value))) replay = True elif is_only_char(value, '+') or is_only_char(value, '>'): seek = max(0, int(time.time() - play_t0 + 10 * len(value))) replay = True elif value in '.,q': value = 0 break elif value == '/': taps = [] else: try: value = int(value) break except ValueError: print(f"invalid input {value!r}") player.terminate() if value: print("setting BPM to", value) id3.settext('TBPM', str(value)) player.wait() print ################################################################################ @action("--audit") def Audit(id3): """ compare title and artist ID3v1 and ID3v2 tags with the file name and interactively resolve discrepancies (also implies --fill-missing) """ # extract potential info from the file name offset = getattr(id3, '_audit_filename_offset', 0) fn = os.path.splitext(os.path.basename(id3.filename))[0][offset:].replace('_', ' ') m = re.match(r'((?P.*?)(--+|\s+-+\s+))?(?P.*)', fn) if m: fn_artist = m.group('artist') if fn_artist: fn_artist = fn_artist.strip() fn_title = m.group('title') if fn_title: fn_title = fn_title.strip() else: fn_artist = None fn_title = fn srcs = ["filename", "ID3v1", "ID3v2"] for name, value0, tag1, tag2 in ( ("title", fn_title, '1title', 'TIT2'), ("artist", fn_artist, '1artist', 'TPE1'), ("album", None, '1album', 'TALB'), ("year", None, '1year', 'TYER'), ("genre", None, '1genre', 'TCON'), ("comment",None, '1comment', 'COMM'), ): v = [value0, id3.gettext(tag1), id3.gettext(tag2)] # special case: strip numeric genre IDs for comparison if name == "genre": for i in (1, 2): m = re.match(r'\(\d+\)(.*)', v[i]) m = m and m.group(1).strip() if m: v[i] = m # generate lower-case versions for comparisons c = [(x or "").replace(' ', '').lower() for x in v] # if the tag is missing, bypass the thorough check if not(v[0] and not(v[1]) and not(v[2])): # check if each of the three combinations match (fn<->V1, V1<->V2, V2<->fn) # for an fn<->V1 match, a partial match is sufficient if a full V2 tag is present, # otherwise only a full match counts match01 = not(v[0]) or not(v[1]) or ((c[1] in c[0]) if v[2] else (c[1] == c[0])) match12 = not(v[1]) or not(v[2]) or (c[1] in c[2]) match02 = not(v[0]) or not(v[2]) or (c[0] == c[2]) if match01 and match02 and match12: continue print(f"{id3.filename}:") print(f"{name} tag is {'ambiguous' if (v[1] or v[2]) else 'missing'}:") for i, x in enumerate(v): if not x: continue print(f"{srcs[i]:<8} [{i}]: {x}") print(f"resolution? [empty = no change, {'/'.join(str(i) for i,x in enumerate(v) if x)} = use existing, - = delete, or type new]") try: res = input(f"({name})=> ").strip() except (EnvironmentError, EOFError, KeyboardInterrupt): print("Aborted by user.") sys.exit(0) if res and (res in "012"): res = v[int(res)] if not(res) or (res == '.'): print(f"{name} tag not changed") elif res == '-': id3.remove(tag1, tag2) print(f"{name} tag removed") else: id3.settext(tag1, res) id3.settext(tag2, res) print(f"{name} tag set to '{res}'") print FillMissingTags(id3) ################################################################################ @action("--show") def DisplayTags(id3): """ display the value of all tags (default if no other modifications or actions) """ print(id3.filename) tag_filter = set(['APIC', 'TXXX', 'WXXX', 'PRIV', 'USLT', 'SYLT']) for name, tag1, tag2 in ( ("group", None, 'TIT1'), ("title", '1title', 'TIT2'), ("subtitle", None, 'TIT3'), ("artist", '1artist', 'TPE1'), ("album", '1album', 'TALB'), ("album artist", None, 'TPE2'), ("conductor", None, 'TPE3'), ("remixed by", None, 'TPE4'), ("year", '1year', 'TYER'), ("date", None, 'TDAT'), ("genre", '1genre', 'TCON'), ("track", '1track', 'TRCK'), ("disc", None, 'TPOS'), ("ISRC", None, 'TSRC'), ("copyright", None, 'TCOP'), ("publisher", None, 'TPUB'), ("composer", None, 'TCOM'), ("orig. album", None, 'TOAL'), ("orig. artist", None, 'TOPE'), ("orig. lyrics", None, 'TOLY'), ("orig. year", None, 'TORY'), ("language", None, 'TLAN'), ("BPM", None, 'TBPM'), ("initial key", None, 'TKEY'), ("encoder", None, 'TENC'), ("software", None, 'TSSE'), ("length (ms)", None, 'TLEN'), ("comment", '1comment','COMM'), ): tag_filter.add(tag2) value1 = id3.gettext(tag1) value2 = id3.gettext(tag2) if not(value1) and not(value2): continue if tag2 == 'TLEN': try: t = int(value2) value2 = f"{t//60000}'{(t//1000)%60}.{t%1000:03d}\"" except ValueError: pass if value1: value1.replace('\r','').replace('\n', ' | ') if value2: value2.replace('\r','').replace('\n', ' | ') if value1 == value2: print(f"ID3v1+2 {name:<12s} : {value2}") else: if value1: print(f"ID3v1 {name:<12s} : {value1}") if value2: print(f"ID3v2 {name:<12s} : {value2}") priv = id3.get('PRIV', []) + id3.get('TXXX', []) + id3.get('WXXX', []) if priv: print(f"ID3v2 private tags : {len(priv)}") for tag in sorted(id3.index): if (tag in tag_filter) or tag.startswith('1'): continue print(f"ID3v2 '{tag:4s}' tag : {id3.index[tag].drepr()}") if id3.get('USLT') or id3.get('SYLT'): print("ID3v2 lyrics present") art = id3.get('APIC') if art: print("ID3v2 embedded artwork:") for img in art: props = [img.mimetype] try: props.append(ID3V2_PICTYPES[img.pictype]) except IndexError: props.append(f"type {img.pictype}") if img.desc: props.append(f"description '{img.desc}'") props.append(f"{0.001*len(img.data):.1f} kB") if have_pil: try: img = Image.open(io.BytesIO(img.data)) mode = {'L':"gray",'1':"mono",'P':"indexed"}.get(img.mode, img.mode) props.append(f"{img.size[0]}x{img.size[1]} {mode} {img.format}") except EnvironmentError: pass print(" -", ", ".join(props)) print() ################################################################################ def Albumize(d, verbosity): # collect files d = os.path.normpath(os.path.abspath(d)) try: tracks = [id3 for id3 in (ID3Info(f, verbosity) for f in (os.path.join(d, f) for f in os.listdir(d)) if os.path.isfile(f) and (os.path.splitext(f)[-1].strip(".").lower() == "mp3") and not(".kjid3-backup." in f)) if id3.good] except EnvironmentError as e: print(f"{d}: can not scan directory -", e, file=sys.stderr) return [] if not tracks: print(f"{d}: directory contains no valid files", file=sys.stderr) return [] # check for track numbers in file names tracknos = [re.match(r'(cd[ ._-]*(?P<discn>\d+)[ ._-]*)?(track[ ._-]*)?(?P<disca>[a-z])?(?P<track>\d\d)[ ._-]+', os.path.basename(id3.filename), re.I) for id3 in tracks] if all(tracknos): tracks[0].log(1, "generating track numbers from file names") else: tracknos = len(tracks) * [None] # collect artists and track numbers artists = collections.defaultdict(int) album_artists = [] maxtrack = collections.defaultdict(int) maxdisc = 0 for trackno, id3 in zip(tracknos, tracks): # artist statistics a1 = id3.gettext('1artist') a2 = id3.gettext('TPE1') n = 1 if (a1 and a2) else 2 if a1: artists[a1] += n if a2: artists[a2] += n # album artist already set? a2 = id3.gettext('TPE2') if a2 and (not(album_artists) or (a2 != album_artists[0])): album_artists.append(a2) # evaluate or generate track and disc numbers discno = (trackno and (trackno.group('discn') or trackno.group('disca'))) if discno: try: discno = int(trackno.group('discn')) except (ValueError, TypeError): discno = trackno.group('disca').upper() - 64 id3.settext('TPOS', str(discno)) else: discno = atoi(id3.gettext('TPOS').split('/')[0]) if trackno: id3._audit_filename_offset = trackno.end(0) trackno = int(trackno.group('track')) id3.settext('TRCK', str(trackno)) id3.settext('1track', str(trackno)) else: trackno = atoi((id3.gettext('TRCK') or id3.gettext('1track')).split('/')[0]) # extract maximum track number maxtrack[discno] = max(maxtrack[discno], trackno) maxdisc = max(maxdisc, discno) # select album artist if len(album_artists) == 1: album_artist = album_artists[0] # we got exactly one existing AA, good! tracks[0].log(1, f"album artist already set to {album_artist!r}") else: artists = sorted((n,a) for a,n in artists.items()) album_artist = None if not artists: tracks[0].log(1, "no artist tags found, can't determine album artist") elif artists[-1][0] < 5: tracks[0].log(1, "album seems to be from various artists, not setting album artist") else: # at least two tracks from most prominent artist, good! album_artist = artists[-1][1] tracks[0].log(1, f"setting album artist to {album_artist!r}") if not(album_artist) and not(maxtrack): print("no --albumize changes to album in", d) return tracks # apply changes for id3 in tracks: if album_artist and not(id3.gettext('TPE2')): id3.settext('TPE2', album_artist) discno = atoi(id3.gettext('TPOS').split('/')[0]) if maxdisc: t = id3.gettext('TPOS') if t and not('/' in t): id3.settext('TPOS', f"{t}/{maxdisc}") if maxtrack[discno]: t = id3.gettext('TRCK') or id3.gettext('1track') if t and not('/' in t): id3.settext('TRCK', f"{t}/{maxtrack[discno]}") FillMissingTags(id3) return tracks ################################################################################ if sys.platform == "win32": def unpack_globs(args): for arg in args: n = 0 for sub in glob.iglob(arg): n += 1 yield sub if not n: yield arg else: def unpack_globs(args): return args def iter_files(seq, recursive, single_dir=False): for item in seq: if os.path.isdir(item): if not recursive: continue for root, dirs, files in os.walk(item): for f in files: if (os.path.splitext(f)[-1].strip(".").lower() == "mp3") \ and not(".kjid3-backup." in f): yield os.path.join(root, f) if single_dir: return else: yield item def unique_files(seq): seen = set() for f in seq: path = os.path.normpath(os.path.abspath(f)) if not(path in seen): seen.add(path) yield f def delayed_iter(seq): multi = False accum = [] for f in seq: if multi: yield f, multi else: accum.append(f) if len(accum) > 1: multi = True for f2 in accum: yield f2, multi accum = [] for f2 in accum: yield f2, multi def restore_backups(f, dry_run=True): root, base = os.path.split(os.path.normpath(os.path.abspath(f))) base, ext = os.path.splitext(base) f = os.path.join(root, base + ext) b = os.path.join(root, f".kjid3-backup.{base}.0{ext}") if os.path.exists(b): if dry_run: print("would restore", b) print(" to", f) else: print("restoring", b) print(" to", f, end=' ') if os.name != 'posix': # Win32 (and potentially others) may not have # replace-on-rename semantics, so delete the target file first try: os.unlink(f) except EnvironmentError: pass try: os.rename(b, f) print("[OK]") except EnvironmentError as e: print("[FAILED]", e) def delete_backups(f, dry_run=True): root, base = os.path.split(os.path.normpath(os.path.abspath(f))) base, ext = os.path.splitext(base) for f in glob.glob(os.path.join(root, f".kjid3-backup.{base}.*{ext}")): if dry_run: print("would delete", f) else: print("deleting", f, end=' ') try: os.unlink(f) print("[OK]") except EnvironmentError as e: print("[FAILED]", e) ################################################################################ if __name__ == "__main__": QuickSetters = { "--title": ("title", "NAME", 'TIT2', '1title'), "--artist": ("artist", "NAME", 'TPE1', '1artist'), "--album": ("album", "NAME", 'TALB', '1album'), "--year": ("year", "YEAR", 'TYER', '1year'), "--genre": ("genre", "TEXT", 'TCON', '1genre'), "--comment": ("comment", "TEXT", 'COMM', '1comment'), "--track": ("track", "A[/B]", 'TRCK', '1track'), "--disc": ("disc", "A[/B]", 'TPOS'), "--composer": ("composer", "NAME", 'TCOM'), "--copy": ("copyright", "TEXT", 'TCOP'), "--bpm": ("tempo", "BPM", 'TBPM'), "--orig-artist": ("original artist", "NAME", 'TOPE'), "--album-artist": ("album artist", "NAME", 'TPE2'), } sys.stdout.reconfigure(errors='replace') parser = argparse.ArgumentParser( usage="%(prog)s [OPTIONS...] INPUTS...", description="ID3 tag manipulation tool") parser.add_argument("-H", "--fullhelp", action='store_true', help="print full help") parser.add_argument("-v", "--verbose", action='count', default=0, help="be more verbose (can be specified multiple times)") parser.add_argument("-V", "--version", action='version', version=os.path.splitext(os.path.basename(sys.argv[0]))[0] + " " + __version__, help="print version and exit") ioopts = parser.add_argument_group("Input/Output Options") ioopts.add_argument("inputs", metavar="INPUTS", nargs='*', help="input file/-s or directory/-ies") ioopts.add_argument("-r", "--recursive", action='store_true', help="recurse into subdirectories (default if sole input argument is a directory)") ioopts.add_argument("-o", "--output", metavar="FILE", help="set output file name (only valid if operating on a single file; default: overwrite input)") ioopts.add_argument("-n", "--dry-run", action='store_true', help="don't write output files even if tags have been modified") ioopts.add_argument("-b", "--backup", action='store_true', help="create backups of all modified files") ioopts.add_argument("-i", "--import", dest='imp', metavar="FILE", help="import ID3 tags from FILE") modes = parser.add_argument_group("Special Modes") modes.add_argument("--albumize", action='store_const', dest='mode', const='albumize', default=None, help="auto-set track count and album artist; INPUTS must be directories with albums (implies --fill-missing)") modes.add_argument("--restore-backups", action='store_const', dest='mode', const='restore', default=None, help="restore original file(s) from backup") modes.add_argument("--delete-backups", action='store_const', dest='mode', const='delete', default=None, help="delete backup files") setters = parser.add_argument_group("Tag Modification", description="sets both ID3v1 and ID3v2 tags where applicable; " + "set to empty string to remove a tag (e.g. --comment \"\"); " + "special syntax for --track and --disc: \"/X\" sets total track/disc number, " + "\"0[/X]\" infers track number from filename") for param in sorted(QuickSetters): desc, metavar = QuickSetters[param][:2] setters.add_argument(param, metavar=metavar, help="set " + desc) setters.add_argument("--set", metavar="TAG=TEXT", action='append', default=[], help="set an arbitrary ID3v2 tag by its frame name") setters.add_argument("--remove", metavar="TAG", action='append', default=[], help="remove an arbitrary ID3v2 tag by its frame name") setters.add_argument("--image", metavar="[TYPE[:DESCRIPTION]=]FILE", action='append', default=[], help="set embedded artwork from a PNG or JPEG file; " + "TYPE can be one of " + ", ".join(sorted(ID3V2_PICTYPE_MAP)) + " or 0..255 [default: cover]; DESCRIPTION is optional") setters.add_argument("--add-image", metavar="[TYPE[:DESCRIPTION]=]FILE", action='append', default=[], help="like --image, but don't replace existing artwork, only add a new image") actions = parser.add_argument_group("Global Modification Actions", description="applied *after* setting individual tags (except --new)") actions.add_argument("--new", action='store_true', help="remove all tags") for param in sorted(AvailableActions): func = AvailableActions[param] actions.add_argument(param, action='append_const', dest='actions', default=[], const=func, help=func.__doc__.strip()) args = parser.parse_args() if args.fullhelp: parser.print_help() print(__doc__) sys.exit(0) if args.imp: if args.new: parser.error("--import and --new are mutually exclusive") imp = ID3Info(args.imp, args.verbose) if not imp.good: print("Error: import source file is not valid", file=sys.stderr) sys.exit(1) else: imp = None setters = [] for param in QuickSetters: value = getattr(args, param.lstrip('-').replace('-', '_')) if not(value is None): for frame_id in QuickSetters[param][2:]: setters.append((frame_id, value)) for raw in args.set: try: frame_id, value = map(str.strip, raw.split('=', 1)) except ValueError: parser.error(f"invalid --set argument {raw!r} (no frame name)") setters.append((frame_id.lower() if frame_id.startswith('1') else frame_id.upper(), value)) for frame_id in args.remove: setters.append((frame_id.lower() if frame_id.startswith('1') else frame_id.upper(), None)) for frame_id, value in setters: if frame_id.startswith('1'): continue if (len(frame_id) != 4) or not(frame_id.isalnum()): print(f"Warning: ID3v2 frame ID {frame_id!r} looks invalid", file=sys.stderr) elif value and not(frame_id.startswith('T')): print(f"Warning: ID3v2 frame ID {frame_id!r} is not a text frame", file=sys.stderr) images = [] for imgspec in args.image + args.add_image: if '=' in imgspec: pictype, imgfile = imgspec.split('=', 1) else: pictype = "" imgfile = imgspec if ':' in pictype: pictype, desc = pictype.split(':', 1) else: desc = None try: images.append(ID3NewImageFrame(imgfile, pictype, desc)) except EnvironmentError as e: print(f"Error: could not import image file '{imgfile}' -", e, file=sys.stderr) sys.exit(1) except ValueError as e: parser.error(e) def process_file(f): if isinstance(f, ID3Info): id3 = f else: id3 = ID3Info(f, args.verbose) if not id3.good: return if args.new: RemoveAll(id3) elif imp: id3.copyfrom(imp) for frame_id, value in setters: try: if not value: id3.remove(frame_id) else: id3.settext_ex(frame_id, value) except Exception as e: print(f"{id3.filename}:", e, file=sys.stderr) if images: if args.image: id3.remove('APIC') for img in images: pictype = ID3V2_PICTYPES[img.pictype] if (img.pictype < len(ID3V2_PICTYPES)) else f"type {img.pictype}" id3.log(1, f"adding new artwork: {img.mimetype}, {pictype}") id3.add(img) for a in args.actions: a(id3) if not(args.actions) and not(setters) and not(images) and (args.mode != 'albumize'): DisplayTags(id3) if not id3.modified: pass elif args.dry_run: print("changes to", id3.filename, "not written to disk on user's request (--dry-run)") else: try: id3.write(args.output, args.backup) except EnvironmentError as e: print(f"{id3.filename}: error while writing new file -", e, file=sys.stderr) inputs = list(unpack_globs(args.inputs)) if not inputs: parser.error("no input files/directories specified") if args.mode == 'albumize': # albumize mode if args.output: parser.error("--output is not compatible with --albumize") if args.imp: parser.error("--import is not compatible with --albumize") if any(os.path.exists(d) and not(os.path.isdir(d)) for d in inputs): parser.error("inputs in --albumize mode must be directories") for d in inputs: for id3 in Albumize(d, args.verbose): process_file(id3) else: # default mode recursive = args.recursive or ((len(inputs) == 1) and os.path.isdir(inputs[0])) for f, is_multi in delayed_iter(unique_files(iter_files(inputs, recursive))): if args.mode == 'delete': delete_backups(f, args.dry_run) elif args.mode == 'restore': restore_backups(f, args.dry_run) else: if is_multi and args.output: parser.error("can only use --output option when specifying a single input file") process_file(f)