hm_qc/checks/file_size_check.py

import os
import json

def run_check(config):
    """
    Check that all present images do not exceed their maximum allowed file size.

    Expected config:
    - working_dir (str): Directory where linkingrecord.json and images reside.
    - linkingrecord_filename (str): Name of the linking record file, default: linkingrecord.json

    Changes:
    - Deduplicate the list of failures by filename.

    Logic:
    - Determine asset type from conditions (viewtype, imagetype, experienceCondition).
    - Map asset type to a max size.
    - Check each existing image file. If any exceed max size, fail and list them.
    - Ignore missing files.
    - If conditions cannot determine asset type, return an error.
    """

    working_dir = config.get("working_dir", "working")
    linkingrecord_filename = config.get("linkingrecord_filename", "linkingrecord.json")
    linkingrecord_path = os.path.join(working_dir, linkingrecord_filename)

    if not os.path.exists(linkingrecord_path):
        return {
            "status": "error",
            "error_message": f"Linking record '{linkingrecord_filename}' not found in {working_dir}."
        }

    # Load the linking record
    with open(linkingrecord_path, 'r', encoding='utf-8') as f:
        linkingrecord = json.load(f)

    if "items" not in linkingrecord or not isinstance(linkingrecord["items"], list):
        return {
            "status": "error",
            "error_message": "Invalid linkingrecord.json: 'items' missing or not a list."
        }

    # Max sizes in KB
    max_sizes_kb = {
        "Base Exterior Images": 600,
        "Base Interior Images": 1024,  # 1MB
        "Engine and Transmission Images": 1024, # 1MB
        "Interior Layered Option Images": 1024,  # 1MB
        "Exterior Layered Option Images": 600,
        "Option Carousel Images": 500,
        "Powertrain Image": 400,
        "Showroom Images": 300,
        "Colour Chips": 50,
        "Bodystyle Images": 200,
        "Series Images": 300,
        "Trim Images": 700
    }

    def get_asset_type(conditions):
        viewtype = conditions.get("viewtype")
        imagetype = conditions.get("imagetype", None)
        experience = conditions.get("experienceCondition")

        # Determine asset type
        if viewtype == "interior" and imagetype == "layeroptint":
            return "Interior Layered Option Images"
        if viewtype == "exterior" and imagetype == "layeroptext":
            return "Exterior Layered Option Images"
        if experience == "2d-background" and viewtype in ["exterior", "interior"] and imagetype is None:
            return "Engine and Transmission Images"
        if viewtype == "exterior" and imagetype is None:
            return "Base Exterior Images"
        if viewtype == "interior" and imagetype is None:
            return "Base Interior Images"
        if viewtype == "carousel":
            if imagetype == "extra":
                return "Option Carousel Images"
            if imagetype == "powertrain":
                return "Powertrain Image"
            if imagetype == "colour":
                return "Colour Chips"
            if imagetype == "bodystyle":
                return "Bodystyle Images"
            if imagetype == "series":
                return "Series Images"
            if imagetype == "trim":
                return "Trim Images"
        if viewtype == "exterior" and imagetype == "showroom":
            return "Showroom Images"

        # No match
        return None

    # Use a dictionary to avoid duplicates (key by filename)
    failed_images_dict = {}

    # Iterate over items
    for item in linkingrecord["items"]:
        conditions = item.get("conditions", {})
        asset_type = get_asset_type(conditions)
        if asset_type is None:
            return {
                "status": "error",
                "error_message": f"Could not determine asset type for conditions: {conditions}"
            }

        max_size_kb = max_sizes_kb[asset_type]
        max_size_bytes = max_size_kb * 1024

        records = item.get("records", [])
        for record in records:
            assets = record.get("assets", [])
            for asset in assets:
                filename = asset.get("filename")
                if not filename:
                    continue
                file_path = os.path.join(working_dir, filename)
                if not os.path.exists(file_path):
                    # Ignore missing files
                    continue

                # Check file size
                try:
                    file_size = os.path.getsize(file_path)
                except OSError:
                    # If we can't get size, skip
                    continue

                if file_size > max_size_bytes:
                    if filename not in failed_images_dict:
                        failed_images_dict[filename] = {
                            "filename": filename,
                            "asset_type": asset_type,
                            "file_size_bytes": file_size,
                            "max_size_bytes": max_size_bytes
                        }

    if failed_images_dict:
        return {
            "status": "failed",
            "details": {
                "message": "Some files exceed the maximum allowed size.",
                "failed_images": list(failed_images_dict.values())
            }
        }

    return {
        "status": "passed",
        "details": {
            "message": "All present images are within the allowed file size."
        }
    }