import hashlib import random from typing import Sequence, Callable from . import NoneBot from .exceptions import CQHttpError from .message import escape from .typing import Context_T, Message_T, Expression_T def context_id(ctx: Context_T, *, mode: str = 'default', use_hash: bool = False) -> str: """ Calculate a unique id representing the current context. mode: default: one id for one context group: one id for one group or discuss user: one id for one user :param ctx: the context dict :param mode: unique id mode: "default", "group", or "user" :param use_hash: use md5 to hash the id or not """ ctx_id = '' if mode == 'default': if ctx.get('group_id'): ctx_id += f'/group/{ctx["group_id"]}' elif ctx.get('discuss_id'): ctx_id += f'/discuss/{ctx["discuss_id"]}' if ctx.get('user_id'): ctx_id += f'/user/{ctx["user_id"]}' elif mode == 'group': if ctx.get('group_id'): ctx_id += f'/group/{ctx["group_id"]}' elif ctx.get('discuss_id'): ctx_id += f'/discuss/{ctx["discuss_id"]}' elif mode == 'user': if ctx.get('user_id'): ctx_id += f'/user/{ctx["user_id"]}' if ctx_id and use_hash: ctx_id = hashlib.md5(ctx_id.encode('ascii')).hexdigest() return ctx_id async def send(bot: NoneBot, ctx: Context_T, message: Message_T, *, ignore_failure: bool = True) -> None: """Send a message ignoring failure by default.""" try: if ctx.get('post_type') == 'message': await bot.send(ctx, message) else: ctx = ctx.copy() if 'message' in ctx: del ctx['message'] if 'group_id' in ctx: await bot.send_group_msg(**ctx, message=message) elif 'discuss_id' in ctx: await bot.send_discuss_msg(**ctx, message=message) elif 'user_id' in ctx: await bot.send_private_msg(**ctx, message=message) except CQHttpError: if not ignore_failure: raise def render_expression(expr: Expression_T, *, escape_args: bool = True, **kwargs) -> str: """ Render an expression to message string. :param expr: expression to render :param escape_args: should escape arguments or not :param kwargs: keyword arguments used in str.format() :return: the rendered message """ if isinstance(expr, Callable): expr = expr(**kwargs) elif isinstance(expr, Sequence) and not isinstance(expr, str): expr = random.choice(expr) if escape_args: for k, v in kwargs.items(): if isinstance(v, str): kwargs[k] = escape(v) return expr.format(**kwargs)