158 lines
5.2 KiB
Python
158 lines
5.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
커밋 flush 훅: git HEAD 가 변했으면 누적 토큰을 Aptabase 로 전송한다.
|
|
|
|
호출 경로 두 가지:
|
|
1. Claude PostToolUse(Bash) 훅 → stdin 에 JSON 입력
|
|
2. git post-commit 훅 → stdin 비어 있음 (직접 git 에서 호출)
|
|
|
|
두 경로 모두 같은 로직 (HEAD 비교 → 변했으면 flush → 리셋).
|
|
|
|
전송 시점:
|
|
- 훅 호출 시마다 HEAD 확인
|
|
- state.last_sent_commit 와 다르면 새 커밋으로 간주 → flush
|
|
|
|
전송 후 동작:
|
|
- state.accum 을 0으로 리셋
|
|
- state.last_sent_commit 를 현재 HEAD 로 갱신
|
|
- 실패 시 state 유지 → 다음 호출에서 재시도
|
|
|
|
전송 필드:
|
|
- claude_oauth_id : Claude OAuth 이메일
|
|
- plan : 구독 플랜
|
|
- user_name : aptabase.json 에서 지정
|
|
- local_ip : 로컬 IP
|
|
- public_ip : 공인 IP
|
|
- commit_hash : 현재 HEAD 해시
|
|
- commit_message : git log -1 --pretty=%B
|
|
- issue_number : 커밋 메시지에서 추출 (없으면 null)
|
|
- repository : owner/repo 또는 디렉터리명
|
|
- repository_url : remote.origin.url
|
|
- total_tokens : 누적 합계
|
|
- input_tokens, cache_creation_tokens, cache_read_tokens, output_tokens
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from aptabase_common import ( # noqa: E402
|
|
EMPTY_TOTALS,
|
|
EVENT_NAME,
|
|
accumulate_from_transcript,
|
|
build_system_props,
|
|
consumed_tokens,
|
|
extract_issue_number,
|
|
get_claude_oauth_id,
|
|
get_local_ip,
|
|
get_plan,
|
|
get_public_ip,
|
|
get_repository_info,
|
|
load_config,
|
|
load_state,
|
|
make_aptabase_session_id,
|
|
now_iso,
|
|
post_event,
|
|
run_git,
|
|
save_state,
|
|
total_tokens,
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
cfg = load_config()
|
|
if cfg is None:
|
|
return
|
|
|
|
# stdin 파싱: Claude 훅 모드(JSON) vs git post-commit 모드(비어 있음)
|
|
hook_input: dict = {}
|
|
try:
|
|
raw = sys.stdin.read()
|
|
if raw.strip():
|
|
hook_input = json.loads(raw)
|
|
except Exception:
|
|
hook_input = {}
|
|
|
|
# Claude 훅 모드면 Bash 툴에서 온 호출만 처리 (다른 툴은 무시)
|
|
# git 훅 모드면 hook_input 이 비어 있어서 이 체크를 통과한다
|
|
if hook_input and hook_input.get("tool_name") != "Bash":
|
|
return
|
|
|
|
cwd = hook_input.get("cwd") or os.getcwd()
|
|
|
|
# git 저장소가 아니면 무시
|
|
head = run_git(["rev-parse", "HEAD"], cwd=cwd)
|
|
if not head:
|
|
return
|
|
|
|
state = load_state(cwd)
|
|
last_sent = state.get("last_sent_commit", "")
|
|
|
|
if head == last_sent:
|
|
return
|
|
|
|
if not last_sent:
|
|
# 첫 실행 — 기준점만 기록하고 다음 커밋을 기다린다
|
|
state["last_sent_commit"] = head
|
|
save_state(state, cwd)
|
|
return
|
|
|
|
# HEAD 가 바뀌었다 — Stop 훅이 놓쳤을 수 있는 최신 토큰을 flush.
|
|
# transcript_path 우선순위:
|
|
# 1) hook_input (Claude PostToolUse 훅 모드)
|
|
# 2) state.last_transcript (git post-commit 모드, Stop 훅이 저장한 값)
|
|
#
|
|
# 이 덕분에 한 턴에 여러 번 커밋하는 경우에도 각 커밋 직전까지 생성된
|
|
# 토큰이 해당 커밋에 flush 된다 (턴 중간 커밋의 0 토큰 문제 해결).
|
|
transcript_path = (
|
|
hook_input.get("transcript_path")
|
|
or state.get("last_transcript", "")
|
|
)
|
|
if transcript_path:
|
|
accumulate_from_transcript(state, transcript_path)
|
|
|
|
commit_message = run_git(["log", "-1", "--pretty=%B", head], cwd=cwd)
|
|
repo = get_repository_info(cwd)
|
|
accum = state.get("accum", dict(EMPTY_TOTALS))
|
|
|
|
payload = {
|
|
"timestamp": now_iso(),
|
|
"sessionId": make_aptabase_session_id(),
|
|
"eventName": EVENT_NAME,
|
|
"systemProps": build_system_props(),
|
|
"props": {
|
|
# 커밋 정보
|
|
"commit_hash": head,
|
|
"commit_message": commit_message,
|
|
"issue_number": extract_issue_number(commit_message),
|
|
"repository": repo["name"] if not repo["url"] else repo["url"],
|
|
# 사용자
|
|
"claude_oauth_id": get_claude_oauth_id(),
|
|
"plan": get_plan(),
|
|
"user_name": cfg.get("user_name", "unknown"),
|
|
# 네트워크
|
|
"local_ip": get_local_ip(),
|
|
"public_ip": get_public_ip(),
|
|
# 토큰 (consumed 가 실제 쿼터 차감량, total 은 원시 합계)
|
|
"consumed_tokens": consumed_tokens(accum),
|
|
"total_tokens": total_tokens(accum),
|
|
"input_tokens": int(accum.get("input_tokens", 0)),
|
|
"output_tokens": int(accum.get("output_tokens", 0)),
|
|
"cache_creation_tokens": int(accum.get("cache_creation_tokens", 0)),
|
|
"cache_read_tokens": int(accum.get("cache_read_tokens", 0)),
|
|
},
|
|
}
|
|
|
|
ok = post_event(cfg["aptabase_host"], cfg["app_key"], payload)
|
|
if ok:
|
|
# 누적 리셋 + 전송 기준점 갱신
|
|
state["accum"] = dict(EMPTY_TOTALS)
|
|
state["last_sent_commit"] = head
|
|
save_state(state, cwd)
|
|
# 실패: state 유지 → 다음 Bash 호출에서 재시도 (누적 + 새 커밋 병합)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|