بناء bot لإشعارات الفريق

بنية روبوت إشعارات الفريق

# Team Notification Bot Flow:
#
# External System (GitHub, PagerDuty, Sentry, etc.)
#    |
#    | HTTP POST (webhook)
#    v
# Flask/FastAPI webhook endpoint
#    |
#    | Parse event
#    v
# Agent: classify, format, route
#    |
#    | Slack API
#    v
# Team channel / DM / thread

print('Webhook -> Agent -> Slack is the core notification pattern')

استقبال Webhooks خارجية

from flask import Flask, request, jsonify
import hmac
import hashlib
import os

flask_app = Flask(__name__)

@flask_app.route('/webhook/github', methods=['POST'])
def github_webhook():
    # Verify GitHub signature
    signature = request.headers.get('X-Hub-Signature-256', '')
    secret = os.environ['GITHUB_WEBHOOK_SECRET'].encode()
    body = request.get_data()
    expected = 'sha256=' + hmac.new(secret, body, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({'error': 'Invalid signature'}), 403

    event_type = request.headers.get('X-GitHub-Event', '')
    payload = request.json

    handle_github_event(event_type, payload)
    return jsonify({'status': 'ok'}), 200

توجيه الأحداث إلى القناة المناسبة

import os

# Channel routing configuration
CHANNEL_MAP = {
    'deploy': os.environ.get('DEPLOY_CHANNEL', 'C0DEPLOY123'),
    'error': os.environ.get('ERROR_CHANNEL', 'C0ERROR456'),
    'pr_review': os.environ.get('PR_CHANNEL', 'C0PR789'),
    'general': os.environ.get('GENERAL_CHANNEL', 'C0GENERAL'),
}

def route_github_event(event_type, payload):
    if event_type == 'push':
        branch = payload.get('ref', '').replace('refs/heads/', '')
        if branch in ('main', 'master'):
            return CHANNEL_MAP['deploy']
        return CHANNEL_MAP['general']
    elif event_type == 'pull_request':
        return CHANNEL_MAP['pr_review']
    elif event_type == 'workflow_run':
        if payload.get('workflow_run', {}).get('conclusion') == 'failure':
            return CHANNEL_MAP['error']
        return CHANNEL_MAP['deploy']
    return CHANNEL_MAP['general']

# --- demo ---
push_payload = {'ref': 'refs/heads/main'}
pr_payload = {}
wf_payload = {'workflow_run': {'conclusion': 'failure'}}

print('push to main       ->', route_github_event('push', push_payload))
print('pull_request        ->', route_github_event('pull_request', pr_payload))
print('failed workflow_run ->', route_github_event('workflow_run', wf_payload))

تنسيق إشعارات النشر

def build_deploy_blocks(repo, branch, commit_sha, deployer, status, env):
    status_emoji = ':white_check_mark:' if status == 'success' else ':x:'
    status_text = 'Success' if status == 'success' else 'Failed'

    blocks = [
        {
            'type': 'header',
            'text': {
                'type': 'plain_text',
                'text': f'{status_emoji} Deploy {status_text}: {repo}'
            }
        },
        {
            'type': 'section',
            'fields': [
                {'type': 'mrkdwn', 'text': f'*Repo:*\n{repo}'},
                {'type': 'mrkdwn', 'text': f'*Environment:*\n{env}'},
                {'type': 'mrkdwn', 'text': f'*Branch:*\n{branch}'},
                {'type': 'mrkdwn', 'text': f'*Deployed by:*\n{deployer}'},
                {'type': 'mrkdwn', 'text': f'*Commit:*\n`{commit_sha[:8]}`'}
            ]
        }
    ]
    return blocks

# --- demo ---
blocks = build_deploy_blocks('coddy-agents', 'main', 'a1b2c3d4e5f6', 'alice', 'success', 'production')
for b in blocks:
    print(b)

إرسال ملخصات مباشرة إلى الأفراد

def send_dm(client, user_id, text, blocks=None):
    # Open DM channel with the user
    dm_result = client.conversations_open(users=[user_id])
    dm_channel = dm_result['channel']['id']

    # Post message to the DM channel
    msg = {'channel': dm_channel, 'text': text}
    if blocks:
        msg['blocks'] = blocks

    return client.chat_postMessage(**msg)

# Example: DM a developer when their PR build fails
def notify_pr_author_of_failure(client, pr_author_slack_id, pr_title, build_url):
    blocks = [
        {
            'type': 'section',
            'text': {
                'type': 'mrkdwn',
                'text': f'Your PR build failed: *{pr_title}*\n<{build_url}|View build logs>'
            }
        }
    ]
    send_dm(client, pr_author_slack_id, f'Build failed: {pr_title}', blocks)

# --- demo: minimal stand-in for the Slack client ---
class _FakeClient:
    def conversations_open(self, users):
        return {'channel': {'id': f'D_{users[0]}'}}
    def chat_postMessage(self, **kwargs):
        print(f"[slack DM] to {kwargs['channel']}: {kwargs['text']}")
        return {'ts': '1700000000.000200'}

notify_pr_author_of_failure(_FakeClient(), 'U_ALICE', 'Add retry logic to fetcher', 'https://ci.example.com/build/42')

تسليم الرسائل وفق جدول زمني

from apscheduler.schedulers.background import BackgroundScheduler
from slack_sdk import WebClient
import os

client = WebClient(token=os.environ['SLACK_BOT_TOKEN'])

def send_daily_summary():
    stats = fetch_daily_stats()  # query your data source
    blocks = [
        {
            'type': 'header',
            'text': {'type': 'plain_text', 'text': 'Daily Team Summary'}
        },
        {
            'type': 'section',
            'text': {'type': 'mrkdwn',
                     'text': f'*PRs merged:* {stats["prs"]}\n'
                             f'*Deploys:* {stats["deploys"]}\n'
                             f'*Incidents:* {stats["incidents"]}'}
        }
    ]
    client.chat_postMessage(
        channel=os.environ['DAILY_CHANNEL'],
        text='Daily Team Summary',
        blocks=blocks
    )

scheduler = BackgroundScheduler()
scheduler.add_job(send_daily_summary, 'cron', hour=9, minute=0)
scheduler.start()
print('Scheduler started: daily summary at 9:00 AM')

تجميع التنبيهات — تجنب إرهاق الإشعارات

import threading
import time
from collections import defaultdict

class AlertAggregator:
    def __init__(self, client, channel, flush_interval=300):
        self.client = client
        self.channel = channel
        self.flush_interval = flush_interval
        self.buffer = defaultdict(int)  # error_type -> count
        self.lock = threading.Lock()
        self._start_flusher()

    def add_alert(self, error_type):
        with self.lock:
            self.buffer[error_type] += 1

    def _flush(self):
        with self.lock:
            if not self.buffer:
                return
            lines = [f'• {err}: {count}x' for err, count in self.buffer.items()]
            self.buffer.clear()

        self.client.chat_postMessage(
            channel=self.channel,
            text=f'Alert summary ({len(lines)} error types):\n' + '\n'.join(lines)
        )

    def _start_flusher(self):
        def loop():
            while True:
                time.sleep(self.flush_interval)
                self._flush()
        threading.Thread(target=loop, daemon=True).start()

# --- demo (flush immediately instead of waiting flush_interval seconds) ---
class _FakeClient:
    def chat_postMessage(self, **kwargs):
        print(f"[slack] postMessage to {kwargs['channel']}: {kwargs['text']}")

agg = AlertAggregator(_FakeClient(), '#alerts', flush_interval=9999)
agg.add_alert('TimeoutError')
agg.add_alert('TimeoutError')
agg.add_alert('ConnectionError')
agg._flush()  # normally the background thread does this every flush_interval seconds

تنسيق كتل التنبيهات الغنية

def build_incident_alert_blocks(service, error_rate, threshold,
                                 runbook_url, pagerduty_url):
    blocks = [
        {
            'type': 'header',
            'text': {'type': 'plain_text', 'text': ':rotating_light: Incident Alert'}
        },
        {
            'type': 'section',
            'text': {
                'type': 'mrkdwn',
                'text': (
                    f'*Service:* `{service}`\n'
                    f'*Error Rate:* {error_rate:.1f}% (threshold: {threshold}%)\n'
                    f'*Status:* Investigating'
                )
            }
        },
        {
            'type': 'actions',
            'elements': [
                {
                    'type': 'button',
                    'text': {'type': 'plain_text', 'text': 'View Runbook'},
                    'url': runbook_url,
                    'action_id': 'view_runbook'
                },
                {
                    'type': 'button',
                    'text': {'type': 'plain_text', 'text': 'PagerDuty'},
                    'url': pagerduty_url,
                    'style': 'danger',
                    'action_id': 'view_pagerduty'
                }
            ]
        }
    ]
    return blocks

# --- demo ---
blocks = build_incident_alert_blocks('checkout-api', 12.4, 5.0,
                                      'https://runbooks.example.com/checkout-api',
                                      'https://pagerduty.example.com/incidents/1')
for b in blocks:
    print(b)

الإشارة إلى فريق المناوبة

import requests
import os

def get_oncall_slack_user():
    # Query PagerDuty for current on-call
    headers = {'Authorization': f'Token token={os.environ["PAGERDUTY_TOKEN"]}'}
    r = requests.get(
        'https://api.pagerduty.com/oncalls?include[]=users&limit=1',
        headers=headers
    )
    oncalls = r.json().get('oncalls', [])
    if not oncalls:
        return None
    email = oncalls[0]['user']['email']
    return email

def send_oncall_alert(client, channel, alert_text):
    oncall_email = get_oncall_slack_user()

    if oncall_email:
        # Look up Slack user by email
        user_result = client.users_lookupByEmail(email=oncall_email)
        user_id = user_result['user']['id']
        mention = f'<@{user_id}>'
    else:
        mention = '<!channel>'

    client.chat_postMessage(
        channel=channel,
        text=f'{mention} - CRITICAL ALERT: {alert_text}'
    )

تجميع التنبيهات ذات الصلة في سلاسل محادثات

class IncidentThread:
    def __init__(self, client, channel):
        self.client = client
        self.channel = channel
        self.active_incidents = {}  # service_name -> thread_ts

    def open_incident(self, service, initial_text, blocks=None):
        msg = self.client.chat_postMessage(
            channel=self.channel,
            text=initial_text,
            blocks=blocks
        )
        self.active_incidents[service] = msg['ts']
        return msg['ts']

    def update_incident(self, service, update_text):
        thread_ts = self.active_incidents.get(service)
        if thread_ts:
            self.client.chat_postMessage(
                channel=self.channel,
                thread_ts=thread_ts,
                text=update_text
            )
        else:
            self.open_incident(service, f'[New] {update_text}')

    def close_incident(self, service, resolution_text):
        thread_ts = self.active_incidents.pop(service, None)
        if thread_ts:
            self.client.chat_postMessage(
                channel=self.channel,
                thread_ts=thread_ts,
                text=f':white_check_mark: RESOLVED: {resolution_text}'
            )

# --- demo: minimal stand-in for the Slack client ---
class _FakeClient:
    def __init__(self):
        self._counter = 0
    def chat_postMessage(self, **kwargs):
        self._counter += 1
        ts = f'ts_{self._counter}'
        print(f"[slack] {kwargs.get('text')} (thread_ts={kwargs.get('thread_ts')})")
        return {'ts': ts}

thread = IncidentThread(_FakeClient(), '#incidents')
thread.open_incident('checkout-api', 'Checkout API error rate spiking')
thread.update_incident('checkout-api', 'Rolled back the last deploy')
thread.close_incident('checkout-api', 'Error rate back to normal')

اختبار روبوت الإشعارات

import requests
import json

def test_webhook(webhook_url, event_type, payload):
    response = requests.post(
        webhook_url,
        json=payload,
        headers={'X-GitHub-Event': event_type, 'Content-Type': 'application/json'}
    )
    print(f'Webhook test {event_type}: {response.status_code}')
    return response

# Test a deploy notification
test_webhook(
    webhook_url='http://localhost:3000/webhook/github',
    event_type='push',
    payload={
        'ref': 'refs/heads/main',
        'pusher': {'name': 'alice'},
        'repository': {'full_name': 'myorg/myapp'},
        'head_commit': {'id': 'abc12345', 'message': 'Fix: auth bug'}
    }
)

print('Check #bot-testing channel for the notification')

تحقق سريع: الرسائل المؤقتة مقابل رسائل القنوات

خلاصة بوت إشعارات الفريق