feat: support ec3 and ac3 codec

pull/1/head
WorldObservationLog 5 months ago
parent 90b271b348
commit da21d6fb1f
  1. 2
      src/cmd.py
  2. 9
      src/mp4.py
  3. 2
      src/rip.py
  4. 2
      src/save.py
  5. 6
      src/types.py
  6. 4
      src/utils.py

@ -35,7 +35,7 @@ class NewInteractiveShell:
download_parser = subparser.add_parser("download") download_parser = subparser.add_parser("download")
download_parser.add_argument("url", type=str) download_parser.add_argument("url", type=str)
download_parser.add_argument("-c", "--codec", download_parser.add_argument("-c", "--codec",
choices=["alac", "ec3", "aac", "aac-binaural", "aac-downmix"], default="alac") choices=["alac", "ec3", "aac", "aac-binaural", "aac-downmix", "ac3"], default="alac")
download_parser.add_argument("-f", "--force", type=bool, default=False) download_parser.add_argument("-f", "--force", type=bool, default=False)
subparser.add_parser("exit") subparser.add_parser("exit")

@ -21,8 +21,6 @@ async def extract_media(m3u8_url: str, codec: str) -> Tuple[str, list[str], str]
if not specifyPlaylist: if not specifyPlaylist:
raise CodecNotFoundException raise CodecNotFoundException
selected_codec = specifyPlaylist.media[0].group_id selected_codec = specifyPlaylist.media[0].group_id
if not specifyPlaylist:
raise
stream = m3u8.load(specifyPlaylist.absolute_uri) stream = m3u8.load(specifyPlaylist.absolute_uri)
skds = [key.uri for key in stream.keys if regex.match('(skd?://[^"]*)', key.uri)] skds = [key.uri for key in stream.keys if regex.match('(skd?://[^"]*)', key.uri)]
keys = [prefetchKey] keys = [prefetchKey]
@ -30,7 +28,7 @@ async def extract_media(m3u8_url: str, codec: str) -> Tuple[str, list[str], str]
match codec: match codec:
case Codec.ALAC: case Codec.ALAC:
key_suffix = CodecKeySuffix.KeySuffixAlac key_suffix = CodecKeySuffix.KeySuffixAlac
case Codec.EC3: case Codec.EC3 | Codec.AC3:
key_suffix = CodecKeySuffix.KeySuffixAtmos key_suffix = CodecKeySuffix.KeySuffixAtmos
case Codec.AAC: case Codec.AAC:
key_suffix = CodecKeySuffix.KeySuffixAAC key_suffix = CodecKeySuffix.KeySuffixAAC
@ -103,6 +101,8 @@ def encapsulate(song_info: SongInfo, decrypted_media: bytes, atmos_convent: bool
f.write(decrypted_media) f.write(decrypted_media)
if song_info.codec == Codec.EC3 and not atmos_convent: if song_info.codec == Codec.EC3 and not atmos_convent:
song_name = Path(tmp_dir.name) / Path(name).with_suffix(".ec3") song_name = Path(tmp_dir.name) / Path(name).with_suffix(".ec3")
elif song_info.codec == Codec.AC3 and not atmos_convent:
song_name = Path(tmp_dir.name) / Path(name).with_suffix(".ac3")
else: else:
song_name = Path(tmp_dir.name) / Path(name).with_suffix(".m4a") song_name = Path(tmp_dir.name) / Path(name).with_suffix(".m4a")
match song_info.codec: match song_info.codec:
@ -122,10 +122,11 @@ def encapsulate(song_info: SongInfo, decrypted_media: bytes, atmos_convent: bool
f"mp4edit --insert moov/trak/mdia/minf/stbl/stsd/alac:{alac_params_atom_name.absolute()} {song_name.absolute()} {final_m4a_name.absolute()}", f"mp4edit --insert moov/trak/mdia/minf/stbl/stsd/alac:{alac_params_atom_name.absolute()} {song_name.absolute()} {final_m4a_name.absolute()}",
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
song_name = final_m4a_name song_name = final_m4a_name
case Codec.EC3: case Codec.EC3 | Codec.AC3:
if not atmos_convent: if not atmos_convent:
with open(song_name.absolute(), "wb") as f: with open(song_name.absolute(), "wb") as f:
f.write(decrypted_media) f.write(decrypted_media)
else:
subprocess.run(f"gpac -i {media.absolute()} -o {song_name.absolute()}", subprocess.run(f"gpac -i {media.absolute()} -o {song_name.absolute()}",
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
case Codec.AAC_BINAURAL | Codec.AAC_DOWNMIX | Codec.AAC: case Codec.AAC_BINAURAL | Codec.AAC_DOWNMIX | Codec.AAC:

@ -38,6 +38,8 @@ async def rip_song(song: Song, auth_params: GlobalAuthParams, codec: str, config
song = encapsulate(song_info, decrypted_song, config.download.atmosConventToM4a) song = encapsulate(song_info, decrypted_song, config.download.atmosConventToM4a)
if codec != Codec.EC3 or (codec == Codec.EC3 and config.download.atmosConventToM4a): if codec != Codec.EC3 or (codec == Codec.EC3 and config.download.atmosConventToM4a):
song = write_metadata(song, song_metadata, config.metadata.embedMetadata, config.download.coverFormat) song = write_metadata(song, song_metadata, config.metadata.embedMetadata, config.download.coverFormat)
if codec != Codec.AC3 or (codec == Codec.AC3 and config.download.atmosConventToM4a):
song = write_metadata(song, song_metadata, config.metadata.embedMetadata, config.download.coverFormat)
save(song, codec, song_metadata, config.download) save(song, codec, song_metadata, config.download)
logger.info(f"Song {song_metadata.artist} - {song_metadata.title} saved!") logger.info(f"Song {song_metadata.artist} - {song_metadata.title} saved!")

@ -14,6 +14,8 @@ def save(song: bytes, codec: str, metadata: SongMetadata, config: Download):
os.makedirs(dir_path.absolute()) os.makedirs(dir_path.absolute())
if codec == Codec.EC3 and not config.atmosConventToM4a: if codec == Codec.EC3 and not config.atmosConventToM4a:
song_path = dir_path / Path(song_name).with_suffix(".ec3") song_path = dir_path / Path(song_name).with_suffix(".ec3")
elif codec == Codec.AC3 and not config.atmosConventToM4a:
song_path = dir_path / Path(song_name).with_suffix(".ac3")
else: else:
song_path = dir_path / Path(song_name).with_suffix(".m4a") song_path = dir_path / Path(song_name).with_suffix(".m4a")
with open(song_path.absolute(), "wb") as f: with open(song_path.absolute(), "wb") as f:

@ -23,6 +23,7 @@ class SongInfo(BaseModel):
class Codec: class Codec:
ALAC = "alac" ALAC = "alac"
EC3 = "ec3" EC3 = "ec3"
AC3 = "ac3"
AAC_BINAURAL = "aac-binaural" AAC_BINAURAL = "aac-binaural"
AAC_DOWNMIX = "aac-downmix" AAC_DOWNMIX = "aac-downmix"
AAC = "aac" AAC = "aac"
@ -38,7 +39,8 @@ class CodecKeySuffix:
class CodecRegex: class CodecRegex:
RegexCodecAtmos = "audio-atmos-\\d{4}$" RegexCodecAtmos = "audio-(atmos|ec3)-\\d{4}$"
RegexCodecAC3 = "audio-ac3-\\d{3}$"
RegexCodecAlac = "audio-alac-stereo-\\d{5}-\\d{2}$" RegexCodecAlac = "audio-alac-stereo-\\d{5}-\\d{2}$"
RegexCodecBinaural = "audio-stereo-\\d{3}-binaural$" RegexCodecBinaural = "audio-stereo-\\d{3}-binaural$"
RegexCodecDownmix = "audio-stereo-\\d{3}-downmix$" RegexCodecDownmix = "audio-stereo-\\d{3}-downmix$"
@ -48,7 +50,7 @@ class CodecRegex:
def get_pattern_by_codec(cls, codec: str): def get_pattern_by_codec(cls, codec: str):
codec_pattern_mapping = {Codec.ALAC: cls.RegexCodecAlac, Codec.EC3: cls.RegexCodecAtmos, codec_pattern_mapping = {Codec.ALAC: cls.RegexCodecAlac, Codec.EC3: cls.RegexCodecAtmos,
Codec.AAC_DOWNMIX: cls.RegexCodecDownmix, Codec.AAC_BINAURAL: cls.RegexCodecBinaural, Codec.AAC_DOWNMIX: cls.RegexCodecDownmix, Codec.AAC_BINAURAL: cls.RegexCodecBinaural,
Codec.AAC: cls.RegexCodecAAC} Codec.AAC: cls.RegexCodecAAC, Codec.AC3: cls.RegexCodecAC3}
return codec_pattern_mapping.get(codec) return codec_pattern_mapping.get(codec)

@ -109,9 +109,11 @@ def check_song_exists(metadata, config: Download, codec: str):
dir_path = Path(config.dirPathFormat.format(**metadata.model_dump())) dir_path = Path(config.dirPathFormat.format(**metadata.model_dump()))
if not config.atmosConventToM4a and codec == Codec.EC3: if not config.atmosConventToM4a and codec == Codec.EC3:
return (Path(dir_path) / Path(song_name).with_suffix(".ec3")).exists() return (Path(dir_path) / Path(song_name).with_suffix(".ec3")).exists()
elif not config.atmosConventToM4a and codec == Codec.AC3:
return (Path(dir_path) / Path(song_name).with_suffix(".ac3")).exists()
else: else:
return (Path(dir_path) / Path(song_name).with_suffix(".m4a")).exists() return (Path(dir_path) / Path(song_name).with_suffix(".m4a")).exists()
def get_valid_filename(filename: str): def get_valid_filename(filename: str):
return "".join(i for i in filename if i not in "\/:*?<>|") return "".join(i for i in filename if i not in r"\/:*?<>|")

Loading…
Cancel
Save