Immich cleanup - Delete assets duplicated by original HEIC file then remove any other duplicates caused by external import

remove_dupes.py

import requests


api_key = "foo"
headers = {
        'Accept': 'application/json',
        'x-api-key': api_key
    }
BASE_URL = 'https://website.com/api'


def get_duplicates():
    """
    Get a list of duplicate assets via the Immich API.
    For full API details, see:
        https://immich.app/docs/api/get-asset-duplicates
    """

    url = f"{BASE_URL}/duplicates"
    payload = {}

    return requests.request("GET", url, headers=headers, data=payload).json()


def delete_assets(asset_ids):
    """
    Delete a list of assets via the Immich API.

    The POST request is sent to the `/assets/delete` endpoint.
    The request body must include:
      - 'force': True (to force deletion)
      - 'ids': a list of asset UUIDs to be deleted.

    For full API details, see:
      https://immich.app/docs/api/delete-assets/
    """
    url = f"{BASE_URL}/assets"
    payload = {
        "force": True,
        "ids": asset_ids
    }
    response = requests.delete(url, headers=headers, json=payload)
    if response.status_code == 204:
        print(f"Successfully deleted assets: {asset_ids}")
    else:
        print(f"Error deleting assets {asset_ids}: {response.status_code} - {response.text}")


def split(arr, size):
    arrs = []
    while len(arr) > size:
        pice = arr[:size]
        arrs.append(pice)
        arr = arr[size:]
    arrs.append(arr)
    return arrs


def process_duplicates(duplicates):
    """
    Process each duplicate group from Immich.

    For every duplicate group retrieved:
      - Check if at least one asset has an originalMimeType of "image/heic".
      - If true, then build a list of asset IDs that do NOT have "image/heic" as their originalMimeType.
      - Call delete_assets() to delete these assets.
      - If no asset in the group is HEIC, skip deletion (to avoid removing all duplicates).
    """
    if not duplicates:
        return
    for_deletion = []

    # Loop through each duplicate group.
    for group in duplicates:
        # Expecting each group to have an "assets" list; fall back to group itself if not.
        assets = group.get("assets", group)

        # Check if the group contains at least one HEIC asset.
        has_heic = any(asset.get("originalMimeType") == "image/heic" for asset in assets)
        if has_heic:
            # Collect IDs of assets that are not HEIC.
            to_delete = [asset["id"] for asset in assets if asset.get("originalMimeType") != "image/heic"]
            if to_delete:
                # print(
                #     f"Deleting non-HEIC assets: {to_delete} from group with asset IDs: {[asset['id'] for asset in assets]}")
                #delete_assets(to_delete)
                for_deletion += to_delete
            else:
                print("All assets in this group are HEIC. Nothing to delete.")
        else:
            # Check if all assets have the same mime type.
            mime_types = {asset.get("originalMimeType") for asset in assets}
            if len(mime_types) == 1:
                # Check if any asset has deviceId of "Library Import".
                library_import_assets = [asset for asset in assets if asset.get("deviceId") == "Library Import"]
                if library_import_assets:
                    if len(library_import_assets) == 1:
                        to_delete = [asset["id"] for asset in library_import_assets]
                        #print(f"Deleting assets with deviceId 'Library Import': {to_delete}")
                        #delete_assets(to_delete)
                        for_deletion += to_delete
                    else:
                        print("Multiple assets with deviceId 'Library Import' found. Skipping deletion.")
                else:
                    print("No assets with deviceId 'Library Import' found. Nothing to delete.")
            else:
                print("Assets have different mime types. Skipping deletion.")

    # Split the list of assets to delete into chunks of 50.
    print(f"Deleting {len(for_deletion)} assets in chunks of 50.")
    for chunk in split(for_deletion, 50):
        delete_assets(chunk)


def run():
    duplicates = get_duplicates()
    process_duplicates(duplicates)

if __name__ == "__main__":
    run()