Lint, typo

This commit is contained in:
Zeecka
2025-12-26 22:58:00 +01:00
parent 796de9c576
commit 9fb4cd70b5
11 changed files with 80 additions and 50 deletions
+1 -1
View File
@@ -7,4 +7,4 @@ DB_URI=postgresql://aperiuser:aperipass@postgres:5432/aperisolve
RQ_DASHBOARD_REDIS_URL=redis://redis:6379/0 RQ_DASHBOARD_REDIS_URL=redis://redis:6379/0
POSTGRES_DB=aperisolve POSTGRES_DB=aperisolve
POSTGRES_USER=aperiuser POSTGRES_USER=aperiuser
POSTGRES_PASSWORD=aperipass POSTGRES_PASSWORD=aperipass
-1
View File
@@ -2,4 +2,3 @@
__pycache__/ __pycache__/
.mypy_cache/ .mypy_cache/
results/ results/
ihdr_crcs.db
+1 -1
View File
@@ -24,7 +24,7 @@ def analyze_foremost(input_img: Path, output_dir: Path) -> None:
timeout=MAX_PENDING_TIME, timeout=MAX_PENDING_TIME,
) )
# Note: foremost use stderr as standard output for results :o) # Note: foremost use stderr as standard output for results :o)
# So we can't use it as error dectection : # So we can't use it as error detection :
# stderr += data.stderr # stderr += data.stderr
# Zip extracted files # Zip extracted files
+1 -1
View File
@@ -11,7 +11,7 @@ from .utils import MAX_PENDING_TIME, update_data
def analyze_outguess( def analyze_outguess(
input_img: Path, output_dir: Path, password: Optional[str] = None input_img: Path, output_dir: Path, password: Optional[str] = None
) -> None: ) -> None:
"""Analyze an image submission using stehide.""" """Analyze an image submission using outguess."""
image_name = input_img.name image_name = input_img.name
try: try:
+4 -4
View File
@@ -7,7 +7,7 @@ from .utils import MAX_PENDING_TIME, update_data
def analyze_pngcheck(input_img: Path, output_dir: Path) -> None: def analyze_pngcheck(input_img: Path, output_dir: Path) -> None:
"""Analyze an image submission using strings.""" """Analyze an image submission using pngcheck."""
try: try:
data = subprocess.run( data = subprocess.run(
@@ -24,15 +24,15 @@ def analyze_pngcheck(input_img: Path, output_dir: Path) -> None:
# ) # )
# return # return
data_strings = data.stdout.split("\n") if data else [] data_pngcheck = data.stdout.split("\n") if data else []
data_strings = [s for s in data_strings if s] data_pngcheck = [s for s in data_pngcheck if s]
update_data( update_data(
output_dir, output_dir,
{ {
"pngcheck": { "pngcheck": {
"status": "ok", "status": "ok",
"output": data_strings, "output": data_pngcheck,
} }
}, },
) )
+1 -1
View File
@@ -12,7 +12,7 @@ from .utils import MAX_PENDING_TIME, update_data
def analyze_steghide( def analyze_steghide(
input_img: Path, output_dir: Path, password: Optional[str] = None input_img: Path, output_dir: Path, password: Optional[str] = None
) -> None: ) -> None:
"""Analyze an image submission using stehide.""" """Analyze an image submission using steghide."""
image_name = input_img.name image_name = input_img.name
try: try:
+43 -11
View File
@@ -15,9 +15,6 @@ _thread_lock = threading.Lock()
MAX_PENDING_TIME = int(os.getenv("MAX_PENDING_TIME", "600")) # 10 minutes by default 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
def update_data( def update_data(
output_dir: Path, new_data: dict[Any, Any], json_filename: str = "results.json" output_dir: Path, new_data: dict[Any, Any], json_filename: str = "results.json"
@@ -56,7 +53,31 @@ def update_data(
def get_resolutions(): def get_resolutions():
BASE_WIDTHS = ( """
Generate a sorted list of common display and image resolutions.
Creates a comprehensive set of resolution tuples by combining various base widths
with standard aspect ratios used in screens, digital displays, photography, and
paper standards. Includes both portrait and landscape orientations.
Base widths range from 16px to 10000px, covering:
- Low resolution: 16-256px (16px increments)
- Standard: 320-1024px (32px increments)
- HD/Full HD: 1280-2560px (64px increments)
- 4K: 3000-4096px (128px increments)
- 5K-8K: 5120-8192px (256px increments)
- Upper bound: 10000px
Aspect ratios include common formats:
- Digital screens: 1:1, 4:3, 3:2, 16:10, 16:9, 21:9
- Photography/print: 5:4, 7:5, 2:3
- Paper standards: ISO A-series, US Letter, Legal, Tabloid
Returns:
list[tuple[int, int]]: Sorted list of (width, height) resolution tuples
with heights constrained between 1 and 10000 pixels.
"""
base_widths = (
list(range(16, 257, 16)) # 16 → 256 list(range(16, 257, 16)) # 16 → 256
+ list(range(320, 1025, 32)) # 320 → 1024 + list(range(320, 1025, 32)) # 320 → 1024
+ list(range(1280, 2561, 64)) # 1280 → 2560 + list(range(1280, 2561, 64)) # 1280 → 2560
@@ -64,7 +85,7 @@ def get_resolutions():
+ list(range(5120, 8193, 256)) # 5K–8K + list(range(5120, 8193, 256)) # 5K–8K
+ [10000] # upper bound + [10000] # upper bound
) )
ASPECT_RATIOS = [ aspect_ratios = [
# Screens / digital # Screens / digital
(1, 1), # square (1, 1), # square
(4, 3), (4, 3),
@@ -84,8 +105,8 @@ def get_resolutions():
(17, 11), # Tabloid (17, 11), # Tabloid
] ]
resolutions = set() resolutions = set()
for w in BASE_WIDTHS: for w in base_widths:
for ar_w, ar_h in ASPECT_RATIOS: for ar_w, ar_h in aspect_ratios:
h = math.floor(round(w * ar_h / ar_w)) h = math.floor(round(w * ar_h / ar_w))
if 1 <= h <= 10000: if 1 <= h <= 10000:
resolutions.add((w, h)) resolutions.add((w, h))
@@ -94,14 +115,25 @@ def get_resolutions():
def get_valid_depth_color_pairs(): def get_valid_depth_color_pairs():
VALID_COLOR_DEPTHS = { """
Generate valid combinations of color and bit depth values.
Yields tuples of (depth, color) representing valid color mode and bit depth
combinations. Supports grayscale (0), RGB (2, 3, 4, 6) color modes with
appropriate bit depths (1, 2, 4, 8, 16 bits).
Yields:
tuple: A tuple of (depth, color) where depth is an integer representing
bit depth and color is an integer representing the color mode.
"""
valid_color_depth = {
0: [1, 2, 4, 8, 16], 0: [1, 2, 4, 8, 16],
2: [8, 16], 2: [8, 16],
3: [1, 2, 4, 8], 3: [1, 2, 4, 8],
4: [8, 16], 4: [8, 16],
6: [8, 16], 6: [8, 16],
} }
for color, depths in VALID_COLOR_DEPTHS.items(): for color, depths in valid_color_depth.items():
for depth in depths: for depth in depths:
yield depth, color yield depth, color
@@ -234,8 +266,8 @@ class PNG:
IndexError: If the bytearray is shorter than expected for the PNG structure. IndexError: If the bytearray is shorter than expected for the PNG structure.
""" """
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" png_signature = b"\x89PNG\r\n\x1a\n"
if data[:8] != PNG_SIGNATURE: if data[:8] != png_signature:
raise ValueError("Invalid PNG signature") raise ValueError("Invalid PNG signature")
offset = 8 # Read IHDR chunk offset = 8 # Read IHDR chunk
length = struct.unpack(">I", data[offset : offset + 4])[0] length = struct.unpack(">I", data[offset : offset + 4])[0]
+7 -7
View File
@@ -1,4 +1,4 @@
"""AperiSolve Flask application.""" """Aperi'Solve Flask application."""
import hashlib import hashlib
import json import json
@@ -41,7 +41,7 @@ def too_large(_: Any) -> tuple[Response, int]:
@app.errorhandler(404) @app.errorhandler(404)
def not_found(_: Any) -> str: def not_found(_: Any) -> str:
"""Error Handler for 404 not found.""" """Error Handler for 404 not found."""
return render_template("error.html", message="Ressource not found", statuscode=404) return render_template("error.html", message="Resource not found", statuscode=404)
@app.route("/") @app.route("/")
@@ -134,12 +134,12 @@ def upload_image() -> tuple[Response, int]:
first_submission_date=datetime.now(timezone.utc), first_submission_date=datetime.now(timezone.utc),
last_submission_date=datetime.now(timezone.utc), last_submission_date=datetime.now(timezone.utc),
) )
db.session.add(new_img) db.session.add(new_img) # pylint: disable=no-member
db.session.commit() db.session.commit() # pylint: disable=no-member
sub_img = Image.query.filter_by(hash=img_hash).first() # type: ignore sub_img = Image.query.filter_by(hash=img_hash).first() # type: ignore
sub_img.upload_count += 1 sub_img.upload_count += 1
db.session.commit() db.session.commit() # pylint: disable=no-member
# Create new Submission entry # Create new Submission entry
submission_path.mkdir(parents=True, exist_ok=True) submission_path.mkdir(parents=True, exist_ok=True)
@@ -152,9 +152,9 @@ def upload_image() -> tuple[Response, int]:
date=time.time(), date=time.time(),
image_hash=sub_img.hash, # type: ignore image_hash=sub_img.hash, # type: ignore
) )
db.session.add(submission) db.session.add(submission) # pylint: disable=no-member
db.session.commit() # Commit to save the new Image and Submission db.session.commit() # pylint: disable=no-member
# Start the analysis jobs # Start the analysis jobs
queue.enqueue("aperisolve.workers.analyze_image", submission.hash, job_timeout=300) queue.enqueue("aperisolve.workers.analyze_image", submission.hash, job_timeout=300)
+6 -6
View File
@@ -43,7 +43,7 @@ def cleanup_old_entries() -> None:
and now - submission.date > MAX_PENDING_TIME # type: ignore and now - submission.date > MAX_PENDING_TIME # type: ignore
): ):
# Processing took too long, delete # Processing took too long, delete
db.session.delete(submission) # type: ignore db.session.delete(submission) # pylint: disable=no-member
elif submission.status == "done": # type: ignore elif submission.status == "done": # type: ignore
# Search for buggy results, delete # Search for buggy results, delete
result_path = ( result_path = (
@@ -51,7 +51,7 @@ def cleanup_old_entries() -> None:
) )
result_file = result_path / "results.json" result_file = result_path / "results.json"
if not result_file.exists(): if not result_file.exists():
db.session.delete(submission) # type: ignore db.session.delete(submission) # pylint: disable=no-member
shutil.rmtree(result_path) shutil.rmtree(result_path)
# Delete "old" images # Delete "old" images
@@ -64,15 +64,15 @@ def cleanup_old_entries() -> None:
img_fold = RESULT_FOLDER / img.hash img_fold = RESULT_FOLDER / img.hash
if delay.total_seconds() > MAX_STORE_TIME: if delay.total_seconds() > MAX_STORE_TIME:
for s in img.submissions: for s in img.submissions:
db.session.delete(s) # type: ignore db.session.delete(s) # pylint: disable=no-member
if img_fold.exists(): if img_fold.exists():
shutil.rmtree(img_fold) # type: ignore shutil.rmtree(img_fold) # type: ignore
db.session.delete(img) db.session.delete(img) # pylint: disable=no-member
# Delete Images with missing submission # Delete Images with missing submission
if len(img.submissions) == 0 and delay.total_seconds() > MAX_PENDING_TIME: if len(img.submissions) == 0 and delay.total_seconds() > MAX_PENDING_TIME:
if img_fold.exists(): if img_fold.exists():
shutil.rmtree(img_fold) # type: ignore shutil.rmtree(img_fold) # type: ignore
db.session.delete(img) db.session.delete(img) # pylint: disable=no-member
db.session.commit() db.session.commit() # pylint: disable=no-member
+1 -1
View File
@@ -1,4 +1,4 @@
"""AperiSolve configuration variables.""" """Aperi'Solve configuration variables."""
from pathlib import Path from pathlib import Path
+15 -16
View File
@@ -2,7 +2,6 @@
import itertools import itertools
from datetime import datetime, timezone from datetime import datetime, timezone
from decimal import Decimal
from flask import Flask from flask import Flask
from flask_sqlalchemy import SQLAlchemy from flask_sqlalchemy import SQLAlchemy
@@ -16,21 +15,21 @@ db: SQLAlchemy = SQLAlchemy()
class Image(db.Model): # type: ignore class Image(db.Model): # type: ignore
"""Model representing an image file in the database.""" """Model representing an image file in the database."""
hash: Column[str] = Column( hash: Column[String] = Column(
String(64), primary_key=True, unique=True, nullable=False String(64), primary_key=True, unique=True, nullable=False
) )
file: Column[str] = Column(String(128), unique=True, nullable=False) file: Column[String] = Column(String(128), unique=True, nullable=False)
size: Column[int] = Column(Integer, nullable=False) size: Column[Integer] = Column(Integer, nullable=False)
first_submission_date: Column[datetime] = Column( first_submission_date: Column[DateTime] = Column(
DateTime, nullable=False, default=lambda: datetime.now(timezone.utc) DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)
) )
last_submission_date: Column[datetime] = Column( last_submission_date: Column[DateTime] = Column(
DateTime, DateTime,
nullable=False, nullable=False,
default=lambda: datetime.now(timezone.utc), default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc),
) )
upload_count: Column[int] = Column(Integer) upload_count: Column[Integer] = Column(Integer)
submissions = db.relationship("Submission", backref="image", lazy=True) submissions = db.relationship("Submission", backref="image", lazy=True)
@@ -40,14 +39,14 @@ class Submission(db.Model): # type: ignore
option. option.
""" """
hash: Column[str] = Column( hash: Column[String] = Column(
String(128), primary_key=True, unique=True, nullable=False String(128), primary_key=True, unique=True, nullable=False
) )
filename: Column[str] = Column(String(128), nullable=False) filename: Column[String] = Column(String(128), nullable=False)
password: Column[str] = Column(String(128)) password: Column[String] = Column(String(128))
deep_analysis: Column[bool] = Column(Boolean, default=False) deep_analysis: Column[Boolean] = Column(Boolean, default=False)
status: Column[str] = Column(String(20), default="pending") status: Column[String] = Column(String(20), default="pending")
date: Column[Decimal] = Column( date: Column[Float] = Column(
Float, nullable=False, default=lambda: datetime.now(timezone.utc) Float, nullable=False, default=lambda: datetime.now(timezone.utc)
) )
@@ -58,9 +57,9 @@ class Submission(db.Model): # type: ignore
class IHDR(db.Model): # type: ignore class IHDR(db.Model): # type: ignore
"""IHDR CRC lookup table""" """IHDR CRC lookup table"""
iid = Column(Integer, primary_key=True, autoincrement=True) iid: Column[Integer] = Column(Integer, primary_key=True, autoincrement=True)
crc = Column(BigInteger, nullable=False) crc: Column[BigInteger] = Column(BigInteger, nullable=False)
packed = Column(BigInteger, nullable=False) packed: Column[BigInteger] = Column(BigInteger, nullable=False)
def create_crc_db() -> None: def create_crc_db() -> None: