suyiiyii/nonebot-bison
1
0
Fork 0
mirror of https://github.com/suyiiyii/nonebot-bison.git synced 2025-06-02 09:26:12 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

♻️ rename scheduler_config -> site

Browse Source
This commit is contained in:
felinae98 2024-06-02 22:36:02 +08:00
parent 2fc11a9653
commit 5bd8b21f19
27 changed files with 135 additions and 169 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
nonebot_bison/apis.py
View File

@ -5,7 +5,7 @@ from .platform import platform_manager
async def check_sub_target(platform_name: str, target: Target):
platform = platform_manager[platform_name]
scheduler_conf_class = platform.scheduler
scheduler_conf_class = platform.site
scheduler = scheduler_dict[scheduler_conf_class]
client = await scheduler.client_mgr.get_query_name_client()

12
nonebot_bison/platform/arknights.py
View File

@ -8,9 +8,9 @@ from pydantic import Field, BaseModel
from nonebot.compat import type_validate_python
from ..post import Post
from ..utils import Site
from ..types import Target, RawPost, Category
from .platform import NewMessage, StatusChange
from ..utils.scheduler_config import SchedulerConfig
class ArkResponseBase(BaseModel):
@ -52,7 +52,7 @@ class ArkBulletinResponse(ArkResponseBase):
data: BulletinData
class ArknightsSchedConf(SchedulerConfig):
class ArknightsSite(Site):
name = "arknights"
schedule_type = "interval"
schedule_setting = {"seconds": 30}
@ -65,7 +65,7 @@ class Arknights(NewMessage):
enable_tag = False
enabled = True
is_common = False
scheduler = ArknightsSchedConf
site = ArknightsSite
has_target = False
default_theme = "arknights"
@ -127,7 +127,7 @@ class AkVersion(StatusChange):
enable_tag = False
enabled = True
is_common = False
scheduler = ArknightsSchedConf
site = ArknightsSite
has_target = False
default_theme = "brief"
@ -172,7 +172,7 @@ class MonsterSiren(NewMessage):
enable_tag = False
enabled = True
is_common = False
scheduler = ArknightsSchedConf
site = ArknightsSite
has_target = False
@classmethod
@ -220,7 +220,7 @@ class TerraHistoricusComic(NewMessage):
enable_tag = False
enabled = True
is_common = False
scheduler = ArknightsSchedConf
site = ArknightsSite
has_target = False
default_theme = "brief"

20
nonebot_bison/platform/bilibili.py
View File

@ -11,12 +11,10 @@ from nonebot.log import logger
from pydantic import Field, BaseModel
from nonebot.compat import PYDANTIC_V2, ConfigDict, type_validate_json, type_validate_python
from nonebot_bison.compat import model_rebuild
from nonebot_bison.utils.scheduler_config import ClientManager
from ..post import Post
from ..compat import model_rebuild
from ..types import Tag, Target, RawPost, ApiError, Category
from ..utils import SchedulerConfig, http_client, text_similarity
from ..utils import Site, ClientManager, http_client, text_similarity
from .platform import NewMessage, StatusChange, CategoryNotSupport, CategoryNotRecognize
TBaseModel = TypeVar("TBaseModel", bound=type[BaseModel])
@ -133,18 +131,18 @@ class BilibiliClient(ClientManager):
return self._client
class BilibiliSchedConf(SchedulerConfig):
class BilibiliSite(Site):
name = "bilibili.com"
schedule_type = "interval"
schedule_setting = {"seconds": 10}
client_man = BilibiliClient
client_mgr = BilibiliClient
class BililiveSchedConf(SchedulerConfig):
class BililiveSchedConf(Site):
name = "live.bilibili.com"
schedule_type = "interval"
schedule_setting = {"seconds": 3}
client_man = BilibiliClient
client_mgr = BilibiliClient
class Bilibili(NewMessage):
@ -160,7 +158,7 @@ class Bilibili(NewMessage):
enable_tag = True
enabled = True
is_common = True
scheduler = BilibiliSchedConf
site = BilibiliSite
name = "B站"
has_target = True
parse_target_promot = "请输入用户主页的链接"
@ -333,7 +331,7 @@ class Bilibililive(StatusChange):
enable_tag = False
enabled = True
is_common = True
scheduler = BililiveSchedConf
site = BililiveSchedConf
name = "Bilibili直播"
has_target = True
use_batch = True
@ -483,7 +481,7 @@ class BilibiliBangumi(StatusChange):
enable_tag = False
enabled = True
is_common = True
scheduler = BilibiliSchedConf
site = BilibiliSite
name = "Bilibili剧集"
has_target = True
parse_target_promot = "请输入剧集主页"

4
nonebot_bison/platform/ff14.py
View File

@ -3,8 +3,8 @@ from typing import Any
from httpx import AsyncClient
from ..post import Post
from ..utils import scheduler
from .platform import NewMessage
from ..utils import anonymous_site
from ..types import Target, RawPost
@ -16,7 +16,7 @@ class FF14(NewMessage):
enabled = True
is_common = False
scheduler_class = "ff14"
scheduler = scheduler("interval", {"seconds": 60})
site = anonymous_site("interval", {"seconds": 60})
has_target = False
@classmethod

8
nonebot_bison/platform/ncm.py
View File

@ -4,12 +4,12 @@ from typing import Any
from httpx import AsyncClient
from ..post import Post
from ..utils import Site
from .platform import NewMessage
from ..utils import SchedulerConfig
from ..types import Target, RawPost, ApiError
class NcmSchedConf(SchedulerConfig):
class NcmSite(Site):
name = "music.163.com"
schedule_type = "interval"
schedule_setting = {"minutes": 1}
@ -21,7 +21,7 @@ class NcmArtist(NewMessage):
enable_tag = False
enabled = True
is_common = True
scheduler = NcmSchedConf
site = NcmSite
name = "网易云-歌手"
has_target = True
parse_target_promot = "请输入歌手主页（包含数字ID）的链接"
@ -78,7 +78,7 @@ class NcmRadio(NewMessage):
enable_tag = False
enabled = True
is_common = False
scheduler = NcmSchedConf
site = NcmSite
name = "网易云-电台"
has_target = True
parse_target_promot = "请输入主播电台主页（包含数字ID）的链接"

10
nonebot_bison/platform/platform.py
View File

@ -14,8 +14,8 @@ from nonebot.log import logger
from nonebot_plugin_saa import PlatformTarget
from ..post import Post
from ..utils import Site, ProcessContext
from ..plugin_config import plugin_config
from ..utils import ProcessContext, SchedulerConfig
from ..types import Tag, Target, RawPost, SubUnit, Category
@ -81,7 +81,7 @@ class PlatformABCMeta(PlatformMeta, ABC): ...
class Platform(metaclass=PlatformABCMeta, base=True):
scheduler: type[SchedulerConfig]
site: type[Site]
ctx: ProcessContext
is_common: bool
enabled: bool
@ -444,7 +444,7 @@ def make_no_target_group(platform_list: list[type[Platform]]) -> type[Platform]:
name = DUMMY_STR
categories_keys = set()
categories = {}
scheduler = platform_list[0].scheduler
site = platform_list[0].site
for platform in platform_list:
if platform.has_target:
@ -458,7 +458,7 @@ def make_no_target_group(platform_list: list[type[Platform]]) -> type[Platform]:
raise RuntimeError(f"Platform categories for {platform_name} duplicate")
categories_keys |= platform_category_key_set
categories.update(platform.categories)
if platform.scheduler != scheduler:
if platform.site != site:
raise RuntimeError(f"Platform scheduler for {platform_name} not fit")
def __init__(self: "NoTargetGroup", ctx: ProcessContext):
@ -490,7 +490,7 @@ def make_no_target_group(platform_list: list[type[Platform]]) -> type[Platform]:
"platform_name": platform_list[0].platform_name,
"name": name,
"categories": categories,
"scheduler": scheduler,
"site": site,
"is_common": platform_list[0].is_common,
"enabled": True,
"has_target": False,

6
nonebot_bison/platform/rss.py
View File

@ -9,10 +9,10 @@ from bs4 import BeautifulSoup as bs
from ..post import Post
from .platform import NewMessage
from ..types import Target, RawPost
from ..utils import SchedulerConfig, text_similarity
from ..utils import Site, text_similarity
class RssSchedConf(SchedulerConfig):
class RssSite(Site):
name = "rss"
schedule_type = "interval"
schedule_setting = {"seconds": 30}
@ -25,7 +25,7 @@ class Rss(NewMessage):
name = "Rss"
enabled = True
is_common = True
scheduler = RssSchedConf
site = RssSite
has_target = True
@classmethod

6
nonebot_bison/platform/weibo.py
View File

@ -12,7 +12,7 @@ from bs4 import BeautifulSoup as bs
from ..post import Post
from .platform import NewMessage
from ..utils import SchedulerConfig, http_client
from ..utils import Site, http_client
from ..types import Tag, Target, RawPost, ApiError, Category
_HEADER = {
@ -35,7 +35,7 @@ _HEADER = {
}
class WeiboSchedConf(SchedulerConfig):
class WeiboSite(Site):
name = "weibo.com"
schedule_type = "interval"
schedule_setting = {"seconds": 3}
@ -53,7 +53,7 @@ class Weibo(NewMessage):
name = "新浪微博"
enabled = True
is_common = True
scheduler = WeiboSchedConf
site = WeiboSite
has_target = True
parse_target_promot = "请输入用户主页（包含数字UID）的链接"

40
nonebot_bison/scheduler/manager.py
View File

@ -1,37 +1,37 @@
from nonebot.log import logger
from ..utils import Site
from ..config import config
from .scheduler import Scheduler
from ..utils import SchedulerConfig
from ..config.db_model import Target
from ..types import Target as T_Target
from ..platform import platform_manager
from ..plugin_config import plugin_config
scheduler_dict: dict[type[SchedulerConfig], Scheduler] = {}
scheduler_dict: dict[type[Site], Scheduler] = {}
async def init_scheduler():
_schedule_class_dict: dict[type[SchedulerConfig], list[Target]] = {}
_schedule_class_platform_dict: dict[type[SchedulerConfig], list[str]] = {}
_schedule_class_dict: dict[type[Site], list[Target]] = {}
_schedule_class_platform_dict: dict[type[Site], list[str]] = {}
for platform in platform_manager.values():
scheduler_config = platform.scheduler
if not hasattr(scheduler_config, "name") or not scheduler_config.name:
scheduler_config.name = f"AnonymousScheduleConfig[{platform.platform_name}]"
site = platform.site
if not hasattr(site, "name") or not site.name:
site.name = f"AnonymousScheduleConfig[{platform.platform_name}]"
platform_name = platform.platform_name
targets = await config.get_platform_target(platform_name)
if scheduler_config not in _schedule_class_dict:
_schedule_class_dict[scheduler_config] = list(targets)
if site not in _schedule_class_dict:
_schedule_class_dict[site] = list(targets)
else:
_schedule_class_dict[scheduler_config].extend(targets)
if scheduler_config not in _schedule_class_platform_dict:
_schedule_class_platform_dict[scheduler_config] = [platform_name]
_schedule_class_dict[site].extend(targets)
if site not in _schedule_class_platform_dict:
_schedule_class_platform_dict[site] = [platform_name]
else:
_schedule_class_platform_dict[scheduler_config].append(platform_name)
for scheduler_config, target_list in _schedule_class_dict.items():
if not plugin_config.bison_use_browser and scheduler_config.require_browser:
logger.warning(f"{scheduler_config.name} requires browser, it will not schedule.")
_schedule_class_platform_dict[site].append(platform_name)
for site, target_list in _schedule_class_dict.items():
if not plugin_config.bison_use_browser and site.require_browser:
logger.warning(f"{site.name} requires browser, it will not schedule.")
continue
schedulable_args = []
@ -39,19 +39,19 @@ async def init_scheduler():
schedulable_args.append(
(target.platform_name, T_Target(target.target), platform_manager[target.platform_name].use_batch)
)
platform_name_list = _schedule_class_platform_dict[scheduler_config]
scheduler_dict[scheduler_config] = Scheduler(scheduler_config, schedulable_args, platform_name_list)
platform_name_list = _schedule_class_platform_dict[site]
scheduler_dict[site] = Scheduler(site, schedulable_args, platform_name_list)
config.register_add_target_hook(handle_insert_new_target)
config.register_delete_target_hook(handle_delete_target)
async def handle_insert_new_target(platform_name: str, target: T_Target):
platform = platform_manager[platform_name]
scheduler_obj = scheduler_dict[platform.scheduler]
scheduler_obj = scheduler_dict[platform.site]
scheduler_obj.insert_new_schedulable(platform_name, target)
async def handle_delete_target(platform_name: str, target: T_Target):
platform = platform_manager[platform_name]
scheduler_obj = scheduler_dict[platform.scheduler]
scheduler_obj = scheduler_dict[platform.site]
scheduler_obj.delete_schedulable(platform_name, target)

6
nonebot_bison/scheduler/scheduler.py
View File

@ -5,13 +5,13 @@ from nonebot.log import logger
from nonebot_plugin_apscheduler import scheduler
from nonebot_plugin_saa.utils.exceptions import NoBotFound
from nonebot_bison.utils.scheduler_config import ClientManager
from nonebot_bison.utils import ClientManager
from ..config import config
from ..send import send_msgs
from ..types import Target, SubUnit
from ..platform import platform_manager
from ..utils import ProcessContext, SchedulerConfig
from ..utils import Site, ProcessContext
@dataclass
@ -30,7 +30,7 @@ class Scheduler:
def __init__(
self,
scheduler_config: type[SchedulerConfig],
scheduler_config: type[Site],
schedulables: list[tuple[str, Target, bool]], # [(platform_name, target, use_batch)]
platform_name_list: list[str],
):

30
nonebot_bison/utils/__init__.py
View File

@ -8,27 +8,17 @@ from bs4 import BeautifulSoup as bs
from nonebot.log import logger, default_format
from nonebot_plugin_saa import Text, Image, MessageSegmentFactory
from .http import http_client
from .context import ProcessContext
from .site import Site as Site
from ..plugin_config import plugin_config
from .image import pic_merge, text_to_image, is_pics_mergable, pic_url_to_image
from .scheduler_config import ClientManager, SchedulerConfig, DefaultClientManager, scheduler
__all__ = [
"http_client",
"Singleton",
"parse_text",
"ProcessContext",
"ClientManager",
"DefaultClientManager",
"html_to_text",
"SchedulerConfig",
"scheduler",
"pic_merge",
"pic_url_to_image",
"is_pics_mergable",
"text_to_image",
]
from .image import pic_merge as pic_merge
from .http import http_client as http_client
from .site import ClientManager as ClientManager
from .image import text_to_image as text_to_image
from .site import anonymous_site as anonymous_site
from .context import ProcessContext as ProcessContext
from .image import is_pics_mergable as is_pics_mergable
from .image import pic_url_to_image as pic_url_to_image
from .site import DefaultClientManager as DefaultClientManager
class Singleton(type):

2
nonebot_bison/utils/context.py
View File

@ -4,7 +4,7 @@ from httpx import Response, AsyncClient
from nonebot_bison.types import Target
from .scheduler_config import ClientManager
from .site import ClientManager
class ProcessContext:

8
nonebot_bison/utils/scheduler_config.py → nonebot_bison/utils/site.py
View File

@ -28,7 +28,7 @@ class DefaultClientManager(ClientManager):
return http_client()
class SchedulerConfig:
class Site:
schedule_type: Literal["date", "interval", "cron"]
schedule_setting: dict
name: str
@ -42,10 +42,10 @@ class SchedulerConfig:
self.default_http_client = http_client()
def scheduler(schedule_type: Literal["date", "interval", "cron"], schedule_setting: dict) -> type[SchedulerConfig]:
def anonymous_site(schedule_type: Literal["date", "interval", "cron"], schedule_setting: dict) -> type[Site]:
return type(
"AnonymousScheduleConfig",
(SchedulerConfig,),
"AnonymousSite",
(Site,),
{
"schedule_type": schedule_type,
"schedule_setting": schedule_setting,

12
tests/platforms/test_arknights.py
View File

@ -11,9 +11,8 @@ from .utils import get_file, get_json
@pytest.fixture()
def arknights(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["arknights"](ProcessContext(DefaultClientManager()))
@ -45,8 +44,7 @@ def monster_siren_list_1():
@respx.mock
async def test_url_parse(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
from nonebot_bison.platform.arknights import Arknights, BulletinData, BulletinListItem, ArkBulletinResponse
cid_router = respx.get("https://ak-webview.hypergryph.com/api/game/bulletin/1")
@ -114,8 +112,7 @@ async def test_url_parse(app: App):
@pytest.mark.asyncio()
async def test_get_date_in_bulletin(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
from nonebot_bison.platform.arknights import Arknights, BulletinListItem
arknights = Arknights(ProcessContext(DefaultClientManager()))
@ -137,8 +134,7 @@ async def test_get_date_in_bulletin(app: App):
@pytest.mark.asyncio()
@respx.mock
async def test_parse_with_breakline(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
from nonebot_bison.platform.arknights import Arknights, BulletinListItem
detail = get_json("arknights-detail-805")

3
tests/platforms/test_bilibili.py
View File

@ -23,9 +23,8 @@ if typing.TYPE_CHECKING:
@pytest.fixture()
def bilibili(app: App) -> "Bilibili":
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["bilibili"](ProcessContext(DefaultClientManager())) # type: ignore

3
tests/platforms/test_bilibili_bangumi.py
View File

@ -13,9 +13,8 @@ if typing.TYPE_CHECKING:
@pytest.fixture()
def bili_bangumi(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["bilibili-bangumi"](ProcessContext(DefaultClientManager()))

3
tests/platforms/test_ff14.py
View File

@ -8,9 +8,8 @@ from .utils import get_json
@pytest.fixture()
def ff14(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["ff14"](ProcessContext(DefaultClientManager()))

3
tests/platforms/test_ncm_artist.py
View File

@ -14,9 +14,8 @@ if typing.TYPE_CHECKING:
@pytest.fixture()
def ncm_artist(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["ncm-artist"](ProcessContext(DefaultClientManager()))

3
tests/platforms/test_ncm_radio.py
View File

@ -14,9 +14,8 @@ if typing.TYPE_CHECKING:
@pytest.fixture()
def ncm_radio(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["ncm-radio"](ProcessContext(DefaultClientManager()))

45
tests/platforms/test_platform.py
View File

@ -84,12 +84,12 @@ def mock_platform_without_cats_tags(app: App):
@pytest.fixture()
def mock_platform(app: App):
from nonebot_bison.post import Post
from nonebot_bison.utils import SchedulerConfig
from nonebot_bison.utils import Site
from nonebot_bison.platform.platform import NewMessage
from nonebot_bison.types import Tag, Target, RawPost, Category
class MockPlatformSchedConf(SchedulerConfig):
name = "mock"
class MockSite(Site):
name = "mock_site"
schedule_type = "interval"
schedule_setting = {"seconds": 100}
@ -100,7 +100,7 @@ def mock_platform(app: App):
is_common = True
enable_tag = True
has_target = True
scheduler = MockPlatformSchedConf
site = MockSite
categories = {
Category(1): "转发",
Category(2): "视频",
@ -144,19 +144,19 @@ def mock_platform(app: App):
@pytest.fixture()
def mock_scheduler_conf(app):
from nonebot_bison.utils import SchedulerConfig
def mock_site(app):
from nonebot_bison.utils import Site
class MockPlatformSchedConf(SchedulerConfig):
name = "mock"
class MockSite(Site):
name = "mock_site"
schedule_type = "interval"
schedule_setting = {"seconds": 100}
return MockPlatformSchedConf
return MockSite
@pytest.fixture()
def mock_platform_no_target(app: App, mock_scheduler_conf):
def mock_platform_no_target(app: App, mock_site):
from nonebot_bison.post import Post
from nonebot_bison.types import Tag, Target, RawPost, Category
from nonebot_bison.platform.platform import NewMessage, CategoryNotSupport
@ -166,7 +166,7 @@ def mock_platform_no_target(app: App, mock_scheduler_conf):
name = "Mock Platform"
enabled = True
is_common = True
scheduler = mock_scheduler_conf
site = mock_site
enable_tag = True
has_target = False
categories = {Category(1): "转发", Category(2): "视频", Category(3): "不支持"}
@ -211,7 +211,7 @@ def mock_platform_no_target(app: App, mock_scheduler_conf):
@pytest.fixture()
def mock_platform_no_target_2(app: App, mock_scheduler_conf):
def mock_platform_no_target_2(app: App, mock_site):
from nonebot_bison.post import Post
from nonebot_bison.platform.platform import NewMessage
from nonebot_bison.types import Tag, Target, RawPost, Category
@ -220,7 +220,7 @@ def mock_platform_no_target_2(app: App, mock_scheduler_conf):
platform_name = "mock_platform"
name = "Mock Platform"
enabled = True
scheduler = mock_scheduler_conf
site = mock_site
is_common = True
enable_tag = True
has_target = False
@ -323,9 +323,8 @@ def mock_status_change(app: App):
@pytest.mark.asyncio
async def test_new_message_target_without_cats_tags(mock_platform_without_cats_tags, user_info_factory):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.types import Target, SubUnit
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
res1 = await mock_platform_without_cats_tags(ProcessContext(DefaultClientManager())).fetch_new_post(
SubUnit(Target("dummy"), [user_info_factory([1, 2], [])])
@ -345,9 +344,8 @@ async def test_new_message_target_without_cats_tags(mock_platform_without_cats_t
@pytest.mark.asyncio
async def test_new_message_target(mock_platform, user_info_factory):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.types import Target, SubUnit
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
res1 = await mock_platform(ProcessContext(DefaultClientManager())).fetch_new_post(
SubUnit(Target("dummy"), [user_info_factory([1, 2], [])])
@ -381,9 +379,8 @@ async def test_new_message_target(mock_platform, user_info_factory):
@pytest.mark.asyncio
async def test_new_message_no_target(mock_platform_no_target, user_info_factory):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.types import Target, SubUnit
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
res1 = await mock_platform_no_target(ProcessContext(DefaultClientManager())).fetch_new_post(
SubUnit(Target("dummy"), [user_info_factory([1, 2], [])])
@ -421,9 +418,8 @@ async def test_new_message_no_target(mock_platform_no_target, user_info_factory)
@pytest.mark.asyncio
async def test_status_change(mock_status_change, user_info_factory):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.types import Target, SubUnit
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
res1 = await mock_status_change(ProcessContext(DefaultClientManager())).fetch_new_post(
SubUnit(Target("dummy"), [user_info_factory([1, 2], [])])
@ -462,10 +458,9 @@ async def test_group(
mock_platform_no_target_2,
user_info_factory,
):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.types import Target, SubUnit
from nonebot_bison.platform.platform import make_no_target_group
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
dummy = Target("dummy")
@ -488,10 +483,10 @@ async def test_batch_fetch_new_message(app: App):
from nonebot_plugin_saa import TargetQQGroup
from nonebot_bison.post import Post
from nonebot_bison.utils import DefaultClientManager
from nonebot_bison.platform.platform import NewMessage
from nonebot_bison.utils.context import ProcessContext
from nonebot_bison.types import Target, RawPost, SubUnit, UserSubInfo
from nonebot_bison.utils.scheduler_config import DefaultClientManager
class BatchNewMessage(NewMessage):
platform_name = "mock_platform"
@ -575,9 +570,9 @@ async def test_batch_fetch_compare_status(app: App):
from nonebot_plugin_saa import TargetQQGroup
from nonebot_bison.post import Post
from nonebot_bison.utils import DefaultClientManager
from nonebot_bison.utils.context import ProcessContext
from nonebot_bison.platform.platform import StatusChange
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.types import Target, RawPost, SubUnit, Category, UserSubInfo
class BatchStatusChange(StatusChange):

6
tests/platforms/test_platform_tag_filter.py
View File

@ -12,9 +12,8 @@ def test_cases():
# 测试正反tag的判断情况
@pytest.mark.asyncio
async def test_filter_user_custom_tag(app: App, test_cases):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
bilibili = platform_manager["bilibili"](ProcessContext(DefaultClientManager()))
for case in test_cases:
@ -25,9 +24,8 @@ async def test_filter_user_custom_tag(app: App, test_cases):
# 测试正反tag的分离情况
@pytest.mark.asyncio
async def test_tag_separator(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
bilibili = platform_manager["bilibili"](ProcessContext(DefaultClientManager()))
tags = ["~111", "222", "333", "~444", "555"]

3
tests/platforms/test_rss.py
View File

@ -34,9 +34,8 @@ def user_info_factory(app: App, dummy_user):
@pytest.fixture()
def rss(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["rss"](ProcessContext(DefaultClientManager()))

3
tests/platforms/test_weibo.py
View File

@ -18,9 +18,8 @@ image_cdn_router = respx.route(host__regex=r"wx\d.sinaimg.cn", path__startswith=
@pytest.fixture()
def weibo(app: App):
from nonebot_bison.utils import ProcessContext
from nonebot_bison.platform import platform_manager
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return platform_manager["weibo"](ProcessContext(DefaultClientManager()))

6
tests/post/test_generate.py
View File

@ -170,9 +170,8 @@ async def test_generate_msg(mock_platform):
from nonebot_plugin_saa import Text, Image
from nonebot_bison.post import Post
from nonebot_bison.utils import ProcessContext
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
post: Post = await mock_platform(ProcessContext(DefaultClientManager())).parse(raw_post_list_1[0])
assert post.platform.default_theme == "basic"
@ -201,9 +200,8 @@ async def test_msg_segments_convert(mock_platform):
from nonebot_plugin_saa import Image
from nonebot_bison.post import Post
from nonebot_bison.utils import ProcessContext
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
plugin_config.bison_use_pic = True

54
tests/scheduler/test_scheduler.py → tests/scheduler/test_site.py
View File

@ -6,13 +6,13 @@ from nonebug import App
from pytest_mock import MockerFixture
if typing.TYPE_CHECKING:
from nonebot_bison.utils.scheduler_config import SchedulerConfig
from nonebot_bison.utils import Site
async def get_schedule_times(scheduler_config: type["SchedulerConfig"], time: int) -> dict[str, int]:
async def get_schedule_times(site: type["Site"], time: int) -> dict[str, int]:
from nonebot_bison.scheduler import scheduler_dict
scheduler = scheduler_dict[scheduler_config]
scheduler = scheduler_dict[site]
res = {}
for _ in range(time):
schedulable = await scheduler.get_next_schedulable()
@ -28,8 +28,8 @@ async def test_scheduler_without_time(init_scheduler):
from nonebot_bison.config import config
from nonebot_bison.types import Target as T_Target
from nonebot_bison.config.db_config import WeightConfig
from nonebot_bison.platform.bilibili import BilibiliSite
from nonebot_bison.scheduler.manager import init_scheduler
from nonebot_bison.platform.bilibili import BilibiliSchedConf
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t1"), "target1", "bilibili", [], [])
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t2"), "target1", "bilibili", [], [])
@ -40,12 +40,12 @@ async def test_scheduler_without_time(init_scheduler):
await init_scheduler()
static_res = await get_schedule_times(BilibiliSchedConf, 6)
static_res = await get_schedule_times(BilibiliSite, 6)
assert static_res["bilibili-t1"] == 1
assert static_res["bilibili-t2"] == 2
assert static_res["bilibili-bangumi-t2"] == 3
static_res = await get_schedule_times(BilibiliSchedConf, 6)
static_res = await get_schedule_times(BilibiliSite, 6)
assert static_res["bilibili-t1"] == 1
assert static_res["bilibili-t2"] == 2
assert static_res["bilibili-bangumi-t2"] == 3
@ -58,14 +58,14 @@ async def test_scheduler_batch_api(init_scheduler, mocker: MockerFixture):
from nonebot_bison.types import UserSubInfo
from nonebot_bison.scheduler import scheduler_dict
from nonebot_bison.types import Target as T_Target
from nonebot_bison.utils import DefaultClientManager
from nonebot_bison.scheduler.manager import init_scheduler
from nonebot_bison.platform.bilibili import BililiveSchedConf
from nonebot_bison.utils.scheduler_config import DefaultClientManager
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t1"), "target1", "bilibili-live", [], [])
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t2"), "target2", "bilibili-live", [], [])
mocker.patch.object(BililiveSchedConf, "client_man", DefaultClientManager)
mocker.patch.object(BililiveSchedConf, "client_mgr", DefaultClientManager)
await init_scheduler()
@ -96,8 +96,8 @@ async def test_scheduler_with_time(app: App, init_scheduler, mocker: MockerFixtu
from nonebot_bison.config import config, db_config
from nonebot_bison.types import Target as T_Target
from nonebot_bison.platform.bilibili import BilibiliSite
from nonebot_bison.scheduler.manager import init_scheduler
from nonebot_bison.platform.bilibili import BilibiliSchedConf
from nonebot_bison.config.db_config import WeightConfig, TimeWeightConfig
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t1"), "target1", "bilibili", [], [])
@ -118,19 +118,19 @@ async def test_scheduler_with_time(app: App, init_scheduler, mocker: MockerFixtu
mocker.patch.object(db_config, "_get_time", return_value=time(1, 30))
static_res = await get_schedule_times(BilibiliSchedConf, 6)
static_res = await get_schedule_times(BilibiliSite, 6)
assert static_res["bilibili-t1"] == 1
assert static_res["bilibili-t2"] == 2
assert static_res["bilibili-bangumi-t2"] == 3
static_res = await get_schedule_times(BilibiliSchedConf, 6)
static_res = await get_schedule_times(BilibiliSite, 6)
assert static_res["bilibili-t1"] == 1
assert static_res["bilibili-t2"] == 2
assert static_res["bilibili-bangumi-t2"] == 3
mocker.patch.object(db_config, "_get_time", return_value=time(10, 30))
static_res = await get_schedule_times(BilibiliSchedConf, 6)
static_res = await get_schedule_times(BilibiliSite, 6)
assert static_res["bilibili-t2"] == 6
@ -139,8 +139,8 @@ async def test_scheduler_add_new(init_scheduler):
from nonebot_bison.config import config
from nonebot_bison.types import Target as T_Target
from nonebot_bison.platform.bilibili import BilibiliSite
from nonebot_bison.scheduler.manager import init_scheduler
from nonebot_bison.platform.bilibili import BilibiliSchedConf
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t1"), "target1", "bilibili", [], [])
@ -148,7 +148,7 @@ async def test_scheduler_add_new(init_scheduler):
await config.add_subscribe(TargetQQGroup(group_id=2345), T_Target("t1"), "target1", "bilibili", [], [])
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t2"), "target2", "bilibili", [], [])
stat_res = await get_schedule_times(BilibiliSchedConf, 1)
stat_res = await get_schedule_times(BilibiliSite, 1)
assert stat_res["bilibili-t2"] == 1
@ -157,34 +157,34 @@ async def test_schedule_delete(init_scheduler):
from nonebot_bison.config import config
from nonebot_bison.types import Target as T_Target
from nonebot_bison.platform.bilibili import BilibiliSite
from nonebot_bison.scheduler.manager import init_scheduler
from nonebot_bison.platform.bilibili import BilibiliSchedConf
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t1"), "target1", "bilibili", [], [])
await config.add_subscribe(TargetQQGroup(group_id=123), T_Target("t2"), "target1", "bilibili", [], [])
await init_scheduler()
stat_res = await get_schedule_times(BilibiliSchedConf, 2)
stat_res = await get_schedule_times(BilibiliSite, 2)
assert stat_res["bilibili-t2"] == 1
assert stat_res["bilibili-t1"] == 1
await config.del_subscribe(TargetQQGroup(group_id=123), T_Target("t1"), "bilibili")
stat_res = await get_schedule_times(BilibiliSchedConf, 2)
stat_res = await get_schedule_times(BilibiliSite, 2)
assert stat_res["bilibili-t2"] == 2
async def test_scheduler_skip_browser(mocker: MockerFixture):
from nonebot_bison.scheduler.scheduler import Site
from nonebot_bison.platform import platform_manager
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.platform.platform import NewMessage
from nonebot_bison.scheduler.scheduler import SchedulerConfig
from nonebot_bison.scheduler import init_scheduler, scheduler_dict
mocker.patch.object(plugin_config, "bison_use_browser", False)
class MockPlatformSchedConf(SchedulerConfig):
name = "mock"
class MockSite(Site):
name = "mock_site"
schedule_type = "interval"
schedule_setting = {"seconds": 100}
require_browser = True
@ -196,26 +196,26 @@ async def test_scheduler_skip_browser(mocker: MockerFixture):
is_common = True
enable_tag = True
has_target = True
scheduler = MockPlatformSchedConf
site = MockSite
mocker.patch.dict(platform_manager, {"mock_platform": MockPlatform})
await init_scheduler()
assert MockPlatformSchedConf not in scheduler_dict.keys()
assert MockSite not in scheduler_dict.keys()
async def test_scheduler_no_skip_not_require_browser(mocker: MockerFixture):
from nonebot_bison.scheduler.scheduler import Site
from nonebot_bison.platform import platform_manager
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.platform.platform import NewMessage
from nonebot_bison.scheduler.scheduler import SchedulerConfig
from nonebot_bison.scheduler import init_scheduler, scheduler_dict
mocker.patch.object(plugin_config, "bison_use_browser", False)
class MockPlatformSchedConf(SchedulerConfig):
name = "mock"
class MockSite(Site):
name = "mock_site"
schedule_type = "interval"
schedule_setting = {"seconds": 100}
@ -226,10 +226,10 @@ async def test_scheduler_no_skip_not_require_browser(mocker: MockerFixture):
is_common = True
enable_tag = True
has_target = True
scheduler = MockPlatformSchedConf
site = MockSite
mocker.patch.dict(platform_manager, {"mock_platform": MockPlatform})
await init_scheduler()
assert MockPlatformSchedConf in scheduler_dict.keys()
assert MockSite in scheduler_dict.keys()

3
tests/test_context.py
View File

@ -5,8 +5,7 @@ from nonebug.app import App
@respx.mock
async def test_http_error(app: App):
from nonebot_bison.utils import ProcessContext, http_client
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager, http_client
example_route = respx.get("https://example.com")
example_route.mock(httpx.Response(403, json={"error": "gg"}))

3
tests/theme/test_themes.py
View File

@ -67,8 +67,7 @@ def mock_platform(app: App):
@pytest.fixture()
def mock_post(app: App, mock_platform):
from nonebot_bison.post import Post
from nonebot_bison.utils import ProcessContext
from nonebot_bison.utils.scheduler_config import DefaultClientManager
from nonebot_bison.utils import ProcessContext, DefaultClientManager
return Post(
m := mock_platform(ProcessContext(DefaultClientManager())),