How to Implement Broadcast Functionality to Send Messages to Multiple Groups in HoshinoBot

You can implement broadcast functionality in HoshinoBot using either the built-in super-user command in hoshino/modules/botmanage/broadcast.py or the Service.broadcast() method defined in hoshino/service.py, both of which iterate through group lists while handling rate limits and errors automatically.

HoshinoBot, the popular QQ bot framework maintained by ice9coffee/hoshinobot, provides two distinct architectural patterns for sending messages to multiple groups simultaneously. Understanding these built-in mechanisms allows you to create custom broadcast commands, scheduled announcements, or administrative alerts without reinventing the wheel.

Understanding HoshinoBot's Broadcast Architecture

The framework separates command-level broadcasting (super-user only) from service-level broadcasting (module-specific). Both approaches rely on the same underlying primitives: retrieving self-IDs via hoshino.get_self_ids(), fetching group lists through bot.get_group_list(), and delivering messages with bot.send_group_msg().

The key difference lies in scope and permissions. Super-user commands bypass service enablement checks, while service broadcasts respect per-group configuration, ensuring messages only reach groups where the specific service is active.

Method 1: Super-User Broadcast Command

Implementation in broadcast.py

The canonical implementation resides in hoshino/modules/botmanage/broadcast.py. This file defines a private command using the @sucmd decorator, which restricts execution to users listed in hoshino.config.SUPERUSERS.

The command iterates over every bot instance (self-ID), retrieves the complete group list for that instance, and transmits the message to each group with a 0.5-second delay to prevent rate limiting. Errors are caught, logged, and summarized for the operator.

import asyncio
import hoshino
from hoshino.service import sucmd

@sucmd('mybc', aliases=('mybroadcast',), force_private=True)
async def my_broadcast(session: hoshino.typing.CommandSession):
    msg = session.current_arg.strip()
    if not msg:
        await session.send('请在指令后面输入要广播的内容。')
        return

    bot = session.bot
    for sid in hoshino.get_self_ids():
        groups = await bot.get_group_list(self_id=sid)
        group_ids = [g['group_id'] for g in groups]

        try:
            await bot.send_private_msg(
                self_id=sid,
                user_id=session.event.user_id,
                message=f'开始向 {len(group_ids)} 个群广播：\n{msg}'
            )
        except Exception:
            hoshino.logger.error('向广播发起者发送概要失败')

        for gid in group_ids:
            await asyncio.sleep(0.5)
            try:
                await bot.send_group_msg(self_id=sid, group_id=gid, message=msg)
                hoshino.logger.info(f'群 {gid} 投递广播成功')
            except hoshino.typing.CQHttpError:
                hoshino.logger.error(f'群 {gid} 投递广播失败')

Key implementation details:

• force_private=True ensures the command only works in private messages, preventing accidental public exposure.
• hoshino.get_self_ids() handles multi-instance deployments where one HoshinoBot process manages multiple QQ accounts.
• The asyncio.sleep(0.5) throttle prevents hitting CQHTTP/OneBot rate limits when sending to hundreds of groups.

Method 2: Service-Level Broadcast Method

Using the Service.broadcast() API

For module-specific announcements that respect service enablement settings, use the broadcast method defined in hoshino/service.py at lines 60-77. This approach is ideal for scheduled jobs, daily news, or game event notifications where you only want to reach groups that have explicitly enabled your service.

The method signature accepts a message or list of messages, a tag for logging, and optional parameters for interval timing and message randomization. It internally calls get_enable_groups() to filter the target list, ensuring compliance with user preferences.

import random
from hoshino import Service

sv = Service('daily_news')

@sv.scheduled_job('cron', hour='9')
async def send_daily_news():
    news = "今天的新闻摘要……"

    await sv.broadcast(
        msgs=news,
        TAG='每日新闻',
        interval_time=0.3,
        randomizer=lambda x: f'{x} {random.choice(["😊","🚀","🌟"])}'
    )

Service-level advantages:

• Respects configuration: Only broadcasts to groups where daily_news is enabled in the service config.
• Simplified API: No need to manually iterate self_ids or fetch group lists; the service handles it.
• Flexible messaging: The randomizer parameter allows you to append random suffixes or vary message content per group.

Step-by-Step Implementation Guide

Creating a Custom Broadcast Command

To add a new broadcast capability to your HoshinoBot instance:

1. Create a new Python file in hoshino/modules/your_module/broadcast_custom.py.
2. Import sucmd from hoshino.service and hoshino for utilities.
3. Define your command function with the @sucmd decorator, specifying force_private=True for security.
4. Extract the message from session.current_arg.
5. Iterate through hoshino.get_self_ids() to handle all bot instances.
6. For each self-ID, fetch groups via bot.get_group_list(self_id=sid).
7. Send messages with bot.send_group_msg and throttle with asyncio.sleep(0.5).

Handling Rate Limits and Errors

The CQHTTP/OneBot protocol imposes rate limits on group messaging. HoshinoBot's built-in implementations use a 0.3-0.5 second delay between messages. For larger deployments (500+ groups), consider:

• Increasing interval_time to 1.0 second or higher.
• Implementing exponential backoff when catching CQHttpError.
• Using the randomizer function in Service.broadcast to vary message content, which may reduce spam detection triggers.

Always wrap send_group_msg calls in try-except blocks to catch hoshino.typing.CQHttpError and log failures without stopping the broadcast loop.

Key Files and Functions Reference

File	Purpose	Key Functions
hoshino/modules/botmanage/broadcast.py	Built-in super-user broadcast command	sucmd('broadcast') implementation
hoshino/service.py	Service class definition	Service.broadcast(), get_enable_groups()
hoshino/__init__.py	Global utilities	get_self_ids(), logger access
hoshino/typing.py	Type definitions and exceptions	CQHttpError, CommandSession

These files demonstrate the official pattern for multi-group messaging in HoshinoBot. The separation between command-level and service-level broadcasting allows you to choose the appropriate permission model and group filtering strategy for your specific use case.

Summary

• Two built-in approaches: Use sucmd in hoshino/modules/botmanage/broadcast.py for admin-only broadcasts to all groups, or Service.broadcast() in hoshino/service.py for module-specific announcements that respect service settings.
• Multi-instance support: Always iterate over hoshino.get_self_ids() when implementing custom broadcasts to support bots running multiple QQ accounts.
• Rate limiting: Implement asyncio.sleep(0.5) between send_group_msg calls to avoid CQHTTP/OneBot rate limits.
• Error resilience: Wrap individual group sends in try-except blocks catching CQHttpError to ensure one failed delivery doesn't abort the entire broadcast.

Frequently Asked Questions

What is the difference between sucmd broadcast and Service.broadcast()?

The sucmd approach in hoshino/modules/botmanage/broadcast.py is a super-user command that sends messages to every group the bot has joined, regardless of module settings. It is designed for administrative emergencies or global announcements. In contrast, Service.broadcast() defined in hoshino/service.py only targets groups where that specific service is enabled, making it ideal for module-specific features like daily news or game notifications that users can opt into or out of.

How do I handle rate limits when broadcasting to hundreds of groups?

When broadcasting to large numbers of groups, implement a throttle using await asyncio.sleep(0.5) between each bot.send_group_msg() call, as shown in the built-in broadcast.py implementation. For Service.broadcast(), set the interval_time parameter to 0.5 or higher. If you encounter CQHttpError exceptions indicating rate limiting, implement exponential backoff by catching the error and increasing the sleep duration dynamically.

Can I filter which groups receive my broadcast message?

Yes, but the approach depends on which broadcast method you use. With Service.broadcast(), filtering happens automatically based on the service's enablement configuration—only groups where your service is enabled will receive the message. For custom sucmd implementations, you must manually implement filtering by checking group_id against an allowlist or blocklist before calling send_group_msg. You can also query group information using bot.get_group_info() to filter by group name, member count, or other attributes.

How do I verify that my broadcast was delivered successfully?

The built-in implementations log delivery results using hoshino.logger. In your custom broadcast code, wrap each send_group_msg call in a try-except block catching hoshino.typing.CQHttpError. Log successful deliveries with logger.info() and failures with logger.error(). For real-time feedback to the administrator, accumulate success/failure counts during the loop and send a summary message to the super-user via bot.send_private_msg() after the broadcast completes, as demonstrated in the official broadcast.py implementation.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}