feat(*): 引入配置模块，将各脚本与配置模块整合
This commit is contained in:
XcantloadX
parent
852385992b
commit
1f387d45bb
|
|
@ -5,6 +5,7 @@ R.py
|
|||
kotonebot-ui/node_modules
|
||||
kotonebot-ui/.vite
|
||||
dumps*/
|
||||
config.json
|
||||
##########################
|
||||
|
||||
# Byte-compiled / optimized / DLL files
|
||||
|
|
|
|||
|
|
@ -30,7 +30,9 @@
|
|||
"-s",
|
||||
"${workspaceFolder}/dumps",
|
||||
"-c",
|
||||
"${command:extension.commandvariable.file.relativeDirDots}.${fileBasenameNoExtension}"
|
||||
"${command:extension.commandvariable.file.relativeDirDots}.${fileBasenameNoExtension}",
|
||||
"-t",
|
||||
"kotonebot.tasks.common.BaseConfig"
|
||||
]
|
||||
},
|
||||
{
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from .backend.context import (
|
|||
image,
|
||||
debug,
|
||||
color,
|
||||
config,
|
||||
rect_expand
|
||||
)
|
||||
from .backend.util import (
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import os
|
||||
import re
|
||||
import time
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import (
|
||||
Callable,
|
||||
|
|
@ -11,6 +12,8 @@ from typing import (
|
|||
Literal,
|
||||
ParamSpec,
|
||||
Concatenate,
|
||||
Generic,
|
||||
Type,
|
||||
)
|
||||
|
||||
import cv2
|
||||
|
|
@ -21,7 +24,6 @@ from kotonebot.backend.util import Rect
|
|||
import kotonebot.backend.image as raw_image
|
||||
from kotonebot.client.device.adb import AdbDevice
|
||||
from kotonebot.backend.image import (
|
||||
CropResult,
|
||||
TemplateMatchResult,
|
||||
MultipleTemplateMatchResult,
|
||||
find_all_crop,
|
||||
|
|
@ -34,10 +36,13 @@ from kotonebot.backend.image import (
|
|||
import kotonebot.backend.color as raw_color
|
||||
from kotonebot.backend.color import find_rgb
|
||||
from kotonebot.backend.ocr import Ocr, OcrResult, jp, en, StringMatchFunction
|
||||
from kotonebot.config.manager import load_config, save_config
|
||||
from kotonebot.config.base_config import UserConfig
|
||||
|
||||
OcrLanguage = Literal['jp', 'en']
|
||||
DEFAULT_TIMEOUT = 120
|
||||
DEFAULT_INTERVAL = 0.4
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# https://stackoverflow.com/questions/74714300/paramspec-for-a-pre-defined-function-without-using-generic-callablep
|
||||
T = TypeVar('T')
|
||||
|
|
@ -386,6 +391,76 @@ class ContextColor:
|
|||
return find_rgb(self.context.device.screenshot(), *args, **kwargs)
|
||||
|
||||
|
||||
V = TypeVar('V')
|
||||
class ContextConfig(Generic[T]):
|
||||
def __init__(self, context: 'Context', config_type: Type[T] = dict[str, Any]):
|
||||
self.context = context
|
||||
self.config_path: str = 'config.json'
|
||||
self.current_key: int | str = 0
|
||||
self.config_type: Type = config_type
|
||||
self.root = load_config(self.config_path, type=config_type)
|
||||
|
||||
def to(self, conf_type: Type[V]) -> 'ContextConfig[V]':
|
||||
self.config_type = conf_type
|
||||
return cast(ContextConfig[V], self)
|
||||
|
||||
def create(self, config: UserConfig[T]):
|
||||
"""创建新用户配置"""
|
||||
self.root.user_configs.append(config)
|
||||
self.save()
|
||||
|
||||
def get(self, key: str | int | None = None) -> UserConfig[T] | None:
|
||||
"""
|
||||
获取指定或当前用户配置数据。
|
||||
|
||||
:param key: 用户配置 ID 或索引（从 0 开始），为 None 时获取当前用户配置
|
||||
:return: 用户配置数据
|
||||
"""
|
||||
if isinstance(key, int):
|
||||
if key < 0 or key >= len(self.root.user_configs):
|
||||
return None
|
||||
return self.root.user_configs[key]
|
||||
elif isinstance(key, str):
|
||||
for user in self.root.user_configs:
|
||||
if user.id == key:
|
||||
return user
|
||||
else:
|
||||
return None
|
||||
else:
|
||||
return self.get(self.current_key)
|
||||
|
||||
def save(self):
|
||||
"""保存所有配置数据到本地"""
|
||||
save_config(self.root, self.config_path)
|
||||
|
||||
def load(self):
|
||||
"""从本地加载所有配置数据"""
|
||||
self.root = load_config(self.config_path, type=self.config_type)
|
||||
|
||||
def switch(self, key: str | int):
|
||||
"""切换到指定用户配置"""
|
||||
self.current_key = key
|
||||
|
||||
@property
|
||||
def current(self) -> UserConfig[T]:
|
||||
"""
|
||||
当前配置数据。
|
||||
|
||||
如果当前配置不存在，则使用默认值自动创建一个新配置。
|
||||
（不推荐，建议在 UI 中启动前要求用户手动创建，或自行创建一个默认配置。）
|
||||
"""
|
||||
c = self.get(self.current_key)
|
||||
if c is None:
|
||||
if not self.config_type:
|
||||
raise ValueError("No config type specified.")
|
||||
logger.warning("No config found, creating a new one using default values. (NOT RECOMMENDED)")
|
||||
c = self.config_type()
|
||||
u = UserConfig(options=c)
|
||||
self.create(u)
|
||||
c = u
|
||||
return c
|
||||
|
||||
|
||||
class Forwarded:
|
||||
def __init__(self, getter: Callable[[], T] | None = None, name: str | None = None):
|
||||
self._FORWARD_getter = getter
|
||||
|
|
@ -406,7 +481,7 @@ class Forwarded:
|
|||
setattr(self._FORWARD_getter(), name, value)
|
||||
|
||||
class Context:
|
||||
def __init__(self):
|
||||
def __init__(self, config_type: Type[T]):
|
||||
# HACK: 暂时写死
|
||||
from adbutils import adb
|
||||
adb.connect('127.0.0.1:5555')
|
||||
|
|
@ -419,6 +494,7 @@ class Context:
|
|||
self.__color = ContextColor(self)
|
||||
self.__vars = ContextGlobalVars()
|
||||
self.__debug = ContextDebug(self)
|
||||
self.__config = ContextConfig[T](self, config_type)
|
||||
self.actions = []
|
||||
|
||||
def inject_device(self, device: DeviceABC):
|
||||
|
|
@ -448,13 +524,19 @@ class Context:
|
|||
def debug(self) -> 'ContextDebug':
|
||||
return self.__debug
|
||||
|
||||
@property
|
||||
def config(self) -> 'ContextConfig':
|
||||
return self.__config
|
||||
|
||||
def rect_expand(rect: Rect, left: int = 0, top: int = 0, right: int = 0, bottom: int = 0) -> Rect:
|
||||
"""
|
||||
向四个方向扩展矩形区域。
|
||||
"""
|
||||
return (rect[0] - left, rect[1] - top, rect[2] + right + left, rect[3] + bottom + top)
|
||||
|
||||
# 暴露 Context 的属性到模块级别
|
||||
# 这里 Context 类还没有初始化，但是 tasks 中的脚本可能已经引用了这里的变量
|
||||
# 为了能够动态更新这里变量的值，这里使用 Forwarded 类再封装一层，
|
||||
# 将调用转发到实际的稍后初始化的 Context 类上
|
||||
_c: Context | None = None
|
||||
device: DeviceABC = cast(DeviceABC, Forwarded(name="device"))
|
||||
"""当前正在执行任务的设备。"""
|
||||
|
|
@ -468,14 +550,25 @@ vars: ContextGlobalVars = cast(ContextGlobalVars, Forwarded(name="vars"))
|
|||
"""全局变量。"""
|
||||
debug: ContextDebug = cast(ContextDebug, Forwarded(name="debug"))
|
||||
"""调试工具。"""
|
||||
config: ContextConfig = cast(ContextConfig, Forwarded(name="config"))
|
||||
"""配置数据。"""
|
||||
|
||||
def init_context():
|
||||
global _c, device, ocr, image, color, vars, debug
|
||||
_c = Context()
|
||||
def init_context(
|
||||
config_type: Type[T] = dict[str, Any]
|
||||
):
|
||||
"""
|
||||
初始化 Context 模块。
|
||||
|
||||
:param config_type:
|
||||
配置数据类类型。配置数据类必须继承自 pydantic 的 `BaseModel`。
|
||||
默认为 `dict[str, Any]`，即普通的 JSON 数据，不包含任何类型信息。
|
||||
"""
|
||||
global _c, device, ocr, image, color, vars, debug, config
|
||||
_c = Context(config_type=config_type)
|
||||
device._FORWARD_getter = lambda: _c.device # type: ignore
|
||||
ocr._FORWARD_getter = lambda: _c.ocr # type: ignore
|
||||
image._FORWARD_getter = lambda: _c.image # type: ignore
|
||||
color._FORWARD_getter = lambda: _c.color # type: ignore
|
||||
vars._FORWARD_getter = lambda: _c.vars # type: ignore
|
||||
debug._FORWARD_getter = lambda: _c.debug # type: ignore
|
||||
|
||||
config._FORWARD_getter = lambda: _c.config # type: ignore
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
import logging
|
||||
from typing import Callable, ParamSpec, TypeVar, overload, Any, NamedTuple
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, ParamSpec, TypeVar, overload
|
||||
|
||||
from kotonebot.backend.context import UserConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,13 +1,13 @@
|
|||
import os
|
||||
import sys
|
||||
import runpy
|
||||
import argparse
|
||||
import shutil
|
||||
from threading import Thread
|
||||
from typing import Callable
|
||||
import argparse
|
||||
import importlib
|
||||
from pathlib import Path
|
||||
from threading import Thread
|
||||
|
||||
from . import debug
|
||||
from kotonebot.backend.context import init_context
|
||||
|
||||
def _task_thread(task_module: str):
|
||||
"""任务线程。"""
|
||||
|
|
@ -27,6 +27,13 @@ def _parse_args():
|
|||
help='Clear the dump folder before running',
|
||||
action='store_true'
|
||||
)
|
||||
parser.add_argument(
|
||||
'-t', '--config-type',
|
||||
help='The full path of the config data type. e.g. `kotonebot.tasks.common.BaseConfig`',
|
||||
type=str,
|
||||
metavar='TYPE',
|
||||
required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'input_module',
|
||||
help='The module to run'
|
||||
|
|
@ -51,6 +58,11 @@ if __name__ == "__main__":
|
|||
if args.clear:
|
||||
if debug.save_to_folder:
|
||||
shutil.rmtree(debug.save_to_folder)
|
||||
|
||||
# 初始化上下文
|
||||
module_name, class_name = args.config_type.rsplit('.', 1)
|
||||
class_ = importlib.import_module(module_name).__getattribute__(class_name)
|
||||
init_context(config_type=class_)
|
||||
|
||||
# 启动服务器
|
||||
from .server import app
|
||||
|
|
|
|||
|
|
@ -39,16 +39,16 @@ def fuzz(text: str) -> Callable[[str], bool]:
|
|||
def regex(regex: str) -> Callable[[str], bool]:
|
||||
"""返回正则表达式字符串匹配函数。"""
|
||||
f = lambda s: re.match(regex, s) is not None
|
||||
f.__repr__ = lambda: f"regex({regex})"
|
||||
f.__name__ = f"regex({regex})"
|
||||
f.__repr__ = lambda: f"regex('{regex}')"
|
||||
f.__name__ = f"regex('{regex}')"
|
||||
return f
|
||||
|
||||
@lru_cache(maxsize=1000)
|
||||
def contains(text: str) -> Callable[[str], bool]:
|
||||
"""返回包含指定文本的函数。"""
|
||||
f = lambda s: text in s
|
||||
f.__repr__ = lambda: f"contains({text})"
|
||||
f.__name__ = f"contains({text})"
|
||||
f.__repr__ = lambda: f"contains('{text}')"
|
||||
f.__name__ = f"contains('{text}')"
|
||||
return f
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
from .base_config import UserConfig
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
import uuid
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
T = TypeVar('T')
|
||||
|
||||
class BackendConfig(BaseModel):
|
||||
adb_ip: str = '127.0.0.1'
|
||||
"""adb 连接的 ip 地址。"""
|
||||
adb_port: int = 5555
|
||||
"""adb 连接的端口。"""
|
||||
|
||||
class UserConfig(BaseModel, Generic[T]):
|
||||
"""用户可以自由添加、删除的配置数据。"""
|
||||
|
||||
name: str = 'default_config'
|
||||
"""显示名称。通常由用户输入。"""
|
||||
id: str = uuid.uuid4().hex
|
||||
"""唯一标识符。"""
|
||||
category: str = 'default'
|
||||
"""类别。如：'global'、'china'、'asia' 等。"""
|
||||
description: str = ''
|
||||
"""描述。通常由用户输入。"""
|
||||
backend: BackendConfig = BackendConfig()
|
||||
"""后端配置。"""
|
||||
options: T
|
||||
"""下游脚本储存的具体数据。"""
|
||||
|
||||
|
||||
class RootConfig(BaseModel, Generic[T]):
|
||||
version: int = 1
|
||||
"""配置版本。"""
|
||||
user_configs: list[UserConfig[T]] = []
|
||||
"""用户配置。"""
|
||||
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
import os
|
||||
from typing import Type, Generic, TypeVar
|
||||
|
||||
from .base_config import RootConfig, UserConfig
|
||||
|
||||
T = TypeVar('T')
|
||||
|
||||
def load_config(
|
||||
config_path: str,
|
||||
*,
|
||||
type: Type[T],
|
||||
use_default_if_not_found: bool = True
|
||||
) -> RootConfig[T]:
|
||||
"""
|
||||
从指定路径读取配置文件
|
||||
|
||||
:param config_path: 配置文件路径
|
||||
:param use_default_if_not_found: 如果配置文件不存在，是否使用默认配置
|
||||
"""
|
||||
if not os.path.exists(config_path):
|
||||
if use_default_if_not_found:
|
||||
return RootConfig[type]()
|
||||
else:
|
||||
raise FileNotFoundError(f"Config file not found: {config_path}")
|
||||
|
||||
with open(config_path, 'r', encoding='utf-8') as f:
|
||||
return RootConfig[type].model_validate_json(f.read())
|
||||
|
||||
def save_config(
|
||||
config: RootConfig[T],
|
||||
config_path: str,
|
||||
):
|
||||
"""将配置保存到指定路径"""
|
||||
RootConfig[T].model_validate(config)
|
||||
with open(config_path, 'w+', encoding='utf-8') as f:
|
||||
f.write(config.model_dump_json())
|
||||
|
|
@ -32,6 +32,7 @@ def initialize(module: str):
|
|||
logger.info(f'{len(task_registry)} task(s) and {len(action_registry)} action(s) loaded.')
|
||||
|
||||
def run(
|
||||
*,
|
||||
no_try: bool = False,
|
||||
):
|
||||
"""
|
||||
|
|
@ -42,6 +43,7 @@ def run(
|
|||
tasks = sorted(task_registry.values(), key=lambda x: x.priority, reverse=True)
|
||||
for task in tasks:
|
||||
logger.info(f'Task started: {task.name}')
|
||||
|
||||
if no_try:
|
||||
task.func()
|
||||
else:
|
||||
|
|
@ -54,10 +56,11 @@ def run(
|
|||
logger.info('All tasks finished.')
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.tasks.common import BaseConfig
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(filename)s:%(lineno)d] - %(message)s')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
logging.getLogger('kotonebot').setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
init_context(config_type=BaseConfig)
|
||||
initialize('kotonebot.tasks')
|
||||
run(no_try=True)
|
||||
|
||||
|
|
|
|||
|
|
@ -2,14 +2,19 @@
|
|||
import logging
|
||||
from time import sleep
|
||||
|
||||
from kotonebot import task, device, image, cropped
|
||||
from .actions.scenes import at_home, goto_home
|
||||
from . import R
|
||||
from .common import conf, BaseConfig
|
||||
from .actions.scenes import at_home, goto_home
|
||||
from kotonebot import task, device, image, cropped
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@task('收取活动费')
|
||||
def acquire_activity_funds():
|
||||
if not conf().activity_funds.enabled:
|
||||
logger.info('Activity funds acquisition is disabled.')
|
||||
return
|
||||
|
||||
if not at_home():
|
||||
goto_home()
|
||||
sleep(1)
|
||||
|
|
@ -23,9 +28,7 @@ def acquire_activity_funds():
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
acquire_activity_funds()
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
"""领取礼物（邮箱）"""
|
||||
import logging
|
||||
from time import sleep
|
||||
|
||||
from . import R
|
||||
from .common import conf
|
||||
from .actions.scenes import at_home, goto_home
|
||||
from kotonebot import device, image, task, color, rect_expand
|
||||
|
||||
|
|
@ -9,6 +11,10 @@ logger = logging.getLogger(__name__)
|
|||
|
||||
@task('领取礼物')
|
||||
def acquire_presents():
|
||||
if not conf().presents.enabled:
|
||||
logger.info('Presents acquisition is disabled.')
|
||||
return
|
||||
|
||||
if not at_home():
|
||||
goto_home()
|
||||
present = image.expect_wait(R.Daily.ButtonPresentsPartial, timeout=1)
|
||||
|
|
@ -30,11 +36,9 @@ def acquire_presents():
|
|||
goto_home()
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
# acquire_presents()
|
||||
print(image.find(R.Common.ButtonIconArrowShort, colored=True))
|
||||
print(image.find(R.Common.ButtonIconArrowShortDisabled, colored=True))
|
||||
|
|
|
|||
|
|
@ -157,8 +157,6 @@ def acquisitions() -> AcquisitionType | None:
|
|||
if __name__ == '__main__':
|
||||
from logging import getLogger
|
||||
import logging
|
||||
from kotonebot.backend.context import init_context
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s')
|
||||
getLogger('kotonebot').setLevel(logging.DEBUG)
|
||||
getLogger(__name__).setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
|
|
|
|||
|
|
@ -57,11 +57,9 @@ def check_and_skip_commu(img: MatLike | None = None) -> bool:
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
print(is_at_commu())
|
||||
# rect = image.expect(R.Common.ButtonCommuFastforward).rect
|
||||
# print(rect)
|
||||
|
|
|
|||
|
|
@ -1,19 +1,21 @@
|
|||
import random
|
||||
import time
|
||||
import cv2
|
||||
import random
|
||||
import logging
|
||||
import unicodedata
|
||||
from time import sleep
|
||||
from typing import Literal
|
||||
from typing_extensions import deprecated
|
||||
|
||||
import cv2
|
||||
|
||||
from .. import R
|
||||
from .scenes import at_home
|
||||
from . import loading
|
||||
from .scenes import at_home
|
||||
from .common import acquisitions
|
||||
from ..common import conf
|
||||
from kotonebot.backend.util import AdaptiveWait, crop_y, cropper_y
|
||||
from kotonebot import ocr, device, contains, image, regex, action, debug, config
|
||||
from .non_lesson_actions import enter_allowance, allowance_available, study_available, enter_study
|
||||
from kotonebot import ocr, device, contains, image, regex, action, debug
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -546,17 +548,17 @@ def produce_end():
|
|||
logger.debug("Click next")
|
||||
device.click(image.expect_wait(R.Common.ButtonNextNoIcon))
|
||||
elif image.find(R.InPurodyuusu.ButtonCancel):
|
||||
# CONFIG: 可选是否关注
|
||||
logger.info("Follow producer dialog found. Click to close.")
|
||||
device.click()
|
||||
# R.InPurodyuusu.ButtonFollowNoIcon
|
||||
if conf().produce.follow_producer:
|
||||
logger.info("Follow producer")
|
||||
device.click(image.expect_wait(R.InPurodyuusu.ButtonFollowNoIcon))
|
||||
else:
|
||||
logger.info("Skip follow producer")
|
||||
device.click()
|
||||
else:
|
||||
device.click_center()
|
||||
sleep(1)
|
||||
logger.info("Produce completed.")
|
||||
# 关注提示
|
||||
# if image.wait_for(R.InPurodyuusu.ButtonFollowProducer, timeout=2):
|
||||
# device.click(image.expect_wait(R.InPurodyuusu.ButtonCancel))
|
||||
|
||||
@action('执行 Regular 培育')
|
||||
def hajime_regular(week: int = -1, start_from: int = 1):
|
||||
|
|
@ -666,12 +668,10 @@ def purodyuusu(
|
|||
__actions__ = [enter_recommended_action]
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
from logging import getLogger
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s')
|
||||
getLogger('kotonebot').setLevel(logging.DEBUG)
|
||||
getLogger(__name__).setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
|
||||
# exam()
|
||||
# until_action_scene()
|
||||
|
|
|
|||
|
|
@ -51,7 +51,5 @@ def wait_loading_end(timeout: float = 60):
|
|||
sleep(1)
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
init_context()
|
||||
print(loading())
|
||||
input()
|
||||
|
|
|
|||
|
|
@ -71,8 +71,6 @@ def goto_shop():
|
|||
|
||||
|
||||
if __name__ == "__main__":
|
||||
from kotonebot.backend.context import init_context
|
||||
init_context()
|
||||
print(at_home())
|
||||
print(at_daily_shop())
|
||||
goto_shop()
|
||||
|
|
|
|||
|
|
@ -2,10 +2,12 @@
|
|||
import logging
|
||||
from time import sleep
|
||||
from typing import Literal
|
||||
from kotonebot import task, device, image, action, ocr, contains, cropped, rect_expand, color
|
||||
from kotonebot.tasks.actions.loading import wait_loading_end
|
||||
from .actions.scenes import at_home, goto_home
|
||||
|
||||
from . import R
|
||||
from .common import conf
|
||||
from .actions.loading import wait_loading_end
|
||||
from .actions.scenes import at_home, goto_home
|
||||
from kotonebot import task, device, image, action, ocr, contains, cropped, rect_expand, color
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -33,14 +35,17 @@ def assign(type: Literal['mini', 'online']) -> bool:
|
|||
:param type: 工作类型。mini=ミニライブ 或 online=ライブ配信。
|
||||
"""
|
||||
# [kotonebot/tasks/assignment.py]
|
||||
target_duration = 12
|
||||
image.expect_wait(R.Daily.IconTitleAssign, timeout=10)
|
||||
if type == 'mini':
|
||||
target_duration = conf().assignment.mini_live_duration
|
||||
if image.find(R.Daily.IconAssignMiniLive):
|
||||
device.click()
|
||||
else:
|
||||
logger.warning('MiniLive already assigned. Skipping...')
|
||||
return False
|
||||
elif type == 'online':
|
||||
target_duration = conf().assignment.online_live_duration
|
||||
if image.find(R.Daily.IconAssignOnlineLive):
|
||||
device.click()
|
||||
else:
|
||||
|
|
@ -88,12 +93,11 @@ def assign(type: Literal['mini', 'online']) -> bool:
|
|||
# 等待页面加载
|
||||
confirm = image.expect_wait(R.Common.ButtonConfirmNoIcon)
|
||||
# 选择时间 [screenshots/assignment/assign_mini_live2.png]
|
||||
# CONFIG: 工作时长
|
||||
if ocr.find(contains('12時間')):
|
||||
logger.info('12時間 selected.')
|
||||
if ocr.find(contains(f'{target_duration}時間')):
|
||||
logger.info(f'{target_duration}時間 selected.')
|
||||
device.click()
|
||||
else:
|
||||
logger.warning('12時間 not found. Using default duration.')
|
||||
logger.warning(f'{target_duration}時間 not found. Using default duration.')
|
||||
sleep(0.5)
|
||||
# 点击 决定する
|
||||
device.click(confirm)
|
||||
|
|
@ -104,6 +108,9 @@ def assign(type: Literal['mini', 'online']) -> bool:
|
|||
@task('工作')
|
||||
def assignment():
|
||||
"""领取工作奖励并重新分配工作"""
|
||||
if not conf().assignment.enabled:
|
||||
logger.info('Assignment is disabled.')
|
||||
return
|
||||
if not at_home():
|
||||
goto_home()
|
||||
btn_assignment = image.expect_wait(R.Daily.ButtonAssignmentPartial)
|
||||
|
|
@ -128,18 +135,22 @@ def assignment():
|
|||
logger.info('Assignment acquired.')
|
||||
# 领取完后会自动进入分配页面
|
||||
image.expect_wait(R.Daily.IconTitleAssign)
|
||||
if image.find(R.Daily.IconAssignMiniLive):
|
||||
assign('mini')
|
||||
sleep(6) # 等待动画结束
|
||||
# TODO: 更好的方法来等待动画结束。
|
||||
if image.find(R.Daily.IconAssignOnlineLive):
|
||||
assign('online')
|
||||
sleep(6) # 等待动画结束
|
||||
if conf().assignment.mini_live_reassign_enabled:
|
||||
if image.find(R.Daily.IconAssignMiniLive):
|
||||
assign('mini')
|
||||
sleep(6) # 等待动画结束
|
||||
# TODO: 更好的方法来等待动画结束。
|
||||
else:
|
||||
logger.info('MiniLive reassign is disabled.')
|
||||
if conf().assignment.online_live_reassign_enabled:
|
||||
if image.find(R.Daily.IconAssignOnlineLive):
|
||||
assign('online')
|
||||
sleep(6) # 等待动画结束
|
||||
else:
|
||||
logger.info('OnlineLive reassign is disabled.')
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
assignment()
|
||||
|
|
|
|||
|
|
@ -1,6 +1,127 @@
|
|||
from enum import IntEnum
|
||||
from enum import IntEnum, Enum
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from kotonebot import config
|
||||
|
||||
class Priority(IntEnum):
|
||||
START_GAME = 1
|
||||
DEFAULT = 0
|
||||
CLAIM_MISSION_REWARD = -1
|
||||
|
||||
|
||||
class APShopItems(IntEnum):
|
||||
PRODUCE_PT_UP = 0
|
||||
"""获取支援强化 Pt 提升"""
|
||||
PRODUCE_NOTE_UP = 1
|
||||
"""获取笔记数提升"""
|
||||
RECHALLENGE = 2
|
||||
"""再挑战券"""
|
||||
REGENERATE_MEMORY = 3
|
||||
"""回忆再生成券"""
|
||||
|
||||
|
||||
class PIdols(Enum):
|
||||
pass
|
||||
|
||||
|
||||
class PurchaseConfig(BaseModel):
|
||||
enabled: bool = False
|
||||
"""是否启用商店购买"""
|
||||
money_enabled: bool = False
|
||||
"""是否启用金币购买"""
|
||||
ap_enabled: bool = False
|
||||
"""是否启用AP购买"""
|
||||
ap_items: list[Literal[0, 1, 2, 3]] = []
|
||||
"""AP商店要购买的物品"""
|
||||
|
||||
|
||||
class ActivityFundsConfig(BaseModel):
|
||||
enabled: bool = False
|
||||
"""是否启用收取活动费"""
|
||||
|
||||
|
||||
class PresentsConfig(BaseModel):
|
||||
enabled: bool = False
|
||||
"""是否启用收取礼物"""
|
||||
|
||||
|
||||
class AssignmentConfig(BaseModel):
|
||||
enabled: bool = False
|
||||
"""是否启用工作"""
|
||||
|
||||
mini_live_reassign_enabled: bool = False
|
||||
"""是否启用重新分配 MiniLive"""
|
||||
mini_live_duration: Literal[4, 6, 12] = 12
|
||||
"""MiniLive 工作时长"""
|
||||
|
||||
online_live_reassign_enabled: bool = False
|
||||
"""是否启用重新分配 OnlineLive"""
|
||||
online_live_duration: Literal[4, 6, 12] = 12
|
||||
"""OnlineLive 工作时长"""
|
||||
|
||||
|
||||
class ContestConfig(BaseModel):
|
||||
enabled: bool = False
|
||||
"""是否启用竞赛"""
|
||||
|
||||
|
||||
class ProduceConfig(BaseModel):
|
||||
enabled: bool = False
|
||||
"""是否启用培育"""
|
||||
mode: Literal['regular'] = 'regular'
|
||||
"""培育模式。"""
|
||||
produce_count: int = 1
|
||||
"""培育的次数。"""
|
||||
idols: list[str] = []
|
||||
"""要培育的偶像。将会按顺序循环选择培育。"""
|
||||
memory_sets: list[int] = []
|
||||
"""要使用的回忆编成编号，从 1 开始。将会按顺序循环选择使用。"""
|
||||
support_card_sets: list[int] = []
|
||||
"""要使用的支援卡编成编号，从 1 开始。将会按顺序循环选择使用。"""
|
||||
auto_set_memory: bool = False
|
||||
"""是否自动编成回忆。此选项优先级高于回忆编成编号。"""
|
||||
auto_set_support_card: bool = False
|
||||
"""是否自动编成支援卡。此选项优先级高于支援卡编成编号。"""
|
||||
use_pt_boost: bool = False
|
||||
"""是否使用支援强化 Pt 提升。"""
|
||||
use_note_boost: bool = False
|
||||
"""是否使用笔记数提升。"""
|
||||
follow_producer: bool = False
|
||||
"""是否关注租借了支援卡的制作人。"""
|
||||
|
||||
|
||||
class MissionRewardConfig(BaseModel):
|
||||
enabled: bool = False
|
||||
"""是否启用领取任务奖励"""
|
||||
|
||||
|
||||
class BaseConfig(BaseModel):
|
||||
purchase: PurchaseConfig = PurchaseConfig()
|
||||
"""商店购买配置"""
|
||||
|
||||
activity_funds: ActivityFundsConfig = ActivityFundsConfig()
|
||||
"""活动费配置"""
|
||||
|
||||
presents: PresentsConfig = PresentsConfig()
|
||||
"""收取礼物配置"""
|
||||
|
||||
assignment: AssignmentConfig = AssignmentConfig()
|
||||
"""工作配置"""
|
||||
|
||||
contest: ContestConfig = ContestConfig()
|
||||
"""竞赛配置"""
|
||||
|
||||
produce: ProduceConfig = ProduceConfig()
|
||||
"""培育配置"""
|
||||
|
||||
mission_reward: MissionRewardConfig = MissionRewardConfig()
|
||||
"""领取任务奖励配置"""
|
||||
|
||||
|
||||
|
||||
def conf() -> BaseConfig:
|
||||
"""获取当前配置数据"""
|
||||
c = config.to(BaseConfig).current
|
||||
return c.options
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ from time import sleep
|
|||
from gettext import gettext as _
|
||||
|
||||
from . import R
|
||||
from .common import conf
|
||||
from .actions.scenes import at_home, goto_home
|
||||
from .actions.loading import wait_loading_end
|
||||
from kotonebot import device, image, ocr, color, action, task, user, rect_expand
|
||||
|
|
@ -106,6 +107,9 @@ def pick_and_contest(has_ongoing_contest: bool = False) -> bool:
|
|||
@task('竞赛')
|
||||
def contest():
|
||||
""""""
|
||||
if not conf().contest.enabled:
|
||||
logger.info('Contest is disabled.')
|
||||
return
|
||||
logger.info('Contest started.')
|
||||
if not at_home():
|
||||
goto_home()
|
||||
|
|
@ -124,11 +128,9 @@ def contest():
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logging.getLogger('kotonebot').setLevel(logging.DEBUG)
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
|
||||
contest()
|
||||
|
|
@ -1,12 +1,13 @@
|
|||
"""领取任务奖励"""
|
||||
from time import sleep
|
||||
import logging
|
||||
from time import sleep
|
||||
|
||||
from kotonebot import device, image, color, task, action, rect_expand
|
||||
from .actions.scenes import at_home, goto_home
|
||||
from .common import Priority
|
||||
from . import R
|
||||
from .common import conf, Priority
|
||||
from .actions.loading import wait_loading_end
|
||||
from .actions.scenes import at_home, goto_home
|
||||
from kotonebot import device, image, color, task, action, rect_expand
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@action('检查任务')
|
||||
|
|
@ -117,6 +118,9 @@ def mission_reward():
|
|||
"""
|
||||
领取任务奖励
|
||||
"""
|
||||
if not conf().mission_reward.enabled:
|
||||
logger.info('Mission reward is disabled.')
|
||||
return
|
||||
logger.info('Claiming mission rewards.')
|
||||
if not at_home():
|
||||
goto_home()
|
||||
|
|
@ -130,12 +134,10 @@ def mission_reward():
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logging.getLogger('kotonebot').setLevel(logging.DEBUG)
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
|
||||
# if image.find(R.Common.CheckboxUnchecked):
|
||||
# logger.debug('Checking skip all.')
|
||||
|
|
|
|||
|
|
@ -1,17 +1,21 @@
|
|||
from time import sleep
|
||||
import logging
|
||||
|
||||
from kotonebot import device, image, ocr, task, action
|
||||
from . import R
|
||||
from .actions.scenes import loading, at_home, goto_home
|
||||
from .common import conf
|
||||
from .actions.loading import wait_loading_end
|
||||
from .actions.in_purodyuusu import hajime_regular
|
||||
from kotonebot import device, image, ocr, task, action
|
||||
from .actions.scenes import loading, at_home, goto_home
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@task('培育')
|
||||
def produce():
|
||||
"""进行培育流程"""
|
||||
if not conf().produce.enabled:
|
||||
logger.info('Produce is disabled.')
|
||||
return
|
||||
if not at_home():
|
||||
goto_home()
|
||||
# [screenshots/produce/home.png]
|
||||
|
|
@ -44,11 +48,12 @@ def produce():
|
|||
device.click(image.expect(R.Common.ButtonNextNoIcon))
|
||||
sleep(0.3)
|
||||
# 选择道具 [screenshots/produce/select_end.png]
|
||||
# CONFIG:
|
||||
device.click(image.expect_wait(R.Produce.CheckboxIconNoteBoost))
|
||||
sleep(0.1)
|
||||
device.click(image.expect_wait(R.Produce.CheckboxIconSupportPtBoost))
|
||||
sleep(0.1)
|
||||
if conf().produce.use_note_boost:
|
||||
device.click(image.expect_wait(R.Produce.CheckboxIconNoteBoost))
|
||||
sleep(0.2)
|
||||
if conf().produce.use_pt_boost:
|
||||
device.click(image.expect_wait(R.Produce.CheckboxIconSupportPtBoost))
|
||||
sleep(0.2)
|
||||
device.click(image.expect_wait(R.Produce.ButtonProduceStart))
|
||||
sleep(0.5)
|
||||
while not loading():
|
||||
|
|
@ -64,12 +69,10 @@ def produce():
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logging.getLogger('kotonebot').setLevel(logging.DEBUG)
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
produce()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -2,11 +2,11 @@
|
|||
import logging
|
||||
from time import sleep
|
||||
|
||||
from kotonebot import task
|
||||
from kotonebot import device, image, ocr, action
|
||||
from kotonebot.backend.util import cropped
|
||||
from .actions.scenes import goto_home, goto_shop, at_daily_shop
|
||||
from . import R
|
||||
from .common import conf
|
||||
from kotonebot.backend.util import cropped
|
||||
from kotonebot import task, device, image, ocr, action
|
||||
from .actions.scenes import goto_home, goto_shop, at_daily_shop
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -44,13 +44,11 @@ def money_items():
|
|||
logger.info(f'Purchasing マニー items completed. {index} items purchased.')
|
||||
|
||||
@action('购买 AP 物品')
|
||||
def ap_items(item_indices: list[int]):
|
||||
def ap_items():
|
||||
"""
|
||||
购买 AP 物品
|
||||
|
||||
前置条件：位于商店页面的 AP Tab
|
||||
|
||||
:param item_indices: 要购买的物品索引列表，从 0 开始。
|
||||
"""
|
||||
# [screenshots\shop\ap1.png]
|
||||
logger.info(f'Purchasing AP items.')
|
||||
|
|
@ -58,6 +56,9 @@ def ap_items(item_indices: list[int]):
|
|||
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.')
|
||||
|
|
@ -86,29 +87,41 @@ def purchase():
|
|||
"""
|
||||
从商店购买物品
|
||||
"""
|
||||
if not conf().purchase.enabled:
|
||||
logger.info('Purchase is disabled.')
|
||||
return
|
||||
if not at_daily_shop():
|
||||
goto_shop()
|
||||
# 进入每日商店 [screenshots\shop\shop.png]
|
||||
# [ap1.png]
|
||||
device.click(image.expect(R.Daily.ButtonDailyShop)) # TODO: memoable
|
||||
sleep(1)
|
||||
|
||||
# 购买マニー物品
|
||||
image.expect_wait(R.Daily.IconShopMoney)
|
||||
money_items()
|
||||
# 点击 AP 选项卡
|
||||
device.click(image.expect_wait(R.Daily.TextTabShopAp, timeout=2)) # TODO: memoable
|
||||
# 等待 AP 选项卡加载完成
|
||||
image.expect_wait(R.Daily.IconShopAp)
|
||||
ap_items([0, 1, 2, 3])
|
||||
sleep(0.5)
|
||||
if conf().purchase.money_enabled:
|
||||
image.expect_wait(R.Daily.IconShopMoney)
|
||||
money_items()
|
||||
sleep(0.5)
|
||||
else:
|
||||
logger.info('Money purchase is disabled.')
|
||||
|
||||
# 购买 AP 物品
|
||||
if conf().purchase.ap_enabled:
|
||||
# 点击 AP 选项卡
|
||||
device.click(image.expect_wait(R.Daily.TextTabShopAp, timeout=2)) # TODO: memoable
|
||||
# 等待 AP 选项卡加载完成
|
||||
image.expect_wait(R.Daily.IconShopAp)
|
||||
ap_items()
|
||||
sleep(0.5)
|
||||
else:
|
||||
logger.info('AP purchase is disabled.')
|
||||
|
||||
goto_home()
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
# money_items()
|
||||
# ap_items([0, 1, 3])
|
||||
purchase()
|
||||
|
|
|
|||
|
|
@ -50,10 +50,8 @@ def start_game():
|
|||
wait()
|
||||
|
||||
if __name__ == '__main__':
|
||||
from kotonebot.backend.context import init_context
|
||||
import logging
|
||||
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
init_context()
|
||||
start_game()
|
||||
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ def info(
|
|||
|
||||
def warning(
|
||||
message: str,
|
||||
images: list[MatLike],
|
||||
images: list[MatLike] | None = None,
|
||||
*,
|
||||
once: bool = False
|
||||
):
|
||||
|
|
|
|||
|
|
@ -13,5 +13,7 @@ python-multipart==0.0.20
|
|||
websockets==14.1
|
||||
numpy==2.2.1
|
||||
psutil==6.1.1
|
||||
# 配置读写
|
||||
pydantic==2.10.4
|
||||
# 其他
|
||||
typing-extensions==4.12.2
|
||||
|
|
|
|||
|
|
@ -94,11 +94,12 @@ class BaseTestCase(unittest.TestCase):
|
|||
cls.device = MockDevice()
|
||||
from kotonebot.backend.debug.server import start_server
|
||||
from kotonebot.backend.debug import debug
|
||||
from kotonebot.tasks.common import BaseConfig
|
||||
debug.enabled = True
|
||||
debug.wait_for_message_sent = True
|
||||
start_server()
|
||||
from kotonebot.backend.context import init_context
|
||||
init_context()
|
||||
init_context(config_type=BaseConfig)
|
||||
from kotonebot.backend.context import _c
|
||||
assert _c is not None, 'context is not initialized'
|
||||
_c.inject_device(cls.device)
|
||||
|
|
|
|||
Loading…
Reference in New Issue