From e27aa700b28223767d448eab31c47e86bdff7eb9 Mon Sep 17 00:00:00 2001
From: Yufeng He <40085740+he-yufeng@users.noreply.github.com>
Date: Thu, 18 Jun 2026 21:33:06 +0800
Subject: [PATCH] fix(preprocess): handle Windows path_mapping rules without
 crashing

The preprocess stage split each path_mapping rule on a bare ":", which
raises "too many values to unpack" for Windows drive-letter rules
(e.g. "C:/a:D:/b") and "not enough values" for rules without a colon.
Delegate to path_Mapping, which already handles Windows/Linux/malformed
rules and is the helper the respond stage already uses.
---
 .../core/pipeline/preprocess_stage/stage.py   | 20 ++++------------
 tests/test_preprocess_stage.py                | 23 +++++++++++++++++++
 2 files changed, 28 insertions(+), 15 deletions(-)

diff --git a/astrbot/core/pipeline/preprocess_stage/stage.py b/astrbot/core/pipeline/preprocess_stage/stage.py
index c284600a62..22ce4bd9cc 100644
--- a/astrbot/core/pipeline/preprocess_stage/stage.py
+++ b/astrbot/core/pipeline/preprocess_stage/stage.py
@@ -12,9 +12,8 @@
     describe_media_ref,
     ensure_jpeg,
     ensure_wav,
-    file_uri_to_path,
-    is_file_uri,
 )
+from astrbot.core.utils.path_util import path_Mapping
 
 from ..context import PipelineContext
 from ..stage import Stage, register_stage
@@ -79,19 +78,10 @@ async def process(
 
             for idx, component in enumerate(message_chain):
                 if isinstance(component, Record | Image) and component.url:
-                    for mapping in mappings:
-                        from_, to_ = mapping.split(":")
-                        from_ = from_.removesuffix("/")
-                        to_ = to_.removesuffix("/")
-
-                        url = (
-                            file_uri_to_path(component.url)
-                            if is_file_uri(component.url)
-                            else component.url
-                        )
-                        if url.startswith(from_):
-                            component.url = url.replace(from_, to_, 1)
-                            logger.debug(f"路径映射: {url} -> {component.url}")
+                    # Delegate to path_Mapping so Windows drive-letter rules
+                    # (e.g. "C:/a:D:/b") and malformed rules are handled instead
+                    # of crashing on a bare ":" split, matching the respond stage.
+                    component.url = path_Mapping(mappings, component.url)
                     message_chain[idx] = component
 
         # Normalize provider-facing media early so downstream code sees local files.
diff --git a/tests/test_preprocess_stage.py b/tests/test_preprocess_stage.py
index 17ff048de7..4cde569b4e 100644
--- a/tests/test_preprocess_stage.py
+++ b/tests/test_preprocess_stage.py
@@ -100,3 +100,26 @@ async def test_preprocess_path_mapping_accepts_file_uri(tmp_path):
     image = event.get_messages()[0]
     assert isinstance(image, Image)
     assert image.file == image.path == image.url == str(target_image)
+
+
+@pytest.mark.asyncio
+async def test_preprocess_path_mapping_tolerates_multi_colon_rules(tmp_path):
+    from PIL import Image as PILImage
+
+    source_root = tmp_path / "source"
+    source_root.mkdir()
+    source_image = source_root / "photo.jpg"
+    PILImage.new("RGB", (2, 2), (255, 0, 0)).save(source_image)
+    event = FakeEvent([Image(file="", url=source_image.as_uri())])
+    stage = PreProcessStage()
+    stage.config = {}
+    # Windows drive-letter rules carry extra ":" (e.g. "C:/a:D:/b"); a bare
+    # ":" split would raise "too many values to unpack". The stage must tolerate
+    # such rules instead of crashing the whole pipeline.
+    stage.platform_settings = {"path_mapping": ["C:/remote:D:/local"]}
+    stage.stt_settings = {"enable": False}
+
+    await stage.process(event)
+
+    image = event.get_messages()[0]
+    assert isinstance(image, Image)