Apprise + シフト管理ツールで OnCall 自作スタックを組む — PyShift・OR-Tools・GoAlert の役割と選び方

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from PyShift.workschedule.work_schedule import WorkSchedule
from PyShift.workschedule.shift import Shift
from datetime import time, timedelta, date

# 8 時間 3 交代制
schedule = WorkSchedule("3 Shift Rotation", "Day-Swing-Night")
day_shift   = schedule.create_shift("Day",   "Day shift",   time(7, 0),  timedelta(hours=8))
swing_shift = schedule.create_shift("Swing", "Swing shift", time(15, 0), timedelta(hours=8))
night_shift = schedule.create_shift("Night", "Night shift", time(23, 0), timedelta(hours=8))

# ローテーション: 5 日勤 → 2 休 → 5 夕勤 → 2 休 → 5 夜勤 → 2 休
rotation = schedule.create_rotation("28-day cycle", "")
rotation.add_segment(day_shift,   5, 2)
rotation.add_segment(swing_shift, 5, 2)
rotation.add_segment(night_shift, 5, 2)

# A チームは 2026-01-01 開始でこのローテに従う
team_a = schedule.create_team("A team", "", rotation, date(2026, 1, 1))

# 「今日、A チームは何のシフト？」を判定
shift_today = team_a.get_shift_instance_for_day(date.today())
print(shift_today)

1
2
3
4
5
6
7

import apprise
# 当番者を PyShift で判定
on_call_team = team_a if team_a.is_working(now) else team_b
# Apprise で通知
apobj = apprise.Apprise()
apobj.add(f"mailto://{on_call_team.email}")
apobj.notify(title="Alert", body="...")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# OR-Tools の例（簡略化）
from ortools.sat.python import cp_model

model = cp_model.CpModel()
# 変数: shifts[(emp, day, shift_type)] = 0 or 1
# 制約: 1 日 1 シフトまで、夜勤明けは休、最低人数確保 ...
# 目的関数: 不公平最小化
solver = cp_model.CpSolver()
solver.Solve(model)
# 結果を schedule.yaml に出力

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

[「Engineering OnCall」というカレンダー（1 つだけ）]
   ├─ 2026-05-08 09:00 〜 2026-05-15 09:00  「[Primary] 田中」
   ├─ 2026-05-15 09:00 〜 2026-05-22 09:00  「[Primary] 佐藤」
   ├─ 2026-05-22 09:00 〜 2026-05-29 09:00  「[Primary] 鈴木」
   ├─ 2026-05-08 09:00 〜 2026-05-15 09:00  「[Secondary] 山田」 ← 別イベント
   └─ ...

   ↓ チーム全員に「閲覧権限」で共有
   ↓ 編集権限は管理者のみ

[Web サービス（Apprise + 自作）]
   API で「今この時刻に該当するイベント」を取得
   → イベントから担当者メールを抽出
   → Apprise で通知

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

from googleapiclient.discovery import build
from google.oauth2 import service_account
from datetime import datetime, timezone

# サービスアカウントで認証
creds = service_account.Credentials.from_service_account_file(
    "sa-key.json",
    scopes=["https://www.googleapis.com/auth/calendar.readonly"],
)
calendar = build("calendar", "v3", credentials=creds)

def get_current_oncall(calendar_id: str) -> str | None:
    """カレンダーから「今この時刻の当番者」のメールを返す"""
    now = datetime.now(timezone.utc)
    events = calendar.events().list(
        calendarId=calendar_id,
        timeMin=now.isoformat(),
        timeMax=now.isoformat(),  # 開始 ≤ now < 終了 が拾える
        singleEvents=True,
    ).execute()

    for event in events.get("items", []):
        summary = event.get("summary", "")
        if "@" in summary:
            return summary.split()[-1]  # "[Primary] tanaka@example.com" → "tanaka@example.com"
    return None

primary   = get_current_oncall("primary-oncall@group.calendar.google.com")
secondary = get_current_oncall("secondary-oncall@group.calendar.google.com")

# Apprise で通知
import apprise
apobj = apprise.Apprise()
apobj.add(f"mailto://{primary}")
apobj.notify(title="Alert", body="...")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

from msal import ConfidentialClientApplication
import requests
from datetime import datetime, timezone, timedelta
import apprise

TENANT_ID     = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
CLIENT_ID     = "yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy"
CLIENT_SECRET = "zzzzzzzzzzzzzzzzzzzzzzzz"
GROUP_ID      = "engineering-oncall@example.onmicrosoft.com"

# クライアント資格情報フローでトークン取得
app = ConfidentialClientApplication(
    CLIENT_ID,
    authority=f"https://login.microsoftonline.com/{TENANT_ID}",
    client_credential=CLIENT_SECRET,
)
result = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
token = result["access_token"]

def get_current_oncall(group_email: str) -> str | None:
    """グループカレンダーから「今の当番」のメールを返す"""
    now = datetime.now(timezone.utc)
    end = now + timedelta(seconds=1)
    url = (
        f"https://graph.microsoft.com/v1.0/groups/{group_email}/calendarView"
        f"?startDateTime={now.isoformat()}&endDateTime={end.isoformat()}"
    )
    r = requests.get(url, headers={"Authorization": f"Bearer {token}"})
    r.raise_for_status()
    for event in r.json().get("value", []):
        subject = event.get("subject", "")
        if "@" in subject:
            return subject.split()[-1]
        for a in event.get("attendees", []):
            return a["emailAddress"]["address"]
    return None

# Apprise で通知（メール + Teams 同時送信）
oncall = get_current_oncall(GROUP_ID)
apobj = apprise.Apprise()
apobj.add(f"mailto://{oncall}")
apobj.add("msteams://TOKEN_A/TOKEN_B/TOKEN_C/")  # Teams Incoming Webhook
apobj.notify(title="Alert", body=f"On-call: {oncall}")

1
2
3
4
5
6
7
8

url = f"https://graph.microsoft.com/v1.0/teams/{TEAM_ID}/schedule/shifts"
r = requests.get(url, headers={"Authorization": f"Bearer {token}"})
shifts = r.json()["value"]
current = next(
    s for s in shifts
    if s["sharedShift"]["startDateTime"] <= now.isoformat() <= s["sharedShift"]["endDateTime"]
)
user_id = current["userId"]

1

apobj.add("msteams://TokenA/TokenB/TokenC/")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

[Grafana Alerting]
      ↓ webhook
[自作 Web サービス（FastAPI）]
      ↓
   ┌─ 当番判定: Google Calendar API or PyShift で「今の担当者は誰か」
   ├─ ack URL 発行
   └─ Apprise で通知（メール + Slack + Pushover）
        ↓
   N 分タイムアウトでエスカレーション
        ↓
[次の当番者]（同様にカレンダーから取得）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# ESCALATION_POLICY を時間帯別に動的取得
def get_escalation_policy_now():
    now = datetime.now()
    # Google Calendar から取得 or PyShift で計算
    primary = get_oncall_from_calendar(now)
    secondary = get_oncall_from_calendar(now + timedelta(hours=1))
    manager = "manager@example.com"
    return [
        f"mailto://{primary}",
        f"mailto://{secondary}",
        f"mailto://{manager}",
    ]