style: enforce the 100-column baseline repo-wide (E501)

prod.example/deploy-state.py

@@ -23,7 +23,9 @@
 SERVICES = ("llbot", "quickquip", "web-admin")
 
 
-def atomic_write(path: Path, data: bytes, mode: int, new_owner: tuple[int, int] | None = None) -> None:
+def atomic_write(
+    path: Path, data: bytes, mode: int, new_owner: tuple[int, int] | None = None
+) -> None:
     missing_parents = []
     parent = path.parent
     while new_owner is not None and not parent.exists():
@@ -122,7 +124,8 @@ def apply_shared(root: Path, incoming: Path, backup: Path) -> None:
                 connections.append({})
             connections[1]["url"] = "ws://quickquip:8080/onebot/v11/ws/"
             connections[1]["token"] = token or ""
-            atomic_write(path, (json.dumps(data, ensure_ascii=False, indent=4) + "\n").encode(), record["mode"])
+            payload = (json.dumps(data, ensure_ascii=False, indent=4) + "\n").encode()
+            atomic_write(path, payload, record["mode"])
 
 
 def restore_shared(root: Path, backup: Path) -> None:
@@ -136,15 +139,24 @@ def restore_shared(root: Path, backup: Path) -> None:
 
 def capture_baseline(root: Path, baseline: Path) -> None:
     """Capture server files and running image IDs without moving live bind sources."""
-    command = ["docker", "compose", "--env-file", str(root / ".env"), "-f", str(root / "prod/docker-compose.yml")]
+    command = [
+        "docker", "compose", "--env-file", str(root / ".env"),
+        "-f", str(root / "prod/docker-compose.yml"),
+    ]
     # No interpolation also preserves env_file references on Compose 2.27+.
     raw = subprocess.check_output(command + ["config", "--no-interpolate", "--format", "json"])
     config = json.loads(raw)
     if set(config["services"]) != set(SERVICES):
-        raise ValueError("migration requires exactly llbot, quickquip and web-admin; review custom services first")
+        raise ValueError(
+            "migration requires exactly llbot, quickquip and web-admin; "
+            "review custom services first"
+        )
     # Keep private runtime files out of the snapshot; copy only mounted app assets.
     baseline.mkdir(mode=0o700)
-    for name in ("src", "config", "llm_about", "frontend/dist", "bot.py", "web_api.py", "pyproject.toml", "requirements.txt", ".dockerignore"):
+    for name in (
+        "src", "config", "llm_about", "frontend/dist", "bot.py", "web_api.py",
+        "pyproject.toml", "requirements.txt", ".dockerignore",
+    ):
         source = checked_path(root, name)
         if source.is_dir():
             validate_tree(source)
@@ -155,7 +167,9 @@ def capture_baseline(root: Path, baseline: Path) -> None:
         container = spec.get("container_name")
         if not container:
             raise ValueError(f"migration needs container_name for {service}")
-        image = subprocess.check_output(["docker", "inspect", "--format", "{{.Image}}", container], text=True).strip()
+        image = subprocess.check_output(
+            ["docker", "inspect", "--format", "{{.Image}}", container], text=True
+        ).strip()
         tag = f"quickquip-{service}:{baseline.name}"
         subprocess.run(["docker", "tag", image, tag], check=True)
         spec["image"] = tag
@@ -173,14 +187,22 @@ def capture_baseline(root: Path, baseline: Path) -> None:
             if not source.is_relative_to(root):
                 raise ValueError(f"external bind mount requires manual migration: {source}")
             relative = source.relative_to(root)
-            if relative.parts[0] == "data" or str(relative) in ("prod/llbot-qq", "prod/llbot-data", ".env"):
+            if relative.parts[0] == "data" or str(relative) in (
+                "prod/llbot-qq", "prod/llbot-data", ".env",
+            ):
                 volume["source"] = str(source)
             elif (baseline / relative).exists():
                 volume["source"] = str(baseline / relative)
             else:
                 raise ValueError(f"unsupported bind mount: {relative}")
     atomic_write(baseline / "prod/docker-compose.yml", json.dumps(config).encode(), 0o600)
-    subprocess.run(["docker", "compose", "--env-file", str(root / ".env"), "-f", str(baseline / "prod/docker-compose.yml"), "config", "--quiet"], check=True)
+    subprocess.run(
+        [
+            "docker", "compose", "--env-file", str(root / ".env"),
+            "-f", str(baseline / "prod/docker-compose.yml"), "config", "--quiet",
+        ],
+        check=True,
+    )
 
 
 def main() -> None:
@@ -201,5 +223,8 @@ def main() -> None:
     try:
         main()
     except PermissionError as exc:
-        print(f"deployment filesystem permission denied: {exc.filename or 'shared files'}", file=sys.stderr)
+        print(
+            f"deployment filesystem permission denied: {exc.filename or 'shared files'}",
+            file=sys.stderr,
+        )
         raise SystemExit(3) from None

pyproject.toml

@@ -24,7 +24,6 @@ exclude = [".venv"]
 
 [tool.ruff.lint]
 select = ["E", "F"]
-ignore = ["E501"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

scripts/backfill_record_identities.py

@@ -20,7 +20,11 @@
 from quickquip.common.record_content import legacy, references, render, validate  # noqa: E402
 from quickquip.common.record_storage import migrate, save_parts  # noqa: E402
 
-DATABASES = {"memories": LLM_DB_PATH, "quotes": QUOTES_DB_PATH, "offline_messages": OFFLINE_MESSAGES_DB_PATH}
+DATABASES = {
+    "memories": LLM_DB_PATH,
+    "quotes": QUOTES_DB_PATH,
+    "offline_messages": OFFLINE_MESSAGES_DB_PATH,
+}
 
 
 class ApplyResult(Enum):
@@ -40,15 +44,23 @@ def _iter_rows(reader, table, group, record_id, batch_size):
         if record_id is not None:
             conditions.append("id=?")
             params.append(record_id)
-        rows = reader.execute(f"SELECT * FROM {table} WHERE {' AND '.join(conditions)} ORDER BY id LIMIT ?", (*params, batch_size)).fetchall()
+        rows = reader.execute(
+            f"SELECT * FROM {table} WHERE {' AND '.join(conditions)} ORDER BY id LIMIT ?",
+            (*params, batch_size),
+        ).fetchall()
         if not rows:
             return
         yield from rows
         last_id = rows[-1]["id"]
 
 
 def _prepare_writer(reader, path, table, report):
-    backup = path.with_name(path.name + ".identities-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + ".bak")
+    backup = path.with_name(
+        path.name
+        + ".identities-"
+        + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
+        + ".bak"
+    )
     with closing(sqlite3.connect(backup)) as target:
         reader.backup(target)
     if report:
@@ -71,20 +83,50 @@ def _preview_row(row, body):
 def _apply_row(writer, table, row, encoded, body):
     with writer:
         writer.execute("BEGIN IMMEDIATE")
-        current = writer.execute(f"SELECT content, content_parts_json FROM {table} WHERE id=?", (row["id"],)).fetchone()
+        current = writer.execute(
+            f"SELECT content, content_parts_json FROM {table} WHERE id=?", (row["id"],)
+        ).fetchone()
         if current is None or current[0] != row["content"] or current[1] != encoded:
             return ApplyResult.CONCURRENT_SKIPPED
         if encoded is None:
             save_parts(writer, table, row["id"], row["group_id"], body)
             return ApplyResult.WRITTEN
-        existing = {r[0] for r in writer.execute(f"SELECT qq FROM {table}_member_refs WHERE record_id=?", (row["id"],))}
+        existing = {
+            r[0]
+            for r in writer.execute(
+                f"SELECT qq FROM {table}_member_refs WHERE record_id=?", (row["id"],)
+            )
+        }
         missing = references(body) - existing
-        writer.executemany(f"INSERT INTO {table}_member_refs(group_id, record_id, qq) VALUES (?, ?, ?)", [(row["group_id"], row["id"], qq) for qq in missing])
+        writer.executemany(
+            f"INSERT INTO {table}_member_refs(group_id, record_id, qq) VALUES (?, ?, ?)",
+            [(row["group_id"], row["id"], qq) for qq in missing],
+        )
         return ApplyResult.INDEX_REPAIRED if missing else ApplyResult.UNCHANGED
 
 
-def backfill(path, table, *, apply=False, group=None, record_id=None, batch_size=200, before_write=None, preview_limit=0, report=None):
-    counts = dict(scanned=0, convertible=0, unparsed=0, existing=0, concurrent_skipped=0, failed=0, written=0, index_repaired=0)
+def backfill(
+    path,
+    table,
+    *,
+    apply=False,
+    group=None,
+    record_id=None,
+    batch_size=200,
+    before_write=None,
+    preview_limit=0,
+    report=None,
+):
+    counts = dict(
+        scanned=0,
+        convertible=0,
+        unparsed=0,
+        existing=0,
+        concurrent_skipped=0,
+        failed=0,
+        written=0,
+        index_repaired=0,
+    )
     path = Path(path).resolve()
     if table not in DATABASES:
         raise ValueError("unsupported table")
@@ -100,9 +142,23 @@ def backfill(path, table, *, apply=False, group=None, record_id=None, batch_size
             for row in _iter_rows(reader, table, group, record_id, batch_size):
                 counts["scanned"] += 1
                 try:
-                    encoded = row["content_parts_json"] if "content_parts_json" in row.keys() else None
-                    body = validate(json.loads(encoded), max_length=1_000_000) if encoded is not None else legacy(row["content"])
-                    category = "existing" if encoded is not None else "convertible" if any(p["type"] != "text" for p in body["parts"]) else "unparsed"
+                    encoded = (
+                        row["content_parts_json"]
+                        if "content_parts_json" in row.keys()
+                        else None
+                    )
+                    body = (
+                        validate(json.loads(encoded), max_length=1_000_000)
+                        if encoded is not None
+                        else legacy(row["content"])
+                    )
+                    category = (
+                        "existing"
+                        if encoded is not None
+                        else "convertible"
+                        if any(p["type"] != "text" for p in body["parts"])
+                        else "unparsed"
+                    )
                     counts[category] += 1
                     if encoded is None and counts["scanned"] <= preview_limit and report:
                         report(_preview_row(row, body))
@@ -123,7 +179,10 @@ def backfill(path, table, *, apply=False, group=None, record_id=None, batch_size
 
 
 def _emit(event):
-    print(json.dumps(event, ensure_ascii=False), file=sys.stderr if "error" in event else sys.stdout)
+    print(
+        json.dumps(event, ensure_ascii=False),
+        file=sys.stderr if "error" in event else sys.stdout,
+    )
 
 
 def main(argv=None):
@@ -134,7 +193,12 @@ def main(argv=None):
     parser.add_argument("--record-id", type=int)
     parser.add_argument("--batch-size", type=int, default=200)
     parser.add_argument("--apply", action="store_true")
-    parser.add_argument("--preview-limit", type=int, default=10, help="Maximum record examples per database; 0 prints counts only")
+    parser.add_argument(
+        "--preview-limit",
+        type=int,
+        default=10,
+        help="Maximum record examples per database; 0 prints counts only",
+    )
     args = parser.parse_args(argv)
     if args.path and args.database == "all":
         parser.error("--path requires a single --database")
@@ -145,7 +209,16 @@ def main(argv=None):
         if args.database not in {table, "all"}:
             continue
         try:
-            counts = backfill(args.path or default, table, apply=args.apply, group=args.group, record_id=args.record_id, batch_size=args.batch_size, preview_limit=args.preview_limit, report=_emit)
+            counts = backfill(
+                args.path or default,
+                table,
+                apply=args.apply,
+                group=args.group,
+                record_id=args.record_id,
+                batch_size=args.batch_size,
+                preview_limit=args.preview_limit,
+                report=_emit,
+            )
             failed |= bool(counts["failed"])
             _emit({"database": table, **counts})
         except Exception as exc:

scripts/ci/mcp_dep_audit.py

@@ -46,7 +46,11 @@ def _install_and_list(pip: str, spec: str) -> set[str] | None:
         capture_output=True,
         text=True,
     )
-    return {line.split("==")[0].lower() for line in result.stdout.strip().splitlines() if "==" in line}
+    return {
+        line.split("==")[0].lower()
+        for line in result.stdout.strip().splitlines()
+        if "==" in line
+    }
 
 
 def main() -> int:

src/quickquip/adapters/nonebot/_forward.py

@@ -49,7 +49,13 @@ def _extract_forward_payload(message) -> tuple[str, list[object]]:
     return "", []
 
 
-def _format_forward_sender(sender_name: str, user_id: str, *, bot_keys: set[str], identities: IdentityIndex) -> str:
+def _format_forward_sender(
+    sender_name: str,
+    user_id: str,
+    *,
+    bot_keys: set[str],
+    identities: IdentityIndex,
+) -> str:
     normalized_user_id = user_id.strip()
     if normalized_user_id and normalized_user_id in bot_keys:
         return f"机器人（QQ {normalized_user_id}）"
@@ -167,7 +173,11 @@ async def _render_forward_content(
                     try:
                         result = await bot.call_api("get_forward_msg", message_id=nested_id)
                     except Exception:
-                        logger.warning("Failed to fetch nested forward message id=%s", nested_id, exc_info=True)
+                        logger.warning(
+                            "Failed to fetch nested forward message id=%s",
+                            nested_id,
+                            exc_info=True,
+                        )
                         nested_text = ""
                     else:
                         nested_nodes = []
@@ -263,5 +273,8 @@ async def extract_forward_content(
         visited_forward_ids={forward_id} if forward_id else set(),
     )
     if len(rendered_text) > MAX_FORWARD_TEXT_CHARS:
-        rendered_text = rendered_text[:MAX_FORWARD_TEXT_CHARS].rstrip() + "…（合并转发内容过长，已截断）"
+        rendered_text = (
+            rendered_text[:MAX_FORWARD_TEXT_CHARS].rstrip()
+            + "…（合并转发内容过长，已截断）"
+        )
     return rendered_text, image_urls

src/quickquip/adapters/nonebot/_llm_reply.py

@@ -135,7 +135,11 @@ async def __call__(self, delivery_id: str, payload: dict[str, Any]) -> DeliveryR
         return DeliveryReceipt(status=DeliveryStatus.UNKNOWN, error_code="missing_message_id")
 
 
-def text_only_message(text: str, Message: type[OneBotMessage], MessageSegment: type[OneBotMessageSegment]) -> OneBotMessage:
+def text_only_message(
+    text: str,
+    Message: type[OneBotMessage],
+    MessageSegment: type[OneBotMessageSegment],
+) -> OneBotMessage:
     """纯文本 Message（§6.2）：分段正文不经 CQ 解析器。
 
     正文中的 ``@QQ 号`` 数字艾特在此出口切分为真实 at 段（分段交付与
@@ -144,15 +148,19 @@ def text_only_message(text: str, Message: type[OneBotMessage], MessageSegment: t
     return Message(split_outbound_at_mentions(text, Message, MessageSegment))
 
 
-def make_matcher_sink(matcher, Message, MessageSegment, *, scope_key: str, interval_ms: int) -> OneBotDeliverySink:
+def make_matcher_sink(
+    matcher, Message, MessageSegment, *, scope_key: str, interval_ms: int
+) -> OneBotDeliverySink:
     return OneBotDeliverySink(
         lambda text: matcher.send(text_only_message(text, Message, MessageSegment)),
         scope_key=scope_key,
         interval_ms=interval_ms,
     )
 
 
-def make_group_bot_sink(bot, Message, MessageSegment, *, group_id: int | str, interval_ms: int) -> OneBotDeliverySink:
+def make_group_bot_sink(
+    bot, Message, MessageSegment, *, group_id: int | str, interval_ms: int
+) -> OneBotDeliverySink:
     async def _send(text: str):
         return await bot.send_group_msg(
             group_id=int(group_id),
@@ -162,7 +170,9 @@ async def _send(text: str):
     return OneBotDeliverySink(_send, scope_key=str(group_id), interval_ms=interval_ms)
 
 
-def make_private_bot_sink(bot, Message, MessageSegment, *, user_id: int | str, interval_ms: int) -> OneBotDeliverySink:
+def make_private_bot_sink(
+    bot, Message, MessageSegment, *, user_id: int | str, interval_ms: int
+) -> OneBotDeliverySink:
     async def _send(text: str):
         return await bot.send_private_msg(
             user_id=int(user_id),

src/quickquip/adapters/nonebot/awakening_plugin.py

@@ -197,7 +197,10 @@ async def _(event):
             lines.append(f"唤醒延长: {settings.extend_duration}s")
             lines.append(f"兴趣话题: {settings.interest_topics or '(未配置)'}")
             lines.append(f"兜底概率: {settings.fallback_probability}")
-            lines.append(f"无聊沉寂: {settings.boredom_silence_seconds}s / 概率 {settings.boredom_probability}")
+            lines.append(
+                f"无聊沉寂: {settings.boredom_silence_seconds}s"
+                f" / 概率 {settings.boredom_probability}"
+            )
             lines.append(f"无聊检查间隔: {settings.boredom_check_interval}s")
             lines.append(f"相关性阈值: {settings.relevance_threshold} (>=1 关闭)")
             lines.append(f"答疑阈值: {settings.qa_threshold} (>=1 关闭)")
@@ -211,7 +214,11 @@ async def _(event):
             if not _is_admin(event):
                 await cmd.finish("仅管理员可执行此操作")
             if len(tokens) < 2:
-                await cmd.finish("用法: /awakening on <规则名>\n可选: awakening_extend, awakening_interest, awakening_fallback, awakening_boredom, awakening_relevance, awakening_qa")
+                await cmd.finish(
+                    "用法: /awakening on <规则名>\n"
+                    "可选: awakening_extend, awakening_interest, awakening_fallback, "
+                    "awakening_boredom, awakening_relevance, awakening_qa"
+                )
             rule_name = tokens[1]
             if rule_name not in AWAKENING_RULE_NAMES:
                 await cmd.finish(f"未知规则: {rule_name}")