Python(FastAPI)으로 통화 완료 웹훅 받아 SQLite에 저장하기

{
  "eventType": "CALL_COMPLETED",
  "timestamp": "2026-03-12T02:45:30.000Z",
  "data": {
    "callHistoryId": "8f3f8fd9-7ab8-4a26-b523-4f8eb7d38b85",
    "groupId": 1,
    "channelId": 100,
    "fromNumber": "01012345678",
    "toNumber": "07012345678",
    "isInbound": true,
    "status": "ANSWERED",
    "tag": "AI_TO_DIRECT",
    "duration": 96,
    "summary": "예약 변경 요청을 접수하고 변경 가능한 시간대를 안내함",
    "sttAI": [
      { "role": "agent", "text": "안녕하세요. KAT Cloud입니다.", "timestamp": 1.2 }
    ],
    "sttDirect": [
      { "role": "customer", "text": "직원 연결 부탁드립니다.", "timestamp": 30.1 }
    ],
    "recordingUrl": "https://storage.example.com/recording.wav",
    "startedAt": "2026-03-12T02:40:00.000Z",
    "completedAt": "2026-03-12T02:41:36.000Z"
  }
}

# db.py
import json
import sqlite3
from contextlib import closing

DB_PATH = "calls.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS calls (
    call_history_id TEXT PRIMARY KEY,
    group_id        INTEGER,
    channel_id      INTEGER,
    from_number     TEXT,
    to_number       TEXT,
    is_inbound      INTEGER,
    status          TEXT,
    tag             TEXT,
    duration        INTEGER,
    summary         TEXT,
    stt_ai          TEXT,
    stt_direct      TEXT,
    recording_url   TEXT,
    started_at      TEXT,
    completed_at    TEXT,
    saved_at        TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_calls_started_at ON calls (started_at);
"""

UPSERT = """
INSERT INTO calls (
    call_history_id, group_id, channel_id, from_number, to_number, is_inbound,
    status, tag, duration, summary, stt_ai, stt_direct, recording_url,
    started_at, completed_at
) VALUES (
    :call_history_id, :group_id, :channel_id, :from_number, :to_number, :is_inbound,
    :status, :tag, :duration, :summary, :stt_ai, :stt_direct, :recording_url,
    :started_at, :completed_at
)
ON CONFLICT(call_history_id) DO UPDATE SET
    status = excluded.status,
    tag = excluded.tag,
    duration = excluded.duration,
    summary = excluded.summary,
    stt_ai = excluded.stt_ai,
    stt_direct = excluded.stt_direct,
    recording_url = excluded.recording_url,
    completed_at = excluded.completed_at
"""


def connect() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH, timeout=10)
    conn.execute("PRAGMA journal_mode=WAL")  # 읽기와 쓰기가 덜 막히게
    return conn


def init_db() -> None:
    with closing(connect()) as conn:
        conn.executescript(SCHEMA)


def save_call(data: dict) -> None:
    row = {
        "call_history_id": data["callHistoryId"],
        "group_id": data.get("groupId"),
        "channel_id": data.get("channelId"),
        "from_number": data.get("fromNumber"),
        "to_number": data.get("toNumber"),
        "is_inbound": 1 if data.get("isInbound") else 0,
        "status": data.get("status"),
        "tag": data.get("tag"),
        "duration": data.get("duration"),
        "summary": data.get("summary"),
        "stt_ai": json.dumps(data.get("sttAI") or [], ensure_ascii=False),
        "stt_direct": json.dumps(data.get("sttDirect") or [], ensure_ascii=False),
        "recording_url": data.get("recordingUrl"),
        "started_at": data.get("startedAt"),
        "completed_at": data.get("completedAt"),
    }
    with closing(connect()) as conn, conn:  # 두 번째 conn이 커밋·롤백을 맡는다
        conn.execute(UPSERT, row)

# main.py
import logging
from contextlib import asynccontextmanager

from fastapi import BackgroundTasks, FastAPI, Request

from db import init_db, save_call

logger = logging.getLogger("katcloud-webhook")


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield


app = FastAPI(lifespan=lifespan)


def store_call_completed(data: dict) -> None:
    try:
        save_call(data)
        logger.info("saved call %s", data.get("callHistoryId"))
    except Exception:
        logger.exception("failed to save call %s", data.get("callHistoryId"))


@app.post("/webhook/call-completed")
async def call_completed(request: Request, background: BackgroundTasks):
    try:
        body = await request.json()
    except ValueError:
        return {"received": False}  # JSON이 아니어도 200으로 끝낸다

    if not isinstance(body, dict) or body.get("eventType") != "CALL_COMPLETED":
        return {"received": True, "skipped": True}

    data = body.get("data") or {}
    if not data.get("callHistoryId"):
        return {"received": True, "skipped": True}

    background.add_task(store_call_completed, data)
    return {"received": True}

pip install fastapi uvicorn
uvicorn main:app --host 0.0.0.0 --port 8000

curl -s -X POST http://localhost:8000/webhook/call-completed \
  -H "Content-Type: application/json" \
  -d '{"eventType":"CALL_COMPLETED","timestamp":"2026-03-12T02:45:30.000Z","data":{"callHistoryId":"8f3f8fd9-7ab8-4a26-b523-4f8eb7d38b85","groupId":1,"channelId":100,"fromNumber":"01012345678","toNumber":"07012345678","isInbound":true,"status":"ANSWERED","tag":"AI_TO_DIRECT","duration":96,"summary":"예약 변경 요청을 접수함","sttAI":[],"sttDirect":[],"startedAt":"2026-03-12T02:40:00.000Z","completedAt":"2026-03-12T02:41:36.000Z"}}'

sqlite3 calls.db "SELECT call_history_id, tag, duration FROM calls;"

# register_webhook.py
import os

import requests

BASE_URL = "https://studio-gateway.zerocall.kr"
HEADERS = {"x-api-key": os.environ["KAT_API_KEY"], "Content-Type": "application/json"}

resp = requests.post(
    f"{BASE_URL}/webhook",
    headers=HEADERS,
    json={
        "eventType": "CALL_COMPLETED",
        "url": "https://hooks.example.com/webhook/call-completed",
    },
    timeout=10,
)
resp.raise_for_status()
webhook = resp.json()["result"]
print("webhook id:", webhook["id"])

test = requests.post(f"{BASE_URL}/webhook/{webhook['id']}/test", headers=HEADERS, timeout=30)
print(test.json()["result"])  # success, statusCode, responseTime, error

# report.py
import json
import sqlite3

conn = sqlite3.connect("calls.db")
conn.row_factory = sqlite3.Row

# 태그별 통화 수와 평균 통화 시간(초)
for row in conn.execute(
    "SELECT tag, COUNT(*) AS calls, ROUND(AVG(duration)) AS avg_sec FROM calls GROUP BY tag"
):
    print(dict(row))

# AI에서 직원으로 넘어간 최근 통화 5건
for row in conn.execute(
    "SELECT call_history_id, summary, stt_ai, stt_direct FROM calls "
    "WHERE tag = 'AI_TO_DIRECT' ORDER BY started_at DESC LIMIT 5"
):
    print(row["call_history_id"], row["summary"])
    for turn in json.loads(row["stt_ai"]) + json.loads(row["stt_direct"]):
        print(f"  [{turn['role']}] {turn['text']}")