multi-group-architecture

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

When to use this skill

何时使用该功能

Use this skill when you need to:

Support multiple LINE groups with independent configurations
Implement group isolation for data and settings
Handle group join/leave events automatically
Manage per-group member lists and schedules
Design scalable multi-tenant bot architecture
Migrate from single-group to multi-group system

当你需要以下功能时使用该技能：

支持多个拥有独立配置的LINE群组
实现数据与设置的群组隔离
自动处理群组加入/退出事件
管理各群组的成员列表与日程
设计可扩展的多租户Bot架构
从单群组系统迁移至多群组系统

How to use it

如何使用该功能

Core Principles

核心原则

Group Isolation: Each group has completely separate data
Automatic Registration: Groups auto-register when bot joins
Independent Scheduling: Each group can have different broadcast times
Context-Aware: All operations require group_id context

群组隔离：每个群组拥有完全独立的数据
自动注册：Bot加入群组时自动完成注册
独立日程：每个群组可设置不同的推送时间
上下文感知：所有操作均需传入group_id上下文

Data Architecture

数据架构

1. Group Registry

1. 群组注册表

python

undefined

python

undefined

List of all registered groups

所有已注册群组的列表

group_ids = [ "C1234567890abcdef1234567890abcdef", "C9876543210fedcba9876543210fedcba" ]

undefined

group_ids = [ "C1234567890abcdef1234567890abcdef", "C9876543210fedcba9876543210fedcba" ]

undefined

2. Per-Group Member Data

2. 群组独立成员数据

python

groups = {
    "C群組ID1": {
        "1": ["Alice", "Bob"],
        "2": ["Charlie"],
        "3": ["David", "Eve"]
    },
    "C群組ID2": {
        "1": ["Frank"],
        "2": ["Grace", "Henry"]
    }
}

python

groups = {
    "C群組ID1": {
        "1": ["Alice", "Bob"],
        "2": ["Charlie"],
        "3": ["David", "Eve"]
    },
    "C群組ID2": {
        "1": ["Frank"],
        "2": ["Grace", "Henry"]
    }
}

3. Per-Group Schedules

3. 群组独立日程

python

group_schedules = {
    "C群組ID1": {
        "days": "mon,wed,fri",
        "hour": 17,
        "minute": 0
    },
    "C群組ID2": {
        "days": "tue,thu",
        "hour": 9,
        "minute": 30
    }
}

python

group_schedules = {
    "C群組ID1": {
        "days": "mon,wed,fri",
        "hour": 17,
        "minute": 0
    },
    "C群組ID2": {
        "days": "tue,thu",
        "hour": 9,
        "minute": 30
    }
}

4. Per-Group Custom Messages (Optional)

4. 群组自定义消息（可选）

python

group_messages = {
    "C群組ID1": {
        "reminder_template": "🗑️ 今天輪到 {members} 收垃圾！"
    },
    "C群組ID2": {
        "reminder_template": "📢 {members} 請記得收垃圾喔！"
    }
}

python

group_messages = {
    "C群組ID1": {
        "reminder_template": "🗑️ 今天輪到 {members} 收垃圾！"
    },
    "C群組ID2": {
        "reminder_template": "📢 {members} 請記得收垃圾喔！"
    }
}

Implementation Patterns

实现模式

1. Extracting Group ID from Event

1. 从事件中提取群组ID

python

def get_group_id_from_event(event):
    """
    Extract group ID from LINE event
    
    Returns:
        str: Group ID or None if not from a group
    """
    try:
        if hasattr(event.source, 'group_id'):
            return event.source.group_id
        else:
            # Private chat, not a group
            return None
    except Exception as e:
        print(f"Failed to get group ID: {e}")
        return None

python

def get_group_id_from_event(event):
    """
    从LINE事件中提取群组ID
    
    返回值：
        str: 群组ID，若不是来自群组则返回None
    """
    try:
        if hasattr(event.source, 'group_id'):
            return event.source.group_id
        else:
            # 私人聊天，非群组
            return None
    except Exception as e:
        print(f"获取群组ID失败: {e}")
        return None

2. Group Join Event Handler

2. 群组加入事件处理器

python

@handler.add(JoinEvent)
def handle_join(event):
    """
    Auto-register group when bot joins
    """
    group_id = get_group_id_from_event(event)
    
    if group_id and group_id not in group_ids:
        # Register new group
        group_ids.append(group_id)
        save_group_ids()
        
        # Initialize empty group data
        groups[group_id] = {}
        save_groups()
        
        # Send welcome message
        welcome_msg = (
            "👋 感謝加入垃圾輪值提醒 Bot！\n\n"
            "📝 請使用以下指令設定：\n"
            "@time 18:00 - 設定推播時間\n"
            "@day mon,wed,fri - 設定推播日期\n"
            "@week 1 小明,小華 - 設定第1週成員\n\n"
            "使用 @help 查看完整指令列表"
        )
        
        messaging_api.reply_message(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text=welcome_msg)]
            )
        )
        
        print(f"✅ New group registered: {group_id}")

python

@handler.add(JoinEvent)
def handle_join(event):
    """
    Bot加入时自动注册群组
    """
    group_id = get_group_id_from_event(event)
    
    if group_id and group_id not in group_ids:
        # 注册新群组
        group_ids.append(group_id)
        save_group_ids()
        
        # 初始化空的群组数据
        groups[group_id] = {}
        save_groups()
        
        # 发送欢迎消息
        welcome_msg = (
            "👋 感謝加入垃圾輪值提醒 Bot！\n\n"
            "📝 請使用以下指令設定：\n"
            "@time 18:00 - 設定推播時間\n"
            "@day mon,wed,fri - 設定推播日期\n"
            "@week 1 小明,小華 - 設定第1週成員\n\n"
            "使用 @help 查看完整指令列表"
        )
        
        messaging_api.reply_message(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text=welcome_msg)]
            )
        )
        
        print(f"✅ 新群组已注册: {group_id}")

3. Group Leave Event Handler

3. 群组退出事件处理器

python

@handler.add(LeaveEvent)
def handle_leave(event):
    """
    Clean up when bot leaves a group
    """
    group_id = get_group_id_from_event(event)
    
    if group_id:
        # Remove from group list
        if group_id in group_ids:
            group_ids.remove(group_id)
            save_group_ids()
        
        # Clean up group data
        if group_id in groups:
            del groups[group_id]
            save_groups()
        
        # Remove schedule
        if group_id in group_schedules:
            del group_schedules[group_id]
            save_group_schedules(group_schedules)
            
            # Remove scheduled job
            job_id = f"reminder_{group_id}"
            if scheduler.get_job(job_id):
                scheduler.remove_job(job_id)
        
        print(f"🗑️ Group cleaned up: {group_id}")

python

@handler.add(LeaveEvent)
def handle_leave(event):
    """
    Bot退出群组时清理数据
    """
    group_id = get_group_id_from_event(event)
    
    if group_id:
        # 从群组列表中移除
        if group_id in group_ids:
            group_ids.remove(group_id)
            save_group_ids()
        
        # 清理群组数据
        if group_id in groups:
            del groups[group_id]
            save_groups()
        
        # 移除日程
        if group_id in group_schedules:
            del group_schedules[group_id]
            save_group_schedules(group_schedules)
            
            # 移除定时任务
            job_id = f"reminder_{group_id}"
            if scheduler.get_job(job_id):
                scheduler.remove_job(job_id)
        
        print(f"🗑️ 群组数据已清理: {group_id}")

4. Context-Aware Operations

4. 上下文感知操作

Always pass group_id to functions:

python

def get_current_group(group_id):
    """Get current week members for specific group"""
    if group_id not in groups:
        return []
    
    group_data = groups[group_id]
    # ... calculation logic
    return current_members

def get_member_schedule(group_id):
    """Get schedule info for specific group"""
    if group_id not in groups:
        return {
            "total_weeks": 0,
            "current_week": 1,
            "group_id": group_id,
            "current_members": []
        }
    
    # ... schedule logic
    return schedule_info

始终向函数传入group_id：

python

def get_current_group(group_id):
    """获取指定群组的当前周成员"""
    if group_id not in groups:
        return []
    
    group_data = groups[group_id]
    # ... 计算逻辑
    return current_members

def get_member_schedule(group_id):
    """获取指定群组的日程信息"""
    if group_id not in groups:
        return {
            "total_weeks": 0,
            "current_week": 1,
            "group_id": group_id,
            "current_members": []
        }
    
    # ... 日程逻辑
    return schedule_info

5. Command Handler with Group Context

5. 带群组上下文的命令处理器

python

@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
    text = event.message.text.strip()
    group_id = get_group_id_from_event(event)
    
    # Require group context
    if not group_id:
        reply_text = "❌ 此 Bot 僅支援群組使用"
        messaging_api.reply_message(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text=reply_text)]
            )
        )
        return
    
    # Process commands with group context
    if text.startswith('@'):
        parts = text[1:].split()
        command = parts[0].lower()
        
        if command == 'schedule':
            # Show THIS group's schedule
            reply_text = get_schedule_info(group_id)
        elif command == 'members':
            # Show THIS group's members
            reply_text = get_member_schedule_summary(group_id)
        # ... more commands

python

@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
    text = event.message.text.strip()
    group_id = get_group_id_from_event(event)
    
    # 要求群组上下文
    if not group_id:
        reply_text = "❌ 此 Bot 僅支援群組使用"
        messaging_api.reply_message(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text=reply_text)]
            )
        )
        return
    
    # 处理带群组上下文的命令
    if text.startswith('@'):
        parts = text[1:].split()
        command = parts[0].lower()
        
        if command == 'schedule':
            # 显示当前群组的日程
            reply_text = get_schedule_info(group_id)
        elif command == 'members':
            # 显示当前群组的成员
            reply_text = get_member_schedule_summary(group_id)
        # ... 更多命令

Migration Strategy

迁移策略

From Single Group to Multi-Group

从单群组迁移至多群组

Step 1: Add Legacy Group Support

python

undefined

步骤1：添加旧版群组支持

python

undefined

Convert old single-group data to multi-group format

将旧版单群组数据转换为多群组格式

def migrate_to_multi_group(): global groups

# If groups is still a dict with week keys
if groups and '1' in groups:
    # This is old format, migrate to new
    legacy_data = groups.copy()
    groups = {
        "legacy": legacy_data
    }
    save_groups()
    print("✅ Migrated to multi-group format")


**Step 2: Backward Compatible Functions**
```python
def get_current_group(group_id=None):
    """
    Support both old and new usage
    
    Args:
        group_id: Group ID (None for legacy mode)
    """
    if group_id is None:
        # Legacy mode: use first available group
        if "legacy" in groups:
            group_data = groups["legacy"]
        elif groups:
            group_data = next(iter(groups.values()))
        else:
            return []
    else:
        # New mode: use specific group
        if group_id not in groups:
            return []
        group_data = groups[group_id]
    
    # ... rest of logic

def migrate_to_multi_group(): global groups

# 如果groups仍是带有周数键的字典
if groups and '1' in groups:
    # 这是旧格式，迁移为新格式
    legacy_data = groups.copy()
    groups = {
        "legacy": legacy_data
    }
    save_groups()
    print("✅ 已迁移至多群组格式")


**步骤2：向后兼容的函数**
```python
def get_current_group(group_id=None):
    """
    同时支持新旧两种使用方式
    
    参数：
        group_id: 群组ID（为None时使用旧版模式）
    """
    if group_id is None:
        # 旧版模式：使用第一个可用群组
        if "legacy" in groups:
            group_data = groups["legacy"]
        elif groups:
            group_data = next(iter(groups.values()))
        else:
            return []
    else:
        # 新版模式：使用指定群组
        if group_id not in groups:
            return []
        group_data = groups[group_id]
    
    # ... 剩余逻辑

Best Practices

最佳实践

Always Validate Group ID

python

if group_id not in groups:
    return error_response("Group not found")

Consistent Storage Keys Use group_id as the primary key across all dictionaries

Atomic Operations Update all related data together:

python

# Update members
groups[group_id] = new_data
save_groups()

# Update schedule
group_schedules[group_id] = new_schedule
save_group_schedules(group_schedules)

Group-Specific Responses Include group context in responses when helpful:

python

reply_text = f"✅ 已更新群組設定\n\n"
reply_text += f"📊 目前有 {len(groups[group_id])} 週輪值"

Testing Multi-Group
- Create multiple test groups
- Verify data isolation between groups
- Test concurrent operations on different groups

始终验证群组ID

python

if group_id not in groups:
    return error_response("群组未找到")

统一存储键名 使用group_id作为所有字典的主键

原子操作 同时更新所有相关数据：

python

# 更新成员数据
groups[group_id] = new_data
save_groups()

# 更新日程
group_schedules[group_id] = new_schedule
save_group_schedules(group_schedules)

群组专属响应 在有帮助时将群组上下文包含在响应中：

python

reply_text = f"✅ 已更新群組設定\n\n"
reply_text += f"📊 目前有 {len(groups[group_id])} 週輪值"

多群组测试
- 创建多个测试群组
- 验证群组间的数据隔离性
- 测试不同群组上的并发操作

Firestore Schema for Multi-Group

多群组的Firestore数据结构

javascript

bot_config/
  groups/
    {
      "C群組ID1": {
        "1": ["Alice", "Bob"],
        "2": ["Charlie"]
      },
      "C群組ID2": {
        "1": ["David"],
        "2": ["Eve", "Frank"]
      }
    }
  
  group_schedules/
    {
      "C群組ID1": {
        "days": "mon,wed,fri",
        "hour": 17,
        "minute": 0
      }
    }

javascript

bot_config/
  groups/
    {
      "C群組ID1": {
        "1": ["Alice", "Bob"],
        "2": ["Charlie"]
      },
      "C群組ID2": {
        "1": ["David"],
        "2": ["Eve", "Frank"]
      }
    }
  
  group_schedules/
    {
      "C群組ID1": {
        "days": "mon,wed,fri",
        "hour": 17,
        "minute": 0
      }
    }

Common Pitfalls

常见误区

❌ Don't: Assume single group

python

current_members = get_current_group()  # Which group?

✅ Do: Always specify group

python

current_members = get_current_group(group_id)

❌ Don't: Share data across groups

python

base_date = date.today()  # Global for all groups

✅ Do: Store per-group if needed

python

group_base_dates = {
    group_id: date.today()
}

❌ 错误做法：假设仅存在单一群组

python

current_members = get_current_group()  # 哪个群组？

✅ 正确做法：始终指定群组

python

current_members = get_current_group(group_id)

❌ 错误做法：在群组间共享数据

python

base_date = date.today()  # 所有群组共用

✅ 正确做法：按需按群组存储

python

group_base_dates = {
    group_id: date.today()
}

Reference Patterns

参考实现

See existing implementation in:

```
main.py
```
: Multi-group command handlers
```
firebase_service.py
```
: Group data storage
```
README.md
```
: Multi-group usage examples

可参考以下现有实现：

```
main.py
```
: 多群组命令处理器
```
firebase_service.py
```
: 群组数据存储
```
README.md
```
: 多群组使用示例