통화 기록 증분 동기화: updatedAfter 커서로 빠짐없이 가져오기

# 1) 최초 조회: 기준 시각부터
GET /call-history?groupId=1&updatedAfter=2026-03-12T00:00:00.000Z&limit=100

# 2) 마지막 항목의 updatedAt이 2026-03-12T02:47:13.000Z였다면 그 값을 그대로 넘긴다
GET /call-history?groupId=1&updatedAfter=2026-03-12T02:47:13.000Z&limit=100

curl -G "https://studio-gateway.zerocall.kr/call-history" \
  -H "x-api-key: $KAT_API_KEY" \
  --data-urlencode "groupId=1" \
  --data-urlencode "updatedAfter=2026-03-12T02:47:13.000Z" \
  --data-urlencode "limit=100"

"""KAT Cloud 통화 기록 증분 동기화 (updatedAfter 커서)

실행: KAT_API_KEY=... python sync_call_history.py --once
"""
import argparse
import json
import os
import random
import sqlite3
import time

import requests

BASE_URL = "https://studio-gateway.zerocall.kr"
PAGE_SIZE = 100                                # limit 최대값
POLL_INTERVAL = 60                             # 새 기록이 없을 때 다음 주기까지 대기(초)
INITIAL_CURSOR = "2026-10-01T00:00:00.000Z"    # 처음 실행할 때의 기준 시각
DB_PATH = "call_history.db"

session = requests.Session()
session.headers.update({"x-api-key": os.environ["KAT_API_KEY"]})


def api_get(path, params=None, max_retries=5):
    for attempt in range(max_retries + 1):
        resp = session.get(BASE_URL + path, params=params, timeout=30)
        if resp.status_code == 429 or resp.status_code >= 500:
            if attempt == max_retries:
                resp.raise_for_status()
            time.sleep(min(60, 2 ** attempt) + random.random())  # 지수 백오프 + 지터
            continue
        if not resp.ok:
            raise RuntimeError(f"{resp.status_code} {resp.text}")
        body = resp.json()
        # 공통 성공 응답 {"code": "0", "message": "Success", "result": ...}
        return body["result"] if isinstance(body, dict) and "result" in body else body


def open_db(path):
    db = sqlite3.connect(path)
    db.executescript("""
    CREATE TABLE IF NOT EXISTS call_history (
        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, is_recorded INTEGER,
        created_at TEXT, updated_at TEXT
    );
    CREATE TABLE IF NOT EXISTS sync_cursor (
        group_id INTEGER PRIMARY KEY,
        updated_after TEXT NOT NULL
    );
    """)
    return db


def load_cursor(db, group_id):
    row = db.execute(
        "SELECT updated_after FROM sync_cursor WHERE group_id = ?", (group_id,)
    ).fetchone()
    return row[0] if row else INITIAL_CURSOR


def save_page(db, group_id, items):
    rows = [(
        it["callHistoryId"], it.get("groupId"), it.get("channelId"),
        it.get("fromNumber"), it.get("toNumber"), int(bool(it.get("isInbound"))),
        it.get("status"), it.get("tag"), it.get("duration"), it.get("summary"),
        json.dumps(it.get("sttAI") or [], ensure_ascii=False),
        json.dumps(it.get("sttDirect") or [], ensure_ascii=False),
        int(bool(it.get("isRecorded"))), it.get("createdAt"), it["updatedAt"],
    ) for it in items]
    with db:  # 기록 저장과 커서 갱신을 하나의 트랜잭션으로 커밋
        db.executemany("""
        INSERT INTO call_history VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
        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,
            is_recorded = excluded.is_recorded, updated_at = excluded.updated_at
        """, rows)
        db.execute("""
        INSERT INTO sync_cursor (group_id, updated_after) VALUES (?, ?)
        ON CONFLICT(group_id) DO UPDATE SET updated_after = excluded.updated_after
        """, (group_id, items[-1]["updatedAt"]))


def sync_group(db, group_id):
    synced = 0
    while True:
        cursor = load_cursor(db, group_id)
        page = api_get("/call-history", {
            "groupId": group_id,
            "updatedAfter": cursor,
            "limit": PAGE_SIZE,
        })
        items = page["data"]
        if not items:
            return synced                      # 갱신된 기록 없음: 커서 유지
        save_page(db, group_id, items)
        synced += len(items)
        if len(items) < PAGE_SIZE:
            return synced                      # 마지막 페이지까지 받음
        if items[-1]["updatedAt"] == cursor:
            print(f"경고: group {group_id} 커서가 진행되지 않아 다음 주기에 재시도합니다.")
            return synced


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--once", action="store_true", help="한 번만 동기화하고 종료")
    args = parser.parse_args()

    db = open_db(DB_PATH)
    while True:
        for group in api_get("/group"):
            n = sync_group(db, group["id"])
            print(time.strftime("%H:%M:%S"), f"group {group['id']}: {n}건 반영")
        if args.once:
            break
        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    main()