feat(cache): add HTTP cache support for image proxy

app/utils/crypto.py

@@ -6,8 +6,8 @@ from typing import Union
 from Crypto import Random
 from Crypto.Cipher import AES
 from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import serialization, hashes
-from cryptography.hazmat.primitives.asymmetric import rsa, padding as asym_padding
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding as asym_padding, rsa
 
 
 class RSAUtils:
@@ -97,7 +97,7 @@ class RSAUtils:
 
 class HashUtils:
     @staticmethod
-    def md5(data: str, encoding: str = "utf-8") -> str:
+    def md5(data: Union[str, bytes], encoding: str = "utf-8") -> str:
         """
         生成数据的MD5哈希值，并以字符串形式返回
 
@@ -105,11 +105,12 @@ class HashUtils:
         :param encoding: 字符串编码类型，默认使用UTF-8
         :return: 生成的MD5哈希字符串
         """
-        encoded_data = data.encode(encoding)
-        return hashlib.md5(encoded_data).hexdigest()
+        if isinstance(data, str):
+            data = data.encode(encoding)
+        return hashlib.md5(data).hexdigest()
 
     @staticmethod
-    def md5_bytes(data: str, encoding: str = "utf-8") -> bytes:
+    def md5_bytes(data: Union[str, bytes], encoding: str = "utf-8") -> bytes:
         """
         生成数据的MD5哈希值，并以字节形式返回
 
@@ -117,8 +118,9 @@ class HashUtils:
         :param encoding: 字符串编码类型，默认使用UTF-8
         :return: 生成的MD5哈希二进制数据
         """
-        encoded_data = data.encode(encoding)
-        return hashlib.md5(encoded_data).digest()
+        if isinstance(data, str):
+            data = data.encode(encoding)
+        return hashlib.md5(data).digest()
 
 
 class CryptoJsUtils:

app/utils/http.py

@@ -223,4 +223,53 @@ class RequestUtils:
                 cookie_dict[cstr[0].strip()] = cstr[1].strip()
         if array:
             return [{"name": k, "value": v} for k, v in cookie_dict.items()]
-        return cookie_dict
+        return cookie_dict
+
+    @staticmethod
+    def parse_cache_control(header: str) -> (str, int):
+        """
+        解析 Cache-Control 头，返回 cache_directive 和 max_age
+        :param header: Cache-Control 头部的字符串
+        :return: cache_directive 和 max_age
+        """
+        cache_directive = ""
+        max_age = None
+
+        if not header:
+            return cache_directive, max_age
+
+        directives = [directive.strip() for directive in header.split(",")]
+        for directive in directives:
+            if directive.startswith("max-age"):
+                try:
+                    max_age = int(directive.split("=")[1])
+                except Exception as e:
+                    logger.debug(f"Invalid max-age directive in Cache-Control header: {directive}, {e}")
+            elif directive in {"no-cache", "private", "public", "no-store", "must-revalidate"}:
+                cache_directive = directive
+
+        return cache_directive, max_age
+
+    @staticmethod
+    def generate_cache_headers(etag: Optional[str], cache_control: Optional[str] = "public",
+                               max_age: Optional[int] = 86400) -> dict:
+        """
+        生成 HTTP 响应的 ETag 和 Cache-Control 头
+        :param etag: 响应的 ETag 值。如果为 None，则不添加 ETag 头部。
+        :param cache_control: Cache-Control 指令，例如 "public"、"private" 等。默认为 "public"
+        :param max_age: Cache-Control 的 max-age 值（秒）。默认为 86400 秒（1天）
+        :return: HTTP 头部的字典
+        """
+        cache_headers = {}
+
+        if etag:
+            cache_headers["ETag"] = etag
+
+        if cache_control and max_age is not None:
+            cache_headers["Cache-Control"] = f"{cache_control}, max-age={max_age}"
+        elif cache_control:
+            cache_headers["Cache-Control"] = cache_control
+        elif max_age is not None:
+            cache_headers["Cache-Control"] = f"max-age={max_age}"
+
+        return cache_headers