refactor(bluebubbles): 重构并新增多项蓝泡泡适配器功能
本次提交对蓝泡泡适配器进行了多维度改进： 1. 新增缓存超限自动清理逻辑，防止内存溢出 2. 增强流式打字同步功能，增加超时自动停止逻辑 3. 重构markdown渲染逻辑，修复格式匹配问题并支持更多语法 4. 新增配置校验工具函数，增加streaming_mode和auth_strategy的合法性校验 5. 新增多项环境配置参数，完善配置项 6. 重构初始化流程，新增多个工具类实例 7. 优化TTS禁用逻辑，新增aiohttp依赖检查 8. 重构流式消息发送逻辑，整合打字指示器逻辑 9. 新增健康状态检查的问题检测功能
This commit is contained in:
parent
fbe47b6393
commit
3cfb7dee25
@ -9,11 +9,15 @@ from typing import Any
|
||||
|
||||
from yuxi.channels.adapters.bluebubbles.accounts import parse_accounts_config
|
||||
from yuxi.channels.adapters.bluebubbles.action_gate import ActionGate
|
||||
from yuxi.channels.adapters.bluebubbles.catchup import CatchupCursorStore, CatchupSummary, run_full_catchup
|
||||
from yuxi.channels.adapters.bluebubbles.catchup import CatchupCursorStore, run_full_catchup
|
||||
from yuxi.channels.adapters.bluebubbles.client import BlueBubblesClient
|
||||
from yuxi.channels.adapters.bluebubbles.contact_resolver import resolve_contact
|
||||
from yuxi.channels.adapters.bluebubbles.conversation_bindings import ConversationBindings
|
||||
from yuxi.channels.adapters.bluebubbles.conversation_id import ConversationID
|
||||
from yuxi.channels.adapters.bluebubbles.conversation_route import ConversationRoute
|
||||
from yuxi.channels.adapters.bluebubbles.directory import list_chats, list_contacts, search_chats, search_contacts
|
||||
from yuxi.channels.adapters.bluebubbles.exceptions import BlueBubblesConnectionError, BlueBubblesHTTPError
|
||||
from yuxi.channels.adapters.bluebubbles.exec_approval import ExecApproval
|
||||
from yuxi.channels.adapters.bluebubbles.group_actions import (
|
||||
add_participant,
|
||||
leave_chat,
|
||||
@ -22,9 +26,9 @@ from yuxi.channels.adapters.bluebubbles.group_actions import (
|
||||
set_group_icon,
|
||||
)
|
||||
from yuxi.channels.adapters.bluebubbles.inbound_dedupe import InboundDedupe
|
||||
from yuxi.channels.adapters.bluebubbles.message_effects import send_with_effect
|
||||
from yuxi.channels.adapters.bluebubbles.monitor import BlueBubblesMonitor, InboundReorderBuffer
|
||||
from yuxi.channels.adapters.bluebubbles.monitor_debounce import MonitorDebounce
|
||||
from yuxi.channels.adapters.bluebubbles.message_effects import send_with_effect
|
||||
from yuxi.channels.adapters.bluebubbles.probe import (
|
||||
is_macos_26_or_higher,
|
||||
probe_imessage_login,
|
||||
@ -49,6 +53,9 @@ from yuxi.channels.adapters.bluebubbles.send import (
|
||||
)
|
||||
from yuxi.channels.adapters.bluebubbles.sent_cache import SentMessageCache
|
||||
from yuxi.channels.adapters.bluebubbles.session import resolve_chat_type
|
||||
from yuxi.channels.adapters.bluebubbles.session_route import SessionRoute
|
||||
from yuxi.channels.adapters.bluebubbles.status_issues import StatusIssues
|
||||
from yuxi.channels.adapters.bluebubbles.streaming_enhanced import StreamingTypingSync
|
||||
from yuxi.channels.adapters.bluebubbles.targets import resolve_chat_guid_with_auto_create
|
||||
from yuxi.channels.adapters.bluebubbles.voice import check_tts_available
|
||||
from yuxi.channels.adapters.bluebubbles.webhook_ingress import WebhookIngress
|
||||
@ -132,6 +139,13 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
group_actions = {"rename_group", "set_group_icon", "add_participant", "remove_participant", "leave_group"}
|
||||
if not group_actions.issubset(set(available)):
|
||||
base.group_management = False
|
||||
|
||||
tts_available = check_tts_available()
|
||||
if not tts_available["espeak"]:
|
||||
base.tts.voice.enabled = False
|
||||
elif not tts_available["ffmpeg"] and not tts_available["pydub"]:
|
||||
base.tts.voice.enabled = False
|
||||
|
||||
return base
|
||||
|
||||
meta = ChannelMeta(id="bluebubbles", label="BlueBubbles")
|
||||
@ -168,9 +182,13 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
self._health_monitor_config: dict = config.get("health_monitor", {}) if config else {}
|
||||
self._network_config: dict = config.get("network", {}) if config else {}
|
||||
self._allow_private_network: bool = (
|
||||
self._network_config.get("dangerously_allow_private_network", False)
|
||||
or config.get("allow_private_network", False)
|
||||
) if config else False
|
||||
(
|
||||
self._network_config.get("dangerously_allow_private_network", False)
|
||||
or config.get("allow_private_network", False)
|
||||
)
|
||||
if config
|
||||
else False
|
||||
)
|
||||
self._media_local_roots: list[str] = config.get("media_local_roots", []) if config else []
|
||||
self._require_mention: bool = config.get("require_mention", False) if config else False
|
||||
self._message_order_check: bool = config.get("message_order_check", True) if config else True
|
||||
@ -187,14 +205,18 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
self._catchup_task: asyncio.Task | None = None
|
||||
self._catchup_cursor_store: CatchupCursorStore | None = None
|
||||
self._sent_cache = SentMessageCache()
|
||||
self._typing_sync = StreamingTypingSync()
|
||||
self._conversation_bindings = ConversationBindings()
|
||||
self._conversation_id = ConversationID()
|
||||
self._conversation_route = ConversationRoute()
|
||||
self._session_route = SessionRoute()
|
||||
self._exec_approval = ExecApproval()
|
||||
self._status_issues = StatusIssues()
|
||||
|
||||
tts_available = check_tts_available()
|
||||
if not tts_available["espeak"]:
|
||||
self.capabilities.tts.voice.enabled = False
|
||||
logger.info("[BlueBubbles] TTS disabled: espeak not found")
|
||||
elif not tts_available["ffmpeg"] and not tts_available["pydub"]:
|
||||
self.capabilities.tts.voice.enabled = False
|
||||
logger.info("[BlueBubbles] TTS disabled: no transcoder (ffmpeg or pydub) found")
|
||||
try:
|
||||
import aiohttp # noqa: F401
|
||||
except ImportError:
|
||||
logger.info("[BlueBubbles] AI Vision unavailable: aiohttp not installed")
|
||||
|
||||
self._text_chunk_limit: int = config.get("text_chunk_limit", 4096) if config else 4096
|
||||
self._chunk_mode: str = config.get("chunk_mode", "newline") if config else "newline"
|
||||
@ -286,6 +308,7 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
)
|
||||
self._secret_contract.audit("password", "connect", {"server_url": self._server_url})
|
||||
await self._client.__aenter__()
|
||||
self._typing_sync.set_client(self._client)
|
||||
|
||||
health = await self.health_check()
|
||||
if health.status == "unhealthy":
|
||||
@ -382,22 +405,40 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
await account_client.__aenter__()
|
||||
self._account_clients[account_id] = account_client
|
||||
logger.info("[BlueBubbles] Connected account: %s", account_id)
|
||||
|
||||
account_monitor = BlueBubblesMonitor(
|
||||
server_url=account_cfg.server_url,
|
||||
password=account_cfg.password,
|
||||
on_event=self._handle_ws_event,
|
||||
on_disconnect=lambda aid=account_id: logger.info("[BlueBubbles] Account WS disconnected: %s", aid),
|
||||
on_reconnect=lambda aid=account_id: logger.info("[BlueBubbles] Account WS reconnected: %s", aid),
|
||||
)
|
||||
await account_monitor.start()
|
||||
self._account_monitors[account_id] = account_monitor
|
||||
logger.info("[BlueBubbles] Account WS monitor started: %s", account_id)
|
||||
except Exception:
|
||||
logger.warning("[BlueBubbles] Failed to connect account: %s", account_id)
|
||||
|
||||
async def disconnect(self) -> None:
|
||||
await self._typing_sync.stop_all()
|
||||
if self._catchup_task and not self._catchup_task.done():
|
||||
self._catchup_task.cancel()
|
||||
self._catchup_task = None
|
||||
self._catchup_cursor_store = None
|
||||
|
||||
for account_id, account_monitor in self._account_monitors.items():
|
||||
try:
|
||||
await account_monitor.stop()
|
||||
except Exception:
|
||||
logger.debug("[BlueBubbles] Error stopping account monitor: %s", account_id)
|
||||
self._account_monitors.clear()
|
||||
|
||||
for account_id, account_client in self._account_clients.items():
|
||||
try:
|
||||
await account_client.__aexit__()
|
||||
except Exception:
|
||||
logger.debug("[BlueBubbles] Error disconnecting account: %s", account_id)
|
||||
self._account_clients.clear()
|
||||
self._account_monitors.clear()
|
||||
|
||||
if self._monitor:
|
||||
await self._monitor.stop()
|
||||
@ -633,8 +674,9 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
except Exception as e:
|
||||
return DeliveryResult(success=False, error=str(e))
|
||||
|
||||
async def send_message_with_effect(self, chat_id: str, content: str, effect: str,
|
||||
reply_to: str | None = None) -> DeliveryResult:
|
||||
async def send_message_with_effect(
|
||||
self, chat_id: str, content: str, effect: str, reply_to: str | None = None
|
||||
) -> DeliveryResult:
|
||||
if not self._client:
|
||||
return DeliveryResult(success=False, error="Client not initialized")
|
||||
try:
|
||||
@ -675,43 +717,31 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
return await self.send(response)
|
||||
|
||||
key = f"{chat_id}:{msg_id}"
|
||||
coalesce_enabled = self._block_streaming_coalesce.get("enabled", False)
|
||||
max_flush_ms = self._block_streaming_coalesce.get("max_flush_interval_ms", 500)
|
||||
|
||||
if coalesce_enabled:
|
||||
self._stream_buffers[key] = self._stream_buffers.get(key, "") + chunk
|
||||
|
||||
now = time.monotonic()
|
||||
last = self._last_edit_at.get(key, 0)
|
||||
if not finished and now - last < max_flush_ms / 1000.0:
|
||||
return DeliveryResult(success=True, message_id=msg_id)
|
||||
|
||||
combined = self._stream_buffers.pop(key, "")
|
||||
prev_sent = self._stream_sent.get(key, "")
|
||||
full_content = prev_sent + combined
|
||||
self._stream_sent[key] = full_content
|
||||
|
||||
lane_content = self._lane_sent.get(key, "")
|
||||
display_content = self._compose_stream_content(full_content, lane_content, finished)
|
||||
|
||||
try:
|
||||
await update_activity(self._client, chat_id, msg_id, display_content)
|
||||
self._last_edit_at[key] = now
|
||||
if finished:
|
||||
self._cleanup_stream_key(key)
|
||||
return DeliveryResult(success=True, message_id=msg_id)
|
||||
except Exception as e:
|
||||
logger.warning(f"[BlueBubbles] Stream chunk update failed: {e}")
|
||||
return DeliveryResult(success=False, error=str(e))
|
||||
if self._streaming_typing_indicator:
|
||||
if finished:
|
||||
await self._typing_sync.stop_typing(chat_id)
|
||||
else:
|
||||
await self._typing_sync.start_typing(chat_id)
|
||||
|
||||
self._stream_buffers[key] = self._stream_buffers.get(key, "") + chunk
|
||||
|
||||
now = time.monotonic()
|
||||
last = self._last_edit_at.get(key, 0)
|
||||
interval_s = self._edit_interval_ms / 1000.0
|
||||
if not finished and now - last < interval_s:
|
||||
return DeliveryResult(success=True, message_id=msg_id)
|
||||
|
||||
coalesce_enabled = self._block_streaming_coalesce.get("enabled", False)
|
||||
if coalesce_enabled:
|
||||
max_flush_ms = self._block_streaming_coalesce.get("max_flush_interval_ms", 500)
|
||||
if not finished and now - last < max_flush_ms / 1000.0:
|
||||
return DeliveryResult(success=True, message_id=msg_id)
|
||||
else:
|
||||
interval_s = self._edit_interval_ms / 1000.0
|
||||
if not finished and now - last < interval_s:
|
||||
return DeliveryResult(success=True, message_id=msg_id)
|
||||
|
||||
return await self._flush_stream_chunk(key, chat_id, msg_id, finished)
|
||||
|
||||
async def _flush_stream_chunk(self, key: str, chat_id: str, msg_id: str, finished: bool) -> DeliveryResult:
|
||||
combined = self._stream_buffers.pop(key, "")
|
||||
prev_sent = self._stream_sent.get(key, "")
|
||||
full_content = prev_sent + combined
|
||||
@ -722,7 +752,7 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
|
||||
try:
|
||||
await update_activity(self._client, chat_id, msg_id, display_content)
|
||||
self._last_edit_at[key] = now
|
||||
self._last_edit_at[key] = time.monotonic()
|
||||
if finished:
|
||||
self._cleanup_stream_key(key)
|
||||
return DeliveryResult(success=True, message_id=msg_id)
|
||||
@ -1111,6 +1141,10 @@ class BlueBubblesAdapter(BaseChannelAdapter):
|
||||
hm_enabled = self._health_monitor_config.get("enabled", True)
|
||||
health.metadata["health_monitor"] = {"enabled": hm_enabled}
|
||||
|
||||
ws_connected = self._monitor is not None and getattr(self._monitor, "_connected", False)
|
||||
self._status_issues.check_connection(str(self._status.value), ws_connected)
|
||||
health.metadata["issues"] = self._status_issues.get_issues()
|
||||
|
||||
return health
|
||||
except Exception as e:
|
||||
return HealthStatus(
|
||||
|
||||
@ -106,6 +106,13 @@ class BlueBubblesConfig(BaseModel):
|
||||
auth_header_name: str = "X-BB-Password"
|
||||
tts_prefer_caf: bool = True
|
||||
|
||||
def __repr__(self) -> str:
|
||||
d = self.model_dump()
|
||||
if d.get("password"):
|
||||
d["password"] = "***"
|
||||
fields = ", ".join(f"{k}={v!r}" for k, v in d.items())
|
||||
return f"{self.__class__.__name__}({fields})"
|
||||
|
||||
@field_validator("allow_private_network", mode="before")
|
||||
@classmethod
|
||||
def _deprecate_allow_private_network(cls, v, info) -> Any:
|
||||
@ -116,6 +123,19 @@ class BlueBubblesConfig(BaseModel):
|
||||
)
|
||||
return v
|
||||
|
||||
@classmethod
|
||||
def _validate_literal(cls, value: str, valid_values: list[str], field_name: str, default: str) -> str:
|
||||
if value not in valid_values:
|
||||
logger.warning(
|
||||
"[BlueBubbles] Invalid value '%s' for '%s', falling back to '%s'. Valid values: %s",
|
||||
value,
|
||||
field_name,
|
||||
default,
|
||||
valid_values,
|
||||
)
|
||||
return default
|
||||
return value
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> BlueBubblesConfig:
|
||||
dm_raw = os.getenv("BLUEBUBBLES_DM_ALLOW_FROM", "")
|
||||
@ -137,7 +157,12 @@ class BlueBubblesConfig(BaseModel):
|
||||
media_max_mb=int(os.getenv("BLUEBUBBLES_MEDIA_MAX_MB", "100")),
|
||||
enable_stickers=os.getenv("BLUEBUBBLES_ENABLE_STICKERS", "true").lower() == "true",
|
||||
media_local_roots=[x.strip() for x in media_roots_raw.split(",") if x.strip()] if media_roots_raw else [],
|
||||
streaming_mode=os.getenv("BLUEBUBBLES_STREAMING_MODE", "partial"),
|
||||
streaming_mode=cls._validate_literal(
|
||||
os.getenv("BLUEBUBBLES_STREAMING_MODE", "partial"),
|
||||
["off", "partial", "block"],
|
||||
"streaming_mode",
|
||||
"partial",
|
||||
),
|
||||
edit_interval_ms=int(os.getenv("BLUEBUBBLES_EDIT_INTERVAL_MS", "500")),
|
||||
tapback_enabled=os.getenv("BLUEBUBBLES_TAPBACK_ENABLED", "true").lower() == "true",
|
||||
require_mention=os.getenv("BLUEBUBBLES_REQUIRE_MENTION", "false").lower() == "true",
|
||||
@ -147,6 +172,42 @@ class BlueBubblesConfig(BaseModel):
|
||||
allow_private_network=os.getenv("BLUEBUBBLES_ALLOW_PRIVATE_NETWORK", "false").lower() == "true",
|
||||
webhook_path=os.getenv("BLUEBUBBLES_WEBHOOK_PATH", "/bluebubbles/webhook"),
|
||||
webhook_secret=os.getenv("BLUEBUBBLES_WEBHOOK_SECRET", ""),
|
||||
auth_strategy=os.getenv("BLUEBUBBLES_AUTH_STRATEGY", "header"),
|
||||
auth_strategy=cls._validate_literal(
|
||||
os.getenv("BLUEBUBBLES_AUTH_STRATEGY", "header"),
|
||||
["header", "query", "both"],
|
||||
"auth_strategy",
|
||||
"header",
|
||||
),
|
||||
tts_prefer_caf=os.getenv("BLUEBUBBLES_TTS_PREFER_CAF", "true").lower() == "true",
|
||||
streaming_typing_indicator=os.getenv("BLUEBUBBLES_STREAMING_TYPING", "true").lower() == "true",
|
||||
message_order_check=os.getenv("BLUEBUBBLES_MESSAGE_ORDER_CHECK", "true").lower() == "true",
|
||||
max_cache_entries=int(os.getenv("BLUEBUBBLES_MAX_CACHE_ENTRIES", "2048")),
|
||||
dm_history_limit=int(os.getenv("BLUEBUBBLES_DM_HISTORY_LIMIT", "100")),
|
||||
coalesce_same_sender_dms=os.getenv("BLUEBUBBLES_COALESCE_DMS", "false").lower() == "true",
|
||||
enrich_group_participants_from_contacts=os.getenv("BLUEBUBBLES_ENRICH_GROUP", "true").lower() == "true",
|
||||
config_writes=os.getenv("BLUEBUBBLES_CONFIG_WRITES", "true").lower() == "true",
|
||||
block_streaming_coalesce=BlockStreamingCoalesceConfig(
|
||||
enabled=os.getenv("BLUEBUBBLES_COALESCE_ENABLED", "false").lower() == "true",
|
||||
max_flush_interval_ms=int(os.getenv("BLUEBUBBLES_COALESCE_MAX_FLUSH_MS", "500")),
|
||||
),
|
||||
network=NetworkConfig(
|
||||
dangerously_allow_private_network=os.getenv(
|
||||
"BLUEBUBBLES_NETWORK_DANGEROUSLY_ALLOW_PRIVATE", "false"
|
||||
).lower()
|
||||
== "true",
|
||||
),
|
||||
markdown=MarkdownConfig(
|
||||
enabled=os.getenv("BLUEBUBBLES_MARKDOWN_ENABLED", "true").lower() == "true",
|
||||
code_blocks=os.getenv("BLUEBUBBLES_MARKDOWN_CODE_BLOCKS", "true").lower() == "true",
|
||||
),
|
||||
catchup=CatchupConfig(
|
||||
enabled=os.getenv("BLUEBUBBLES_CATCHUP_ENABLED", "true").lower() == "true",
|
||||
max_age_minutes=int(os.getenv("BLUEBUBBLES_CATCHUP_MAX_AGE_MINUTES", "120")),
|
||||
per_run_limit=int(os.getenv("BLUEBUBBLES_CATCHUP_PER_RUN_LIMIT", "50")),
|
||||
first_run_lookback_minutes=int(os.getenv("BLUEBUBBLES_CATCHUP_FIRST_LOOKBACK", "30")),
|
||||
max_failure_retries=int(os.getenv("BLUEBUBBLES_CATCHUP_MAX_FAILURES", "10")),
|
||||
),
|
||||
health_monitor=HealthMonitorConfig(
|
||||
enabled=os.getenv("BLUEBUBBLES_HEALTH_MONITOR_ENABLED", "true").lower() == "true",
|
||||
),
|
||||
)
|
||||
|
||||
@ -24,3 +24,7 @@ class MonitorDebounce:
|
||||
expired = [k for k, v in self._cache.items() if now - v > self.WINDOW_SECONDS]
|
||||
for k in expired:
|
||||
del self._cache[k]
|
||||
if len(self._cache) > self.MAX_ENTRIES:
|
||||
oldest = sorted(self._cache.items(), key=lambda x: x[1])[: len(self._cache) - self.MAX_ENTRIES]
|
||||
for k, _ in oldest:
|
||||
del self._cache[k]
|
||||
|
||||
@ -77,8 +77,6 @@ def render_markdown_to_imessage(text: str) -> str:
|
||||
if not text:
|
||||
return ""
|
||||
|
||||
import re
|
||||
|
||||
codes: dict[str, str] = {}
|
||||
|
||||
def _save_code(m: re.Match) -> str:
|
||||
@ -88,12 +86,58 @@ def render_markdown_to_imessage(text: str) -> str:
|
||||
|
||||
text = re.sub(r"`[^`]+`", _save_code, text)
|
||||
|
||||
text = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", text)
|
||||
text = re.sub(r"__(.+?)__", r"<b>\1</b>", text)
|
||||
|
||||
text = re.sub(r"(?<!\*)\*(?!\*)(.+?)(?<!\*)\*(?!\*)", r"<i>\1</i>", text)
|
||||
tokens = _tokenize_markdown(text)
|
||||
result = _render_tokens(tokens)
|
||||
|
||||
for key, val in codes.items():
|
||||
text = text.replace(key, val)
|
||||
result = result.replace(key, val)
|
||||
|
||||
return text
|
||||
return result
|
||||
|
||||
|
||||
def _tokenize_markdown(text: str) -> list[tuple[str, str]]:
|
||||
TOKEN_RE = re.compile(r"(\*\*|__|(?<!\*)\*(?!\*)|(?<!\\)~{1,2})")
|
||||
tokens: list[tuple[str, str]] = []
|
||||
pos = 0
|
||||
for m in TOKEN_RE.finditer(text):
|
||||
if m.start() > pos:
|
||||
tokens.append(("text", text[pos : m.start()]))
|
||||
tokens.append(("marker", m.group()))
|
||||
pos = m.end()
|
||||
if pos < len(text):
|
||||
tokens.append(("text", text[pos:]))
|
||||
return tokens
|
||||
|
||||
|
||||
def _render_tokens(tokens: list[tuple[str, str]]) -> str:
|
||||
stack: list[tuple[str, str]] = []
|
||||
result: list[str] = []
|
||||
|
||||
MARKER_MAP = {
|
||||
"**": ("<b>", "</b>"),
|
||||
"__": ("<b>", "</b>"),
|
||||
"*": ("<i>", "</i>"),
|
||||
"~": ("<s>", "</s>"),
|
||||
"~~": ("<s>", "</s>"),
|
||||
}
|
||||
|
||||
for kind, value in tokens:
|
||||
if kind == "text":
|
||||
result.append(value)
|
||||
elif kind == "marker":
|
||||
if value in MARKER_MAP:
|
||||
if stack and stack[-1][0] == value:
|
||||
stack.pop()
|
||||
result.append(MARKER_MAP[value][1])
|
||||
else:
|
||||
stack.append((value, MARKER_MAP[value][0]))
|
||||
result.append(MARKER_MAP[value][0])
|
||||
else:
|
||||
result.append(value)
|
||||
|
||||
for marker, _ in reversed(stack):
|
||||
mapped = MARKER_MAP.get(marker)
|
||||
if mapped:
|
||||
result.append(mapped[1])
|
||||
|
||||
return "".join(result)
|
||||
|
||||
@ -12,13 +12,20 @@ from yuxi.utils.logging_config import logger
|
||||
|
||||
class StreamingTypingSync:
|
||||
TYPING_REFRESH_INTERVAL = 8.0
|
||||
DEFAULT_STREAM_TIMEOUT = 60.0
|
||||
|
||||
def __init__(self, client: BlueBubblesClient | None = None):
|
||||
def __init__(
|
||||
self,
|
||||
client: BlueBubblesClient | None = None,
|
||||
stream_timeout: float = DEFAULT_STREAM_TIMEOUT,
|
||||
):
|
||||
self._client = client
|
||||
self._active_chats: set[str] = set()
|
||||
self._last_typing: dict[str, float] = {}
|
||||
self._keepalive_task: asyncio.Task | None = None
|
||||
self._running = False
|
||||
self._stream_timeout = stream_timeout
|
||||
self._stream_started: dict[str, float] = {}
|
||||
|
||||
def set_client(self, client: BlueBubblesClient) -> None:
|
||||
self._client = client
|
||||
@ -28,6 +35,7 @@ class StreamingTypingSync:
|
||||
return
|
||||
|
||||
self._active_chats.add(chat_id)
|
||||
self._stream_started.setdefault(chat_id, time.monotonic())
|
||||
try:
|
||||
await send_typing_indicator(self._client, chat_id, display=True)
|
||||
self._last_typing[chat_id] = time.monotonic()
|
||||
@ -40,6 +48,7 @@ class StreamingTypingSync:
|
||||
|
||||
async def stop_typing(self, chat_id: str) -> None:
|
||||
self._active_chats.discard(chat_id)
|
||||
self._stream_started.pop(chat_id, None)
|
||||
if not self._client:
|
||||
return
|
||||
try:
|
||||
@ -74,6 +83,14 @@ class StreamingTypingSync:
|
||||
except Exception:
|
||||
logger.debug(f"[BlueBubbles] Typing keepalive failed for {chat_id}")
|
||||
|
||||
stream_started = self._stream_started.get(chat_id)
|
||||
if stream_started and (now - stream_started) > self._stream_timeout:
|
||||
logger.warning(
|
||||
f"[BlueBubbles] Stream timeout for {chat_id}, "
|
||||
f"({now - stream_started:.0f}s), auto-stopping typing"
|
||||
)
|
||||
await self.stop_typing(chat_id)
|
||||
|
||||
|
||||
async def send_stream_with_typing(
|
||||
adapter: Any,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user