optimize direct download I/O and regex compilation / 优化直接下载 I/O 并预编译正则表达式)

app/core/media/direct_downloader.py

@@ -3,6 +3,7 @@
 import time
 from typing import Optional
 
+import aiofiles
 import httpx
 
 from ...utils.logger import logger
@@ -58,12 +59,12 @@ async def _download_stream(self) -> None:
                         logger.error(f"Request Stream Failed, Status Code: {response.status_code}")
                         return
 
-                    with open(self.save_path, "wb") as f:
+                    async with aiofiles.open(self.save_path, "wb") as f:
                         async for chunk in response.aiter_bytes(self.chunk_size):
                             if self.stop_event.is_set():
                                 break
 
-                            f.write(chunk)
+                            await f.write(chunk)
                             self.total_bytes += len(chunk)
 
                             # Please don't remove this comment code

app/utils/utils.py

@@ -22,6 +22,32 @@
 OptionalStr = str | None
 OptionalDict = dict | None
 
+# Pre-compiled regular expressions for performance
+EMOJI_PATTERN = re.compile(
+    "["
+    "\U0001f1e0-\U0001f1ff"  # flags (iOS)
+    "\U0001f300-\U0001f5ff"  # symbols & pictographs
+    "\U0001f600-\U0001f64f"  # emoticons
+    "\U0001f680-\U0001f6ff"  # transport & map symbols
+    "\U0001f700-\U0001f77f"  # alchemical symbols
+    "\U0001f780-\U0001f7ff"  # Geometric Shapes Extended
+    "\U0001f800-\U0001f8ff"  # Supplemental Arrows-C
+    "\U0001f900-\U0001f9ff"  # Supplemental Symbols and Pictographs
+    "\U0001fa00-\U0001fa6f"  # Chess Symbols
+    "\U0001fa70-\U0001faff"  # Symbols and Pictographs Extended-A
+    "\U00002702-\U000027b0"  # Dingbats
+    "]+",
+    flags=re.UNICODE,
+)
+
+URL_PATTERN = re.compile(
+    r"^(https?://)" r"([a-zA-Z0-9-]+\.)+[a-zA-Z0-9]{1,6}" r"(:\d+)?" r"(/\S*)?$"
+)
+
+CONTAINS_URL_PATTERN = re.compile(
+    r"(?i)\bhttps?://" r"(?:[a-zA-Z0-9-]+\.)+[a-zA-Z0-9]{1,6}" r"(?::\d+)?" r"(?:/\S*)?"
+)
+
 
 def is_web_session_alive(page) -> bool:
     """Return True if the flet page session/connection is still healthy."""
@@ -109,23 +135,7 @@ def get_file_paths(directory: str) -> list:
 
 
 def remove_emojis(text: str, replace_text: str = "") -> str:
-    emoji_pattern = re.compile(
-        "["
-        "\U0001f1e0-\U0001f1ff"  # flags (iOS)
-        "\U0001f300-\U0001f5ff"  # symbols & pictographs
-        "\U0001f600-\U0001f64f"  # emoticons
-        "\U0001f680-\U0001f6ff"  # transport & map symbols
-        "\U0001f700-\U0001f77f"  # alchemical symbols
-        "\U0001f780-\U0001f7ff"  # Geometric Shapes Extended
-        "\U0001f800-\U0001f8ff"  # Supplemental Arrows-C
-        "\U0001f900-\U0001f9ff"  # Supplemental Symbols and Pictographs
-        "\U0001fa00-\U0001fa6f"  # Chess Symbols
-        "\U0001fa70-\U0001faff"  # Symbols and Pictographs Extended-A
-        "\U00002702-\U000027b0"  # Dingbats
-        "]+",
-        flags=re.UNICODE,
-    )
-    return emoji_pattern.sub(replace_text, text)
+    return EMOJI_PATTERN.sub(replace_text, text)
 
 
 def check_disk_capacity(file_path: str | Path, show: bool = False) -> float:
@@ -253,26 +263,14 @@ def is_valid_url(url):
         result = urlparse(url)
         if not all([result.scheme, result.netloc]):
             return False
-        url_pattern = re.compile(
-            r"^(https?://)"
-            r"([a-zA-Z0-9-]+\.)+[a-zA-Z0-9]{1,6}"
-            r"(:\d+)?"
-            r"(/\S*)?$"
-        )
-        return bool(url_pattern.match(url))
+        return bool(URL_PATTERN.match(url))
     except ValueError:
         return False
 
 
 def contains_url(text):
-    url_pattern = re.compile(
-        r"(?i)\bhttps?://"
-        r"(?:[a-zA-Z0-9-]+\.)+[a-zA-Z0-9]{1,6}"
-        r"(?::\d+)?"
-        r"(?:/\S*)?"
-    )
     try:
-        return bool(url_pattern.search(text))
+        return bool(CONTAINS_URL_PATTERN.search(text))
     except ValueError:
         return False