init: NoneBot2 机器人项目（夜间禁言 + B站动态推送插件）
This commit is contained in:
snowy-yang committed
1 parent
702f722e74
commit
320ecfc0a7
11 files changed
+3090
No files matched your search
@@ -0,0 +1,324 @@
|
||||
"""B站动态推送插件
|
||||
|
||||
定时轮询 BILI_PUSH_UIDS 中B站用户的动态，
|
||||
按可自定义的模板渲染消息（含动态图片）后推送到
|
||||
NIGHT_MUTE_GROUPS 白名单群（与夜间禁言插件共用）。
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
|
||||
import httpx
|
||||
from nonebot import get_bots, get_driver, get_plugin_config, logger, require
|
||||
from nonebot.adapters.onebot.v11 import (
|
||||
ActionFailed,
|
||||
Bot,
|
||||
Message,
|
||||
MessageSegment,
|
||||
NetworkError,
|
||||
)
|
||||
from nonebot.plugin import PluginMetadata
|
||||
|
||||
require("nonebot_plugin_apscheduler")
|
||||
require("nonebot_plugin_localstore")
|
||||
|
||||
from nonebot_plugin_apscheduler import scheduler
|
||||
from nonebot_plugin_localstore import get_data_dir
|
||||
|
||||
from .bilibili import BiliClient
|
||||
from .config import Config
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="B站动态推送",
|
||||
description="监控B站用户动态，按自定义模板推送（含图片）到白名单群",
|
||||
usage=(
|
||||
"在 .env 中配置 BILI_PUSH_UIDS（要监控的B站用户UID）；"
|
||||
"推送目标群与夜间禁言共用 NIGHT_MUTE_GROUPS 白名单；"
|
||||
"消息格式通过 BILI_PUSH_TEMPLATE 模板自定义"
|
||||
),
|
||||
config=Config,
|
||||
)
|
||||
|
||||
config = get_plugin_config(Config)
|
||||
|
||||
if not config.bili_push_uids:
|
||||
logger.warning("[bili_push] 未配置 BILI_PUSH_UIDS，插件不会监控任何用户")
|
||||
if not config.night_mute_groups:
|
||||
logger.warning("[bili_push] 未配置 NIGHT_MUTE_GROUPS 群白名单，推送没有目标群")
|
||||
|
||||
client = BiliClient(config.bili_push_cookie)
|
||||
get_driver().on_shutdown(client.aclose)
|
||||
|
||||
_state_file = get_data_dir("bili_push") / "state.json"
|
||||
|
||||
_MAX_TEXT_LEN = 400
|
||||
_IMAGE_MARKER = "\x00IMAGES\x00" # 模板内 {images} 的内部占位符
|
||||
|
||||
_ACTION = {
|
||||
"DYNAMIC_TYPE_AV": "投稿了新视频",
|
||||
"DYNAMIC_TYPE_ARTICLE": "发布了新专栏",
|
||||
"DYNAMIC_TYPE_MUSIC": "发布了新音频",
|
||||
"DYNAMIC_TYPE_FORWARD": "转发了一条动态",
|
||||
"DYNAMIC_TYPE_LIVE_RCMD": "发布了直播分享",
|
||||
}
|
||||
|
||||
_TITLE_PREFIX = {
|
||||
"MAJOR_TYPE_ARCHIVE": "[视频]",
|
||||
"MAJOR_TYPE_ARTICLE": "[专栏]",
|
||||
"MAJOR_TYPE_MUSIC": "[音频]",
|
||||
"MAJOR_TYPE_LIVE_RCMD": "[直播]",
|
||||
}
|
||||
|
||||
|
||||
# ------------------------ 消息格式化 ------------------------
|
||||
|
||||
|
||||
def _truncate(text: str) -> str:
|
||||
text = text.strip()
|
||||
if len(text) > _MAX_TEXT_LEN:
|
||||
return text[:_MAX_TEXT_LEN] + "..."
|
||||
return text
|
||||
|
||||
|
||||
def _card_title(major: dict) -> str:
|
||||
"""视频/专栏/音频/番剧/直播卡片的标题"""
|
||||
for mtype, key in (
|
||||
("MAJOR_TYPE_ARCHIVE", "archive"),
|
||||
("MAJOR_TYPE_ARTICLE", "article"),
|
||||
("MAJOR_TYPE_MUSIC", "music"),
|
||||
("MAJOR_TYPE_PGC", "pgc"),
|
||||
):
|
||||
if major.get("type") == mtype:
|
||||
return str((major.get(key) or {}).get("title") or "").strip()
|
||||
if major.get("type") == "MAJOR_TYPE_LIVE_RCMD":
|
||||
info = major.get("live_rcmd") or {}
|
||||
return str((info.get("live_play_info") or {}).get("title") or "").strip()
|
||||
return ""
|
||||
|
||||
|
||||
def _major_images(major: dict) -> list[str]:
|
||||
draw = major.get("draw") or {}
|
||||
return [
|
||||
item["img_src"] for item in (draw.get("items") or []) if item.get("img_src")
|
||||
]
|
||||
|
||||
|
||||
def _images_from_major(major: dict) -> list[str]:
|
||||
"""卡片里的图片：图文动态的正文图 / 视频封面"""
|
||||
images = _major_images(major)
|
||||
archive = major.get("archive") or {}
|
||||
if archive.get("cover"):
|
||||
images.append(str(archive["cover"]))
|
||||
return images
|
||||
|
||||
|
||||
def _collect_images(item: dict) -> list[str]:
|
||||
"""动态图片：正文图/封面；转发动态没有自己的图时用原动态的"""
|
||||
dynamic = item["modules"]["module_dynamic"]
|
||||
images = _images_from_major(dynamic.get("major") or {})
|
||||
if not images:
|
||||
orig_dynamic = (
|
||||
(item.get("orig") or {}).get("modules") or {}
|
||||
).get("module_dynamic") or {}
|
||||
images = _images_from_major(orig_dynamic.get("major") or {})
|
||||
return images
|
||||
|
||||
|
||||
def _dynamic_link(item: dict) -> str:
|
||||
dynamic = item["modules"]["module_dynamic"]
|
||||
dtype = dynamic.get("dynamic_type") or ""
|
||||
major = dynamic.get("major") or {}
|
||||
if dtype == "DYNAMIC_TYPE_AV" and (major.get("archive") or {}).get("bvid"):
|
||||
return f"https://www.bilibili.com/video/{major['archive']['bvid']}"
|
||||
if dtype == "DYNAMIC_TYPE_ARTICLE" and (major.get("article") or {}).get("id"):
|
||||
return f"https://www.bilibili.com/read/cv{major['article']['id']}"
|
||||
return f"https://t.bilibili.com/{item['id_str']}"
|
||||
|
||||
|
||||
def _collect_values(item: dict) -> dict[str, str]:
|
||||
"""收集一条动态的全部模板变量"""
|
||||
modules = item["modules"]
|
||||
dynamic = modules["module_dynamic"]
|
||||
dtype = dynamic.get("dynamic_type") or ""
|
||||
major = dynamic.get("major") or {}
|
||||
desc = str((dynamic.get("desc") or {}).get("text") or "").strip()
|
||||
title = _card_title(major)
|
||||
|
||||
orig_author = orig_text = ""
|
||||
if dtype == "DYNAMIC_TYPE_FORWARD":
|
||||
orig_modules = (item.get("orig") or {}).get("modules") or {}
|
||||
orig_author = str(
|
||||
(orig_modules.get("module_author") or {}).get("name") or "未知用户"
|
||||
)
|
||||
orig_dynamic = orig_modules.get("module_dynamic") or {}
|
||||
orig_text = str((orig_dynamic.get("desc") or {}).get("text") or "").strip()
|
||||
if not orig_text:
|
||||
orig_text = _card_title(orig_dynamic.get("major") or "")
|
||||
|
||||
# 默认的 {text}：转发带原动态摘要，带卡片的先标题后正文
|
||||
if dtype == "DYNAMIC_TYPE_FORWARD":
|
||||
lines = [f"> @{orig_author}"]
|
||||
if orig_text:
|
||||
lines.append(_truncate(orig_text))
|
||||
if desc:
|
||||
lines += ["", _truncate(desc)]
|
||||
text = "\n".join(lines)
|
||||
elif title:
|
||||
head = f"{_TITLE_PREFIX.get(major.get('type') or '', '')} {title}".strip()
|
||||
text = f"{head}\n{_truncate(desc)}" if desc else head
|
||||
else:
|
||||
text = _truncate(desc)
|
||||
|
||||
return {
|
||||
"author": str(modules["module_author"]["name"]),
|
||||
"action": _ACTION.get(dtype, "发布了新动态"),
|
||||
"text": text,
|
||||
"title": title,
|
||||
"orig_author": orig_author,
|
||||
"orig_text": orig_text,
|
||||
"link": _dynamic_link(item),
|
||||
}
|
||||
|
||||
|
||||
def _format_dynamic(item: dict) -> Message:
|
||||
"""按模板把一条动态渲染成群消息"""
|
||||
values = _collect_values(item)
|
||||
rendered = config.bili_push_template
|
||||
for key, value in values.items():
|
||||
rendered = rendered.replace(f"{{{key}}}", value)
|
||||
rendered = rendered.replace("{images}", _IMAGE_MARKER)
|
||||
rendered = rendered.replace("\\n", "\n") # 兼容 .env 未转义的情况
|
||||
rendered = re.sub(r"[ \t]+\n", "\n", rendered)
|
||||
rendered = re.sub(r"\n{3,}", "\n\n", rendered).strip()
|
||||
|
||||
parts = rendered.split(_IMAGE_MARKER, 1)
|
||||
before = parts[0].rstrip("\n")
|
||||
after = parts[1].lstrip("\n") if len(parts) > 1 else ""
|
||||
|
||||
message = Message()
|
||||
if before:
|
||||
message += before
|
||||
limit = config.bili_push_max_images
|
||||
if limit > 0:
|
||||
for src in _collect_images(item)[:limit]:
|
||||
if message:
|
||||
message += "\n"
|
||||
message += MessageSegment.image(src)
|
||||
if after:
|
||||
if message:
|
||||
message += "\n"
|
||||
message += after
|
||||
return message
|
||||
|
||||
|
||||
# ------------------------ 轮询与推送 ------------------------
|
||||
|
||||
|
||||
def _load_state() -> dict[str, str]:
|
||||
"""读取上次推送进度 {uid: 最新动态id}"""
|
||||
try:
|
||||
data = json.loads(_state_file.read_text(encoding="utf-8"))
|
||||
return data if isinstance(data, dict) else {}
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
def _save_state(state: dict[str, str]) -> None:
|
||||
try:
|
||||
_state_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
_state_file.write_text(
|
||||
json.dumps(state, ensure_ascii=False), encoding="utf-8"
|
||||
)
|
||||
except OSError as e:
|
||||
logger.warning(f"[bili_push] 推送状态保存失败: {e}")
|
||||
|
||||
|
||||
def _connected_bots() -> list[Bot]:
|
||||
"""当前已连接的所有 OneBot v11 Bot"""
|
||||
return [b for b in get_bots().values() if isinstance(b, Bot)]
|
||||
|
||||
|
||||
async def _send_to_group(bot: Bot, group_id: int, message: Message) -> bool:
|
||||
try:
|
||||
await bot.send_group_msg(group_id=group_id, message=message)
|
||||
except (ActionFailed, NetworkError) as e:
|
||||
logger.warning(f"[bili_push] 推送失败 group={group_id}: {e}")
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
async def _broadcast(message: Message) -> bool:
|
||||
"""广播到白名单所有群，返回是否有至少一个群发送成功"""
|
||||
success = False
|
||||
for bot in _connected_bots():
|
||||
for group_id in config.night_mute_groups:
|
||||
if await _send_to_group(bot, group_id, message):
|
||||
success = True
|
||||
return success
|
||||
|
||||
|
||||
async def _poll_uid(uid: int, state: dict[str, str]) -> None:
|
||||
key = str(uid)
|
||||
try:
|
||||
items = await client.get_space_dynamics(uid)
|
||||
except (httpx.HTTPError, RuntimeError, KeyError, TypeError, ValueError) as e:
|
||||
logger.warning(f"[bili_push] 获取 uid={uid} 动态失败: {e}")
|
||||
return
|
||||
if not items:
|
||||
return
|
||||
last_id = int(state[key]) if key in state else 0
|
||||
if last_id == 0:
|
||||
# 首次监控该用户：记录当前最新动态作为基线，不推送历史
|
||||
state[key] = items[0]["id_str"]
|
||||
_save_state(state)
|
||||
logger.info(f"[bili_push] uid={uid} 已记录基线动态 {state[key]}")
|
||||
return
|
||||
new_items = sorted(
|
||||
(it for it in items if int(it["id_str"]) > last_id),
|
||||
key=lambda it: int(it["id_str"]),
|
||||
)
|
||||
if not new_items:
|
||||
return
|
||||
if len(new_items) > config.bili_push_batch_limit:
|
||||
logger.info(
|
||||
f"[bili_push] uid={uid} 积压 {len(new_items)} 条动态，"
|
||||
f"仅推送最新 {config.bili_push_batch_limit} 条"
|
||||
)
|
||||
new_items = new_items[-config.bili_push_batch_limit :]
|
||||
# 逐条推送，推送失败的动态不推进进度，下轮轮询自动重试，
|
||||
# 避免机器人离线/接口故障期间漏推
|
||||
pushed_count = 0
|
||||
for item in new_items:
|
||||
if not await _broadcast(_format_dynamic(item)):
|
||||
logger.warning(
|
||||
f"[bili_push] uid={uid} 动态 {item['id_str']} 推送失败，"
|
||||
"剩余动态将在下轮轮询重试"
|
||||
)
|
||||
break
|
||||
pushed_count += 1
|
||||
if pushed_count:
|
||||
state[key] = new_items[pushed_count - 1]["id_str"]
|
||||
_save_state(state)
|
||||
|
||||
|
||||
async def poll_dynamics() -> None:
|
||||
"""轮询所有监控用户，推送新动态"""
|
||||
if not config.bili_push_uids or not config.night_mute_groups:
|
||||
return
|
||||
state = _load_state()
|
||||
for index, uid in enumerate(sorted(config.bili_push_uids)):
|
||||
if index:
|
||||
await asyncio.sleep(2)
|
||||
await _poll_uid(uid, state)
|
||||
|
||||
|
||||
scheduler.add_job(
|
||||
poll_dynamics,
|
||||
"interval",
|
||||
seconds=config.bili_push_interval,
|
||||
id="bili_push_poll",
|
||||
replace_existing=True,
|
||||
max_instances=1,
|
||||
misfire_grace_time=60,
|
||||
)
|
||||
@@ -0,0 +1,187 @@
|
||||
"""B站动态 API 客户端
|
||||
|
||||
使用 web-dynamic v1 接口（feed/space）拉取用户动态：
|
||||
先通过 finger/spi 接口获取 buvid3/buvid4 cookie 绕过基础风控，
|
||||
再对请求参数做 wbi 签名，并附带 dm_* 浏览器行为占位参数
|
||||
（缺少它们会返回 code=0 但内容为空的软风控响应）。
|
||||
接口细节参考 bilibili-API-collect 社区文档。
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import re
|
||||
import time
|
||||
from typing import Any
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import httpx
|
||||
|
||||
_UA = (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
|
||||
)
|
||||
|
||||
_DYN_API = "https://api.bilibili.com/x/polymer/web-dynamic/v1/feed/space"
|
||||
|
||||
# wbi 签名的固定混淆表
|
||||
_MIXIN_KEY_ENC_TAB = [
|
||||
46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,
|
||||
33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,
|
||||
61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,
|
||||
36, 20, 34, 44, 52,
|
||||
]
|
||||
|
||||
# 触发风控的返回码，遇到后刷新 wbi key 重试一次
|
||||
_RISK_CODES = (-352, -403, -412)
|
||||
|
||||
# wbi key 缓存时长（官方每天更换，30 分钟足够）
|
||||
_WBI_CACHE_SECONDS = 1800
|
||||
|
||||
# 单次拉取最多尝试次数。B站会对部分请求随机返回软风控
|
||||
# （code=0 但列表为空 / HTTP 412），换身份重试通常能成功
|
||||
_MAX_ATTEMPTS = 3
|
||||
|
||||
# HTTP 层风控状态码
|
||||
_HTTP_RISK_STATUS = 412
|
||||
|
||||
# 风控要求的浏览器行为参数，固定占位值即可
|
||||
_DM_PARAMS = {
|
||||
"dm_img_list": "[]",
|
||||
"dm_img_str": "base64",
|
||||
"dm_cover_img_str": "base64",
|
||||
"dm_img_inter": '{"ds":[],"wh":[0,0,0],"of":[0,0,0]}',
|
||||
}
|
||||
|
||||
|
||||
class BiliApiError(RuntimeError):
|
||||
"""B站接口返回非 0 code"""
|
||||
|
||||
def __init__(self, code: int, message: str) -> None:
|
||||
super().__init__(f"B站接口返回异常 code={code} message={message}")
|
||||
self.code = code
|
||||
|
||||
|
||||
def _mixin_key(img_key: str, sub_key: str) -> str:
|
||||
return "".join((img_key + sub_key)[i] for i in _MIXIN_KEY_ENC_TAB)[:32]
|
||||
|
||||
|
||||
def _wbi_sign(params: dict[str, Any], img_key: str, sub_key: str) -> dict[str, Any]:
|
||||
params = {**params, "wts": round(time.time())}
|
||||
query = urlencode(
|
||||
{
|
||||
key: "".join(c for c in str(value) if c not in "!'()*")
|
||||
for key, value in sorted(params.items())
|
||||
}
|
||||
)
|
||||
params["w_rid"] = hashlib.md5(
|
||||
(query + _mixin_key(img_key, sub_key)).encode()
|
||||
).hexdigest()
|
||||
return params
|
||||
|
||||
|
||||
class BiliClient:
|
||||
"""轮询B站用户动态的最小客户端，进程内共享一个实例"""
|
||||
|
||||
def __init__(self, cookie: str = "") -> None:
|
||||
self._extra_cookie = cookie.strip()
|
||||
self._bootstrapped = False
|
||||
self._wbi_keys: tuple[str, str] | None = None
|
||||
self._wbi_time = 0.0
|
||||
self._client = httpx.AsyncClient(
|
||||
headers={"User-Agent": _UA, "Accept": "application/json, text/plain, */*"},
|
||||
timeout=15,
|
||||
follow_redirects=True,
|
||||
)
|
||||
|
||||
async def aclose(self) -> None:
|
||||
await self._client.aclose()
|
||||
|
||||
async def _fetch_buvid(self) -> None:
|
||||
"""通过 finger/spi 获取并设置新的 buvid3/buvid4 cookie"""
|
||||
try:
|
||||
resp = await self._client.get(
|
||||
"https://api.bilibili.com/x/frontend/finger/spi"
|
||||
)
|
||||
data = resp.json().get("data") or {}
|
||||
except (httpx.HTTPError, ValueError):
|
||||
return
|
||||
buvid3 = data.get("b_3")
|
||||
if buvid3:
|
||||
self._client.cookies.set("buvid3", buvid3, domain=".bilibili.com")
|
||||
self._client.cookies.set(
|
||||
"buvid4", data.get("b_4", ""), domain=".bilibili.com"
|
||||
)
|
||||
|
||||
async def _bootstrap(self) -> None:
|
||||
"""首次请求前获取 buvid cookie，绕过基础风控"""
|
||||
if self._bootstrapped:
|
||||
return
|
||||
self._bootstrapped = True
|
||||
if self._extra_cookie:
|
||||
self._client.headers["Cookie"] = self._extra_cookie
|
||||
return
|
||||
await self._fetch_buvid()
|
||||
|
||||
async def _ensure_wbi(self, *, force: bool = False) -> tuple[str, str]:
|
||||
"""获取 wbi key（nav 接口未登录也返回），缓存 30 分钟"""
|
||||
if not force and self._wbi_keys and time.time() - self._wbi_time < (
|
||||
_WBI_CACHE_SECONDS
|
||||
):
|
||||
return self._wbi_keys
|
||||
resp = await self._client.get("https://api.bilibili.com/x/web-interface/nav")
|
||||
resp.raise_for_status()
|
||||
try:
|
||||
wbi = resp.json()["data"]["wbi_img"]
|
||||
keys = (
|
||||
re.search(r"/(\w+)\.png", wbi["img_url"]).group(1),
|
||||
re.search(r"/(\w+)\.png", wbi["sub_url"]).group(1),
|
||||
)
|
||||
except (KeyError, TypeError, AttributeError) as e:
|
||||
raise BiliApiError(-1, "nav 接口响应结构异常，无法获取 wbi key") from e
|
||||
self._wbi_keys = keys
|
||||
self._wbi_time = time.time()
|
||||
return keys
|
||||
|
||||
async def get_space_dynamics(self, uid: int) -> list[dict]:
|
||||
"""拉取用户最新动态列表（新→旧），失败时抛出异常"""
|
||||
await self._bootstrap()
|
||||
params: dict[str, Any] = {
|
||||
"host_mid": uid,
|
||||
"timezone_offset": -480,
|
||||
"platform": "web",
|
||||
"features": "itemOpusStyle",
|
||||
**_DM_PARAMS,
|
||||
}
|
||||
headers = {"Referer": f"https://space.bilibili.com/{uid}/dynamic"}
|
||||
last_error = BiliApiError(0, "no response")
|
||||
for attempt in range(_MAX_ATTEMPTS):
|
||||
if attempt:
|
||||
await asyncio.sleep(1)
|
||||
if not self._extra_cookie:
|
||||
await self._fetch_buvid() # 换一个新身份再试
|
||||
img_key, sub_key = await self._ensure_wbi(force=bool(attempt))
|
||||
try:
|
||||
resp = await self._client.get(
|
||||
_DYN_API,
|
||||
params=_wbi_sign(params, img_key, sub_key),
|
||||
headers=headers,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError as e:
|
||||
if e.response.status_code != _HTTP_RISK_STATUS:
|
||||
raise
|
||||
last_error = BiliApiError(_HTTP_RISK_STATUS, "HTTP 412")
|
||||
continue
|
||||
data = resp.json()
|
||||
if data.get("code") != 0:
|
||||
last_error = BiliApiError(
|
||||
int(data.get("code", 0)), str(data.get("message", ""))
|
||||
)
|
||||
if last_error.code not in _RISK_CODES:
|
||||
raise last_error # 非风控错误，重试没有意义
|
||||
else:
|
||||
items = (data.get("data") or {}).get("items") or []
|
||||
if items:
|
||||
return items
|
||||
last_error = BiliApiError(0, "empty feed (可能被软风控)")
|
||||
raise last_error
|
||||
@@ -0,0 +1,38 @@
|
||||
"""bili_push B站动态推送插件配置"""
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class Config(BaseModel):
|
||||
"""全部配置项均可在 .env 文件中覆盖"""
|
||||
|
||||
# 推送目标群：与夜间禁言插件共用同一个白名单（NIGHT_MUTE_GROUPS）
|
||||
night_mute_groups: set[int] = Field(default_factory=set)
|
||||
|
||||
# 要监控的B站用户 UID，为空时不监控
|
||||
bili_push_uids: set[int] = Field(default_factory=set)
|
||||
|
||||
# 轮询间隔（秒）
|
||||
bili_push_interval: int = Field(default=300, ge=30)
|
||||
|
||||
# 单个用户一次轮询最多推送的动态条数（防止积压刷屏）
|
||||
bili_push_batch_limit: int = Field(default=3, ge=1, le=10)
|
||||
|
||||
# 一条动态最多附带的图片张数，0 表示不发图
|
||||
bili_push_max_images: int = Field(default=3, ge=0, le=9)
|
||||
|
||||
# 推送消息模板，可用变量：
|
||||
# {author} UP主名字
|
||||
# {action} 动作（投稿了新视频 / 发布了新动态 / 转发了一条动态 等）
|
||||
# {text} 动态正文（转发动态会自动带上原动态摘要）
|
||||
# {title} 视频/专栏/音频等卡片标题
|
||||
# {orig_author} 转发的原动态作者
|
||||
# {orig_text} 转发的原动态内容
|
||||
# {link} 动态/视频链接
|
||||
# {images} 图片插入位置（不写则图片追加在消息末尾）
|
||||
# 换行写作 \n
|
||||
bili_push_template: str = "[B站动态] {author} {action}\n\n{text}\n{images}\n{link}"
|
||||
|
||||
# 可选：登录B站后的完整 Cookie 字符串，
|
||||
# 动态接口被风控（返回 -352）时需要配置才能拉取
|
||||
bili_push_cookie: str = ""
|
||||
@@ -0,0 +1,192 @@
|
||||
"""夜间禁言插件
|
||||
|
||||
禁言时段内（默认 0 点到 6 点）在白名单群里「谁发言就禁言谁」，
|
||||
禁言时长 = 当前时间到结束时刻的秒数，由 QQ 服务器到期自动解禁，
|
||||
即使结束时机器人离线也能解。
|
||||
另在开始前 / 开始时 / 结束时向白名单群发送可自定义的提示消息。
|
||||
"""
|
||||
|
||||
import math
|
||||
from datetime import date, datetime, timedelta
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from nonebot import get_bots, get_plugin_config, logger, on_message, require
|
||||
from nonebot.adapters import Event
|
||||
from nonebot.adapters.onebot.v11 import (
|
||||
ActionFailed,
|
||||
Bot,
|
||||
GroupMessageEvent,
|
||||
NetworkError,
|
||||
)
|
||||
from nonebot.plugin import PluginMetadata
|
||||
from nonebot.rule import Rule
|
||||
|
||||
require("nonebot_plugin_apscheduler")
|
||||
|
||||
from nonebot_plugin_apscheduler import scheduler
|
||||
|
||||
from .config import Config
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="夜间禁言",
|
||||
description=(
|
||||
"0点到6点谁发言就禁言谁（禁到早上6点），"
|
||||
"并在开始前/开始时/结束时发送群提示"
|
||||
),
|
||||
usage="在 .env 中配置 NIGHT_MUTE_GROUPS 群白名单后自动生效，无需命令",
|
||||
config=Config,
|
||||
)
|
||||
|
||||
config = get_plugin_config(Config)
|
||||
|
||||
# 与 nonebot_plugin_apscheduler 默认的 cron 时区保持一致
|
||||
TZ = ZoneInfo("Asia/Shanghai")
|
||||
|
||||
if not config.night_mute_groups:
|
||||
logger.warning("[night_mute] 未配置 NIGHT_MUTE_GROUPS 群白名单，插件不会生效")
|
||||
if config.night_mute_start_hour == config.night_mute_end_hour:
|
||||
logger.warning(
|
||||
"[night_mute] NIGHT_MUTE_START_HOUR 与 NIGHT_MUTE_END_HOUR 相同，"
|
||||
"禁言窗口无效，插件不会禁言任何人"
|
||||
)
|
||||
|
||||
|
||||
# ------------------------ 时间窗口 ------------------------
|
||||
|
||||
|
||||
def _in_window(now: datetime) -> bool:
|
||||
"""当前时间是否在禁言窗口 [start, end) 内，支持跨午夜窗口"""
|
||||
start, end = config.night_mute_start_hour, config.night_mute_end_hour
|
||||
if start == end:
|
||||
return False
|
||||
if start < end:
|
||||
return start <= now.hour < end
|
||||
return now.hour >= start or now.hour < end
|
||||
|
||||
|
||||
def _seconds_until_end(now: datetime) -> int:
|
||||
"""距离禁言窗口结束时刻的秒数，至少 1 秒"""
|
||||
end_dt = now.replace(
|
||||
hour=config.night_mute_end_hour, minute=0, second=0, microsecond=0
|
||||
)
|
||||
if end_dt <= now:
|
||||
end_dt += timedelta(days=1)
|
||||
return max(1, math.ceil((end_dt - now).total_seconds()))
|
||||
|
||||
|
||||
# ------------------------ 发言即禁言 ------------------------
|
||||
|
||||
# 当晚已禁言名单：日期 -> {(group_id, user_id)}
|
||||
_muted: dict[date, set[tuple[int, int]]] = {}
|
||||
|
||||
|
||||
async def _is_target(event: Event) -> bool:
|
||||
"""白名单群 + 禁言窗口内 + 可被禁言的普通成员发言"""
|
||||
if not isinstance(event, GroupMessageEvent):
|
||||
return False
|
||||
if event.group_id not in config.night_mute_groups:
|
||||
return False
|
||||
if event.anonymous is not None: # 匿名成员不处理
|
||||
return False
|
||||
if event.sender.role in ("owner", "admin"): # 群主/管理员无法被禁言
|
||||
return False
|
||||
return _in_window(datetime.now(TZ))
|
||||
|
||||
|
||||
night_mute = on_message(rule=Rule(_is_target), priority=15, block=False)
|
||||
|
||||
|
||||
@night_mute.handle()
|
||||
async def _(bot: Bot, event: GroupMessageEvent):
|
||||
now = datetime.now(TZ)
|
||||
today = now.date()
|
||||
muted_today = _muted.get(today)
|
||||
if muted_today is None: # 新的一天，清掉昨夜名单
|
||||
_muted.clear()
|
||||
muted_today = _muted.setdefault(today, set())
|
||||
key = (event.group_id, event.user_id)
|
||||
if key in muted_today: # 当晚已禁言过，不重复调 API
|
||||
return
|
||||
duration = _seconds_until_end(now)
|
||||
try:
|
||||
await bot.set_group_ban(
|
||||
group_id=event.group_id, user_id=event.user_id, duration=duration
|
||||
)
|
||||
except ActionFailed as e:
|
||||
logger.warning(
|
||||
f"[night_mute] 禁言失败 group={event.group_id} "
|
||||
f"user={event.user_id}: {e}（请确认机器人有群管理员身份）"
|
||||
)
|
||||
return
|
||||
muted_today.add(key)
|
||||
logger.info(
|
||||
f"[night_mute] 已禁言 group={event.group_id} "
|
||||
f"user={event.user_id} 时长={duration}s"
|
||||
)
|
||||
|
||||
|
||||
# ------------------------ 定时提示 ------------------------
|
||||
|
||||
|
||||
async def _send_to_group(bot: Bot, group_id: int, text: str) -> None:
|
||||
try:
|
||||
await bot.send_group_msg(group_id=group_id, message=text)
|
||||
except (ActionFailed, NetworkError) as e:
|
||||
logger.warning(f"[night_mute] 提示发送失败 group={group_id}: {e}")
|
||||
|
||||
|
||||
async def _broadcast(text: str) -> None:
|
||||
"""向白名单所有群发送提示，机器人不在群内时仅记警告"""
|
||||
if not config.night_mute_groups:
|
||||
return
|
||||
bots = [b for b in get_bots().values() if isinstance(b, Bot)]
|
||||
if not bots:
|
||||
logger.warning("[night_mute] 定时提示未发送：没有已连接的 OneBot v11 Bot")
|
||||
return
|
||||
for bot in bots:
|
||||
for group_id in config.night_mute_groups:
|
||||
await _send_to_group(bot, group_id, text)
|
||||
|
||||
|
||||
def _schedule_notice(job_id: str, hour: int, minute: int, text: str) -> None:
|
||||
"""注册一个定时群提示任务"""
|
||||
|
||||
async def _job() -> None:
|
||||
await _broadcast(text)
|
||||
|
||||
scheduler.add_job(
|
||||
_job,
|
||||
"cron",
|
||||
hour=hour,
|
||||
minute=minute,
|
||||
timezone=str(TZ),
|
||||
id=job_id,
|
||||
replace_existing=True,
|
||||
misfire_grace_time=300,
|
||||
)
|
||||
|
||||
|
||||
def _pre_notice_cron() -> tuple[int, int]:
|
||||
"""预告时刻 = 开始时刻 - 预告分钟数，自动处理跨日（0点前10分钟 -> 23:50）"""
|
||||
minutes = (
|
||||
config.night_mute_start_hour * 60 - config.night_mute_pre_notice_minutes
|
||||
) % (24 * 60)
|
||||
return divmod(minutes, 60)
|
||||
|
||||
|
||||
_pre_hour, _pre_minute = _pre_notice_cron()
|
||||
_schedule_notice(
|
||||
"night_mute_pre_notice", _pre_hour, _pre_minute, config.night_mute_notice_before
|
||||
)
|
||||
_schedule_notice(
|
||||
"night_mute_start_notice",
|
||||
config.night_mute_start_hour,
|
||||
0,
|
||||
config.night_mute_notice_start,
|
||||
)
|
||||
_schedule_notice(
|
||||
"night_mute_end_notice",
|
||||
config.night_mute_end_hour,
|
||||
0,
|
||||
config.night_mute_notice_end,
|
||||
)
|
||||
@@ -0,0 +1,28 @@
|
||||
"""night_mute 夜间禁言插件配置"""
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class Config(BaseModel):
|
||||
"""全部配置项均可在 .env 文件中覆盖"""
|
||||
|
||||
# 群白名单：只有这里写了的群才会执行夜间禁言
|
||||
night_mute_groups: set[int] = Field(default_factory=set)
|
||||
|
||||
# 禁言时段（整点小时，0-23），默认 0 点到 6 点
|
||||
night_mute_start_hour: int = Field(default=0, ge=0, le=23)
|
||||
night_mute_end_hour: int = Field(default=6, ge=0, le=23)
|
||||
|
||||
# 禁言开始前多少分钟发送预告提示
|
||||
night_mute_pre_notice_minutes: int = Field(default=10, ge=1, le=1440)
|
||||
|
||||
# 三个时间点的提示消息
|
||||
night_mute_notice_before: str = (
|
||||
"夜间禁言将在10分钟后开启，有重要消息的抓紧说，"
|
||||
"过点发言就会被禁言到早上6点哦！"
|
||||
)
|
||||
night_mute_notice_start: str = (
|
||||
"夜间禁言已开启：现在起谁发言就禁言到早上6点，"
|
||||
"重要消息可以说一条再睡！"
|
||||
)
|
||||
night_mute_notice_end: str = "早上好！夜间禁言已解除，大家畅所欲言！"
|
||||
Reference in new issue
Block a user