mirror of
https://github.com/Benexl/FastAnime.git
synced 2025-12-05 20:40:09 -08:00
fix(cli-image-utils): pyright errors
This commit is contained in:
@@ -3,6 +3,8 @@ import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from viu_media.core.exceptions import DependencyNotFoundError
|
||||
import importlib.util
|
||||
|
||||
import click
|
||||
import httpx
|
||||
@@ -43,67 +45,74 @@ def resize_image_from_url(
|
||||
"""
|
||||
from io import BytesIO
|
||||
|
||||
from PIL import Image
|
||||
if importlib.util.find_spec("PIL"):
|
||||
from PIL import Image # pyright: ignore[reportMissingImports]
|
||||
|
||||
if not return_bytes and output_path is None:
|
||||
raise ValueError("output_path must be provided if return_bytes is False.")
|
||||
if not return_bytes and output_path is None:
|
||||
raise ValueError("output_path must be provided if return_bytes is False.")
|
||||
|
||||
try:
|
||||
# Use the provided synchronous client
|
||||
response = client.get(url)
|
||||
response.raise_for_status() # Raise an exception for bad status codes
|
||||
try:
|
||||
# Use the provided synchronous client
|
||||
response = client.get(url)
|
||||
response.raise_for_status() # Raise an exception for bad status codes
|
||||
|
||||
image_bytes = response.content
|
||||
image_stream = BytesIO(image_bytes)
|
||||
img = Image.open(image_stream)
|
||||
image_bytes = response.content
|
||||
image_stream = BytesIO(image_bytes)
|
||||
img = Image.open(image_stream)
|
||||
|
||||
if maintain_aspect_ratio:
|
||||
img_copy = img.copy()
|
||||
img_copy.thumbnail((new_width, new_height), Image.Resampling.LANCZOS)
|
||||
resized_img = img_copy
|
||||
else:
|
||||
resized_img = img.resize((new_width, new_height), Image.Resampling.LANCZOS)
|
||||
|
||||
if return_bytes:
|
||||
# Determine the output format. Default to JPEG if original is unknown or problematic.
|
||||
# Handle RGBA to RGB conversion for JPEG output.
|
||||
output_format = (
|
||||
img.format if img.format in ["JPEG", "PNG", "WEBP"] else "JPEG"
|
||||
)
|
||||
if output_format == "JPEG":
|
||||
if resized_img.mode in ("RGBA", "P"):
|
||||
resized_img = resized_img.convert("RGB")
|
||||
|
||||
byte_arr = BytesIO()
|
||||
resized_img.save(byte_arr, format=output_format)
|
||||
logger.info(
|
||||
f"Image from {url} resized to {resized_img.width}x{resized_img.height} and returned as bytes ({output_format} format)."
|
||||
)
|
||||
return byte_arr.getvalue()
|
||||
else:
|
||||
# Ensure the directory exists before saving
|
||||
if output_path:
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
resized_img.save(output_path)
|
||||
logger.info(
|
||||
f"Image from {url} resized to {resized_img.width}x{resized_img.height} and saved as '{output_path}'"
|
||||
if maintain_aspect_ratio:
|
||||
img_copy = img.copy()
|
||||
img_copy.thumbnail((new_width, new_height), Image.Resampling.LANCZOS)
|
||||
resized_img = img_copy
|
||||
else:
|
||||
resized_img = img.resize(
|
||||
(new_width, new_height), Image.Resampling.LANCZOS
|
||||
)
|
||||
return None
|
||||
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"An error occurred while requesting {url}: {e}")
|
||||
return None
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(
|
||||
f"HTTP error occurred: {e.response.status_code} - {e.response.text}"
|
||||
if return_bytes:
|
||||
# Determine the output format. Default to JPEG if original is unknown or problematic.
|
||||
# Handle RGBA to RGB conversion for JPEG output.
|
||||
output_format = (
|
||||
img.format if img.format in ["JPEG", "PNG", "WEBP"] else "JPEG"
|
||||
)
|
||||
if output_format == "JPEG":
|
||||
if resized_img.mode in ("RGBA", "P"):
|
||||
resized_img = resized_img.convert("RGB")
|
||||
|
||||
byte_arr = BytesIO()
|
||||
resized_img.save(byte_arr, format=output_format)
|
||||
logger.info(
|
||||
f"Image from {url} resized to {resized_img.width}x{resized_img.height} and returned as bytes ({output_format} format)."
|
||||
)
|
||||
return byte_arr.getvalue()
|
||||
else:
|
||||
# Ensure the directory exists before saving
|
||||
if output_path:
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
resized_img.save(output_path)
|
||||
logger.info(
|
||||
f"Image from {url} resized to {resized_img.width}x{resized_img.height} and saved as '{output_path}'"
|
||||
)
|
||||
return None
|
||||
|
||||
except httpx.RequestError as e:
|
||||
logger.error(f"An error occurred while requesting {url}: {e}")
|
||||
return None
|
||||
except httpx.HTTPStatusError as e:
|
||||
logger.error(
|
||||
f"HTTP error occurred: {e.response.status_code} - {e.response.text}"
|
||||
)
|
||||
return None
|
||||
except ValueError as e:
|
||||
logger.error(f"Configuration error: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"An unexpected error occurred: {e}")
|
||||
return None
|
||||
else:
|
||||
raise DependencyNotFoundError(
|
||||
"Pillow library is required for image processing. Please install it via 'uv pip install Pillow'."
|
||||
)
|
||||
return None
|
||||
except ValueError as e:
|
||||
logger.error(f"Configuration error: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"An unexpected error occurred: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def render(url: str, capture: bool = False, size: str = "30x30") -> Optional[str]:
|
||||
@@ -123,17 +132,12 @@ def render(url: str, capture: bool = False, size: str = "30x30") -> Optional[str
|
||||
If capture is False, prints directly to the terminal and returns None.
|
||||
Returns None on any failure.
|
||||
"""
|
||||
# --- Common subprocess arguments ---
|
||||
subprocess_kwargs = {
|
||||
"check": False, # We will handle errors manually
|
||||
"capture_output": capture,
|
||||
"text": capture, # Decode stdout/stderr as text if capturing
|
||||
}
|
||||
|
||||
# --- Try icat (Kitty terminal) first ---
|
||||
if icat_executable := shutil.which("icat"):
|
||||
process = subprocess.run(
|
||||
[icat_executable, "--align", "left", url], **subprocess_kwargs
|
||||
[icat_executable, "--align", "left", url],
|
||||
capture_output=capture,
|
||||
text=capture,
|
||||
)
|
||||
if process.returncode == 0:
|
||||
return process.stdout if capture else None
|
||||
@@ -148,11 +152,11 @@ def render(url: str, capture: bool = False, size: str = "30x30") -> Optional[str
|
||||
response.raise_for_status()
|
||||
img_bytes = response.content
|
||||
|
||||
# Add stdin input to the subprocess arguments
|
||||
subprocess_kwargs["input"] = img_bytes
|
||||
|
||||
process = subprocess.run(
|
||||
[chafa_executable, f"--size={size}", "-"], **subprocess_kwargs
|
||||
[chafa_executable, f"--size={size}", "-"],
|
||||
capture_output=capture,
|
||||
text=capture,
|
||||
input=img_bytes,
|
||||
)
|
||||
if process.returncode == 0:
|
||||
return process.stdout if capture else None
|
||||
|
||||
Reference in New Issue
Block a user