Studio Booking

用于通过 Telegram 机器人自动完成录音棚预订的技能。

适用场景

• 用户希望预订录音棚（日期、时间、时长）
• 管理员管理预订（创建、取消、改期）
• 客户咨询空闲时段、价格、服务内容
• 需生成 TelegaPay 支付链接
• 需交叉销售附加服务（录音师、母带处理、现场乐手等）
• 客户查询其当前预订状态或历史到访记录

操作指南

1. 潜在线索资质判定

开始预订流程前，先识别客户类型：

新客户：

• 确认项目类型（单曲录制、混音、母带处理、排练等）
• 了解预算范围及偏好的时间段
• 如尚未提供联系方式，则请求用于确认的联系信息
• 建议在正式预订前安排一次录音棚参观

老客户：

• 以姓名问候，并提及过往合作项目
• 确认是否需要沿用相同配置（麦克风、设备等）
• 若累计预订超 5 次，可提供忠诚度折扣

2. 时段可用性检查

使用以下 SQL 模板检查指定时段是否可用：

SELECT s.id, s.start_time, s.end_time, s.price
FROM slots s
WHERE s.date = :date
  AND s.is_available = 1
  AND s.id NOT IN (
    SELECT b.slot_id FROM bookings b
    WHERE b.status IN ('pending', 'confirmed', 'paid')
      AND b.cancelled_at IS NULL
  )
  AND (
    CAST(strftime('%s', s.start_time) AS INTEGER) >= CAST(strftime('%s', :time_from) AS INTEGER)
  )
ORDER BY s.start_time;

参数说明：

• :date — 日期，格式为 YYYY-MM-DD
• :time_from — 允许的最早开始时间（例如 10:00）

若无可用时段，则推荐邻近日期（±3 天）。

3. 创建预订（SAGA 模式）

共包含 5 个步骤的 SAGA 流程：

STEPS = [
    ("validate_slot", lambda: check_slot_available(db, slot_id)),
    ("create_booking", lambda: create_booking_record(db, user_id, slot_id, service_id)),
    ("hold_payment", lambda: hold_invoice(payment, booking_id, amount)),
    ("send_confirmation", lambda: send_booking_notification(bot, chat_id, booking_id)),
    ("notify_admin", lambda: notify_admin_new_booking(bot, admin_chat_id, booking_id)),
]

COMPENSATIONS = {
    "validate_slot": None,
    "create_booking": lambda: delete_booking_record(db, booking_id),
    "hold_payment": lambda: cancel_invoice(payment, invoice_id),
    "send_confirmation": lambda: True,
    "notify_admin": lambda: True,
}

每个步骤必须设置 30 秒超时。任一步骤失败时，按逆序执行对应补偿操作。

4. TelegaPay — 支付链接生成

import hashlib
import hmac
import json
from urllib.parse import urlencode


def generate_telegapay_link(
    amount: int,
    order_id: str,
    description: str,
    secret_key: str,
    shop_id: str,
    callback_url: str,
    user_phone: str = "",
    expires_in: int = 3600,
) -> str:
    payload = {
        "amount": str(amount),
        "order_id": order_id,
        "description": description,
        "shop_id": shop_id,
        "callback_url": callback_url,
        "expires_in": str(expires_in),
    }
    if user_phone:
        payload["phone"] = user_phone

    sign_str = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    signature = hmac.new(
        secret_key.encode(), sign_str.encode(), hashlib.sha256
    ).hexdigest()
    payload["signature"] = signature
    return f"https://telegapay.com/pay?{urlencode(payload)}"

5. 交叉销售矩阵

根据客户选择的核心服务，推荐相应附加服务：

百分比表示相对于基础价格的加价幅度。

6. 预订管理

取消预订：

async def cancel_booking(db, booking_id: int, reason: str = "", refund: bool = False):
    booking = await db.fetch_one(
        "SELECT * FROM bookings WHERE id = ?", (booking_id,)
    )
    if not booking:
        raise ValueError("Booking not found")

    async with db.transaction():
        await db.execute(
            "UPDATE bookings SET status = 'cancelled', cancelled_at = datetime('now'), "
            "cancel_reason = ?, refunded = ? WHERE id = ?",
            (reason, int(refund), booking_id),
        )
        if refund and booking.get("paid_at"):
            await refund_telegapay(booking["invoice_id"], booking["amount"])

        await db.execute(
            "UPDATE slots SET is_available = 1 WHERE id = ?",
            (booking["slot_id"],),
        )

月度占用日历：

async def get_occupancy_calendar(db, year: int, month: int) -> dict:
    """返回当月每日预订数量字典 {日期: 数量}。"""
    rows = await db.fetch_all(
        """SELECT CAST(strftime('%d', s.date) AS INTEGER) as day, COUNT(*) as count
           FROM bookings b
           JOIN slots s ON b.slot_id = s.id
           WHERE strftime('%Y-%m', s.date) = ?
             AND b.status IN ('confirmed', 'paid')
           GROUP BY day""",
        (f"{year:04d}-{month:02d}",),
    )
    return {row["day"]: row["count"] for row in rows}

7. 客户行为追踪

到访历史记录：

SELECT b.id, s.date, s.start_time, s.end_time,
       sv.name as service, b.amount,
       b.status, b.created_at
FROM bookings b
JOIN slots s ON b.slot_id = s.id
JOIN services sv ON b.service_id = sv.id
WHERE b.user_id = :user_id
  AND b.status IN ('paid', 'confirmed')
ORDER BY s.date DESC, s.start_time DESC
LIMIT 20;

客户统计信息：

async def client_stats(db, user_id: int) -> dict:
    row = await db.fetch_one(
        """SELECT COUNT(*) as total_bookings,
                  SUM(amount) as total_spent,
                  COUNT(DISTINCT strftime('%Y-%m', s.date)) as months_active,
                  MAX(s.date) as last_visit
           FROM bookings b
           JOIN slots s ON b.slot_id = s.id
           WHERE b.user_id = ? AND b.status = 'paid'""",
        (user_id,),
    )
    return dict(row) if row else {}

8. 通知策略

• 提前 2 小时通知客户：预订提醒
• 预订成功后 1 小时通知客户：邀请提交评价
• 通知管理员：新预订、取消、付款超时
• 预订后第 1 天通知客户：复购邀约（嵌入交叉销售推荐）

9. 错误处理

10. 定价与套餐规则

PRICING_RULES = {
    "base_hourly": 2500_00,
    "min_booking": 2,
    "discounts": {
        "3_hours": 0.05,
        "5_hours": 0.10,
        "night_rate": 0.80,
        "weekend": 1.15,
        "loyalty_5plus": 0.10,
    },
    "extras": {
        "engineer": 800_00,
        "mixing": 3500_00,
        "mastering": 2000_00,
        "backup_copy": 500_00,
    },
}

11. 与 Telegram 机器人的集成

async def booking_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()

    action, *params = query.data.split(":")
    booking_flow = BookingCoordinator(db, payment, bot)

    match action:
        case "calendar":
            await booking_flow.show_calendar(query)
        case "select_date":
            await booking_flow.show_time_slots(query, params[0])
        case "select_time":
            await booking_flow.show_services(query, params[0])
        case "select_service":
            await booking_flow.show_extras(query, params[0], params[1])
        case "confirm":
            await booking_flow.run_saga(query, params)
        case "pay":
            await booking_flow.generate_payment(query, params[0])

参考文档

• TelegaPay 文档：https://telegapay.com/docs
• python-telegram-bot：https://docs.python-telegram-bot.org/
• 时间格式：HH:MM（24 小时制）
• 日期格式：YYYY-MM-DD（ISO 8601）
• 价格单位：整型，最小单位为“分”（1 元 = 100 分）
• 时区：Europe/Moscow（UTC+3，不启用夏令时）