656 lines
28 KiB
Python
656 lines
28 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""会话身份改用昵称之后，身份必须稳：同一个人永远同一个 ID。"""
|
|
|
|
import io
|
|
from contextlib import redirect_stdout
|
|
from unittest import TestCase, main, mock
|
|
|
|
import numpy as np
|
|
|
|
import session_name
|
|
from session_name import NameReader, crop_to_ink, ink_bounds, normalize_name, session_id_for
|
|
from test_support import queue_log_redirect
|
|
|
|
|
|
def setUpModule():
|
|
queue_log_redirect.start()
|
|
|
|
|
|
def tearDownModule():
|
|
queue_log_redirect.stop()
|
|
|
|
|
|
def panel_with_text(width=200, height=40, box=(20, 10, 120, 30)) -> np.ndarray:
|
|
"""浅底深字的一块面板，box 是那行字占的矩形。"""
|
|
image = np.full((height, width, 3), 245, dtype=np.uint8)
|
|
x1, y1, x2, y2 = box
|
|
image[y1:y2, x1:x2] = 30
|
|
return image
|
|
|
|
|
|
class FakeEngine:
|
|
"""替身 OCR：按调用顺序吐出预先排好的（文本, 置信度）。"""
|
|
|
|
def __init__(self, *results):
|
|
self.results = list(results)
|
|
self.calls = []
|
|
|
|
def __call__(self, image, **kwargs):
|
|
self.calls.append((image.shape, kwargs))
|
|
if not self.results:
|
|
return None, None
|
|
text, score = self.results.pop(0)
|
|
return [[text, score]], None
|
|
|
|
|
|
def reader_with(*results) -> NameReader:
|
|
reader = NameReader()
|
|
reader._engine = FakeEngine(*results)
|
|
return reader
|
|
|
|
|
|
class NameNormalisationTest(TestCase):
|
|
def test_the_title_and_the_list_row_agree_on_one_person(self):
|
|
# 实测标题读出来带空格、列表行不带。不抹平的话同一个人两个 ID，
|
|
# 又回到一人多档、客户收不到回复的老问题
|
|
self.assertEqual(normalize_name("高瑞 @微信"), normalize_name("高瑞@微信"))
|
|
self.assertEqual(
|
|
session_id_for("高瑞 @微信"), session_id_for("高瑞@微信")
|
|
)
|
|
|
|
def test_full_width_characters_fold_to_the_same_id(self):
|
|
self.assertEqual(session_id_for("高瑞＠微信"), session_id_for("高瑞@微信"))
|
|
|
|
def test_the_id_is_a_plain_md5_hex(self):
|
|
sid = session_id_for("高瑞@微信")
|
|
self.assertEqual(len(sid), 32)
|
|
self.assertTrue(all(ch in "0123456789abcdef" for ch in sid))
|
|
|
|
def test_two_different_people_never_share_an_id(self):
|
|
self.assertNotEqual(
|
|
session_id_for("高瑞@微信"), session_id_for("一个小迷糊@微信")
|
|
)
|
|
|
|
def test_no_name_means_no_id(self):
|
|
# 读不到名字时给个空 ID，调用方好判断"这一轮先别认"，
|
|
# 而不是拿空字符串的 md5 去当成某个真实会话
|
|
self.assertEqual(session_id_for(""), "")
|
|
self.assertEqual(session_id_for(" "), "")
|
|
|
|
|
|
class InkCropTest(TestCase):
|
|
def test_it_finds_the_line_of_text_in_the_panel(self):
|
|
image = panel_with_text(box=(20, 10, 120, 30))
|
|
self.assertEqual(ink_bounds(image), (20, 10, 120, 30))
|
|
|
|
def test_a_blank_panel_has_no_text(self):
|
|
blank = np.full((40, 200, 3), 245, dtype=np.uint8)
|
|
self.assertIsNone(ink_bounds(blank))
|
|
self.assertIsNone(crop_to_ink(blank))
|
|
|
|
def test_the_crop_keeps_a_margin_around_the_glyphs(self):
|
|
# 贴着字裁会切掉首字的一竖，实测「一个小迷糊@微信」会读成「个小迷糊@微信」
|
|
crop = crop_to_ink(panel_with_text(box=(20, 10, 120, 30)), pad=5)
|
|
self.assertEqual(crop.shape[1], 110)
|
|
self.assertEqual(crop.shape[0], 30)
|
|
|
|
def test_the_crop_never_runs_off_the_panel(self):
|
|
# 字顶到左右边框时，留白不能把裁剪框推到画面外面去
|
|
crop = crop_to_ink(panel_with_text(box=(0, 12, 200, 28)), pad=8)
|
|
self.assertEqual(crop.shape[:2], (32, 200))
|
|
|
|
def test_a_dark_themed_panel_still_yields_its_text(self):
|
|
image = np.full((40, 200, 3), 30, dtype=np.uint8)
|
|
image[10:30, 20:120] = 240
|
|
self.assertEqual(ink_bounds(image), (20, 10, 120, 30))
|
|
|
|
|
|
class NameReadingTest(TestCase):
|
|
def test_full_page_layout_ocr_returns_text_with_coordinates(self):
|
|
class LayoutEngine:
|
|
def __init__(self):
|
|
self.kwargs = None
|
|
|
|
def __call__(self, image, **kwargs):
|
|
self.kwargs = kwargs
|
|
return [
|
|
[[[10, 20], [70, 20], [70, 42], [10, 42]], "消息", 0.96],
|
|
[[[80, 80], [150, 80], [150, 102], [80, 102]], "噪声", 0.20],
|
|
], None
|
|
|
|
reader = NameReader()
|
|
reader._engine = LayoutEngine()
|
|
|
|
items = reader.read_layout(np.zeros((160, 240, 3), dtype=np.uint8))
|
|
|
|
self.assertEqual(
|
|
items,
|
|
[
|
|
{
|
|
"text": "消息",
|
|
"score": 0.96,
|
|
"x1": 10,
|
|
"y1": 20,
|
|
"x2": 70,
|
|
"y2": 42,
|
|
}
|
|
],
|
|
)
|
|
self.assertTrue(reader._engine.kwargs["use_det"])
|
|
self.assertFalse(reader._engine.kwargs["use_cls"])
|
|
|
|
def test_a_confident_read_becomes_the_session_identity(self):
|
|
reader = reader_with(("高瑞@微信", 0.94))
|
|
self.assertEqual(reader.read(panel_with_text()), "高瑞@微信")
|
|
|
|
def test_the_detector_is_skipped_so_it_stays_fast_enough_to_poll(self):
|
|
# 带检测要 1.6 秒，放不进轮询；框自己算好了就别让模型再找一遍
|
|
reader = reader_with(("高瑞@微信", 0.94))
|
|
reader.read(panel_with_text())
|
|
_shape, kwargs = reader._engine.calls[0]
|
|
self.assertFalse(kwargs["use_det"])
|
|
self.assertFalse(kwargs["use_cls"])
|
|
|
|
def test_a_shaky_read_is_refused_rather_than_guessed(self):
|
|
# 实测「打卡」被读成「江卡」只有 0.708。宁可这一轮不认，
|
|
# 也不能凭一个错名字开出一份新档案
|
|
reader = reader_with(("江卡", 0.70))
|
|
self.assertEqual(reader.read(panel_with_text()), "")
|
|
|
|
def test_a_blank_panel_never_reaches_the_engine(self):
|
|
reader = reader_with(("不该被调用", 0.99))
|
|
blank = np.full((40, 200, 3), 245, dtype=np.uint8)
|
|
self.assertEqual(reader.read(blank), "")
|
|
self.assertEqual(reader._engine.calls, [])
|
|
|
|
def test_the_same_screen_is_only_read_once(self):
|
|
# 同一块画面在一轮里会被反复问到；每次都重识别既慢，又可能这次读对
|
|
# 下次读错，会话身份就在同一轮里跳变
|
|
reader = reader_with(("高瑞@微信", 0.94))
|
|
image = panel_with_text()
|
|
first = reader.read(image, signature=b"same-screen")
|
|
second = reader.read(image, signature=b"same-screen")
|
|
self.assertEqual(first, "高瑞@微信")
|
|
self.assertEqual(second, "高瑞@微信")
|
|
self.assertEqual(len(reader._engine.calls), 1)
|
|
|
|
def test_a_different_screen_is_read_again(self):
|
|
reader = reader_with(("高瑞@微信", 0.94), ("一个小迷糊@微信", 0.98))
|
|
image = panel_with_text()
|
|
self.assertEqual(reader.read(image, signature=b"screen-a"), "高瑞@微信")
|
|
self.assertEqual(reader.read(image, signature=b"screen-b"), "一个小迷糊@微信")
|
|
|
|
def test_a_refused_read_is_not_cached_as_an_answer(self):
|
|
# 低置信度不该被记成"这块画面就是没名字"，下一轮画面清楚了要能读出来
|
|
reader = reader_with(("江卡", 0.70), ("打卡记录", 0.95))
|
|
image = panel_with_text()
|
|
self.assertEqual(reader.read(image, signature=b"row"), "")
|
|
self.assertEqual(reader.read(image, signature=b"row"), "打卡记录")
|
|
|
|
def test_the_cache_is_capped_so_a_day_long_run_cannot_eat_memory(self):
|
|
# 每一行的未读态/选中态/悬停态各占一个条目，机器人一跑就是一整天
|
|
reader = NameReader()
|
|
reader._engine = FakeEngine(
|
|
*[("高瑞@微信", 0.94)] * (session_name.MAX_CACHE_ENTRIES + 40)
|
|
)
|
|
image = panel_with_text()
|
|
for index in range(session_name.MAX_CACHE_ENTRIES + 30):
|
|
reader.read(image, signature=f"screen-{index}".encode())
|
|
self.assertLessEqual(len(reader._cache), session_name.MAX_CACHE_ENTRIES)
|
|
self.assertGreater(len(reader._cache), 0)
|
|
|
|
def test_a_missing_engine_reports_no_name_instead_of_crashing(self):
|
|
reader = NameReader()
|
|
reader._engine_failed = True
|
|
self.assertFalse(reader.available)
|
|
self.assertEqual(reader.read(panel_with_text()), "")
|
|
|
|
def test_an_engine_that_throws_does_not_take_the_round_down(self):
|
|
class Exploding:
|
|
def __call__(self, image, **kwargs):
|
|
raise RuntimeError("模型崩了")
|
|
|
|
reader = NameReader()
|
|
reader._engine = Exploding()
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertEqual(reader.read(panel_with_text()), "")
|
|
|
|
|
|
class MisreadCorrectionTest(TestCase):
|
|
def test_a_one_character_slip_is_pulled_back_to_the_known_contact(self):
|
|
# 「行业资讯」被读成「亏业资讯」。不纠的话这一轮就成了另一个客户，
|
|
# 上下文断掉、档案分家
|
|
reader = NameReader()
|
|
reader.remember("行业资讯")
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertEqual(reader.canonical("亏业资讯"), "行业资讯")
|
|
|
|
def test_a_genuinely_new_contact_is_not_folded_into_an_old_one(self):
|
|
reader = NameReader()
|
|
reader.remember("高瑞@微信")
|
|
self.assertEqual(reader.canonical("一个小迷糊@微信"), "一个小迷糊@微信")
|
|
|
|
def test_two_characters_apart_is_too_far_to_assume(self):
|
|
reader = NameReader()
|
|
reader.remember("甄养堂助理-江莉莉")
|
|
self.assertEqual(reader.canonical("甄养堂助理-王菲菲"), "甄养堂助理-王菲菲")
|
|
|
|
def test_short_names_are_never_corrected(self):
|
|
# 「小王」和「小李」也只差一个字，但那是两个人
|
|
reader = NameReader()
|
|
reader.remember("小王")
|
|
self.assertEqual(reader.canonical("小李"), "小李")
|
|
|
|
def test_the_contact_seen_most_often_wins_a_tie(self):
|
|
# 偶发误读只会出现一两次，常客才是真身
|
|
reader = NameReader()
|
|
reader._known = {"客户联系": 40, "客尸联系": 1}
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertEqual(reader.canonical("客户联糸"), "客户联系")
|
|
|
|
def test_an_exact_known_name_is_returned_untouched(self):
|
|
reader = NameReader()
|
|
reader.remember("高瑞@微信")
|
|
self.assertEqual(reader.canonical("高瑞@微信"), "高瑞@微信")
|
|
|
|
def test_correction_keeps_the_id_stable_across_a_slip(self):
|
|
# 这才是纠正的意义：读错一个字，会话 ID 依然不变
|
|
reader = NameReader()
|
|
reader.remember("行业资讯")
|
|
with redirect_stdout(io.StringIO()):
|
|
corrected = reader.canonical("亏业资讯")
|
|
self.assertEqual(session_id_for(corrected), session_id_for("行业资讯"))
|
|
|
|
def test_a_corrected_read_does_not_register_the_wrong_spelling(self):
|
|
reader = NameReader()
|
|
reader.remember("行业资讯")
|
|
with redirect_stdout(io.StringIO()):
|
|
reader.read(panel_with_text(), signature=b"")
|
|
reader._engine = FakeEngine(("亏业资讯", 0.89))
|
|
with redirect_stdout(io.StringIO()):
|
|
reader.read(panel_with_text())
|
|
self.assertNotIn("亏业资讯", reader.known_names())
|
|
|
|
|
|
class BotIdentityReadingTest(TestCase):
|
|
"""机器人读「当前打开的是谁」，是进度提示和会话身份共同的入口。"""
|
|
|
|
def _bot(self):
|
|
from wechat_bot import WeChatBot
|
|
|
|
bot = WeChatBot.__new__(WeChatBot)
|
|
bot._progress_text = ""
|
|
bot.progress_cb = None
|
|
return bot
|
|
|
|
def test_the_open_chat_is_named_from_its_title(self):
|
|
bot = self._bot()
|
|
bot._name_reader_instance = reader_with(("高瑞@微信", 0.94))
|
|
bot._chat_title_panel = lambda: panel_with_text()
|
|
bot._chat_identity_signature = lambda: b"title-sig"
|
|
self.assertEqual(bot._open_chat_display_name(), "高瑞@微信")
|
|
|
|
def test_the_title_pixels_key_the_cache_so_one_chat_reads_once(self):
|
|
# 同一个会话在一轮里会被问到好几次，每次都重识别既慢、又可能这次读对
|
|
# 下次读错，身份就在同一轮里跳变
|
|
bot = self._bot()
|
|
reader = reader_with(("高瑞@微信", 0.94))
|
|
bot._name_reader_instance = reader
|
|
bot._chat_title_panel = lambda: panel_with_text()
|
|
bot._chat_identity_signature = lambda: b"title-sig"
|
|
bot._open_chat_display_name()
|
|
bot._open_chat_display_name()
|
|
self.assertEqual(len(reader._engine.calls), 1)
|
|
|
|
def test_a_missing_title_panel_names_nobody(self):
|
|
bot = self._bot()
|
|
bot._name_reader_instance = reader_with(("不该被调用", 0.99))
|
|
bot._chat_title_panel = lambda: None
|
|
bot._chat_identity_signature = lambda: b""
|
|
self.assertEqual(bot._open_chat_display_name(), "")
|
|
|
|
def test_the_progress_line_falls_back_to_a_fingerprint_stub(self):
|
|
# 读不出名字时进度提示也不能变成空白，用户会以为它死了
|
|
bot = self._bot()
|
|
bot._chat_title_panel = lambda: None
|
|
bot._chat_identity_signature = lambda: b""
|
|
bot._name_reader_instance = reader_with()
|
|
self.assertEqual(bot._session_label(bytes.fromhex("ab" * 20)), "会话 abababab")
|
|
|
|
def test_the_progress_line_prefers_the_nickname(self):
|
|
bot = self._bot()
|
|
bot._name_reader_instance = reader_with(("一个小迷糊@微信", 0.98))
|
|
bot._chat_title_panel = lambda: panel_with_text()
|
|
bot._chat_identity_signature = lambda: b"title-sig"
|
|
self.assertEqual(bot._session_label(b"\x01" * 40), "一个小迷糊@微信")
|
|
|
|
|
|
class RowIdentityTest(TestCase):
|
|
"""会话身份改由昵称派生：同一行在任何渲染下都必须是同一个人。"""
|
|
|
|
def _bot(self, name_by_row=None):
|
|
from wechat_bot import WeChatBot
|
|
|
|
bot = WeChatBot.__new__(WeChatBot)
|
|
bot.identity_by_name = True
|
|
bot.scale = 1.0
|
|
bot.session_item_h = 64
|
|
bot._session_identity_trustworthy = lambda: True
|
|
if name_by_row is not None:
|
|
bot._row_display_name = (
|
|
lambda img, rel_y, row_center=False: name_by_row(rel_y)
|
|
)
|
|
return bot
|
|
|
|
@staticmethod
|
|
def _row_image(width=230, height=64, bold=False, selected=False):
|
|
"""一行会话。未读是加粗、选中是蓝底白字——像素差别正是老 bug 的根源。"""
|
|
background = (230, 120, 40) if selected else (245, 245, 245)
|
|
image = np.full((height, width, 3), background, dtype=np.uint8)
|
|
ink = 255 if selected else (10 if bold else 60)
|
|
image[16:30, 70:150] = ink
|
|
if bold:
|
|
image[16:32, 70:154] = ink
|
|
image[8:24, 210:226] = (60, 60, 220) # 未读红点
|
|
return image
|
|
|
|
def test_the_same_contact_matches_across_unread_and_selected_renderings(self):
|
|
# 这就是现场那句「实际打开对象与目标不一致，已停止后续发送」：
|
|
# 记下目标时它是未读（加粗+红点），点开后它是选中（蓝底白字），
|
|
# 像素哈希算出两个身份，消息复制到了也不敢发
|
|
bot = self._bot(lambda rel_y: "一个小迷糊@微信")
|
|
unread_fp = bot._session_fingerprint(
|
|
self._row_image(bold=True), 32, row_center=True
|
|
)
|
|
selected_fp = bot._session_fingerprint(
|
|
self._row_image(selected=True), 32, row_center=True
|
|
)
|
|
self.assertTrue(unread_fp)
|
|
self.assertEqual(unread_fp, selected_fp)
|
|
self.assertTrue(bot._session_fp_matches(unread_fp, selected_fp))
|
|
|
|
def test_two_different_contacts_stay_apart(self):
|
|
bot = self._bot(lambda rel_y: "高瑞@微信" if rel_y < 50 else "一个小迷糊@微信")
|
|
first = bot._session_fingerprint(self._row_image(), 32, row_center=True)
|
|
second = bot._session_fingerprint(self._row_image(), 96, row_center=True)
|
|
self.assertNotEqual(first, second)
|
|
self.assertFalse(bot._session_fp_matches(first, second))
|
|
|
|
def test_the_session_id_is_exactly_the_nickname_md5(self):
|
|
from wechat_bot import WeChatBot
|
|
|
|
fp = WeChatBot._fp_from_name("高瑞@微信")
|
|
self.assertEqual(len(fp), 40)
|
|
self.assertEqual(WeChatBot.session_id_of(fp), session_id_for("高瑞@微信"))
|
|
|
|
def test_the_title_and_the_row_produce_one_identity(self):
|
|
# 标题里带空格、列表行不带；归一之后必须是同一个人
|
|
from wechat_bot import WeChatBot
|
|
|
|
self.assertEqual(
|
|
WeChatBot._fp_from_name("高瑞 @微信"),
|
|
WeChatBot._fp_from_name("高瑞@微信"),
|
|
)
|
|
|
|
def test_no_nickname_means_no_identity(self):
|
|
from wechat_bot import WeChatBot
|
|
|
|
self.assertEqual(WeChatBot._fp_from_name(""), b"")
|
|
self.assertEqual(WeChatBot._fp_from_name(" "), b"")
|
|
|
|
def test_an_unreadable_row_falls_back_to_the_pixel_path(self):
|
|
# OCR 读不出来（空行、引擎缺失）时不能整个瘫掉，还得按老办法认
|
|
bot = self._bot(lambda rel_y: "")
|
|
fallback = mock.Mock(return_value=b"pixel-path")
|
|
bot._canonical_fp = fallback
|
|
bot._raw_session_fingerprint = mock.Mock(return_value=b"raw")
|
|
bot._session_name_fingerprint = mock.Mock(return_value=b"")
|
|
bot._session_fingerprint(self._row_image(), 32, row_center=True)
|
|
fallback.assert_called_once()
|
|
|
|
def test_the_pixel_path_is_untouched_when_the_switch_is_off(self):
|
|
bot = self._bot(lambda rel_y: "高瑞@微信")
|
|
bot.identity_by_name = False
|
|
bot._canonical_fp = mock.Mock(return_value=b"a" * 8)
|
|
bot._raw_session_fingerprint = mock.Mock(return_value=b"raw")
|
|
bot._session_name_fingerprint = mock.Mock(return_value=b"n" * 32)
|
|
bot._canonical_session_fp = lambda raw: raw
|
|
bot._session_render_fingerprint = mock.Mock(return_value=b"")
|
|
fp = bot._session_fingerprint(self._row_image(), 32, row_center=True)
|
|
self.assertEqual(fp, b"a" * 8 + b"n" * 32)
|
|
|
|
|
|
class RowNameCropTest(TestCase):
|
|
def _bot(self, width=460, item_h=128):
|
|
from wechat_bot import WeChatBot
|
|
|
|
bot = WeChatBot.__new__(WeChatBot)
|
|
bot.scale = 2.0
|
|
bot.session_item_h = item_h
|
|
bot._avatar_anchor = lambda img, y: y
|
|
return bot
|
|
|
|
def test_the_crop_skips_the_avatar_and_stops_before_the_timestamp(self):
|
|
# 实测 460px 宽的列表：头像占到 112，昵称 136..318，时间戳更靠右。
|
|
# 时间戳一旦进框，「1分钟前」变「12:08」就会让同一个人换 md5
|
|
bot = self._bot()
|
|
image = np.zeros((1208, 460, 3), dtype=np.uint8)
|
|
panel = bot._row_name_panel(image, 78, row_center=True)
|
|
self.assertIsNotNone(panel)
|
|
left = int(460 * 0.26)
|
|
right = int(460 * 0.78)
|
|
self.assertEqual(panel.shape[1], right - left)
|
|
self.assertGreater(left, 112) # 头像在外
|
|
self.assertLess(left, 136) # 昵称起点没被切
|
|
self.assertGreater(right, 318) # 最长的昵称还在
|
|
self.assertLess(right, 460) # 时间戳在外
|
|
|
|
def test_the_crop_follows_a_narrow_sidebar_instead_of_fixed_pixels(self):
|
|
# 侧栏宽度是动态识别出来的，写死像素在窄侧栏下会切掉字
|
|
narrow = self._bot()
|
|
image = np.zeros((600, 300, 3), dtype=np.uint8)
|
|
panel = narrow._row_name_panel(image, 78, row_center=True)
|
|
self.assertEqual(panel.shape[1], int(300 * 0.78) - int(300 * 0.26))
|
|
|
|
def test_the_crop_takes_the_name_line_not_the_message_preview(self):
|
|
bot = self._bot()
|
|
image = np.zeros((1208, 460, 3), dtype=np.uint8)
|
|
panel = bot._row_name_panel(image, 200, row_center=True)
|
|
self.assertLessEqual(panel.shape[0], int(128 * 0.34))
|
|
|
|
def test_a_row_too_small_to_hold_a_name_is_refused(self):
|
|
bot = self._bot(item_h=8)
|
|
image = np.zeros((40, 20, 3), dtype=np.uint8)
|
|
self.assertIsNone(bot._row_name_panel(image, 10, row_center=True))
|
|
|
|
def test_a_missing_image_never_raises(self):
|
|
bot = self._bot()
|
|
self.assertIsNone(bot._row_name_panel(None, 10, row_center=True))
|
|
|
|
|
|
class ChatEdgeRuleTest(TestCase):
|
|
"""聊天区左边缘那条分隔线，是「对着同一句话连回三遍」的真正根因。"""
|
|
|
|
@staticmethod
|
|
def _grid(rows=36, cols=48):
|
|
return np.zeros((rows, cols), dtype=bool)
|
|
|
|
def _bot(self):
|
|
from wechat_bot import WeChatBot
|
|
|
|
return WeChatBot
|
|
|
|
def test_a_full_height_line_at_the_left_edge_is_erased(self):
|
|
# 会话列表和聊天区之间的分隔线被裁进了聊天区，网格最左一列每行都有内容
|
|
grid = self._grid()
|
|
grid[3:, 0] = True
|
|
cleaned = self._bot()._without_edge_rules(grid)
|
|
self.assertFalse(cleaned.any())
|
|
|
|
def test_the_bottom_most_message_is_found_again_once_the_line_is_gone(self):
|
|
# 有这条线时，「最底下有内容的行」永远是最后一行；判定区间落在空白处，
|
|
# 只看见左边那条线，于是任何会话都被判成「最后一条是客户发的」
|
|
grid = self._grid()
|
|
grid[3:, 0] = True # 边缘竖线
|
|
grid[17:21, 26:48] = True # 我方气泡，靠右
|
|
cleaned = self._bot()._without_edge_rules(grid)
|
|
bottom = int(np.where(cleaned.any(axis=1))[0][-1])
|
|
self.assertEqual(bottom, 20)
|
|
|
|
def test_a_real_bubble_touching_the_left_edge_is_kept(self):
|
|
# 客户消息本来就靠左，不能连它一起抹掉
|
|
grid = self._grid()
|
|
grid[17:21, 0:22] = True
|
|
cleaned = self._bot()._without_edge_rules(grid)
|
|
self.assertTrue(cleaned[17:21, 0].all())
|
|
|
|
def test_a_long_message_spanning_the_screen_is_not_mistaken_for_a_rule(self):
|
|
# 中间列再长也不清；只有贴边且几乎贯穿全高的才算 UI 线条
|
|
grid = self._grid()
|
|
grid[:, 20] = True
|
|
cleaned = self._bot()._without_edge_rules(grid)
|
|
self.assertTrue(cleaned[:, 20].all())
|
|
|
|
def test_a_rule_on_the_right_edge_is_erased_too(self):
|
|
grid = self._grid()
|
|
grid[:, -1] = True
|
|
cleaned = self._bot()._without_edge_rules(grid)
|
|
self.assertFalse(cleaned.any())
|
|
|
|
def test_an_empty_grid_survives_untouched(self):
|
|
cleaned = self._bot()._without_edge_rules(self._grid())
|
|
self.assertFalse(cleaned.any())
|
|
|
|
def test_a_degenerate_grid_never_raises(self):
|
|
bot = self._bot()
|
|
self.assertIsNone(bot._without_edge_rules(None))
|
|
tiny = np.zeros((2, 4), dtype=bool)
|
|
self.assertEqual(bot._without_edge_rules(tiny).shape, (2, 4))
|
|
|
|
def test_the_original_grid_is_not_mutated(self):
|
|
grid = self._grid()
|
|
grid[:, 0] = True
|
|
self._bot()._without_edge_rules(grid)
|
|
self.assertTrue(grid[:, 0].all())
|
|
|
|
|
|
class NoRepeatReplyTest(TestCase):
|
|
"""一条来消息只回一次。现场出过对着同一句「你好」连回三条的事故。"""
|
|
|
|
def _bot(self, last_bubble_outgoing, pending=None):
|
|
from wechat_bot import WeChatBot
|
|
|
|
bot = WeChatBot.__new__(WeChatBot)
|
|
bot._last_visible_bubble_is_outgoing = lambda: last_bubble_outgoing
|
|
bot._pending_reply_sessions = dict(pending or {})
|
|
bot._pending_exchanges = {}
|
|
bot._clear_reply_pending = mock.Mock()
|
|
bot._commit_staged_exchange = mock.Mock()
|
|
bot._remember_active_surface = mock.Mock()
|
|
bot._pending_reply_state = lambda fp: bot._pending_reply_sessions.get(fp.hex())
|
|
return bot
|
|
|
|
FP = bytes.fromhex("ab" * 20)
|
|
|
|
def test_our_own_reply_on_screen_stops_another_one(self):
|
|
# 客户只发了一句「你好」，回执没确认，任务留在队列里 —— 这时候绝不能
|
|
# 再生成一条措辞不同的问候发出去
|
|
bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "你好呀，我在呢"}})
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertTrue(bot._nothing_left_to_answer(self.FP, "你好\n你好呀，我在呢"))
|
|
|
|
def test_seeing_our_reply_settles_the_task_instead_of_retrying_forever(self):
|
|
# 自己写的那句就在屏幕上，发送成功是确凿的：落档、清任务，别再重试
|
|
bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "你好呀，我在呢"}})
|
|
with redirect_stdout(io.StringIO()):
|
|
bot._nothing_left_to_answer(self.FP, "你好\n你好呀，我在呢")
|
|
bot._commit_staged_exchange.assert_called_once_with(self.FP.hex())
|
|
bot._clear_reply_pending.assert_called_once_with(self.FP)
|
|
|
|
def test_a_human_reply_also_ends_the_task_without_archiving_our_draft(self):
|
|
# 右边最后那句不是我们写的，说明人工接手回过了
|
|
bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "机器人拟的稿"}})
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertTrue(bot._nothing_left_to_answer(self.FP, "你好\n我是人工，这就帮您看"))
|
|
bot._commit_staged_exchange.assert_not_called()
|
|
bot._clear_reply_pending.assert_called_once_with(self.FP)
|
|
|
|
def test_a_waiting_customer_still_gets_answered(self):
|
|
bot = self._bot(False, {self.FP.hex(): {"staged_reply_text": "上一轮的回复"}})
|
|
self.assertFalse(bot._nothing_left_to_answer(self.FP, "你好"))
|
|
bot._clear_reply_pending.assert_not_called()
|
|
|
|
def test_an_unreadable_bubble_side_never_silences_the_bot(self):
|
|
# 宽气泡、深色主题判不出左右时宁可多回一句，不能因为看不清就把客户晾着
|
|
bot = self._bot(None, {self.FP.hex(): {"staged_reply_text": "上一轮的回复"}})
|
|
self.assertFalse(bot._nothing_left_to_answer(self.FP, "你好"))
|
|
|
|
def test_an_idle_chat_with_no_task_is_simply_left_alone(self):
|
|
bot = self._bot(True)
|
|
self.assertTrue(bot._nothing_left_to_answer(self.FP, ""))
|
|
bot._clear_reply_pending.assert_not_called()
|
|
|
|
def test_the_guard_survives_a_blank_transcript(self):
|
|
# 剪贴板没抓到文字时也不能把"我方最后发言"当成要回复
|
|
bot = self._bot(True, {self.FP.hex(): {"staged_reply_text": "你好呀"}})
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertTrue(bot._nothing_left_to_answer(self.FP, ""))
|
|
bot._commit_staged_exchange.assert_not_called()
|
|
bot._clear_reply_pending.assert_called_once_with(self.FP)
|
|
|
|
|
|
class ProgressReportingTest(TestCase):
|
|
def _bot(self):
|
|
from wechat_bot import WeChatBot
|
|
|
|
bot = WeChatBot.__new__(WeChatBot)
|
|
bot._progress_text = ""
|
|
return bot
|
|
|
|
def test_the_same_line_is_not_sent_twice(self):
|
|
# 轮询每秒都在跑，重复内容会把界面刷爆
|
|
bot = self._bot()
|
|
seen = []
|
|
bot.progress_cb = seen.append
|
|
bot.report_progress("正在回复 高瑞@微信")
|
|
bot.report_progress("正在回复 高瑞@微信")
|
|
bot.report_progress("已回复 高瑞@微信")
|
|
self.assertEqual(seen, ["正在回复 高瑞@微信", "已回复 高瑞@微信"])
|
|
|
|
def test_a_broken_callback_never_takes_the_round_down(self):
|
|
bot = self._bot()
|
|
|
|
def explode(_text):
|
|
raise RuntimeError("界面没了")
|
|
|
|
bot.progress_cb = explode
|
|
bot.report_progress("正在回复 高瑞@微信")
|
|
|
|
def test_a_headless_run_reports_to_nobody_without_complaining(self):
|
|
bot = self._bot()
|
|
bot.progress_cb = None
|
|
bot.report_progress("正在回复 高瑞@微信")
|
|
|
|
def test_a_partially_built_bot_still_accepts_progress(self):
|
|
# 不少代码路径用 __new__ 造机器人，绕过了 __init__
|
|
from wechat_bot import WeChatBot
|
|
|
|
bot = WeChatBot.__new__(WeChatBot)
|
|
seen = []
|
|
bot.progress_cb = seen.append
|
|
bot.report_progress("等待新消息")
|
|
self.assertEqual(seen, ["等待新消息"])
|
|
|
|
|
|
class CorrectionDistanceTest(TestCase):
|
|
def test_the_distance_helper_stops_early_instead_of_scanning_everything(self):
|
|
far = session_name._edit_distance("abcdefgh", "zzzzzzzz", 1)
|
|
self.assertGreater(far, 1)
|
|
|
|
def test_identical_strings_are_zero_apart(self):
|
|
self.assertEqual(session_name._edit_distance("高瑞@微信", "高瑞@微信", 1), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|