kotones-auto-assistant/kotonebot/kaa/tasks/daily/purchase.py

219 lines
7.4 KiB
Python

"""从商店购买物品"""
import logging
from typing import Optional
from kotonebot.backend.loop import Loop
from kotonebot.kaa.tasks import R
from kotonebot.kaa.common import conf, DailyMoneyShopItems
from kotonebot.primitives.geometry import Point
from kotonebot.util import Countdown, cropped
from kotonebot import task, device, image, action, sleep
from kotonebot.backend.dispatch import SimpleDispatcher
from ..actions.scenes import goto_home, goto_shop, at_daily_shop
logger = logging.getLogger(__name__)
@action('购买 Money 物品', screenshot_mode='manual-inherit')
def money_items2(items: Optional[list[DailyMoneyShopItems]] = None):
"""
购买 Money 物品
前置条件:商店页面的 マニー Tab\n
结束状态:-
:param items: 要购买的物品列表,默认为 None。为 None 时使用配置文件里的设置。
"""
# 前置条件:[screenshots\shop\money1.png]
logger.info(f'Purchasing マニー items.')
if items is None:
items = conf().purchase.money_items
device.screenshot()
if DailyMoneyShopItems.Recommendations in items:
dispatch_recommended_items()
items.remove(DailyMoneyShopItems.Recommendations)
finished = []
max_scroll = 3
scroll = 0
while items:
for item in items:
if ret := image.find(item.to_resource(), colored=True):
logger.info(f'Purchasing {item.to_ui_text(item)}...')
device.click()
confirm_purchase(ret.position)
finished.append(item)
items = [item for item in items if item not in finished]
# 全都买完了
if not items:
break
# 还有,翻页后继续
else:
device.swipe_scaled(x1=0.5, x2=0.5, y1=0.8, y2=0.5)
sleep(0.5)
device.screenshot()
scroll += 1
if scroll >= max_scroll:
break
logger.info(f'Purchasing money items completed. {len(finished)} item(s) purchased.')
if items:
logger.info(f'{len(items)} item(s) not purchased/already purchased: {", ".join([item.to_ui_text(item) for item in items])}')
@action('购买推荐商品', screenshot_mode='manual-inherit')
def dispatch_recommended_items():
"""
购买推荐商品
前置条件:商店页面的 マニー Tab\n
结束状态:-
"""
# 前置条件:[screenshots\shop\money1.png]
logger.info(f'Start purchasing recommended items.')
while True:
device.screenshot()
if rec := image.find(R.Daily.TextShopRecommended):
logger.info(f'Clicking on recommended item.') # TODO: 计数
device.click()
confirm_purchase(rec.position)
sleep(2.5) #
elif image.find(R.Daily.IconTitleDailyShop) and not image.find(R.Daily.TextShopRecommended):
logger.info(f'No recommended item found. Finished.')
break
@action('确认购买', screenshot_mode='manual-inherit')
def confirm_purchase(target_item_pos: Point | None = None):
"""
确认购买
前置条件:点击某个商品后的瞬间\n
结束状态:对话框关闭后原来的界面
"""
# 前置条件:[screenshots\shop\dialog.png]
# TODO: 需要有个更好的方式检测是否已购买
purchased = False
cd = Countdown(sec=3)
for _ in Loop():
if cd.expired():
purchased = True
break
if image.find(R.Daily.TextShopItemSoldOut):
logger.info('Item sold out.')
purchased = True
break
elif image.find(R.Daily.TextShopItemPurchased):
logger.info('Item already purchased.')
purchased = True
break
elif image.find(R.Common.ButtonConfirm):
logger.info('Confirming purchase...')
device.click()
sleep(0.5)
else:
if target_item_pos:
device.click(target_item_pos)
if purchased:
logger.info('Item sold out.')
sleep(1) # 等待售罄提示消失
return
else:
device.screenshot()
while image.find(R.Daily.ButtonShopCountAdd, colored=True):
logger.debug('Adjusting quantity(+1)...')
device.click()
sleep(0.2)
device.screenshot()
logger.debug('Confirming purchase...')
device.click(image.expect_wait(R.Common.ButtonConfirm))
# 等待对话框动画结束
image.expect_wait(R.Daily.IconTitleDailyShop)
@action('购买 AP 物品')
def ap_items():
"""
购买 AP 物品
前置条件:位于商店页面的 AP Tab
"""
# [screenshots\shop\ap1.png]
logger.info(f'Purchasing AP items.')
results = image.find_all(R.Daily.IconShopAp, threshold=0.7)
sleep(1)
# 按 X, Y 坐标排序从小到大
results = sorted(results, key=lambda x: (x.position[0], x.position[1]))
# 按照配置文件里的设置过滤
item_indices = conf().purchase.ap_items
logger.info(f'Purchasing AP items: {item_indices}')
for index in item_indices:
if index <= len(results):
logger.info(f'Purchasing #{index} AP item.')
device.click(results[index])
sleep(0.5)
with cropped(device, y1=0.3):
purchased = image.wait_for(R.Daily.TextShopItemSoldOut, timeout=1)
if purchased is not None:
logger.info(f'AP item #{index} already purchased.')
continue
comfirm = image.expect_wait(R.Common.ButtonConfirm, timeout=2)
# 如果数量不是最大,调到最大
while image.find(R.Daily.ButtonShopCountAdd, colored=True):
logger.debug('Adjusting quantity(+1)...')
device.click()
sleep(0.3)
logger.debug(f'Confirming purchase...')
device.click(comfirm)
sleep(1.5)
else:
logger.warning(f'AP item #{index} not found')
logger.info(f'Purchasing AP items completed. {len(item_indices)} items purchased.')
@task('商店购买')
def purchase():
"""
从商店购买物品
"""
if not conf().purchase.enabled:
logger.info('Purchase is disabled.')
return
goto_shop()
# 进入每日商店 [screenshots\shop\shop.png]
device.click(image.expect(R.Daily.ButtonDailyShop)) # TODO: memoable
# 等待载入
ap_tab = image.expect_wait(R.Daily.TextTabShopAp)
# 购买マニー物品
if conf().purchase.money_enabled:
image.expect_wait(R.Daily.IconShopMoney)
money_items2()
sleep(0.5)
if image.find(R.Daily.ButtonRefreshMoneyShop):
logger.info('Refreshing money shop.')
device.click()
sleep(0.5)
money_items2()
sleep(0.5)
else:
logger.info('Money purchase is disabled.')
# 购买 AP 物品
if conf().purchase.ap_enabled:
# 点击 AP 选项卡
device.click(ap_tab)
# 等待 AP 选项卡加载完成
image.expect_wait(R.Daily.IconShopAp, threshold=0.7)
ap_items()
sleep(0.5)
else:
logger.info('AP purchase is disabled.')
goto_home()
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)
purchase()