__init__.py 8.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266
  1. from __future__ import annotations
  2. import os
  3. import re
  4. import io
  5. import base64
  6. from io import BytesIO
  7. from pathlib import Path
  8. try:
  9. from PIL.Image import open as open_image, new as new_image
  10. from PIL.Image import FLIP_LEFT_RIGHT, ROTATE_180, ROTATE_270, ROTATE_90
  11. has_requirements = True
  12. except ImportError:
  13. has_requirements = False
  14. from ..typing import ImageType, Union, Image
  15. from ..errors import MissingRequirementsError
  16. ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'webp', 'svg'}
  17. EXTENSIONS_MAP: dict[str, str] = {
  18. "image/png": "png",
  19. "image/jpeg": "jpg",
  20. "image/gif": "gif",
  21. "image/webp": "webp",
  22. }
  23. def to_image(image: ImageType, is_svg: bool = False) -> Image:
  24. """
  25. Converts the input image to a PIL Image object.
  26. Args:
  27. image (Union[str, bytes, Image]): The input image.
  28. Returns:
  29. Image: The converted PIL Image object.
  30. """
  31. if not has_requirements:
  32. raise MissingRequirementsError('Install "pillow" package for images')
  33. if isinstance(image, str) and image.startswith("data:"):
  34. is_data_uri_an_image(image)
  35. image = extract_data_uri(image)
  36. if is_svg:
  37. try:
  38. import cairosvg
  39. except ImportError:
  40. raise MissingRequirementsError('Install "cairosvg" package for svg images')
  41. if not isinstance(image, bytes):
  42. image = image.read()
  43. buffer = BytesIO()
  44. cairosvg.svg2png(image, write_to=buffer)
  45. return open_image(buffer)
  46. if isinstance(image, bytes):
  47. is_accepted_format(image)
  48. return open_image(BytesIO(image))
  49. elif not isinstance(image, Image):
  50. image = open_image(image)
  51. image.load()
  52. return image
  53. return image
  54. def is_allowed_extension(filename: str) -> bool:
  55. """
  56. Checks if the given filename has an allowed extension.
  57. Args:
  58. filename (str): The filename to check.
  59. Returns:
  60. bool: True if the extension is allowed, False otherwise.
  61. """
  62. return '.' in filename and \
  63. filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
  64. def is_data_uri_an_media(data_uri: str) -> str:
  65. return is_data_an_wav(data_uri) or is_data_uri_an_image(data_uri)
  66. def is_data_an_wav(data_uri: str, filename: str = None) -> str:
  67. """
  68. Checks if the given data URI represents an image.
  69. Args:
  70. data_uri (str): The data URI to check.
  71. Raises:
  72. ValueError: If the data URI is invalid or the image format is not allowed.
  73. """
  74. if filename and filename.endswith(".wav"):
  75. return "audio/wav"
  76. # Check if the data URI starts with 'data:image' and contains an image format (e.g., jpeg, png, gif)
  77. if isinstance(data_uri, str) and re.match(r'data:audio/wav;base64,', data_uri):
  78. return "audio/wav"
  79. def is_data_uri_an_image(data_uri: str) -> bool:
  80. """
  81. Checks if the given data URI represents an image.
  82. Args:
  83. data_uri (str): The data URI to check.
  84. Raises:
  85. ValueError: If the data URI is invalid or the image format is not allowed.
  86. """
  87. # Check if the data URI starts with 'data:image' and contains an image format (e.g., jpeg, png, gif)
  88. if not re.match(r'data:image/(\w+);base64,', data_uri):
  89. raise ValueError("Invalid data URI image.")
  90. # Extract the image format from the data URI
  91. image_format = re.match(r'data:image/(\w+);base64,', data_uri).group(1).lower()
  92. # Check if the image format is one of the allowed formats (jpg, jpeg, png, gif)
  93. if image_format not in ALLOWED_EXTENSIONS and image_format != "svg+xml":
  94. raise ValueError("Invalid image format (from mime file type).")
  95. def is_accepted_format(binary_data: bytes) -> str:
  96. """
  97. Checks if the given binary data represents an image with an accepted format.
  98. Args:
  99. binary_data (bytes): The binary data to check.
  100. Raises:
  101. ValueError: If the image format is not allowed.
  102. """
  103. if binary_data.startswith(b'\xFF\xD8\xFF'):
  104. return "image/jpeg"
  105. elif binary_data.startswith(b'\x89PNG\r\n\x1a\n'):
  106. return "image/png"
  107. elif binary_data.startswith(b'GIF87a') or binary_data.startswith(b'GIF89a'):
  108. return "image/gif"
  109. elif binary_data.startswith(b'\x89JFIF') or binary_data.startswith(b'JFIF\x00'):
  110. return "image/jpeg"
  111. elif binary_data.startswith(b'\xFF\xD8'):
  112. return "image/jpeg"
  113. elif binary_data.startswith(b'RIFF') and binary_data[8:12] == b'WEBP':
  114. return "image/webp"
  115. else:
  116. raise ValueError("Invalid image format (from magic code).")
  117. def extract_data_uri(data_uri: str) -> bytes:
  118. """
  119. Extracts the binary data from the given data URI.
  120. Args:
  121. data_uri (str): The data URI.
  122. Returns:
  123. bytes: The extracted binary data.
  124. """
  125. data = data_uri.split(",")[-1]
  126. data = base64.b64decode(data)
  127. return data
  128. def get_orientation(image: Image) -> int:
  129. """
  130. Gets the orientation of the given image.
  131. Args:
  132. image (Image): The image.
  133. Returns:
  134. int: The orientation value.
  135. """
  136. exif_data = image.getexif() if hasattr(image, 'getexif') else image._getexif()
  137. if exif_data is not None:
  138. orientation = exif_data.get(274) # 274 corresponds to the orientation tag in EXIF
  139. if orientation is not None:
  140. return orientation
  141. def process_image(image: Image, new_width: int, new_height: int) -> Image:
  142. """
  143. Processes the given image by adjusting its orientation and resizing it.
  144. Args:
  145. image (Image): The image to process.
  146. new_width (int): The new width of the image.
  147. new_height (int): The new height of the image.
  148. Returns:
  149. Image: The processed image.
  150. """
  151. # Fix orientation
  152. orientation = get_orientation(image)
  153. if orientation:
  154. if orientation > 4:
  155. image = image.transpose(FLIP_LEFT_RIGHT)
  156. if orientation in [3, 4]:
  157. image = image.transpose(ROTATE_180)
  158. if orientation in [5, 6]:
  159. image = image.transpose(ROTATE_270)
  160. if orientation in [7, 8]:
  161. image = image.transpose(ROTATE_90)
  162. # Resize image
  163. image.thumbnail((new_width, new_height))
  164. # Remove transparency
  165. if image.mode == "RGBA":
  166. image.load()
  167. white = new_image('RGB', image.size, (255, 255, 255))
  168. white.paste(image, mask=image.split()[-1])
  169. return white
  170. # Convert to RGB for jpg format
  171. elif image.mode != "RGB":
  172. image = image.convert("RGB")
  173. return image
  174. def to_bytes(image: ImageType) -> bytes:
  175. """
  176. Converts the given image to bytes.
  177. Args:
  178. image (ImageType): The image to convert.
  179. Returns:
  180. bytes: The image as bytes.
  181. """
  182. if isinstance(image, bytes):
  183. return image
  184. elif isinstance(image, str) and image.startswith("data:"):
  185. is_data_uri_an_media(image)
  186. return extract_data_uri(image)
  187. elif isinstance(image, Image):
  188. bytes_io = BytesIO()
  189. image.save(bytes_io, image.format)
  190. image.seek(0)
  191. return bytes_io.getvalue()
  192. elif isinstance(image, (str, os.PathLike)):
  193. return Path(image).read_bytes()
  194. elif isinstance(image, Path):
  195. return image.read_bytes()
  196. else:
  197. try:
  198. image.seek(0)
  199. except (AttributeError, io.UnsupportedOperation):
  200. pass
  201. return image.read()
  202. def to_data_uri(image: ImageType) -> str:
  203. if not isinstance(image, str):
  204. data = to_bytes(image)
  205. data_base64 = base64.b64encode(data).decode()
  206. return f"data:{is_accepted_format(data)};base64,{data_base64}"
  207. return image
  208. class ImageDataResponse():
  209. def __init__(
  210. self,
  211. images: Union[str, list],
  212. alt: str,
  213. ):
  214. self.images = images
  215. self.alt = alt
  216. def get_list(self) -> list[str]:
  217. return [self.images] if isinstance(self.images, str) else self.images
  218. class ImageRequest:
  219. def __init__(
  220. self,
  221. options: dict = {}
  222. ):
  223. self.options = options
  224. def get(self, key: str):
  225. return self.options.get(key)