fix(task): 修复某些情况下会跳过未读交流时会发生误触

本来在快进,结果又点了下快进按钮,导致取消快进。
修改后直接改用跳过按钮而不是快进。
This commit is contained in:
XcantloadX 2025-03-17 20:50:48 +08:00
parent 69a59879ff
commit 46c73544b0
4 changed files with 27 additions and 49 deletions

Binary file not shown.

Before

Width:  |  Height:  |  Size: 188 B

View File

Before

Width:  |  Height:  |  Size: 924 KiB

After

Width:  |  Height:  |  Size: 924 KiB

View File

@ -0,0 +1 @@
{"definitions":{"f1f21925-3e22-4dd1-b53b-bb52bcf26c2b":{"name":"Common.ButtonCommuSkip","displayName":"跳过交流按钮","type":"template","annotationId":"f1f21925-3e22-4dd1-b53b-bb52bcf26c2b","useHintRect":false}},"annotations":[{"id":"f1f21925-3e22-4dd1-b53b-bb52bcf26c2b","type":"rect","data":{"x1":180,"y1":1187,"x2":204,"y2":1215}}]}

View File

@ -5,6 +5,7 @@ from cv2.typing import MatLike
from .. import R
from ..game_ui import WhiteFilter
from kotonebot.util import Countdown, Interval
from kotonebot import device, image, color, user, rect_expand, until, action, sleep, use_screenshot
@ -13,7 +14,7 @@ logger = logging.getLogger(__name__)
@action('检查是否处于交流')
def is_at_commu():
return image.find(R.Common.ButtonCommuFastforward) is not None
return image.find(R.Common.ButtonCommuSkip, preprocessors=[WhiteFilter()]) is not None
@action('跳过交流')
def skip_commu():
@ -30,56 +31,32 @@ def handle_unread_commu(img: MatLike | None = None) -> bool:
ret = False
logger.info('Check and skip commu')
img = use_screenshot(img)
skip_btn = image.find(R.Common.ButtonCommuFastforward)
skip_btn = image.find(R.Common.ButtonCommuFastforward, preprocessors=[WhiteFilter()])
if skip_btn is None:
logger.info('No fast forward button found. Not at a commu.')
return ret
ret = True
logger.debug('Fast forward button found. Check commu')
button_bg_rect = rect_expand(skip_btn.rect, 10, 10, 50, 10)
def is_fastforwarding():
nonlocal img
assert img is not None
colors = color.raw().dominant_color(img, 2, rect=button_bg_rect)
RANGE = ((20, 65, 95), (180, 100, 100))
return any(color.raw().in_range(c, RANGE) for c in colors)
# 防止截图速度过快时,截图到了未加载完全的画面
cd = Interval(seconds=0.6)
hit = 0
HIT_THRESHOLD = 2
it = Interval()
while True:
if image.find(R.Common.ButtonCommuFastforward) and not is_fastforwarding():
logger.debug("Unread commu hit %d/%d", hit, HIT_THRESHOLD)
hit += 1
else:
hit = 0
if not is_at_commu():
break
if hit >= HIT_THRESHOLD:
break
cd.wait()
img = device.screenshot()
should_skip = hit >= HIT_THRESHOLD
if not should_skip:
logger.info('Fast forwarding. No action needed.')
return False
if should_skip:
user.info('发现未读交流', images=[img])
logger.debug('Not fast forwarding. Click fast forward button')
device.click(skip_btn)
sleep(0.7)
if image.wait_for(R.Common.ButtonConfirm, timeout=5):
logger.debug('Click confirm button')
if image.find(R.Common.ButtonCommuSkip, preprocessors=[WhiteFilter()]):
device.click()
else:
logger.info('Fast forwarding. No action needed.')
logger.debug('Wait until not at commu')
# TODO: 这里改用 while 循环,避免点击未生效的情况
until(lambda: not is_at_commu(), interval=0.3)
logger.info('Fast forward done')
logger.debug('Clicked skip button.')
if image.find(R.Common.ButtonConfirm):
logger.info('Unread commu found.')
device.click()
logger.debug('Clicked confirm button.')
logger.debug('Pushing notification...')
user.info('发现未读交流', images=[img])
logger.debug('Fast forwarding...')
it.wait()
logger.info('Fast forward done.')
return ret
@ -87,12 +64,12 @@ if __name__ == '__main__':
import logging
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
logger.setLevel(logging.DEBUG)
from kotonebot.backend.context import manual_context, inject_context
from kotonebot.backend.debug.mock import MockDevice
manual_context().begin()
device = MockDevice()
device.load_image(r"D:\current_screenshot.png")
inject_context(device=device)
print(is_at_commu())
# rect = image.expect(R.Common.ButtonCommuFastforward).rect
# print(rect)
# rect = rect_expand(rect, 10, 10, 50, 10)
# print(rect)
# img = device.screenshot()
# print(color.raw().dominant_color(img, 2, rect=rect))
# skip_commu()
# check_and_skip_commu()
# while True:
# print(handle_unread_commu())