feat(weibo): 增加外部单协作台账

web/backend/services/weibo_collab_events.py

@@ -0,0 +1,82 @@
+"""微博外部单协作的进程内实时失效通知。"""
+
+from __future__ import annotations
+
+import asyncio
+import itertools
+import threading
+from datetime import UTC, datetime
+from typing import Any
+
+
+class CollabEventHub:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._revisions = itertools.count(1)
+        self._current_revision = 0
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
+
+    def subscribe(self) -> asyncio.Queue[dict[str, Any]]:
+        loop = asyncio.get_running_loop()
+        queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=100)
+        with self._lock:
+            self._loop = loop
+            self._subscribers.add(queue)
+        return queue
+
+    def unsubscribe(self, queue: asyncio.Queue[dict[str, Any]]) -> None:
+        with self._lock:
+            self._subscribers.discard(queue)
+
+    def ready_event(self) -> dict[str, Any]:
+        with self._lock:
+            return {"type": "collab.ready", "revision": self._current_revision}
+
+    def ping_event(self) -> dict[str, Any]:
+        with self._lock:
+            return {"type": "collab.ping", "revision": self._current_revision}
+
+    def publish(
+        self,
+        *,
+        scope: str,
+        action: str,
+        actor_id: int | None = None,
+    ) -> dict[str, Any]:
+        with self._lock:
+            revision = next(self._revisions)
+            self._current_revision = revision
+            loop = self._loop
+            subscribers = list(self._subscribers)
+        event = {
+            "type": "collab.invalidate",
+            "revision": revision,
+            "scope": scope,
+            "action": action,
+            "actor_id": actor_id,
+            "at": datetime.now(UTC).isoformat(),
+        }
+        if loop is None:
+            return event
+        for queue in subscribers:
+            try:
+                loop.call_soon_threadsafe(self._offer, queue, event)
+            except RuntimeError:
+                continue
+        return event
+
+    @staticmethod
+    def _offer(queue: asyncio.Queue[dict[str, Any]], event: dict[str, Any]) -> None:
+        while queue.full():
+            try:
+                queue.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+        try:
+            queue.put_nowait(event)
+        except asyncio.QueueFull:
+            pass
+
+
+collab_events = CollabEventHub()

web/backend/services/weibo_service.py

@@ -115,6 +115,7 @@ def refresh_weibo_account(
     account: WeiboAccount,
     *,
     activity: str = "",
+    commit: bool = True,
 ) -> WeiboHomeSnapshot:
     """刷新一个微博账号；失败时保留旧资料并记录本次检查状态。"""
 
@@ -126,12 +127,14 @@ def refresh_weibo_account(
         account.ck_check_status = _error_status(exc)
         account.ck_check_message = str(exc)
         account.ck_checked_at = _now()
-        db.commit()
+        if commit:
+            db.commit()
         raise
 
     _apply_snapshot(account, snapshot, account.cookie or "")
-    db.commit()
-    db.refresh(account)
+    if commit:
+        db.commit()
+        db.refresh(account)
     return snapshot