⚗️ add regex and shell command di function

This commit is contained in:
yanyongyu 2021-12-14 22:40:47 +08:00
parent 329a1fd226
commit de7c51a518
4 changed files with 68 additions and 30 deletions

View File

@ -1,15 +1,6 @@
# used by Params
WRAPPER_ASSIGNMENTS = (
"__module__",
"__name__",
"__qualname__",
"__doc__",
"__annotations__",
"__globals__",
)
# used by Matcher
RECEIVE_KEY = "_receive_{id}"
LAST_RECEIVE_KEY = "_last_receive"
ARG_KEY = "_arg_{key}"
ARG_STR_KEY = "{key}"
REJECT_TARGET = "_current_target"

View File

@ -8,7 +8,6 @@ from pydantic.typing import ForwardRef, evaluate_forwardref
def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
signature = inspect.signature(call)
globalns = getattr(call, "__globals__", {})
print(signature.parameters)
typed_params = [
inspect.Parameter(
name=param.name,

View File

@ -29,7 +29,6 @@ from nonebot.log import logger
from nonebot.utils import CacheDict
from nonebot.dependencies import Dependent
from nonebot.permission import USER, Permission
from nonebot.consts import ARG_KEY, ARG_STR_KEY, RECEIVE_KEY, REJECT_TARGET
from nonebot.adapters import (
Bot,
Event,
@ -37,6 +36,13 @@ from nonebot.adapters import (
MessageSegment,
MessageTemplate,
)
from nonebot.consts import (
ARG_KEY,
ARG_STR_KEY,
RECEIVE_KEY,
REJECT_TARGET,
LAST_RECEIVE_KEY,
)
from nonebot.typing import (
Any,
T_State,
@ -422,7 +428,7 @@ class Matcher(metaclass=MatcherMeta):
@classmethod
def receive(
cls, id: str = "", parameterless: Optional[List[Any]] = None
cls, id: Optional[str] = None, parameterless: Optional[List[Any]] = None
) -> Callable[[T_Handler], T_Handler]:
"""
:说明:
@ -611,11 +617,15 @@ class Matcher(metaclass=MatcherMeta):
await bot.send(event=event, message=_prompt, **kwargs)
raise RejectedException
def get_receive(self, id: str, default: T = None) -> Union[Event, T]:
def get_receive(self, id: Optional[str], default: T = None) -> Union[Event, T]:
if id is None:
return self.state.get(LAST_RECEIVE_KEY, default)
return self.state.get(RECEIVE_KEY.format(id=id), default)
def set_receive(self, id: str, event: Event) -> None:
self.state[RECEIVE_KEY.format(id=id)] = event
def set_receive(self, id: Optional[str], event: Event) -> None:
if id is not None:
self.state[RECEIVE_KEY.format(id=id)] = event
self.state[LAST_RECEIVE_KEY] = event
def get_arg(self, key: str, default: T = None) -> Union[Event, T]:
return self.state.get(ARG_KEY.format(key=key), default)

View File

@ -1,6 +1,5 @@
import inspect
from functools import wraps, partial
from typing import Any, Tuple, Union, TypeVar, Callable, Optional, cast
from typing import Any, Dict, List, Tuple, Callable, Optional, cast
from contextlib import AsyncExitStack, contextmanager, asynccontextmanager
from pydantic.fields import Required, Undefined
@ -18,7 +17,6 @@ from nonebot.consts import (
RAW_CMD_KEY,
REGEX_GROUP,
REGEX_MATCHED,
WRAPPER_ASSIGNMENTS,
)
from nonebot.utils import (
CacheDict,
@ -31,8 +29,6 @@ from nonebot.utils import (
generic_check_issubclass,
)
T = TypeVar("T")
class DependsInner:
def __init__(
@ -53,7 +49,7 @@ class DependsInner:
def Depends(
dependency: Optional[T_Handler] = None,
*,
use_cache: bool = True,
use_cache: bool = False,
) -> Any:
"""
:说明:
@ -266,6 +262,46 @@ def CommandArg() -> Message:
return Depends(_command_arg)
def _shell_command_args(state=State()) -> Any:
return state[SHELL_ARGS]
def ShellCommandArgs():
return Depends(_shell_command_args)
def _shell_command_argv(state=State()) -> List[str]:
return state[SHELL_ARGV]
def ShellCommandArgv() -> Any:
return Depends(_shell_command_argv)
def _regex_matched(state=State()) -> str:
return state[REGEX_MATCHED]
def RegexMatched() -> str:
return Depends(_regex_matched)
def _regex_group(state=State()):
return state[REGEX_GROUP]
def RegexGroup() -> Tuple[Any, ...]:
return Depends(_regex_group)
def _regex_dict(state=State()):
return state[REGEX_DICT]
def RegexDict() -> Dict[str, Any]:
return Depends(_regex_dict)
class MatcherParam(Param):
@classmethod
def _check_param(
@ -280,16 +316,18 @@ class MatcherParam(Param):
return matcher
def _received(matcher: "Matcher", id: str = "", default: T = None) -> Union[Event, T]:
return matcher.get_receive(id, default)
def Received(id: str, default: Any = None) -> Any:
def _received(matcher: "Matcher"):
return matcher.get_receive(id, default)
return Depends(_received)
def Received(id: str = "", default: Any = None) -> Any:
return Depends(
wraps(_received, assigned=WRAPPER_ASSIGNMENTS)(
partial(_received, id=id, default=default)
)
)
def LastReceived(default: Any = None) -> Any:
def _last_received(matcher: "Matcher") -> Any:
return matcher.get_receive(None, default)
return Depends(_last_received)
class ExceptionParam(Param):