From 705e19701fb56a0ab4c36d69af2cc383ed691b95 Mon Sep 17 00:00:00 2001 From: Philip Zimmermann Date: Sat, 13 Dec 2025 16:37:46 +0100 Subject: [PATCH] Linter --- aperisolve/analyzers/image_resize.py | 17 ++++++----- aperisolve/analyzers/utils.py | 45 ++++++++++++++++++---------- 2 files changed, 39 insertions(+), 23 deletions(-) diff --git a/aperisolve/analyzers/image_resize.py b/aperisolve/analyzers/image_resize.py index 549d588..45a6194 100644 --- a/aperisolve/analyzers/image_resize.py +++ b/aperisolve/analyzers/image_resize.py @@ -3,6 +3,7 @@ import zlib import sqlite3 from pathlib import Path +from typing import List from .utils import update_data # Full list of expected sizes @@ -182,7 +183,7 @@ def calc_checksum( return bytearray((zlib.crc32(new_header) & 0xFFFFFFFF).to_bytes(4, byteorder="big")) -def lookup_crc(crc_bytes: bytes, logs: list) -> list: +def lookup_crc(crc_bytes: bytes, logs: List[str]) -> List[tuple[int, int]]: """ Queries the SQLite DB for the CRC and returns a list of (width, height) tuples. """ @@ -196,7 +197,7 @@ def lookup_crc(crc_bytes: bytes, logs: list) -> list: try: target_crc = int.from_bytes(crc_bytes, byteorder="big") except ValueError: - logs.append(f"Invalid CRC bytes: {crc_bytes}") + logs.append(f"Invalid CRC bytes: {crc_bytes!r}") return [] conn = sqlite3.connect(db_path) @@ -210,9 +211,9 @@ def lookup_crc(crc_bytes: bytes, logs: list) -> list: if results: logs.append(f"Database: Found {len(results)} match(es) for CRC {target_crc}") return results - else: - logs.append(f"Database: No match found for CRC {target_crc}") - return [] + + logs.append(f"Database: No match found for CRC {target_crc}") + return [] def analyze_image_resize(input_img: Path, output_dir: Path) -> None: @@ -269,7 +270,9 @@ def analyze_image_resize(input_img: Path, output_dir: Path) -> None: # Check if this size matches the file's CRC if target_crc == calc_checksum(header_chunk, width_bytes, height_bytes): match_found = True - logs.append(f"Success (Common List): Match found! Dimensions: {width}x{height}") + logs.append( + f"Success (Common List): Match found! Dimensions: {width}x{height}" + ) candidates.append((width, height)) break # Stop after first match @@ -314,4 +317,4 @@ def analyze_image_resize(input_img: Path, output_dir: Path) -> None: update_data(output_dir, output_data) except Exception as e: - update_data(output_dir, {"image_resize": {"status": "error", "error": str(e)}}) \ No newline at end of file + update_data(output_dir, {"image_resize": {"status": "error", "error": str(e)}}) diff --git a/aperisolve/analyzers/utils.py b/aperisolve/analyzers/utils.py index 693ea94..55fdbaa 100644 --- a/aperisolve/analyzers/utils.py +++ b/aperisolve/analyzers/utils.py @@ -14,6 +14,14 @@ _thread_lock = threading.Lock() MAX_PENDING_TIME = int(os.getenv("MAX_PENDING_TIME", "600")) # 10 minutes by default +# PATH CONFIGURATION +# Prioritize /data for Docker persistence, fallback to local directory for bare metal +DB_NAME = "ihdr_crcs.db" +if os.path.exists("/data"): + DB_PATH = os.path.join("/data", DB_NAME) +else: + DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), DB_NAME) + def update_data( output_dir: Path, new_data: dict[Any, Any], json_filename: str = "results.json" @@ -51,7 +59,11 @@ def update_data( fcntl.flock(lock, fcntl.LOCK_UN) -def create_crc_db(): +def create_crc_db() -> None: + """Creates and populates CRC lookup database with CRC, Width and Heigth tuples""" + if os.path.exists(DB_PATH): + print(f"Database already exists at {DB_PATH}. Skipping generation.") + return db_filename = "ihdr_crcs.db" # Connect to database @@ -64,51 +76,53 @@ def create_crc_db(): # Create table # We store the CRC as an integer for faster indexing/storage - cursor.execute(""" + cursor.execute( + """ CREATE TABLE IF NOT EXISTS ihdr ( crc INTEGER, width INTEGER, height INTEGER ) - """) + """ + ) # Drop index if exists (rebuilding it at the end is faster) cursor.execute("DROP INDEX IF EXISTS idx_crc") max_width = 2000 max_height = 2000 - W = range(1, max_width + 1) - H = range(1, max_height + 1) + w = range(1, max_width + 1) + h = range(1, max_height + 1) # Standard PNG IHDR parameters bit_depth = [1, 2, 4, 8, 16] color_type = [0, 2, 3, 4, 6] - compression_method = [0] # This is always 0 according to the PNG spec - filter_method = [0] # This is always 0 according to the PNG spec + compression_method = [0] # This is always 0 according to the PNG spec + filter_method = [0] # This is always 0 according to the PNG spec interlace_method = [0, 1] # Create the generator all_combinations = itertools.product( - W, H, bit_depth, color_type, compression_method, filter_method, interlace_method + w, h, bit_depth, color_type, compression_method, filter_method, interlace_method ) batch_size = 100000 batch = [] count = 0 - for w, h, bd, ct, comp, filt, inter in all_combinations: + for w_val, h_val, bd, ct, comp, filt, inter in all_combinations: # Construct IHDR chunk data ihdr_data = ( - w.to_bytes(4, "big") + - h.to_bytes(4, "big") + - bytes([bd, ct, comp, filt, inter]) + w_val.to_bytes(4, "big") + + h_val.to_bytes(4, "big") + + bytes([bd, ct, comp, filt, inter]) ) # Calculate CRC - crc = zlib.crc32(b"IHDR" + ihdr_data) & 0xffffffff + crc = zlib.crc32(b"IHDR" + ihdr_data) & 0xFFFFFFFF # Add to batch - batch.append((crc, w, h)) + batch.append((crc, w_val, h_val)) if len(batch) >= batch_size: cursor.executemany("INSERT INTO ihdr VALUES (?,?,?)", batch) @@ -129,5 +143,4 @@ def create_crc_db(): conn.commit() conn.close() - print(f"Database ready.") - return \ No newline at end of file + print("Database ready.")