trueroad/find_zipped_flac_groups.py

## find_zipped_flac_groups.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Find groups of equivalent FLACs that are zipped.

https://gist.github.com/trueroad/e7c0852f2a5822df05bde9e5578743b7

Copyright (C) 2024 Masamichi Hosoda.
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:

* Redistributions of source code must retain the above copyright notice,
  this list of conditions and the following disclaimer.

* Redistributions in binary form must reproduce the above copyright notice,
  this list of conditions and the following disclaimer in the documentation
  and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED.
IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.

複数の ZIP された FLAC を入力として、等価 FLAC のグループを見つける。

等価というのは、
サンプリングレート、フレーム数、チャンネル数、コンテンツ
がまったく同じとし、それ以外のメタデータ等は考慮しない。

FLAC のコンテンツはフレーム毎に 16 bit 整数として比較する
（定数 `DTYPE_FOR_COMPARE` にて設定）。
ZIP をまたいだ比較はせず、ZIP 内の FLAC 同士での比較となる。

入力はコマンドライン引数に比較したい FLAC を含んだ
ZIP ファイル名をすべて記述する。

出力は標準出力に、
グループ間に空行を入れつつ、
グループ内は所属する FLAC の
サンプリングレート、フレーム数、チャンネル数、ファイル名
をタブ区切りで 1 行ずつ、という形式となる。
グループに所属しない FLAC については出力しない。

あわせて標準エラー出力に処理中の ZIP ファイル名を出力する。
"""

from __future__ import annotations
import hashlib
import os
from pathlib import Path
import sys
from types import TracebackType
from typing import Any, Final, IO, Optional, Union
import zipfile

import soundfile as sf  # type: ignore[import-untyped]

# コンテンツ比較用フォーマット
DTYPE_FOR_COMPARE: Final[str] = 'int16'


class compare_zipped_flacs:
    """Compare zipped FLACs class."""

    def __init__(self,
                 zip_filename: Union[str, os.PathLike[Any]]
                 ) -> None:
        """
        __init__.

        Args:
          zip_filename (Union[str, os.PathLike[Any]]):
            比較する FLAC を格納した ZIP ファイル名
        """
        # ZIP ファイル
        self.zf: zipfile.ZipFile = zipfile.ZipFile(zip_filename, 'r')
        # FLAC 辞書
        self.flac_dict: dict[str, tuple[IO[bytes], sf.SoundFile, str]] = {}

        # ZIP ファイル内の FLAC ファイル名
        flac_filename: str
        for flac_filename in self.zf.namelist():
            # 拡張子が .flac ではないのでスキップ
            if Path(flac_filename).suffix.lower() != '.flac':
                continue
            # FLAC ファイルを開く
            zflac: IO[bytes] = self.zf.open(flac_filename, 'r')
            # FLAC を読み込む
            zsf: sf.SoundFile = sf.SoundFile(zflac)
            # FLAC 辞書に登録（ハッシュは遅延させる）
            self.flac_dict[flac_filename] = (zflac, zsf, '')

    def __enter__(self) -> compare_zipped_flacs:
        """
        コンテキストマネージャ（with 構文）に入る.

        with 構文のブロックに入ったので、
        内部の変数の __enter__() を呼んでいく。

        Returns:
          complare_zipped_flacs: 自インスタンス（with 構文の `as` へ渡される）
        """
        self.zf = self.zf.__enter__()
        filename: str
        zflac: IO[bytes]
        zsf: sf.SoundFile
        hs: str
        for filename, (zflac, zsf, hs) in self.flac_dict.items():
            zflac = zflac.__enter__()
            zsf = zsf.__enter__()
            self.flac_dict[filename] = (zflac, zsf, hs)
        return self

    def __exit__(self,
                 exc_type: Optional[type[BaseException]],
                 exc_value: Optional[BaseException],
                 traceback: Optional[TracebackType]) -> None:
        """
        コンテキストマネージャ（with 構文）から出る.

        with 構文のブロックから抜けるので、
        内部の変数の __exit__() を呼んでいく。

        Args:
          exc_type (Optional[type[BaseException]]): 例外情報
          exc_value (Optional[BaseException]): 例外情報
          traceback (Optional[TracebackType]): 例外情報
        """
        zflac: IO[bytes]
        zsf: sf.SoundFile
        for zflac, zsf, _ in self.flac_dict.values():
            zsf.__exit__(exc_type, exc_value, traceback)
            zflac.__exit__(exc_type, exc_value, traceback)
        self.zf.__exit__(exc_type, exc_value, traceback)

    def close(self) -> None:
        """
        クローズする.

        クローズするために、
        内部の変数の close() を呼んでいく。
        """
        zflac: IO[bytes]
        zsf: sf.SoundFile
        for zflac, zsf, _ in self.flac_dict.values():
            zsf.close()
            zflac.close()
        self.zf.close()

    def __load_hash(self, filename: str) -> None:
        """
        指定された FLAC のコンテンツからハッシュを計算してロードする.

        Args:
          filename (str): FLAC ファイル名
        """
        # FLAC 辞書に登録されている情報を取り出す
        zflac: IO[bytes]
        zsf: sf.SoundFile
        zflac, zsf, _ = self.flac_dict[filename]

        # FLAC のコンテンツを読み込み、ハッシュを計算して、
        # FLAC 辞書に登録する
        self.flac_dict[filename] = \
            (zflac,
             zsf,
             hashlib.sha256(zsf.read(dtype=DTYPE_FOR_COMPARE)).hexdigest())

    def compare(self) -> None:
        """比較する."""
        # 既に等価 FLAC グループに入っている FLAC のインデックス格納用
        already_grouped: set[int] = set()
        # FLAC 辞書に登録されている FLAC ファイル名
        flac_filenames: list[str] = list(self.flac_dict.keys())

        # 外側の判定ループ i
        i: int
        for i in range(len(flac_filenames)):
            if i in already_grouped:
                # 既に等価 FLAC グループに入っているインデックスはスキップ
                continue

            # 外側インデックス i に紐づく FLAC ファイル名
            filename_i: str = flac_filenames[i]
            # 外側インデックス i に紐づく FLAC 読み込みクラス
            zsf_i: sf.SoundFile = self.flac_dict[filename_i][1]
            # 等価 FLAC グループ開始フラグを未開始に初期化
            b_group_started: bool = False

            # 内側の判定ループ j
            j: int
            for j in range(len(flac_filenames)):
                if i >= j:
                    # 内側外側で同じインデックス同士は比較せずにスキップ、
                    # 内側外側が逆のインデックス組み合わせもスキップ
                    continue
                if j in already_grouped:
                    # 既に等価 FLAC グループに入っているのでスキップ
                    continue

                # 内側インデックス j に紐づく FLAC ファイル名
                filename_j: str = flac_filenames[j]
                # 内側インデックス j に紐づく FLAC 読み込みクラス
                zsf_j: sf.SoundFile = self.flac_dict[filename_j][1]

                # FLAC のパラメータを比較する
                if (((zsf_i.samplerate != zsf_j.samplerate) or
                     (zsf_i.channels != zsf_j.channels) or
                     (zsf_i.frames != zsf_j.frames))):
                    # サンプリングレート、チャンネル数、フレーム数
                    # いずれかが異なる
                    continue

                # パラメータが同じなのでコンテンツを比較する

                # コンテンツハッシュがロードされているかチェック
                if self.flac_dict[filename_i][2] == '':
                    # i に紐づくコンテンツハッシュが未ロードなのでロードする
                    self.__load_hash(filename_i)
                if self.flac_dict[filename_j][2] == '':
                    # j に紐づくコンテンツハッシュが未ロードなのでロードする
                    self.__load_hash(filename_j)

                # コンテンツハッシュ同士を比較する
                if self.flac_dict[filename_i][2] != \
                   self.flac_dict[filename_j][2]:
                    # コンテンツハッシュが異なる＝コンテンツが異なる
                    continue

                # コンテンツハッシュが同一＝コンテンツが同一と判断
                # FLAC 同士が一致！

                if not b_group_started:
                    # まだグループを開始していなかったので開始処理
                    # 最初の FLAC の情報を出力
                    print(f'\n{zsf_i.samplerate}\t{zsf_i.frames}\t'
                          f'{zsf_i.channels}\t{filename_i}')
                    # 等価 FLAC グループ開始済フラグを立てる
                    b_group_started = True
                # 2 番目以降の FLAC の情報を出力
                print(f'{zsf_j.samplerate}\t{zsf_j.frames}\t'
                      f'{zsf_j.channels}\t{filename_j}')
                # このインデックスはグループ化されたので次以降はスキップ
                already_grouped.add(i)
                already_grouped.add(j)


def main() -> None:
    """Do main."""
    # コマンドラインで指定された ZIP ファイル名でループ
    zip_filename: str
    for zip_filename in sys.argv[1:]:
        sys.stdout.flush()
        print(f'*** {zip_filename} ***', file=sys.stderr, flush=True)

        # 比較クラスで ZIP ファイルを開き、中の FLAC を比較する
        csf: compare_zipped_flacs
        with compare_zipped_flacs(zip_filename) as csf:
            csf.compare()


if __name__ == '__main__':
    main()
	#!/usr/bin/env python3
	# -- coding: utf-8 --
	"""
	Find groups of equivalent FLACs that are zipped.

	https://gist.github.com/trueroad/e7c0852f2a5822df05bde9e5578743b7

	Copyright (C) 2024 Masamichi Hosoda.
	All rights reserved.

	Redistribution and use in source and binary forms, with or without
	modification, are permitted provided that the following conditions
	are met:

	* Redistributions of source code must retain the above copyright notice,
	this list of conditions and the following disclaimer.

	* Redistributions in binary form must reproduce the above copyright notice,
	this list of conditions and the following disclaimer in the documentation
	and/or other materials provided with the distribution.

	THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
	AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
	IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
	ARE DISCLAIMED.
	IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
	FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
	DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
	OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
	HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
	LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
	OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
	SUCH DAMAGE.

	複数の ZIP された FLAC を入力として、等価 FLAC のグループを見つける。

	等価というのは、
	サンプリングレート、フレーム数、チャンネル数、コンテンツ
	がまったく同じとし、それ以外のメタデータ等は考慮しない。

	FLAC のコンテンツはフレーム毎に 16 bit 整数として比較する
	（定数 `DTYPE_FOR_COMPARE` にて設定）。
	ZIP をまたいだ比較はせず、ZIP 内の FLAC 同士での比較となる。

	入力はコマンドライン引数に比較したい FLAC を含んだ
	ZIP ファイル名をすべて記述する。

	出力は標準出力に、
	グループ間に空行を入れつつ、
	グループ内は所属する FLAC の
	サンプリングレート、フレーム数、チャンネル数、ファイル名
	をタブ区切りで 1 行ずつ、という形式となる。
	グループに所属しない FLAC については出力しない。

	あわせて標準エラー出力に処理中の ZIP ファイル名を出力する。
	"""

	from __future__ import annotations
	import hashlib
	import os
	from pathlib import Path
	import sys
	from types import TracebackType
	from typing import Any, Final, IO, Optional, Union
	import zipfile

	import soundfile as sf # type: ignore[import-untyped]

	# コンテンツ比較用フォーマット
	DTYPE_FOR_COMPARE: Final[str] = 'int16'


	class compare_zipped_flacs:
	"""Compare zipped FLACs class."""

	def __init__(self,
	zip_filename: Union[str, os.PathLike[Any]]
	) -> None:
	"""
	__init__.

	Args:
	zip_filename (Union[str, os.PathLike[Any]]):
	比較する FLAC を格納した ZIP ファイル名
	"""
	# ZIP ファイル
	self.zf: zipfile.ZipFile = zipfile.ZipFile(zip_filename, 'r')
	# FLAC 辞書
	self.flac_dict: dict[str, tuple[IO[bytes], sf.SoundFile, str]] = {}

	# ZIP ファイル内の FLAC ファイル名
	flac_filename: str
	for flac_filename in self.zf.namelist():
	# 拡張子が .flac ではないのでスキップ
	if Path(flac_filename).suffix.lower() != '.flac':
	continue
	# FLAC ファイルを開く
	zflac: IO[bytes] = self.zf.open(flac_filename, 'r')
	# FLAC を読み込む
	zsf: sf.SoundFile = sf.SoundFile(zflac)
	# FLAC 辞書に登録（ハッシュは遅延させる）
	self.flac_dict[flac_filename] = (zflac, zsf, '')

	def __enter__(self) -> compare_zipped_flacs:
	"""
	コンテキストマネージャ（with 構文）に入る.

	with 構文のブロックに入ったので、
	内部の変数の __enter__() を呼んでいく。

	Returns:
	complare_zipped_flacs: 自インスタンス（with 構文の `as` へ渡される）
	"""
	self.zf = self.zf.__enter__()
	filename: str
	zflac: IO[bytes]
	zsf: sf.SoundFile
	hs: str
	for filename, (zflac, zsf, hs) in self.flac_dict.items():
	zflac = zflac.__enter__()
	zsf = zsf.__enter__()
	self.flac_dict[filename] = (zflac, zsf, hs)
	return self

	def __exit__(self,
	exc_type: Optional[type[BaseException]],
	exc_value: Optional[BaseException],
	traceback: Optional[TracebackType]) -> None:
	"""
	コンテキストマネージャ（with 構文）から出る.

	with 構文のブロックから抜けるので、
	内部の変数の __exit__() を呼んでいく。

	Args:
	exc_type (Optional[type[BaseException]]): 例外情報
	exc_value (Optional[BaseException]): 例外情報
	traceback (Optional[TracebackType]): 例外情報
	"""
	zflac: IO[bytes]
	zsf: sf.SoundFile
	for zflac, zsf, _ in self.flac_dict.values():
	zsf.__exit__(exc_type, exc_value, traceback)
	zflac.__exit__(exc_type, exc_value, traceback)
	self.zf.__exit__(exc_type, exc_value, traceback)

	def close(self) -> None:
	"""
	クローズする.

	クローズするために、
	内部の変数の close() を呼んでいく。
	"""
	zflac: IO[bytes]
	zsf: sf.SoundFile
	for zflac, zsf, _ in self.flac_dict.values():
	zsf.close()
	zflac.close()
	self.zf.close()

	def __load_hash(self, filename: str) -> None:
	"""
	指定された FLAC のコンテンツからハッシュを計算してロードする.

	Args:
	filename (str): FLAC ファイル名
	"""
	# FLAC 辞書に登録されている情報を取り出す
	zflac: IO[bytes]
	zsf: sf.SoundFile
	zflac, zsf, _ = self.flac_dict[filename]

	# FLAC のコンテンツを読み込み、ハッシュを計算して、
	# FLAC 辞書に登録する
	self.flac_dict[filename] = \
	(zflac,
	zsf,
	hashlib.sha256(zsf.read(dtype=DTYPE_FOR_COMPARE)).hexdigest())

	def compare(self) -> None:
	"""比較する."""
	# 既に等価 FLAC グループに入っている FLAC のインデックス格納用
	already_grouped: set[int] = set()
	# FLAC 辞書に登録されている FLAC ファイル名
	flac_filenames: list[str] = list(self.flac_dict.keys())

	# 外側の判定ループ i
	i: int
	for i in range(len(flac_filenames)):
	if i in already_grouped:
	# 既に等価 FLAC グループに入っているインデックスはスキップ
	continue

	# 外側インデックス i に紐づく FLAC ファイル名
	filename_i: str = flac_filenames[i]
	# 外側インデックス i に紐づく FLAC 読み込みクラス
	zsf_i: sf.SoundFile = self.flac_dict[filename_i][1]
	# 等価 FLAC グループ開始フラグを未開始に初期化
	b_group_started: bool = False

	# 内側の判定ループ j
	j: int
	for j in range(len(flac_filenames)):
	if i >= j:
	# 内側外側で同じインデックス同士は比較せずにスキップ、
	# 内側外側が逆のインデックス組み合わせもスキップ
	continue
	if j in already_grouped:
	# 既に等価 FLAC グループに入っているのでスキップ
	continue

	# 内側インデックス j に紐づく FLAC ファイル名
	filename_j: str = flac_filenames[j]
	# 内側インデックス j に紐づく FLAC 読み込みクラス
	zsf_j: sf.SoundFile = self.flac_dict[filename_j][1]

	# FLAC のパラメータを比較する
	if (((zsf_i.samplerate != zsf_j.samplerate) or
	(zsf_i.channels != zsf_j.channels) or
	(zsf_i.frames != zsf_j.frames))):
	# サンプリングレート、チャンネル数、フレーム数
	# いずれかが異なる
	continue

	# パラメータが同じなのでコンテンツを比較する

	# コンテンツハッシュがロードされているかチェック
	if self.flac_dict[filename_i][2] == '':
	# i に紐づくコンテンツハッシュが未ロードなのでロードする
	self.__load_hash(filename_i)
	if self.flac_dict[filename_j][2] == '':
	# j に紐づくコンテンツハッシュが未ロードなのでロードする
	self.__load_hash(filename_j)

	# コンテンツハッシュ同士を比較する
	if self.flac_dict[filename_i][2] != \
	self.flac_dict[filename_j][2]:
	# コンテンツハッシュが異なる＝コンテンツが異なる
	continue

	# コンテンツハッシュが同一＝コンテンツが同一と判断
	# FLAC 同士が一致！

	if not b_group_started:
	# まだグループを開始していなかったので開始処理
	# 最初の FLAC の情報を出力
	print(f'\n{zsf_i.samplerate}\t{zsf_i.frames}\t'
	f'{zsf_i.channels}\t{filename_i}')
	# 等価 FLAC グループ開始済フラグを立てる
	b_group_started = True
	# 2 番目以降の FLAC の情報を出力
	print(f'{zsf_j.samplerate}\t{zsf_j.frames}\t'
	f'{zsf_j.channels}\t{filename_j}')
	# このインデックスはグループ化されたので次以降はスキップ
	already_grouped.add(i)
	already_grouped.add(j)


	def main() -> None:
	"""Do main."""
	# コマンドラインで指定された ZIP ファイル名でループ
	zip_filename: str
	for zip_filename in sys.argv[1:]:
	sys.stdout.flush()
	print(f'* {zip_filename} *', file=sys.stderr, flush=True)

	# 比較クラスで ZIP ファイルを開き、中の FLAC を比較する
	csf: compare_zipped_flacs
	with compare_zipped_flacs(zip_filename) as csf:
	csf.compare()


	if __name__ == '__main__':
	main()