写真が趣味で10年くらい前までは銀塩カメラで撮った写真をPCに取り込んでExifを付けたりしていました。しかし、最近はフィルム価格の高騰、そもそも手に入らないなど、完全デジタルへ移行することになりました。
色々写真を気ままに撮っていたら、気づけば85,000枚以上。
「後で整理しよう」と思いながら15年以上、もう完全に手に負えない状態になっていました。手作業でなんとかしようとかそういうタームではありません。
Windowsでは写真管理にdigikamを使っていますが、重複検出はできても、8万枚規模となると時間的に無理があります。
そこで今回は考えました。Pythonで自動検出して、削除は自分で判断しながらしよう。
ファイルをPythonが自動的に消すのは怖い、だから重複ファイルを別のフォルダに退避させる方法を考えました。あくまで削除は自分です。
そんなふわっとした設計思想で作ったのが、今回紹介したいスクリプトです。
榊式 Photo Deduplicator(完成版コード)
使用言語はPythonです。単純は複雑に勝る、明示は暗黙に勝るという設計が好きです。
そして画像処理には以下3つを利用しています。
- Pillow:Python用の画像処理ライブラリ。対応形式はJPEG, PNG, BMP, TIFFなど。
- pillow-heif:Pillow向けのHEIF/HEIC画像デコーダ/エンコーダ。
- ImageHash:近くハッシュで画像の類似性を比較してくれます。見た目の違いを数値的に扱う現実派で機械学習、画像検索、重複画像検出などができます。
そしてこちらが今回完成させたものです。2人でちまちま作っていたので時間がかかりました。画像の重複で困っている方がいたら、改変して使ってください。
PhotoDeduplicator.py(メモ帳で書いて大丈夫です)
import os
import shutil
import pickle
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor
from PIL import Image
import imagehash
from pillow_heif import register_heif_opener
from tqdm import tqdm
# =====================================================
# 設定
# =====================================================
SOURCE_DIR = r'E:\Photograph\Photos'
DEST_DIR = r'E:\Photograph\duplication'
CACHE_FILE = r'E:\Photograph\analysis_cache.pkl' # 解析結果の保存先
EXTENSIONS = {'.jpg', '.jpeg', '.png', '.heic', '.tiff', '.webp'}
HASH_THRESHOLD = 5
BUCKET_BITS = 8
CHECK_NEIGHBOR = True
DRY_RUN = True
WORKERS = os.cpu_count()
# =====================================================
# 画像解析(並列フェーズ)
# =====================================================
def analyze_image(path: Path):
try:
register_heif_opener()
# HDD負荷軽減のため、一度に開いて必要な情報をすべて抜き出す
file_size = os.path.getsize(path)
with Image.open(path) as img:
ph = imagehash.phash(img)
# imagehashオブジェクトを数値(int)に変換してシリアライズ可能にする
ph_int = int(str(ph), 16)
w, h = img.size
score = (w * h) + file_size
return (str(path), ph_int, score)
except Exception:
return None
# =====================================================
# move安全処理
# =====================================================
def safe_move(src_path_str, dest_dir):
src = Path(src_path_str)
base = src.name
name, ext = src.stem, src.suffix
target = Path(dest_dir) / base
i = 1
while target.exists():
target = Path(dest_dir) / f"{name}_dup_{i}{ext}"
i += 1
shutil.move(str(src), str(target))
# =====================================================
# ハミング距離計算(数値用)
# =====================================================
def hamming_distance(h1, h2):
# 排他的論理和をとって、立っているビット数を数える(ハミング距離)
return bin(h1 ^ h2).count('1')
# =====================================================
# メイン
# =====================================================
def run():
os.makedirs(DEST_DIR, exist_ok=True)
# Phase 0: キャッシュ確認
results = []
if os.path.exists(CACHE_FILE):
print(f"キャッシュファイルを発見しました: {CACHE_FILE}")
with open(CACHE_FILE, 'rb') as f:
results = pickle.load(f)
print(f"解析済みデータ {len(results)} 件をロードしました。")
else:
files = [
p for p in Path(SOURCE_DIR).rglob('*')
if p.suffix.lower() in EXTENSIONS
]
print(f"対象枚数 : {len(files)}")
print(f"CPU並列数 : {WORKERS}")
print("=" * 50)
# Phase 1: 解析
print("Phase1: 解析中(並列)")
with ProcessPoolExecutor(WORKERS) as exe:
for r in tqdm(exe.map(analyze_image, files), total=len(files)):
if r:
results.append(r)
# 解析結果を保存(HDDのボトルネック対策:再計算を防ぐ)
with open(CACHE_FILE, 'wb') as f:
pickle.dump(results, f)
print(f"解析結果をキャッシュに保存しました。")
# Phase 2: 重複検出
print("Phase2: 重複検出中")
buckets = {}
dup_count = 0
for path_str, ph_int, score in tqdm(results):
# バケットキー生成(int型で計算)
key = ph_int >> (64 - BUCKET_BITS)
keys = [key]
if CHECK_NEIGHBOR:
keys += [key - 1, key + 1]
matched = False
for k in keys:
bucket = buckets.get(k)
if not bucket:
continue
for i, (old_hash, old_score, old_path_str) in enumerate(bucket):
# 数値同士のハミング距離で比較
if hamming_distance(ph_int, old_hash) <= HASH_THRESHOLD:
matched = True
# スコア比較
if score > old_score:
loser = old_path_str
bucket[i] = (ph_int, score, path_str)
else:
loser = path_str
dup_count += 1
if DRY_RUN:
pass # 大量のprintはコンソールを重くするため省略可
else:
safe_move(loser, DEST_DIR)
break
if matched:
break
if not matched:
buckets.setdefault(key, []).append((ph_int, score, path_str))
print("\n" + "=" * 50)
print(f"完了 重複検出数: {dup_count}")
if DRY_RUN:
print("※DRY_RUNモードのため、実際の移動は行っていません。")
print("=" * 50)
if __name__ == "__main__":
import multiprocessing
multiprocessing.freeze_support()
run()
トライ&エラーと実行後の検証を何度も行うことで私たちは書いています。それでも無駄な部分や解釈違いがたくさんあります。おま環かもしれませんが、そこはカスタマイズしてください。
なぜ自作スクリプトにしたのか
自分のために自作スクリプトを考えたりするときが一番楽しいからです。というのが本音ですが、それよりも8万枚以上の重複写真を探すのは手作業では不可能だからです。
手作業は物理的に不可能
およそ85,000枚を1秒1枚で確認しても約24時間かかります。無理です。それに現実での手作業はもっと時間がかかります。一か月くらい時間が飛んでしまうかもしれません。
既存ソフトは遅いし重い
これは完璧におま環なのですが、私が使ってきた写真管理アプリはどれも遅いし重かったです。いくつかあったのですが、有名なのはこの2つでしょうか。
写真.appは2万枚を超えたあたりから重くなってきました。しかし、同期するのはとても楽で良かったです。難点は取り込んだ写真が、ライブラリの中のMastersフォルダにあり、直接管理したい私には不便でした。
Lightroomはとても便利で軽く、すばらしいソフトでした。しかしサブスクが嫌いということと、買切るにも高いという点が足かせとなりました。
確かにこれらGUI系のソフトはとても便利です。
- メモリを大量消費する
- フリーズの可能性がある
- それらによる長時間の待ちがある
枚数が増えると起動に時間がかかり、処理をするにも遅くなってしまう。ハイエンド機を買うなんて本末転倒ですし、要するに枚数が多くなるに比例してストレスも溜まるのです。
でしたら、私たちで軽いCLIツールを作った方が速いという結論に至りました。
設計思想
これまでいろいろ書いてきましたが、ここからはまじめな話になります。
pHashを採用した理由(完全一致は使わない)
重複検出というとMD5やSHAを思い浮かべると思います。しかし、それは完全一致専用のものです。あいまいな部分のある写真整理には向きません。
そこで採用したのは近くハッシュ(pHash)です。
- リサイズされても大丈夫
- 圧縮されても大丈夫
- 色味変化でも大丈夫
- その上軽量
実測でもそれらが分かります。
- CPU使用率:0~15%
- CPU温度:32~35℃
- 周波数:最大5.2GHz維持
つまるところ、CPU使用率をガリガリ上げていく時に、CPUがボトルネックにならない利点がありました。写真用途ではほぼ最適解と言えます。
HEIC対応は2026年時点ではすでに必須条件
数世代前からiPhoneの写真は、設定で変えない限り、HEICです。また2025年9月の段階でiPhoneユーザー(18~60歳)はスマートフォン全体の48.3%もいるとMMD研究所の調査でわかっています。無視できない数字です。
つまりHEICを読めないツールは実用外といっていいでしょう。でも、そんな悩みも一行で終わります。
register_heif_opener()
これだけでHEICが普通に扱えます。入れない理由がありません。
O(n²) → O(n) にしたアルゴリズム
最初に話に出てきた実装はこれです。なんだか素朴な感じがします。
「全部総当たり比較」です。
しかし80,000 * 80,000 * 1/2 = 32億回の比較
絶対無理としか思えませんでした。
そこで次に検討したのが、バケット分割です。
ph.hash >> 上位bit
これでバケット分割をすることで、256グループに分けて探索範囲を激減させました。これで億を超える比較を数百万回に減らすことができました。
結果として、ほぼO(n)となり速度アップに成功しました。
スコア方式で「残す方の写真」を決める
小手先のことは何もしません。
score = 解像度 + ファイルサイズ
これだけですが、実際に走らせてみるとこれが一番良い結果を出しました。
- 解像度が高い=情報が多い
- サイズが大きい=圧縮が弱い
ここが大事なのですが、「だいたい良い写真」を見つけることができます。
シンプル=速い=壊れない、これが私たちの真理です。
実際に83,923枚で回した実測結果
理論と実測は必ずしも一致しない、そういうことは私たちのような素人にはよくあることです。だから何度も検証しますし、なんだか検証している方が楽しくなってきたりします。
期待していた事:CPU使用率90%から100%で爆速
正直なところ、本当はCPUが全力で働いているところを見て笑いたかったです。
CPUは真っ赤、排風が熱風、ファンの音が部屋中に響く、そんなことを想定していました。
でも現実は違いました。
現実:CPUが暇している
まずは実測値から示したいと思います。開始から1時間、およそ5,500枚検知時の計測です。
- CPU使用率:0~15%
- CPU温度:32~35℃
- クロック:5,287MHz
- ファン:最大回転でも余裕がある
CPUが全然働いていない。アルゴリズムが軽すぎたきらいがあるようです。しかし、真犯人は別にいたのです。
真犯人はHDD(I/O wait地獄)
もっとばーっと進捗バーが進んでいるはず、CPUも悲鳴をあげているはず、しかし現実は鈍重でした。
検証として、Dドライブに1,000枚程のファイルをコピーして、走らせてみました。すると、驚くほど軽快に進み、323枚の重複を見つけてくれました。
EドライブとDドライブ、何が違うのかと考えましたが、すぐにわかりました。DドライブはSSDでEドライブは外付けHDDでした。
タスクマネージャーを見てみると、案の定外付けHDDは100%で死にそうになっていました。ディスク使用率100%に張り付いているということは、遅いのは完全にI/O待ちです。
つまりHDDの仕様も兼ねて、小ファイル大量処理*外付けHDDという最悪の組み合わせでした。
処理が止まっているのではなく、I/O待機でした。これは完全に失敗談です。
I/O待機に対する改善策
HDDの処理が間に合わなくてCPUがデータを送れない、そのためCPUの使用率が上がらなかったということでした。
もうこうなってしまっては、何を書き加えても、変更しても、スクリプトではどうしようもありません。ハードウェアの問題です。
最適解:SSDにファイルを置く
これだけで世界は変わります。アルゴリズム最適化よりもずっと効果があります。苦肉の策ですが、爆速とは言えないけれど体感で数倍速くなります。
- SSDへコピー
- Defender除外
- 同一ドライブ内でrename
これだけで最低でも倍は速さが変わります。またGPUも使えばいいんじゃないかという話もでましたが、あまり変わらない結果でした。
ここまでの失敗点(真犯人HDDボトルネック対策)
最初に走らせたときはエラーがでました。7時間かけて83,923枚、全て吹き飛びましたが、あきらめません。
元エラーの正体
TypeError: cannot use 'numpy.ndarray' as a dict key
dictキーは、・int、・str、・tuple、・frozensetみたいなhashable型のみでした。しかしnumpyはmutable、ハッシュ負荷でした。このため、100%落ちる結果になりました。
では修正するにはどうしたら?次のところに気を付けてみました。
- imagehash → 16進文字列 → int
- 完全不変
- pickle可能
- dictキー大丈夫
- XORで速度アップ
ph_int = int(str(ph), 16)
これは、imagehashが内部でおこなっていることをまねしてみました。仕様的にも正統派な感じがします。
HDDボトルネック対策
もうPhase 1で全部読み込むことにしました。
file_size = os.path.getsize(path)
with Image.open(path):
phash
size取得
I/Oは1回だけに圧縮してしまおうという魂胆です。
HDDはその特性から、シーク地獄やランダムアクセス最悪です。そのため、I/O回数制限で高速化を図りました。ストレージ最適化の基本原則に立ち返りました。
キャッシュ保存
pickle.dump(results)
さすがに7時間が吹き飛んだのは心にくるものがありました。実質チェックポイント再開ができるようになりました。これでエラーが起きてもPhase 2から始められます。
地味だけれどがんばるポイント
Hamming distanceをint XORで実装してみました。
bin(h1 ^ h2).count('1')
imagehash同士の比較より結構速くなっているのではないかと思っています。
上書き事故の禁止
safe_moveの衝突回避をしておきました。重複除去スクリプトで上書き事故はしゃれになりません。
きちんとプロテクトしておく方がいいと思いました。
検証結果で良かった点のまとめ
- pHashは驚くほど軽い
- HEICには簡単に完全対応
- ほぼO(n)で速さが出る
- DRY_RUNで安全走行
- 自動削除をしない安心設計
実用ツールとしては使えるレベルに達したと思います。
注意点と限界
やはり全ての重複画像を拾うことはできませんでした。もっと速度重視で100%一致のみのスクリプトを一度走らせてから、今回の榊式Photo Deduplicator.pyを走らすべきです。
バケット方式は理論上100%ではない
極端なケースで取りこぼす可能性がありました。しかしそれも±1探索でほぼ解決できたと思っています。
GPUは意味がない
本当の意味で意味が無いわけではないのですが、ボトルネックとなる原因がI/Oなので速くはなりません。
HDDは本当に遅い
HDDは本当に遅いです。大容量で安かったから買いましたが、シークタイム、回転待ち、毎秒120回転程度の限界、「7,200回転いいじゃん」とかダメでした。小さなファイルの大量処理、今回のような写真整理には本当に不向きでした。
公式リファレンス
- Python公式:https://www.python.org/
- Pillow:https://python-pillow.org/
- ImageHash:https://github.com/JohannesBuchner/imagehash
- pillow-heif:https://github.com/bigcat88/pillow_heif
ライブラリの信頼性も高く、メンテナンスの頻度も高いので長期運用可能です。
よくある質問
- Q完全一致ではなくpHash(知覚ハッシュ)を使う理由は何ですか?
- A
MD5やSHAなど完全一致ハッシュは「1bitでも違うと別物」と判定されてしまいます。しかし、写真整理では、リサイズ・圧縮率・EXIF変更・HEICからJPEG変換などが頻繁に起こるため、同じ写真でもバイナリが一致しないケースばかりでした。
その点、pHashは「見た目の類似度」で判定するため、先に述べた小さな違いも「同一画像」として検出できます。実用性、軽さ、速さの点で採用しました。
- QCPU使用率が低いのですが、PC性能が足りないのでしょうか?
- A
いいえ、正常だと思います。実測結果はさんざんなものでしたが、それはボトルネックとなっている部分がCPUではなく、ストレージ速度(I/O)にあるからです。SSDで処理してください。
- QHEIC(iPhoneの写真)も処理できますか?
- A
はい、処理できます。本スクリプトでは、pillow-heifを使用し、PillowからHEICを透過的に読み込めるようにしています。日本のスマホユーザーの48%はiPhoneのようなので、HEICを扱えないツールは致命的です。HEICの処理は必須といっていいです。
まとめ
色々話し合いましたが、今回のような知覚的な問題や怒涛の数量、つまり大量の写真の重複整理の最適解は次の三つにあると結論付けました。
- 軽いアルゴリズム
- I/O最適化
- 安全設計
高級ソフトもいいと思います。GPUで爆速もいいと思います。しかし、このブログの方針でもありますが、自分で作った道具が一番楽しいです。
同じように写真の重複で困っている方がいましたら、自分の環境に調整して使ってみてください。

コメント