wehub-resource-sync
680 lines
25 KiB
Python
680 lines
25 KiB
Python
"""
|
||
SMS app state accessor.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import re
|
||
from typing import Any
|
||
|
||
from bench_env.task.base import BaseApp
|
||
from bench_env.task.common_tasks import match_value
|
||
from bench_env.task.tencent_meeting.app import TencentMeeting
|
||
from bench_env.task.utils import norm
|
||
|
||
SMS_EXISTING_SENDERS = [
|
||
"华为云",
|
||
"抖音月付",
|
||
"10690955998",
|
||
"建设银行",
|
||
"中国电信",
|
||
"中国联通",
|
||
]
|
||
|
||
SMS_UNREAD_SENDERS = {
|
||
"华为云": "华为云",
|
||
"抖音月付": "抖音月付",
|
||
"建设银行": "建设银行",
|
||
"中国电信": "中国电信",
|
||
"中国联通": "中国联通",
|
||
}
|
||
|
||
SMS_NEW_RECIPIENTS = ["张三", "李四", "王五"]
|
||
|
||
SMS_RECIPIENT_PARAM = {
|
||
"type": "enum",
|
||
"values": SMS_NEW_RECIPIENTS,
|
||
"default": "张三",
|
||
"description": "短信收件人（真人联系人，发送到新会话）",
|
||
}
|
||
|
||
SMS_COMPARE_MESSAGE_COUNT_PAIRS = [
|
||
("中国电信", "中国联通"),
|
||
("中国电信", "10690955998"),
|
||
("华为云", "中国联通"),
|
||
]
|
||
|
||
SMS_SEND_CHANGES = [
|
||
"os.providers.sms.conversations",
|
||
"os.providers.sms.messagesByConversationId",
|
||
]
|
||
|
||
|
||
def _provider(input: Any, *, init: bool = False) -> dict:
|
||
"""Get SMS provider state from os.providers.sms."""
|
||
src = input.os_init if init else input.os
|
||
return (src.get("providers") or {}).get("sms") or {}
|
||
|
||
|
||
def _contacts_list(input: Any) -> list[dict[str, Any]]:
|
||
"""Extract contacts list from os.providers.contacts."""
|
||
try:
|
||
return (input.os.get("providers") or {}).get("contacts", {}).get("contacts") or []
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def sms_from_input(input: Any) -> "Sms":
|
||
"""Create Sms accessor with current + init provider state."""
|
||
return Sms(
|
||
_provider(input),
|
||
init=_provider(input, init=True),
|
||
contacts=_contacts_list(input),
|
||
)
|
||
|
||
|
||
def sms_init_from_input(input: Any) -> "Sms":
|
||
"""Create Sms accessor from init provider state only."""
|
||
return Sms(_provider(input, init=True))
|
||
|
||
|
||
class Sms(BaseApp):
|
||
"""
|
||
SMS state accessor.
|
||
|
||
Usage:
|
||
sms = Sms(input.apps["sms"])
|
||
sms.conversations
|
||
sms.messages_by_conversation_id
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
state: dict[str, Any],
|
||
init: dict[str, Any] | None = None,
|
||
contacts: list[dict[str, Any]] | None = None,
|
||
):
|
||
super().__init__(state, init=init)
|
||
self._contacts = contacts or []
|
||
|
||
def _phone_numbers_for_contact(self, name: str) -> list[str]:
|
||
"""Look up phone numbers for a contact by display name."""
|
||
target = norm(name)
|
||
numbers: list[str] = []
|
||
for contact in self._contacts:
|
||
if target != norm(str(contact.get("displayName") or "")):
|
||
continue
|
||
for phone in contact.get("phones") or []:
|
||
num = re.sub(r"[^\d+]", "", str(phone.get("number") or ""))
|
||
if num:
|
||
numbers.append(num)
|
||
return numbers
|
||
|
||
def _contact_names_for_phone(self, phone: str) -> list[str]:
|
||
"""通过号码反查联系人名，只接受规范化后的等价号码匹配。"""
|
||
target_variants = set(self._phone_variants(phone))
|
||
if not target_variants:
|
||
return []
|
||
|
||
names: list[str] = []
|
||
for contact in self._contacts:
|
||
display = str(contact.get("displayName") or "")
|
||
if not display:
|
||
continue
|
||
for ph in contact.get("phones") or []:
|
||
contact_variants = set(self._phone_variants(str(ph.get("number") or "")))
|
||
if target_variants & contact_variants:
|
||
names.append(display)
|
||
break
|
||
return names
|
||
|
||
@property
|
||
def conversations(self) -> list[dict[str, Any]]:
|
||
return self.get_list("conversations")
|
||
|
||
@property
|
||
def messages_by_conversation_id(self) -> dict[str, list[dict[str, Any]]]:
|
||
return self.get("messagesByConversationId", {})
|
||
|
||
@property
|
||
def unread_conversation_count(self) -> int:
|
||
return sum(1 for conversation in self.conversations if conversation["isUnread"])
|
||
|
||
@property
|
||
def all_conversations_read(self) -> bool:
|
||
return self.unread_conversation_count == 0
|
||
|
||
def latest_unread_sender(self) -> str:
|
||
for conversation in self.conversations:
|
||
if conversation.get("isUnread"):
|
||
return str(conversation["sender"])
|
||
raise ValueError("No unread conversations found")
|
||
|
||
def conversation_by_sender(self, sender_name: str) -> dict[str, Any]:
|
||
target = norm(sender_name)
|
||
for conversation in self.conversations:
|
||
if target == norm(conversation["sender"]):
|
||
return conversation
|
||
raise ValueError(f"Conversation for sender '{sender_name}' not found")
|
||
|
||
def messages_for_sender(self, sender_name: str) -> list[dict[str, Any]]:
|
||
conversation = self.conversation_by_sender(sender_name)
|
||
conversation_id = str(conversation["id"])
|
||
messages = self.messages_by_conversation_id[conversation_id]
|
||
if not messages:
|
||
raise ValueError(f"No messages found for sender '{sender_name}'")
|
||
return messages
|
||
|
||
def message_count_for(self, sender_name: str) -> int:
|
||
return len(self.messages_for_sender(sender_name))
|
||
|
||
def latest_message_from(self, sender_name: str) -> dict[str, Any]:
|
||
return self.messages_for_sender(sender_name)[-1]
|
||
|
||
def latest_incoming_message_from(self, sender_name: str) -> dict[str, Any]:
|
||
messages = self.messages_for_sender(sender_name)
|
||
for message in reversed(messages):
|
||
if not message["isOutgoing"]:
|
||
return message
|
||
raise ValueError(f"No incoming message found for sender '{sender_name}'")
|
||
|
||
def latest_incoming_content_from(self, sender_name: str) -> str:
|
||
return str(self.latest_incoming_message_from(sender_name)["content"])
|
||
|
||
@staticmethod
|
||
def _generated_conversation_id(sender_name: str) -> str:
|
||
return f"conv_{str(sender_name).encode('utf-8').hex()[:24]}"
|
||
|
||
def ensure_conversation_with_sender(self, sender_name: str) -> dict[str, Any]:
|
||
for conversation in self.conversations:
|
||
if norm(str(conversation.get("sender") or "")) == norm(sender_name):
|
||
return conversation
|
||
|
||
conversation = {
|
||
"id": self._generated_conversation_id(sender_name),
|
||
"sender": str(sender_name),
|
||
"timestamp": "",
|
||
"avatarColor": "#3482FF",
|
||
"avatarText": str(sender_name)[:1] or "#",
|
||
"isUnread": False,
|
||
"simSlot": 1,
|
||
"messageCount": 0,
|
||
}
|
||
self.conversations.insert(0, conversation)
|
||
self.messages_by_conversation_id[conversation["id"]] = []
|
||
return conversation
|
||
|
||
def prepare_state_with_incoming_message(
|
||
self,
|
||
sender_name: str,
|
||
content: str,
|
||
*,
|
||
message_id: str,
|
||
timestamp: str,
|
||
is_unread: bool = True,
|
||
) -> dict[str, Any]:
|
||
next_state = copy.deepcopy(self.raw)
|
||
next_state.setdefault("conversations", copy.deepcopy(self.get_list("conversations")))
|
||
next_state.setdefault(
|
||
"messagesByConversationId",
|
||
copy.deepcopy(self.get("messagesByConversationId", {})),
|
||
)
|
||
next_sms = Sms(next_state)
|
||
conversation = next_sms.ensure_conversation_with_sender(sender_name)
|
||
conversation_id = str(conversation["id"])
|
||
messages = list(next_sms.messages_by_conversation_id.get(conversation_id) or [])
|
||
messages.append(
|
||
{
|
||
"id": message_id,
|
||
"content": str(content),
|
||
"timestamp": str(timestamp),
|
||
"isOutgoing": False,
|
||
}
|
||
)
|
||
next_sms.messages_by_conversation_id[conversation_id] = messages
|
||
conversation["timestamp"] = str(timestamp)
|
||
conversation["isUnread"] = bool(is_unread)
|
||
conversation["messageCount"] = len(messages)
|
||
next_sms.raw["conversations"] = [
|
||
conversation,
|
||
*[
|
||
item
|
||
for item in next_sms.conversations
|
||
if str(item.get("id") or "") != conversation_id
|
||
],
|
||
]
|
||
return next_state
|
||
|
||
def sender_with_most_messages(self) -> str:
|
||
pairs = [
|
||
(conversation["sender"], self.message_count_for(conversation["sender"]))
|
||
for conversation in self.conversations
|
||
]
|
||
best_sender, best_count = max(pairs, key=lambda item: item[1])
|
||
if sum(1 for _, count in pairs if count == best_count) > 1:
|
||
raise ValueError("Multiple senders are tied for most messages")
|
||
return str(best_sender)
|
||
|
||
def sender_with_more_messages(self, sender1: str, sender2: str) -> str:
|
||
count1 = self.message_count_for(sender1)
|
||
count2 = self.message_count_for(sender2)
|
||
if count1 == count2:
|
||
raise ValueError(f"Message counts are tied for '{sender1}' and '{sender2}'")
|
||
return sender1 if count1 > count2 else sender2
|
||
|
||
@staticmethod
|
||
def sample_compare_pair(_env_state: dict[str, Any], rng: Any) -> dict[str, str]:
|
||
sender1, sender2 = rng.choice(SMS_COMPARE_MESSAGE_COUNT_PAIRS)
|
||
return {"sender1": sender1, "sender2": sender2}
|
||
|
||
def new_outgoing_messages_to(self, sender_name: str) -> list[dict[str, Any]]:
|
||
if not self.has_init:
|
||
raise ValueError("Init state is required to diff new outgoing messages")
|
||
|
||
current_ids = self.find_conversation_ids(sender_name)
|
||
if not current_ids:
|
||
return []
|
||
|
||
initial_map = self.init.messages_by_conversation_id
|
||
messages: list[dict[str, Any]] = []
|
||
for conversation_id in current_ids:
|
||
current_messages = self.messages_by_conversation_id[conversation_id]
|
||
initial_ids = {
|
||
str(message["id"])
|
||
for message in initial_map.get(conversation_id, [])
|
||
}
|
||
messages.extend(
|
||
message
|
||
for message in current_messages
|
||
if message["isOutgoing"] and str(message["id"]) not in initial_ids
|
||
)
|
||
return messages
|
||
|
||
def latest_message_to(self, sender_name: str) -> dict[str, Any] | None:
|
||
target = norm(sender_name)
|
||
best = None
|
||
for conversation in self.conversations:
|
||
if target and target not in norm(conversation.get("sender")):
|
||
continue
|
||
conversation_id = str(conversation.get("id") or "")
|
||
messages = self.messages_by_conversation_id.get(conversation_id) or []
|
||
if not messages:
|
||
continue
|
||
last = messages[-1]
|
||
if best is None:
|
||
best = last
|
||
continue
|
||
if (last.get("timestamp") or 0) >= (best.get("timestamp") or 0):
|
||
best = last
|
||
return best
|
||
|
||
def find_conversation_ids(self, sender_name: str) -> list[str]:
|
||
target = norm(sender_name)
|
||
ids: list[str] = []
|
||
for conversation in self.conversations:
|
||
if target and target not in norm(conversation.get("sender")):
|
||
continue
|
||
conversation_id = str(conversation.get("id") or "")
|
||
if conversation_id:
|
||
ids.append(conversation_id)
|
||
if ids:
|
||
return ids
|
||
|
||
# Fallback 1: match via conversation.phoneNumber field (set by contact selection flow)
|
||
phones = self._phone_numbers_for_contact(sender_name)
|
||
if phones:
|
||
for conversation in self.conversations:
|
||
conv_phone = re.sub(r"[^\d+]", "", str(conversation.get("phoneNumber") or ""))
|
||
if not conv_phone:
|
||
continue
|
||
for phone in phones:
|
||
if conv_phone == phone or conv_phone.endswith(phone) or phone.endswith(conv_phone):
|
||
conversation_id = str(conversation.get("id") or "")
|
||
if conversation_id:
|
||
ids.append(conversation_id)
|
||
break
|
||
if ids:
|
||
return ids
|
||
|
||
# Fallback 2: resolve contact name to phone numbers and match against sender
|
||
if not phones:
|
||
phones = []
|
||
for conversation in self.conversations:
|
||
sender = re.sub(r"[^\d+]", "", str(conversation.get("sender") or ""))
|
||
if not sender:
|
||
continue
|
||
for phone in phones:
|
||
if sender == phone or sender.endswith(phone) or phone.endswith(sender):
|
||
conversation_id = str(conversation.get("id") or "")
|
||
if conversation_id:
|
||
ids.append(conversation_id)
|
||
break
|
||
if ids:
|
||
return ids
|
||
|
||
# Fallback 3: 任务参数给的是手机号，但实际通过联系人选择后，会话 sender 是联系人名。
|
||
for contact_name in self._contact_names_for_phone(sender_name):
|
||
contact_target = norm(contact_name)
|
||
for conversation in self.conversations:
|
||
if contact_target and contact_target not in norm(conversation.get("sender")):
|
||
continue
|
||
conversation_id = str(conversation.get("id") or "")
|
||
if conversation_id and conversation_id not in ids:
|
||
ids.append(conversation_id)
|
||
return ids
|
||
|
||
def new_outgoing_message(
|
||
self, sender_name: str, expected_message: str
|
||
) -> dict[str, Any] | None:
|
||
expected = str(expected_message or "")
|
||
|
||
def _sms_norm(value: str) -> str:
|
||
return re.sub(r"[\s\.\,\!\?，。！？；;：:]", "", value or "")
|
||
|
||
current_map = self.messages_by_conversation_id or {}
|
||
try:
|
||
initial_map = self.init.messages_by_conversation_id or {}
|
||
except Exception:
|
||
initial_map = {}
|
||
|
||
for conversation_id in self.find_conversation_ids(sender_name):
|
||
current_messages = current_map.get(conversation_id) or []
|
||
initial_messages = initial_map.get(conversation_id) or []
|
||
initial_ids = {
|
||
str((message or {}).get("id") or "")
|
||
for message in initial_messages
|
||
if isinstance(message, dict)
|
||
}
|
||
for message in current_messages:
|
||
if not isinstance(message, dict):
|
||
continue
|
||
message_id = str(message.get("id") or "")
|
||
if message_id and message_id in initial_ids:
|
||
continue
|
||
if not bool(message.get("isOutgoing")):
|
||
continue
|
||
content = str(message.get("content") or "")
|
||
if expected in content or _sms_norm(expected) == _sms_norm(content):
|
||
return message
|
||
return None
|
||
|
||
def check_new_sent_to(
|
||
self, sender_name: str, *keywords: str, field: str | None = None
|
||
) -> dict[str, Any]:
|
||
if field is None:
|
||
field = f"sent_to_{sender_name}"
|
||
|
||
messages = self.new_outgoing_messages_to(sender_name)
|
||
matched = next(
|
||
(
|
||
message
|
||
for message in messages
|
||
if all(keyword in str(message["content"]) for keyword in keywords)
|
||
),
|
||
None,
|
||
)
|
||
actual = str(matched["content"]) if matched else None
|
||
return {
|
||
"field": field,
|
||
"expected": f"new message to '{sender_name}' containing {list(keywords)}",
|
||
"actual": actual,
|
||
"passed": matched is not None,
|
||
}
|
||
|
||
def check_new_sent_any_of(
|
||
self,
|
||
sender_name: str,
|
||
labels: list[str],
|
||
*extra_keywords: str,
|
||
field: str | None = None,
|
||
) -> dict[str, Any]:
|
||
if field is None:
|
||
field = f"sent_to_{sender_name}"
|
||
|
||
messages = self.new_outgoing_messages_to(sender_name)
|
||
matched = next(
|
||
(
|
||
message
|
||
for message in messages
|
||
if any(label and label in str(message.get("content") or "") for label in labels)
|
||
and all(keyword in str(message.get("content") or "") for keyword in extra_keywords)
|
||
),
|
||
None,
|
||
)
|
||
actual = str(matched.get("content") or "") if matched else None
|
||
return {
|
||
"field": field,
|
||
"expected": {
|
||
"recipient": sender_name,
|
||
"any_of": labels,
|
||
"keywords": list(extra_keywords),
|
||
},
|
||
"actual": actual,
|
||
"passed": matched is not None,
|
||
}
|
||
|
||
def check_new_sent_contains_number(
|
||
self,
|
||
sender_name: str,
|
||
expected: float,
|
||
*extra_keywords: str,
|
||
field: str | None = None,
|
||
) -> dict[str, Any]:
|
||
if field is None:
|
||
field = f"sent_to_{sender_name}_number"
|
||
|
||
messages = self.new_outgoing_messages_to(sender_name)
|
||
matched = next(
|
||
(
|
||
message
|
||
for message in messages
|
||
if match_value(expected, str(message.get("content") or ""))
|
||
and all(keyword in str(message.get("content") or "") for keyword in extra_keywords)
|
||
),
|
||
None,
|
||
)
|
||
actual = str(matched.get("content") or "") if matched else None
|
||
return {
|
||
"field": field,
|
||
"expected": {
|
||
"recipient": sender_name,
|
||
"number": expected,
|
||
"keywords": list(extra_keywords),
|
||
},
|
||
"actual": actual,
|
||
"passed": matched is not None,
|
||
}
|
||
|
||
@staticmethod
|
||
def _normalize_phone_for_match(value: str) -> str:
|
||
return "".join(ch for ch in str(value or "") if ch.isdigit())
|
||
|
||
@staticmethod
|
||
def _phone_variants(phone: str) -> list[str]:
|
||
"""生成电话号码的国内/国际两种纯数字形式，用于宽松匹配。
|
||
|
||
固话（国内带前导 0，国际不带）：
|
||
国内 010 8854 4114 → 01088544114 (11 位，0 开头)
|
||
国际 +86 10 8854 4114 → 861088544114 (12 位，86+10 位)
|
||
手机（国内不带 0，国际加 86）：
|
||
国内 138 0000 0000 → 13800000000 (11 位，1 开头)
|
||
国际 +86 138 0000 0000 → 8613800000000 (13 位，86+11 位)
|
||
|
||
过短的数字串（< 7 位）不产生任何变体，避免 substring 匹配误过。
|
||
"""
|
||
digits = "".join(ch for ch in str(phone or "") if ch.isdigit())
|
||
if len(digits) < 7:
|
||
return []
|
||
variants: set[str] = {digits}
|
||
if digits.startswith("86"):
|
||
rest = digits[2:]
|
||
if len(rest) == 10: # 国际固话 → 国内补 "0"
|
||
variants.add("0" + rest)
|
||
elif len(rest) == 11 and rest[0] == "1": # 国际手机 → 国内直接去掉 86
|
||
variants.add(rest)
|
||
elif len(digits) == 11:
|
||
if digits[0] == "0": # 国内固话 → 国际去 "0" 加 "86"
|
||
variants.add("86" + digits[1:])
|
||
elif digits[0] == "1": # 国内手机 → 国际加 "86"
|
||
variants.add("86" + digits)
|
||
return list(variants)
|
||
|
||
def check_new_sent_contains_phone(
|
||
self,
|
||
sender_name: str,
|
||
phone: str,
|
||
*,
|
||
field: str | None = None,
|
||
) -> dict[str, Any]:
|
||
if field is None:
|
||
field = f"sent_phone_to_{sender_name}"
|
||
|
||
variants = self._phone_variants(phone)
|
||
messages = self.new_outgoing_messages_to(sender_name)
|
||
matched = next(
|
||
(
|
||
message
|
||
for message in messages
|
||
if variants
|
||
and any(
|
||
v in self._normalize_phone_for_match(str(message.get("content") or ""))
|
||
for v in variants
|
||
)
|
||
),
|
||
None,
|
||
)
|
||
actual = str(matched.get("content") or "") if matched else None
|
||
return {
|
||
"field": field,
|
||
"expected": f"new message to '{sender_name}' containing phone {phone}",
|
||
"actual": actual,
|
||
"passed": matched is not None,
|
||
}
|
||
|
||
def check_no_new_sent_to(
|
||
self, sender_name: str, *, field: str | None = None
|
||
) -> dict[str, Any]:
|
||
if field is None:
|
||
field = f"no_new_sent_to_{sender_name}"
|
||
messages = self.new_outgoing_messages_to(sender_name)
|
||
return {
|
||
"field": field,
|
||
"expected": f"no new message to '{sender_name}'",
|
||
"actual": [str(message.get("content") or "") for message in messages],
|
||
"passed": len(messages) == 0,
|
||
}
|
||
|
||
def check_all_read(self, field: str = "all_read") -> dict[str, Any]:
|
||
return {
|
||
"field": field,
|
||
"expected": True,
|
||
"actual": self.all_conversations_read,
|
||
"passed": self.all_conversations_read,
|
||
}
|
||
|
||
def senders_with_keyword(self, keyword: str) -> list[str]:
|
||
"""返回所有收到过含 keyword 消息的会话 sender 名称列表。"""
|
||
init_map = self.init.messages_by_conversation_id if self.has_init else {}
|
||
curr_map = self.messages_by_conversation_id
|
||
senders: list[str] = []
|
||
for conv in self.conversations:
|
||
cid = str(conv["id"])
|
||
all_msgs = list(init_map.get(cid) or []) + list(curr_map.get(cid) or [])
|
||
has_kw = any(
|
||
not m.get("isOutgoing") and keyword in str(m.get("content") or "")
|
||
for m in all_msgs
|
||
if isinstance(m, dict)
|
||
)
|
||
if has_kw:
|
||
senders.append(str(conv["sender"]))
|
||
return senders
|
||
|
||
def check_replied_to_all(
|
||
self,
|
||
senders: list[str],
|
||
reply: str,
|
||
*,
|
||
field: str = "replied_to_all",
|
||
) -> dict[str, Any]:
|
||
"""验证是否给所有指定 sender 都发了包含 reply 的新消息。"""
|
||
sent_count = sum(
|
||
1 for sender in senders
|
||
if self.new_outgoing_messages_to(sender)
|
||
and any(reply in str(m["content"]) for m in self.new_outgoing_messages_to(sender))
|
||
)
|
||
return {
|
||
"field": field,
|
||
"expected": f"reply '{reply}' to all {len(senders)} senders",
|
||
"actual": f"{sent_count}/{len(senders)}",
|
||
"passed": sent_count == len(senders) and len(senders) > 0,
|
||
}
|
||
|
||
def check_conversation_deleted(self, sender_name: str, field: str = "deleted") -> dict[str, Any]:
|
||
target = norm(sender_name)
|
||
still_exists = any(target == norm(c["sender"]) for c in self.conversations)
|
||
return {
|
||
"field": field,
|
||
"expected": f"conversation '{sender_name}' deleted",
|
||
"actual": f"conversation '{sender_name}' still exists" if still_exists else f"conversation '{sender_name}' deleted",
|
||
"passed": not still_exists,
|
||
}
|
||
|
||
def check_new_outgoing_contains_meeting_id(
|
||
self,
|
||
recipient_display: str,
|
||
meeting_id: str,
|
||
*,
|
||
field: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""最新一条新发往外短信正文中含规范化会议号（忽略空格，兼容 digit 间空白 pattern）。"""
|
||
if field is None:
|
||
field = f"sms_meeting_{recipient_display}"
|
||
mid = re.sub(r"\s+", "", str(meeting_id))
|
||
msgs = self.new_outgoing_messages_to(recipient_display)
|
||
text = str(msgs[-1]["content"]) if msgs else ""
|
||
if not mid:
|
||
return {
|
||
"field": field,
|
||
"expected": f"短信含会议号",
|
||
"actual": text[:200] or "(none)",
|
||
"passed": False,
|
||
}
|
||
pat = TencentMeeting.meeting_id_pattern(mid)
|
||
passed = bool(msgs) and bool(pat.search(text))
|
||
return {
|
||
"field": field,
|
||
"expected": f"短信含会议号 {mid}",
|
||
"actual": text[:200] or "(none)",
|
||
"passed": passed,
|
||
}
|
||
|
||
def check_new_outgoing_contains_meeting_id_and_password(
|
||
self,
|
||
recipient_display: str,
|
||
meeting_id: str,
|
||
password: str,
|
||
*,
|
||
field: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""最新一条新发往外短信正文中含会议号（忽略空格）和密码。"""
|
||
if field is None:
|
||
field = f"sms_meeting_share_{recipient_display}"
|
||
mid = re.sub(r"\s+", "", str(meeting_id))
|
||
msgs = self.new_outgoing_messages_to(recipient_display)
|
||
text = str(msgs[-1]["content"]) if msgs else ""
|
||
if not mid:
|
||
passed = False
|
||
else:
|
||
passed = (
|
||
bool(msgs)
|
||
and bool(TencentMeeting.meeting_id_pattern(mid).search(text))
|
||
and str(password) in text
|
||
)
|
||
return {
|
||
"field": field,
|
||
"expected": f"短信含会议号 {mid} 和密码 {password}",
|
||
"actual": text[:200] or "(none)",
|
||
"passed": passed,
|
||
}
|