diff --git a/spug_api/apps/exec/transfer.py b/spug_api/apps/exec/transfer.py
index f893e30..42fcd5e 100644
--- a/spug_api/apps/exec/transfer.py
+++ b/spug_api/apps/exec/transfer.py
@@ -10,6 +10,7 @@ from apps.account.utils import has_host_perm
 from apps.host.models import Host
 from apps.setting.utils import AppSetting
 from libs import json_response, JsonParser, Argument, auth
+from libs.utils import str_decode
 from concurrent import futures
 from threading import Thread
 import subprocess
@@ -96,7 +97,7 @@ def _dispatch_sync(task):
         for t in futures.as_completed(threads):
             exc = t.exception()
             if exc:
-                rds.publish(t.token, json.dumps({'key': t.key, 'status': -1, 'data': f'Exception: {exc}'}))
+                rds.publish(t.token, json.dumps({'key': t.key, 'status': -1, 'data': f'\x1b[31mException: {exc}\x1b[0m'}))
     if task.host_id:
         command = f'umount -f {task.src_dir} && rm -rf {task.src_dir}'
     else:
@@ -120,7 +121,7 @@ def _do_sync(rds, task, host):
             message = task.stdout.readline()
             if not message:
                 break
-            message = message.decode().rstrip('\r\n')
+            message = str_decode(message).rstrip('\r\n')
             if 'rsync: command not found' in message:
                 data = '\r\n\x1b[31m检测到该主机未安装rsync，可通过批量执行/执行任务模块进行以下命令批量安装\x1b[0m'
                 data += '\r\nCentos/Redhat: yum install -y rsync'
diff --git a/spug_api/consumer/consumers.py b/spug_api/consumer/consumers.py
index 34715df..91b5554 100644
--- a/spug_api/consumer/consumers.py
+++ b/spug_api/consumer/consumers.py
@@ -7,6 +7,7 @@ from asgiref.sync import async_to_sync
 from apps.host.models import Host
 from consumer.utils import BaseConsumer
 from apps.account.utils import has_host_perm
+from libs.utils import str_decode
 from threading import Thread
 import time
 import json
@@ -64,10 +65,7 @@ class SSHConsumer(BaseConsumer):
             if not data:
                 self.close(3333)
                 break
-            try:
-                text = data.decode()
-            except UnicodeDecodeError:
-                text = data.decode(encoding='GBK', errors='ignore')
+            text = str_decode(data)
             if not is_ready:
                 self.send(text_data='\033[2J\033[3J\033[1;1H')
                 is_ready = True
@@ -138,7 +136,7 @@ class PubSubConsumer(BaseConsumer):
     def receive(self, **kwargs):
         response = self.p.get_message(timeout=10)
         while response:
-            data = response['data'].decode()
+            data = str_decode(response['data'])
             self.send(text_data=data)
             response = self.p.get_message(timeout=10)
         self.send(text_data='pong')
diff --git a/spug_api/libs/utils.py b/spug_api/libs/utils.py
index 05aff4f..788b3ce 100644
--- a/spug_api/libs/utils.py
+++ b/spug_api/libs/utils.py
@@ -37,6 +37,14 @@ def human_time(date=None):
     return date.strftime('%H:%M:%S')
 
 
+def str_decode(data):
+    try:
+        data = data.decode()
+    except UnicodeDecodeError:
+        data = data.decode(encoding='GBK', errors='ignore')
+    return data
+
+
 # 解析时间类型的数据
 def parse_time(value):
     if isinstance(value, datetime):