afterhours-shift-manager/scripts/seed_schedule.py

#!/usr/bin/env python3
"""Seed the afterhours-shifts DynamoDB table with initial roster and schedule.

Usage:
    python scripts/seed_schedule.py                    # uses default table name
    python scripts/seed_schedule.py --table my-table   # custom table name
"""

import argparse

import boto3


TABLE_NAME = "afterhours-shifts"

ROSTER = [
    {"extension": "100", "name": "Adam Moussa"},
    {"extension": "111", "name": "Alyssa Ficarra"},
    {"extension": "113", "name": "Sarah May"},
    {"extension": "114", "name": "Ashley Fedner"},
    {"extension": "115", "name": "Derrick Smith"},
    {"extension": "116", "name": "Cindy Vallecillo"},
]

WEEKLY_SCHEDULE = {
    "Monday": {"extension": "114", "name": "Ashley Fedner"},
    "Tuesday": {"extension": "115", "name": "Derrick Smith"},
    "Wednesday": {"extension": "114", "name": "Ashley Fedner"},
    "Thursday": {"extension": "115", "name": "Derrick Smith"},
    "Friday": {"extension": "116", "name": "Cindy Vallecillo"},
    "Saturday": {"extension": "100", "name": "Fallback"},
    "Sunday": {"extension": "100", "name": "Fallback"},
}

CONFIG = {
    "fallback_extension": "100",
    "holiday_multiplier": "1.5",  # default holiday pay multiplier (Decimal string)
    "holiday_queue": "802",  # 3CX queue repointed to holiday assignees during a holiday window
    "ivr_number": "800",  # 3CX IVR whose key-0 + no-input/timeout routes are repointed on a holiday
}


def seed(table_name: str):
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table(table_name)

    print(f"Seeding table: {table_name}")

    # Roster
    for emp in ROSTER:
        table.put_item(
            Item={
                "PK": "ROSTER",
                "SK": emp["extension"],
                "name": emp["name"],
                "extension": emp["extension"],
                "slack_user_id": "",
            }
        )
        print(f"  Roster: Ext {emp['extension']} — {emp['name']}")

    # Weekly schedule
    for day, info in WEEKLY_SCHEDULE.items():
        table.put_item(
            Item={
                "PK": "WEEKLY",
                "SK": day,
                "extension": info["extension"],
                "name": info["name"],
            }
        )
        print(f"  Weekly: {day} — {info['name']} (Ext {info['extension']})")

    # Config — set each key only if absent, so re-running never clobbers live
    # operational values (admin_users, shift_rate, captured_ivr_routes).
    set_clause = ", ".join(f"#{k} = if_not_exists(#{k}, :{k})" for k in CONFIG)
    table.update_item(
        Key={"PK": "CONFIG", "SK": "CONFIG"},
        UpdateExpression="SET " + set_clause,
        ExpressionAttributeNames={f"#{k}": k for k in CONFIG},
        ExpressionAttributeValues={f":{k}": v for k, v in CONFIG.items()},
    )
    print(
        f"  Config (set-if-absent): fallback={CONFIG['fallback_extension']}, "
        f"holiday_queue={CONFIG['holiday_queue']}, ivr_number={CONFIG['ivr_number']}, "
        f"holiday_multiplier={CONFIG['holiday_multiplier']}"
    )

    print(
        "\nDone! Employees should run `/oncall register <extension>` in Slack to link their accounts."
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--table", default=TABLE_NAME)
    args = parser.parse_args()
    seed(args.table)