동료가 만든 자동화를 '내 것'으로 포크하기 — 음성 기록을 Obsidian에 의미 있게 올리기

#!/usr/bin/env python3
"""
L1 Ingest — Limitless -> Obsidian

하루치 Limitless lifelog '전체'를 가져와 날짜별 .md 한 파일로 저장하고,
별표(starred)를 frontmatter로 구조화한다.
=> PLL(Process Life Logs)이 수정 없이 바로 소비할 수 있는 입력 형식.

설계:
  - 별표만 받지 않고 '하루 전체'를 받는다. (PLL의 전후 맥락 수집 기능을 살리기 위해)
  - 별표 대화 -> "# ⭐ {title}"  H1 섹션
  - 그 외 대화   -> "# {title}"    H1 섹션
  - frontmatter에 starred_list / starred_conversations 를 기록 -> PLL Phase 0가 이걸 읽는다.

사용:
  export LIMITLESS_API_KEY="sk-..."
  python3 l1_ingest.py                       # 오늘(America/Los_Angeles) 분
  python3 l1_ingest.py --date 2026-06-01     # 특정 날짜
  python3 l1_ingest.py --vault "/path/to/Vault" --tz America/Los_Angeles

출력:
  {vault}/Ingest/Limitless/{date}.md
"""

import os
import sys
import json
import argparse
import urllib.parse
import urllib.request
import urllib.error
from datetime import datetime

try:
    from zoneinfo import ZoneInfo
except ImportError:  # py<3.9 fallback
    ZoneInfo = None

API_BASE = "https://api.limitless.ai/v1/lifelogs"


def today_in(tz_name):
    if ZoneInfo is not None:
        return datetime.now(ZoneInfo(tz_name)).strftime("%Y-%m-%d")
    return datetime.now().strftime("%Y-%m-%d")


def fetch_day(api_key, date, tz_name, per_page=100):
    """해당 날짜의 모든 lifelog를 페이지네이션으로 끝까지 수집."""
    lifelogs = []
    cursor = None
    while True:
        params = {
            "date": date,
            "timezone": tz_name,
            "includeMarkdown": "true",
            "direction": "asc",     # 시간순(전후 맥락 보존에 중요)
            "limit": str(per_page),
        }
        if cursor:
            params["cursor"] = cursor

        url = API_BASE + "?" + urllib.parse.urlencode(params)
        req = urllib.request.Request(url, headers={"X-API-Key": api_key})
        try:
            with urllib.request.urlopen(req, timeout=60) as resp:
                payload = json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            sys.exit(f"[L1] API 오류 {e.code}: {e.read().decode('utf-8', 'ignore')}")
        except urllib.error.URLError as e:
            sys.exit(f"[L1] 네트워크 오류: {e.reason}")

        page = payload.get("data", {}).get("lifelogs", []) or []
        lifelogs.extend(page)

        meta = payload.get("meta", {}).get("lifelogs", {}) or {}
        cursor = meta.get("nextCursor")
        if not cursor:
            break

    return lifelogs


def yaml_quote(s):
    """frontmatter 리스트 항목용 안전 따옴표 처리."""
    return '"' + str(s).replace("\\", "\\\\").replace('"', '\\"') + '"'


def build_markdown(lifelogs, date):
    starred_titles = []
    body = []

    for lg in lifelogs:
        title = (lg.get("title") or "(untitled)").strip()
        is_starred = bool(lg.get("isStarred"))
        if is_starred:
            starred_titles.append(title)
            body.append(f"# ⭐ {title}")
        else:
            body.append(f"# {title}")

        start = lg.get("startTime", "")
        end = lg.get("endTime", "")
        if start or end:
            body.append(f"*{start} — {end}*")
        body.append("")
        body.append((lg.get("markdown") or "").rstrip())
        body.append("")

    generated = datetime.now().astimezone().isoformat(timespec="seconds")

    fm = ["---"]
    fm.append(f"date: {date}")
    fm.append("source: limitless")
    fm.append(f"total_conversations: {len(lifelogs)}")
    fm.append(f"starred_conversations: {len(starred_titles)}")
    if starred_titles:
        fm.append("starred_list:")
        for t in starred_titles:
            fm.append(f"  - {yaml_quote(t)}")
    else:
        fm.append("starred_list: []")
    fm.append(f"generated: {generated}")
    fm.append("---")
    fm.append("")

    return "\n".join(fm) + "\n".join(body), len(starred_titles), len(lifelogs)


def main():
    ap = argparse.ArgumentParser(description="Limitless -> Obsidian L1 ingest")
    ap.add_argument("--date", help="YYYY-MM-DD (기본: 오늘)")
    ap.add_argument("--tz", default="America/Los_Angeles", help="IANA timezone")
    ap.add_argument("--vault", default=".", help="Obsidian vault 루트 경로")
    ap.add_argument("--per-page", type=int, default=100)
    args = ap.parse_args()

    api_key = os.environ.get("LIMITLESS_API_KEY")
    if not api_key:
        sys.exit("[L1] LIMITLESS_API_KEY 환경변수가 필요합니다.")

    date = args.date or today_in(args.tz)
    lifelogs = fetch_day(api_key, date, args.tz, args.per_page)

    md, n_star, n_total = build_markdown(lifelogs, date)

    out_dir = os.path.join(args.vault, "Ingest", "Limitless")
    os.makedirs(out_dir, exist_ok=True)
    out_path = os.path.join(out_dir, f"{date}.md")
    with open(out_path, "w", encoding="utf-8") as f:
        f.write(md)

    print(f"[L1] {out_path}")
    print(f"[L1] 대화 {n_total}건 / 별표 {n_star}건 기록 완료")
    if n_star == 0:
        print("[L1] 별표 0건 -> PLL은 Phase 0에서 정상 EXIT 합니다.")


if __name__ == "__main__":
    main()

---
date: 2026-06-02
source: limitless
total_conversations: 3
starred_conversations: 1
starred_list:
  - "프로젝트 운영 아이디어"
generated: 2026-06-02T...
---
# 아침 산책 중 어제 일어난 일 소회
*08:10 — 08:25*

> (맥락용 비별표 대화)

# ⭐ 프로젝트 운영 아이디어 팀 미칭
*10:05 — 10:30*

> (별표 누른 핵심 대화)

# 가족과 점심대화
*12:00 — 12:30*

> (맥락용 비별표 대화)