caret_backup.py

gistfile1.txt

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# ///
"""Caret 노트·녹취·로컬 녹음 백업 유틸리티.

실행 방법:
    export CARET_API_KEY="Caret notes:read API 키"
    export CARET_BACKUP_DIR="/Users/me/Documents/caret_backup"
    uv run caret_backup.py

또는:
    uv run caret_backup.py \
        --api-key "Caret notes:read API 키" \
        --backup-dir "/백업/경로"

백업 대상:
    - Caret Public API에서 접근 가능한 모든 노트
    - 노트의 요약, 메모, 참석자 및 기타 메타데이터
    - 최종 병합 transcript 원본 JSON
    - transcript CSV, Markdown, TXT
    - ~/Documents/Caret에 저장된 M4A·MP3 등 로컬 녹음
    - 로컬 녹음과 서버 노트의 note ID 매칭 목록

주의사항:
    - Caret 데이터는 읽기만 하며 수정하거나 삭제하지 않습니다.
    - 반복 실행할 수 있으며 변경되지 않은 녹음은 다시 복사하지 않습니다.
    - API 키는 CARET_API_KEY 환경변수 사용을 권장합니다. --api-key로
      전달하면 shell history나 프로세스 목록에 노출될 수 있습니다.
    - Public API는 서버에 보관된 녹음의 다운로드 URL을 제공하지 않습니다.
      따라서 서버에만 존재하는 원격 녹음은 이 도구로 백업되지 않습니다.
    - transcript는 Caret의 최종 병합본입니다. 실시간 STT packet,
      interim 결과 및 refine 이전 원문은 포함되지 않습니다.
    - 서버에서 이미 삭제된 노트와 보존 기간이 지난 녹음은 복구할 수 없습니다.

환경변수:
    CARET_API_KEY: Caret Settings > 워크스페이스 >  개발자에서 생성한 API 키, notes:read 권한 필요
    CARET_BACKUP_DIR: 백업 결과를 저장할 디렉터리
    CARET_AUDIO_DIR: 로컬 녹음 디렉터리(기본값: ~/Documents/Caret)

주요 옵션:
    --local-only: API를 호출하지 않고 로컬 녹음만 백업합니다.
    --notes-only: 로컬 녹음을 복사하지 않고 서버 노트와 녹취만 백업합니다.

출력 구조:
    <backup-dir>/
    ├── export_summary.json
    ├── notes_index.csv
    ├── local_audio_inventory.csv
    ├── notes/<note-id>/
    │   ├── note.json
    │   ├── transcript.csv
    │   ├── transcript.md
    │   └── transcript.txt
    └── audio/
        └── Meeting_<note-id>.m4a 등
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import re
import shutil
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Iterable
from datetime import datetime
from pathlib import Path
from typing import Any


API_BASE_URL = "https://api.caret.so/v1"
AUDIO_FILE_RE = re.compile(
    r"^(?P<prefix>Meeting|Note)_(?P<note_id>[0-9a-fA-F-]+)(?P<extension>\.[^.]+)$"
)
INVALID_FILENAME_CHARS_RE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Back up all accessible Caret notes, transcripts, and local audio."
    )
    parser.add_argument(
        "--backup-dir",
        type=Path,
        default=os.environ.get("CARET_BACKUP_DIR"),
        help="Backup directory (or CARET_BACKUP_DIR).",
    )
    parser.add_argument(
        "--api-key",
        default=os.environ.get("CARET_API_KEY"),
        help="Caret notes:read API key (or CARET_API_KEY). Environment is safer.",
    )
    parser.add_argument(
        "--audio-dir",
        type=Path,
        default=Path(os.environ.get("CARET_AUDIO_DIR", "~/Documents/Caret")),
        help="Local Caret audio directory (default: ~/Documents/Caret).",
    )
    parser.add_argument(
        "--local-only",
        action="store_true",
        help="Back up local recordings without calling the Caret API.",
    )
    parser.add_argument(
        "--notes-only",
        action="store_true",
        help="Export server notes without copying local recordings.",
    )
    args = parser.parse_args()

    if args.backup_dir is None:
        parser.error("--backup-dir or CARET_BACKUP_DIR is required")
    if not args.local_only and not str(args.api_key or "").strip():
        parser.error("--api-key or CARET_API_KEY is required unless --local-only is used")
    if args.local_only and args.notes_only:
        parser.error("--local-only and --notes-only cannot be used together")
    args.api_key = str(args.api_key or "").strip()
    return args


def safe_filename(value: str, fallback: str = "untitled") -> str:
    value = INVALID_FILENAME_CHARS_RE.sub("_", value)
    value = re.sub(r"\s+", " ", value).strip().rstrip(". ")
    return (value or fallback)[:100]


def atomic_write_text(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    temporary = Path(temporary_name)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="") as output:
            output.write(content)
        temporary.replace(path)
    except BaseException:
        temporary.unlink(missing_ok=True)
        raise


def atomic_write_json(path: Path, value: Any) -> None:
    atomic_write_text(path, json.dumps(value, ensure_ascii=False, indent=2) + "\n")


def atomic_write_csv(
    path: Path, rows: Iterable[dict[str, Any]], fields: list[str]
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    temporary = Path(temporary_name)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="") as output:
            writer = csv.DictWriter(output, fieldnames=fields, extrasaction="ignore")
            writer.writeheader()
            writer.writerows(rows)
        temporary.replace(path)
    except BaseException:
        temporary.unlink(missing_ok=True)
        raise


class CaretApi:
    def __init__(self, api_key: str) -> None:
        self.api_key = api_key

    def get(self, path: str, query: dict[str, Any] | None = None) -> Any:
        url = f"{API_BASE_URL}{path}"
        if query:
            url += "?" + urllib.parse.urlencode(query)

        for attempt in range(6):
            request = urllib.request.Request(
                url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Accept": "application/json",
                    "User-Agent": "caret-backup/1.0",
                },
            )
            try:
                with urllib.request.urlopen(request, timeout=60) as response:
                    return json.load(response)
            except urllib.error.HTTPError as error:
                body = error.read().decode("utf-8", errors="replace")
                if error.code == 429 and attempt < 5:
                    retry_after = error.headers.get("Retry-After", "")
                    delay = int(retry_after) if retry_after.isdigit() else 2**attempt
                    time.sleep(min(max(delay, 1), 60))
                    continue
                raise RuntimeError(
                    f"Caret API returned HTTP {error.code} for {path}: {body[:500]}"
                ) from error
            except urllib.error.URLError as error:
                if attempt < 5:
                    time.sleep(min(2**attempt, 30))
                    continue
                raise RuntimeError(f"Caret API request failed for {path}: {error}") from error
        raise AssertionError("unreachable")

    def list_notes(self) -> list[dict[str, Any]]:
        notes: list[dict[str, Any]] = []
        offset = 0
        while True:
            payload = self.get(
                "/notes",
                {
                    "limit": 100,
                    "offset": offset,
                    "sortBy": "createdAt",
                    "sortOrder": "asc",
                },
            )
            page = payload.get("items")
            if not isinstance(page, list):
                raise RuntimeError("Unexpected Caret API response: items is not an array")
            notes.extend(note for note in page if isinstance(note, dict))
            pagination = payload.get("pagination") or {}
            if pagination.get("isLast", True):
                return notes
            next_offset = pagination.get("nextOffset")
            if not isinstance(next_offset, int) or next_offset <= offset:
                raise RuntimeError("Unexpected Caret API pagination response")
            offset = next_offset

    def get_note(self, note_id: str) -> dict[str, Any]:
        encoded_id = urllib.parse.quote(note_id, safe="")
        payload = self.get(f"/notes/{encoded_id}")
        note = payload.get("note")
        if not isinstance(note, dict):
            raise RuntimeError(f"Unexpected Caret API response for note {note_id}")
        return note


def scan_audio(audio_dir: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    if not audio_dir.is_dir():
        raise RuntimeError(f"Caret audio directory not found: {audio_dir}")
    for path in sorted(audio_dir.iterdir(), key=lambda item: item.name.lower()):
        if not path.is_file():
            continue
        match = AUDIO_FILE_RE.fullmatch(path.name)
        stat = path.stat()
        rows.append(
            {
                "file_name": path.name,
                "prefix": match.group("prefix") if match else "",
                "note_id": match.group("note_id") if match else "",
                "extension": path.suffix.lower(),
                "bytes": stat.st_size,
                "modified_at": datetime.fromtimestamp(stat.st_mtime)
                .astimezone()
                .isoformat(),
                "source_path": str(path),
            }
        )
    return rows


def copy_file_safely(source: Path, destination: Path) -> bool:
    if destination.exists():
        source_stat = source.stat()
        destination_stat = destination.stat()
        if (
            source_stat.st_size == destination_stat.st_size
            and source_stat.st_mtime_ns == destination_stat.st_mtime_ns
        ):
            return False
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_name(f".{destination.name}.copying")
    temporary.unlink(missing_ok=True)
    try:
        shutil.copy2(source, temporary)
        temporary.replace(destination)
    except BaseException:
        temporary.unlink(missing_ok=True)
        raise
    return True


def transcript_rows(note: dict[str, Any]) -> list[dict[str, str]]:
    result: list[dict[str, str]] = []
    transcripts = note.get("transcripts") or []
    if not isinstance(transcripts, list):
        return result
    for segment in transcripts:
        if not isinstance(segment, dict):
            continue
        result.append(
            {
                "speaker": str(segment.get("speaker") or ""),
                "start": str(
                    segment.get("startTimestamp") or segment.get("start") or ""
                ),
                "end": str(segment.get("endTimestamp") or segment.get("end") or ""),
                "text": str(segment.get("text") or ""),
                "translation": str(segment.get("translation") or ""),
            }
        )
    return result


def transcript_text(rows: list[dict[str, str]], markdown: bool) -> str:
    paragraphs: list[str] = []
    for row in rows:
        timestamp = row["start"]
        if row["end"]:
            timestamp = f"{timestamp} - {row['end']}" if timestamp else row["end"]
        prefix = f"[{timestamp}] " if timestamp else ""
        speaker = row["speaker"] or "Unknown"
        line = f"{prefix}**{speaker}**: {row['text']}" if markdown else f"{prefix}{speaker}: {row['text']}"
        if row["translation"]:
            line += f"\n\n> {row['translation']}" if markdown else f"\n  Translation: {row['translation']}"
        paragraphs.append(line)
    return "\n\n".join(paragraphs) + ("\n" if paragraphs else "")


def export_notes(api: CaretApi, backup_dir: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    listed_notes = api.list_notes()
    notes_dir = backup_dir / "notes"
    index: list[dict[str, Any]] = []
    detailed_notes: list[dict[str, Any]] = []

    for number, listed_note in enumerate(listed_notes, start=1):
        note_id = str(listed_note.get("id") or "")
        if not note_id:
            continue
        note = api.get_note(note_id)
        detailed_notes.append(note)
        title = str(note.get("title") or "untitled")
        # Use the immutable note ID so a title change does not create stale duplicates.
        note_dir = notes_dir / safe_filename(note_id)
        rows = transcript_rows(note)
        atomic_write_json(note_dir / "note.json", note)
        atomic_write_csv(
            note_dir / "transcript.csv",
            rows,
            ["speaker", "start", "end", "text", "translation"],
        )
        atomic_write_text(note_dir / "transcript.md", transcript_text(rows, True))
        atomic_write_text(note_dir / "transcript.txt", transcript_text(rows, False))
        index.append(
            {
                "id": note_id,
                "title": title,
                "kind": note.get("kind") or "",
                "status": note.get("status") or "",
                "created_at": note.get("createdAt") or "",
                "updated_at": note.get("updatedAt") or "",
                "duration_sec": note.get("totalDurationSec") or "",
                "transcript_segments": len(rows),
                "directory": str(note_dir.relative_to(backup_dir)),
            }
        )
        print(f"[notes {number}/{len(listed_notes)}] {title}")

    atomic_write_csv(
        backup_dir / "notes_index.csv",
        index,
        [
            "id",
            "title",
            "kind",
            "status",
            "created_at",
            "updated_at",
            "duration_sec",
            "transcript_segments",
            "directory",
        ],
    )
    return detailed_notes, index


def write_audio_inventory(
    audio_rows: list[dict[str, Any]], notes: list[dict[str, Any]], backup_dir: Path
) -> None:
    notes_by_id = {str(note.get("id")): note for note in notes if note.get("id")}
    rows = []
    for audio in audio_rows:
        note = notes_by_id.get(audio["note_id"], {})
        rows.append(
            {
                **audio,
                "matched_server_note": bool(note),
                "server_title": note.get("title") or "",
                "server_status": note.get("status") or "",
                "server_created_at": note.get("createdAt") or "",
            }
        )
    atomic_write_csv(
        backup_dir / "local_audio_inventory.csv",
        rows,
        [
            "file_name",
            "prefix",
            "note_id",
            "extension",
            "bytes",
            "modified_at",
            "source_path",
            "matched_server_note",
            "server_title",
            "server_status",
            "server_created_at",
        ],
    )


def main() -> int:
    args = parse_args()
    backup_dir = args.backup_dir.expanduser().resolve()
    audio_dir = args.audio_dir.expanduser().resolve()
    if backup_dir == audio_dir or audio_dir in backup_dir.parents:
        raise RuntimeError("The backup directory cannot be inside the source audio directory")
    backup_dir.mkdir(parents=True, exist_ok=True)

    audio_rows: list[dict[str, Any]] = []
    copied = 0
    if not args.notes_only:
        audio_rows = scan_audio(audio_dir)
        for number, row in enumerate(audio_rows, start=1):
            source = Path(row["source_path"])
            if copy_file_safely(source, backup_dir / "audio" / source.name):
                copied += 1
            print(f"[audio {number}/{len(audio_rows)}] {source.name}")

    notes: list[dict[str, Any]] = []
    note_index: list[dict[str, Any]] = []
    if not args.local_only:
        notes, note_index = export_notes(CaretApi(args.api_key), backup_dir)

    if not args.notes_only:
        write_audio_inventory(audio_rows, notes, backup_dir)
    summary = {
        "generated_at": datetime.now().astimezone().isoformat(),
        "backup_directory": str(backup_dir),
        "audio_source_directory": str(audio_dir),
        "local_audio_files": len(audio_rows),
        "local_audio_files_copied_or_updated": copied,
        "local_audio_bytes": sum(int(row["bytes"]) for row in audio_rows),
        "unique_local_note_ids": len(
            {row["note_id"] for row in audio_rows if row["note_id"]}
        ),
        "server_notes_exported": len(note_index),
        "transcript_segments": sum(
            int(row["transcript_segments"]) for row in note_index
        ),
    }
    atomic_write_json(backup_dir / "export_summary.json", summary)
    print("\nBackup complete")
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        print("\nBackup interrupted", file=sys.stderr)
        raise SystemExit(130)
    except Exception as error:
        print(f"Backup failed: {error}", file=sys.stderr)
        raise SystemExit(1)