""" 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, }