mirror of
https://github.com/snowy-yang/qqbotsdk.git
synced 2026-10-21 07:59:07 +00:00
snowy-yang
1552c5207b
以「语义可读、逻辑清晰、减少心智负担」为目标，不追求极致速度：
- **移除 pyee 依赖**：EventEmitter 改为内置的 `事件名 → handler 列表` 注册表
（on() 追加、emit() 顺序遍历）。SDK 用不到 pyee 的 once/new_listener/错误
事件机制，去掉后少一个依赖、分发语义一眼可读。注册语义不变（同函数重复
注册只生效一次），并补测试锁定「重复注册只生效一次」「按注册顺序执行」。
破坏性：once/remove_listener 等装饰性 API 不再提供。
- **去掉热路径上的重复计算**（纯缓存，不改语义）：
· payloads 按类缓存解析计划（字段名 + 是否嵌套 Author），不再每次事件重跑
dataclasses.fields 与联合类型解包；未收录字段照旧忽略（新增回归测试，
线上 payload 会带 author.union_openid 这类字段）。实测 7.5µs → 2.5µs。
· crypto 按 AppSecret 缓存派生私钥（每个 webhook 请求都要验签，派生一次
约 40µs）；公钥派生只差 3µs，不再多缓存一层。
- **命名与接口一致性**：
· WebsocketConnecter.reply_helper → send_helper（它从出站缓冲取帧发送，
原名会让人误以为还有 reply 队列；model.py 的对应注释同步）
· Connecter 由 typing.Protocol 改为 abc.ABC，与 BaseProtocol 风格统一
- **webhook 去重改摊还清理**：过期判定按当前 id 做（O(1)，且 TTL 外重发的
同一 id 仍按新事件处理——原先的写法在阈值清理下会误判为重复，已由测试
抓出并修正），整表清过期项只在表长到阈值时执行，不再每请求 O(n) 扫描。
测试 85 全绿（新增未知字段/解析计划缓存/注册语义/去重摊还与 TTL 过期），
ruff check/format + pyright 干净。文档 ARCHITECTURE/API/CHANGELOG 同步。
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
from time import monotonic
|
|
|
|
from qqbotsdk.config import Config
|
|
from qqbotsdk.connecter import WebhookConnecter
|
|
from qqbotsdk.queue import EventQueue
|
|
|
|
|
|
def make_connecter() -> WebhookConnecter:
|
|
config = Config(
|
|
app_id="app",
|
|
app_secret="secret",
|
|
base_url="https://api.bot.qq.com",
|
|
timeout=None, # type: ignore[arg-type]
|
|
connecter="webhook",
|
|
intents_file="intents.toml",
|
|
webhook_host="0.0.0.0",
|
|
webhook_port=8080,
|
|
webhook_path="/qqbot/webhook",
|
|
)
|
|
|
|
async def noop_handle(payload):
|
|
return None
|
|
|
|
return WebhookConnecter(config, EventQueue(), noop_handle)
|
|
|
|
|
|
def test_first_push_is_not_duplicate():
|
|
c = make_connecter()
|
|
assert not c._duplicated("evt-1")
|
|
|
|
|
|
def test_same_id_within_ttl_is_duplicate():
|
|
c = make_connecter()
|
|
assert not c._duplicated("evt-1")
|
|
assert c._duplicated("evt-1")
|
|
|
|
|
|
def test_none_or_empty_id_never_duplicates():
|
|
c = make_connecter()
|
|
assert not c._duplicated(None)
|
|
assert not c._duplicated("")
|
|
assert not c._duplicated(None)
|
|
|
|
|
|
def test_expired_entries_are_evicted():
|
|
c = make_connecter()
|
|
assert not c._duplicated("evt-1")
|
|
# 人为把记录改成过期
|
|
for key in list(c._seen):
|
|
c._seen[key] -= c._DEDUP_TTL + 1
|
|
assert not c._duplicated("evt-1")
|
|
assert "evt-1" in c._seen # 以新时间戳重新记录
|
|
|
|
|
|
def test_expired_id_is_treated_as_new():
|
|
"""TTL 之外重发的同一 id 是新事件，不能被当成重复丢掉。"""
|
|
c = make_connecter()
|
|
assert not c._duplicated("evt-x")
|
|
c._seen["evt-x"] -= c._DEDUP_TTL + 1 # 只把这一个改成过期
|
|
assert not c._duplicated("evt-x")
|
|
|
|
|
|
def test_table_is_swept_only_at_threshold():
|
|
"""表长未到阈值时不做全表清理（摊还 O(1)），过期项留在表里也不影响判定。"""
|
|
c = make_connecter()
|
|
assert not c._duplicated("old")
|
|
c._seen["old"] -= c._DEDUP_TTL + 1
|
|
assert not c._duplicated("new") # 未到阈值：不清理
|
|
assert "old" in c._seen
|
|
# 灌到阈值，下一次判定触发清理，过期项被移除
|
|
for i in range(c._DEDUP_SWEEP_AT):
|
|
c._seen[f"filler-{i}"] = monotonic()
|
|
assert not c._duplicated("another")
|
|
assert "old" not in c._seen
|