Merge pull request #3320 from InfinityPacer/feature/subscribe

app/chain/subscribe.py

@@ -1,6 +1,7 @@
 import copy
 import random
 import time
+import threading
 from datetime import datetime
 from typing import Dict, List, Optional, Union, Tuple
 
@@ -28,15 +29,17 @@ from app.log import logger
 from app.schemas import NotExistMediaInfo, Notification, SubscrbieInfo, SubscribeEpisodeInfo, SubscribeDownloadFileInfo, \
     SubscribeLibraryFileInfo
 from app.schemas.types import MediaType, SystemConfigKey, MessageChannel, NotificationType, EventType
+from app.utils.singleton import Singleton
 
 
-class SubscribeChain(ChainBase):
+class SubscribeChain(ChainBase, metaclass=Singleton):
     """
     订阅管理处理链
     """
 
     def __init__(self):
         super().__init__()
+        self._rlock = threading.RLock()
         self.downloadchain = DownloadChain()
         self.downloadhis = DownloadHistoryOper()
         self.searchchain = SearchChain()
@@ -238,8 +241,11 @@ class SubscribeChain(ChainBase):
         :param manual: 是否手动搜索
         :return: 更新订阅状态为R或删除订阅
         """
+        with self._rlock:
+            logger.debug(f"search lock acquired at {datetime.now()}")
             if sid:
-            subscribes = [self.subscribeoper.get(sid)]
+                subscribe = self.subscribeoper.get(sid)
+                subscribes = [subscribe] if subscribe else []
             else:
                 subscribes = self.subscribeoper.list(state)
             # 遍历订阅
@@ -404,10 +410,14 @@ class SubscribeChain(ChainBase):
 
             # 手动触发时发送系统消息
             if manual:
+                if subscribes:
                     if sid:
                         self.message.put(f'{subscribes[0].name} 搜索完成！', title="订阅搜索", role="system")
                     else:
                         self.message.put('所有订阅搜索完成！', title="订阅搜索", role="system")
+                else:
+                    self.message.put('没有找到订阅！', title="订阅搜索", role="system")
+            logger.debug(f"search Lock released at {datetime.now()}")
 
     def update_subscribe_priority(self, subscribe: Subscribe, meta: MetaInfo,
                                   mediainfo: MediaInfo, downloads: List[Context]):
@@ -535,6 +545,8 @@ class SubscribeChain(ChainBase):
         # 记录重新识别过的种子
         _recognize_cached = []
 
+        with self._rlock:
+            logger.debug(f"match lock acquired at {datetime.now()}")
             # 所有订阅
             subscribes = self.subscribeoper.list('R')
             # 遍历订阅
@@ -779,6 +791,7 @@ class SubscribeChain(ChainBase):
                 # 判断是否要完成订阅
                 self.finish_subscribe_or_not(subscribe=subscribe, meta=meta, mediainfo=mediainfo,
                                              downloads=downloads, lefts=lefts)
+            logger.debug(f"match Lock released at {datetime.now()}")
 
     def check(self):
         """

app/db/subscribe_oper.py

@@ -83,6 +83,7 @@ class SubscribeOper(DbOper):
         更新订阅
         """
         subscribe = self.get(sid)
+        if subscribe:
             subscribe.update(self._db, payload)
         return subscribe