Unable to connect to Anthropic services
无法连接到 Anthropic 服务

Failed to connect to api.anthropic.com: ERR_BAD_REQUEST
连接到 api.anthropic.com 失败：ERR_BAD_REQUEST

Please check your internet connection and network settings.
请检查您的互联网连接和网络设置。

Note: Claude Code might not be available in your country. Check supported countries at：https://anthropic.com/supported-countries
注意：Claude Code 可能在您的国家/地区不可用。请查看支持的国家/地区：https://anthropic.com/supported-countries

{
  "env": {
    "ENABLE_TOOL_SEARCH": "true",
    "CLAUDE_CODE_ATTRIBUTION_HEADER": "0"
  },
  "hooks": {
    "UserPromptSubmit": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "$HOME/.claude/ccnotify/ccnotify.py UserPromptSubmit"
          }
        ]
      }
    ],
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "$HOME/.claude/ccnotify/ccnotify.py Stop"
          }
        ]
      }
    ],
    "Notification": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "$HOME/.claude/ccnotify/ccnotify.py Notification"
          }
        ]
      }
    ]
  },
  "language": "chinese",
  "statusLine": {
    "type": "command",
    "command": "bash $HOME/.claude/statusline.sh",
    "refreshInterval": 15
  },
  "permissions": {
    "defaultMode": "acceptEdits",
    "allow": [
      "Read",
      "Glob",
      "Grep",
      "Edit",
      "Write",
      "Bash(node *)",
      "Bash(npm *)",
      "Bash(pnpm *)",
      "Bash(yarn *)",
      "Bash(bun *)",
      "Bash(python3 *)",
      "Bash(python *)",
      "Bash(npx *)",
      "Bash(git *)",
      "Bash(ls *)",
      "Bash(cat *)",
      "Bash(echo *)",
      "Bash(wc *)",
      "Bash(head *)",
      "Bash(tail *)",
      "Bash(grep *)",
      "Bash(find *)",
      "Bash(mkdir *)",
      "Bash(touch *)",
      "Bash(clear)",
      "Bash(curl *)",
      "Bash(sed *)",
      "Bash(mdfind *)",
      "Bash(system_profiler *)",
      "Bash(fc-list *)",
      "Bash(brew *)",
      "Bash(codesign *)",
      "Bash(plutil -p -)",
      "Bash(kill *)",
      "Bash(claude *)",
      "Bash(tsc *)",
      "WebSearch"
    ]
  }
}

~/.claude/statusline.sh

#!/bin/bash
# Claude Code Statusline - 模型 / 上下文进度条

input=$(cat)

model=$(echo "$input" | /usr/bin/jq -r '.model.display_name // "?"')
ctx_pct=$(echo "$input" | /usr/bin/jq -r '.context_window.used_percentage // 0')
total_in=$(echo "$input" | /usr/bin/jq -r '.context_window.total_input_tokens // 0')
total_out=$(echo "$input" | /usr/bin/jq -r '.context_window.total_output_tokens // 0')

# --- 获取上下文窗口总量 ---
ctx_total=$(echo "$input" | /usr/bin/jq -r '
  .context_window.total_tokens // 
  .context_window.max_tokens // 
  .context_window.context_window_size // 
  .context_window.token_limit // 0
')

if [ "$ctx_total" = "0" ] || [ -z "$ctx_total" ]; then
    model_window=$(echo "$model" | grep -oE '[0-9]+[kK]' | head -1)
    if [ -n "$model_window" ]; then
        ctx_total=$(echo "$model_window" | sed 's/[kK]//' | awk '{print $1 * 1000}')
    fi
    
    if [ "$ctx_total" = "0" ] && [ "$(echo "$ctx_pct > 0" | bc -l 2>/dev/null || echo 0)" = "1" ]; then
        total_used=$((total_in + total_out))
        if [ "$total_used" -gt 0 ]; then
            ctx_total=$(echo "$total_used * 100 / $ctx_pct" | bc -l | awk '{printf "%.0f", $1}')
        fi
    fi
    
    if [ "$ctx_total" = "0" ]; then
        ctx_total=$((total_in + total_out))
    fi
fi

# --- Token 格式化 ---
fmt_tok() {
  local n=$1
  case $n in
    ''|*[!0-9]*) 
      echo "0"
      return
      ;;
  esac
  
  if [ "$n" -ge 1000000000 ]; then
    echo "$n" | awk '{printf "%.1fG", $1/1000000000}'
  elif [ "$n" -ge 1000000 ]; then
    echo "$n" | awk '{printf "%.1fM", $1/1000000}'
  elif [ "$n" -ge 1000 ]; then
    echo "$n" | awk '{printf "%.1fK", $1/1000}'
  else
    echo "$n"
  fi
}

# --- 将 ctx_pct 转为整数 ---
ctx_pct_int=$(echo "$ctx_pct" | awk '{printf "%.0f", $1}')
if [ -z "$ctx_pct_int" ] || [ "$ctx_pct_int" -lt 0 ] 2>/dev/null; then
  ctx_pct_int=0
elif [ "$ctx_pct_int" -gt 100 ] 2>/dev/null; then
  ctx_pct_int=100
fi

# --- 计算当前已用 token 数 ---
if [ "$ctx_total" -gt 0 ] 2>/dev/null; then
    if [ "$(echo "$ctx_pct > 0" | bc -l 2>/dev/null || echo 0)" = "1" ]; then
        ctx_used=$(echo "$ctx_total * $ctx_pct / 100" | bc -l 2>/dev/null | awk '{printf "%.0f", $1}')
    else
        ctx_used=$((total_in + total_out))
    fi
else
    ctx_used=$((total_in + total_out))
    ctx_total=$ctx_used
fi

if [ -z "$ctx_used" ] || ! [[ "$ctx_used" =~ ^[0-9]+$ ]] 2>/dev/null; then
    ctx_used=0
fi

ctx_used_fmt=$(fmt_tok "$ctx_used")
ctx_total_fmt=$(fmt_tok "$ctx_total")

# --- 上下文进度条 10格 ---
filled=$((ctx_pct_int * 10 / 100))
[ $filled -gt 10 ] && filled=10
[ $filled -lt 0 ] && filled=0
empty=$((10 - filled))

bar=""
for ((i=0; i<filled; i++)); do bar="${bar}█"; done
for ((i=0; i<empty; i++)); do bar="${bar}░"; done

# --- 颜色 ---
if [ "$ctx_pct_int" -gt 80 ] 2>/dev/null; then
    ctx_color="31"
elif [ "$ctx_pct_int" -gt 50 ] 2>/dev/null; then
    ctx_color="33"
else
    ctx_color="32"
fi

# --- 输出（模型 | 上下文进度条）---
printf "\033[35m%s\033[0m ｜ 上下文 \033[%sm%s\033[0m \033[%sm%d%%\033[0m \033[37m%s/%s\033[0m\n" \
  "$model" "$ctx_color" "$bar" "$ctx_color" "$ctx_pct_int" \
  "$ctx_used_fmt" "$ctx_total_fmt"

brew install terminal-notifier

~/.claude/ccnotify/ccnotify.py

#!/usr/bin/env python3
"""
Claude Code Notify
https://github.com/dazuiba/CCNotify
"""

import os
import sys
import json
import sqlite3
import subprocess
import logging
from logging.handlers import TimedRotatingFileHandler
from datetime import datetime


class ClaudePromptTracker:
    def __init__(self):
        """Initialize the prompt tracker with database setup"""
        script_dir = os.path.dirname(os.path.abspath(__file__))
        self.db_path = os.path.join(script_dir, "ccnotify.db")
        self.setup_logging()
        self.init_database()

    def setup_logging(self):
        """Setup logging to file with daily rotation"""

        script_dir = os.path.dirname(os.path.abspath(__file__))
        log_path = os.path.join(script_dir, "ccnotify.log")

        # Create a timed rotating file handler
        handler = TimedRotatingFileHandler(
            log_path,
            when="midnight",  # Rotate at midnight
            interval=1,  # Every 1 day
            backupCount=1,  # Keep 1 days of logs
            encoding="utf-8",
        )

        # Set the log format
        formatter = logging.Formatter(
            "%(asctime)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
        )
        handler.setFormatter(formatter)

        # Configure the root logger
        logger = logging.getLogger()
        logger.setLevel(logging.INFO)
        logger.addHandler(handler)

    def init_database(self):
        """Create tables and triggers if they don't exist"""
        with sqlite3.connect(self.db_path) as conn:
            # Create main table
            conn.execute("""
                CREATE TABLE IF NOT EXISTS prompt (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    session_id TEXT NOT NULL,
                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                    prompt TEXT,
                    cwd TEXT,
                    seq INTEGER,
                    stoped_at DATETIME,
                    lastWaitUserAt DATETIME
                )
            """)

            # Create trigger for auto-incrementing seq
            conn.execute("""
                CREATE TRIGGER IF NOT EXISTS auto_increment_seq
                AFTER INSERT ON prompt
                FOR EACH ROW
                BEGIN
                    UPDATE prompt
                    SET seq = (
                        SELECT COALESCE(MAX(seq), 0) + 1
                        FROM prompt
                        WHERE session_id = NEW.session_id
                    )
                    WHERE id = NEW.id;
                END
            """)

            conn.commit()

    def handle_user_prompt_submit(self, data):
        """Handle UserPromptSubmit event - insert new prompt record"""
        session_id = data.get("session_id")
        prompt = data.get("prompt", "")
        cwd = data.get("cwd", "")

        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """
                INSERT INTO prompt (session_id, prompt, cwd)
                VALUES (?, ?, ?)
            """,
                (session_id, prompt, cwd),
            )
            conn.commit()

        logging.info(f"Recorded prompt for session {session_id}")

    def handle_stop(self, data):
        """Handle Stop event - update completion time and send notification"""
        session_id = data.get("session_id")

        with sqlite3.connect(self.db_path) as conn:
            # Find the latest unfinished record for this session
            cursor = conn.execute(
                """
                SELECT id, created_at, cwd
                FROM prompt
                WHERE session_id = ? AND stoped_at IS NULL
                ORDER BY created_at DESC
                LIMIT 1
            """,
                (session_id,),
            )

            row = cursor.fetchone()
            if row:
                record_id, created_at, cwd = row

                # Update completion time
                conn.execute(
                    """
                    UPDATE prompt
                    SET stoped_at = CURRENT_TIMESTAMP
                    WHERE id = ?
                """,
                    (record_id,),
                )
                conn.commit()

                # Get seq number and calculate duration
                cursor = conn.execute(
                    "SELECT seq FROM prompt WHERE id = ?", (record_id,)
                )
                seq_row = cursor.fetchone()
                seq = seq_row[0] if seq_row else 1

                duration = self.calculate_duration_from_db(record_id)
                self.send_notification(
                    title=os.path.basename(cwd) if cwd else "Claude Task",
                    subtitle=f"job#{seq} done, duration: {duration}",
                    cwd=cwd,
                )

                logging.info(
                    f"Task completed for session {session_id}, job#{seq}, duration: {duration}"
                )

    def handle_notification(self, data):
        """Handle Notification event - check for various notification types and send notifications"""
        session_id = data.get("session_id")
        message = data.get("message", "")
        cwd = data.get("cwd", "")

        # Log all notifications for debugging
        logging.info(f"[NOTIFICATION] session={session_id}, message='{message}'")

        # Determine notification type and subtitle
        message_lower = message.lower()
        subtitle = None
        should_update_db = False
        should_notify = True

        if (
            "waiting for your input" in message_lower
            or "waiting for input" in message_lower
        ):
            subtitle = "Waiting for input"
            should_update_db = True
            should_notify = (
                False  # Suppress notification - Stop handler will send "job done"
            )
        elif "permission" in message_lower:
            subtitle = "Permission Required"
        elif "approval" in message_lower or "choose an option" in message_lower:
            subtitle = "Action Required"
        else:
            # For other notifications, use a generic subtitle
            subtitle = "Notification"

        # Update database for waiting notifications
        if should_update_db:
            with sqlite3.connect(self.db_path) as conn:
                # Fix: Use subquery instead of ORDER BY/LIMIT in UPDATE
                conn.execute(
                    """
                    UPDATE prompt
                    SET lastWaitUserAt = CURRENT_TIMESTAMP
                    WHERE id = (
                        SELECT id FROM prompt
                        WHERE session_id = ?
                        ORDER BY created_at DESC
                        LIMIT 1
                    )
                """,
                    (session_id,),
                )
                conn.commit()
            logging.info(f"Updated lastWaitUserAt for session {session_id}")

        # Send notification only if should_notify is True
        if should_notify:
            self.send_notification(
                title=os.path.basename(cwd) if cwd else "Claude Task",
                subtitle=subtitle,
                cwd=cwd,
            )
            logging.info(f"Notification sent for session {session_id}: {subtitle}")
        else:
            logging.info(
                f"Notification suppressed for session {session_id}: {subtitle}"
            )

    def calculate_duration_from_db(self, record_id):
        """Calculate duration for a completed record"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                """
                SELECT created_at, stoped_at
                FROM prompt
                WHERE id = ?
            """,
                (record_id,),
            )

            row = cursor.fetchone()
            if row and row[1]:
                return self.calculate_duration(row[0], row[1])

        return "Unknown"

    def calculate_duration(self, start_time, end_time):
        """Calculate human-readable duration between two timestamps"""
        try:
            if isinstance(start_time, str):
                start_dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
            else:
                start_dt = datetime.fromisoformat(start_time)

            if isinstance(end_time, str):
                end_dt = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
            else:
                end_dt = datetime.fromisoformat(end_time)

            duration = end_dt - start_dt
            total_seconds = int(duration.total_seconds())

            if total_seconds < 60:
                return f"{total_seconds}s"
            elif total_seconds < 3600:
                minutes = total_seconds // 60
                seconds = total_seconds % 60
                if seconds > 0:
                    return f"{minutes}m{seconds}s"
                else:
                    return f"{minutes}m"
            else:
                hours = total_seconds // 3600
                minutes = (total_seconds % 3600) // 60
                if minutes > 0:
                    return f"{hours}h{minutes}m"
                else:
                    return f"{hours}h"
        except Exception as e:
            logging.error(f"Error calculating duration: {e}")
            return "Unknown"

    def send_notification(self, title, subtitle, cwd=None):
        """Send macOS notification using terminal-notifier"""
        from datetime import datetime

        current_time = datetime.now().strftime("%B %d, %Y at %H:%M")

        try:
            cmd = [
                "terminal-notifier",
                "-sound",
                "default",
                "-title",
                title,
                "-subtitle",
                f"{subtitle}\n{current_time}",
            ]

            if cwd:
                cmd.extend(["-execute", f'/usr/local/bin/code "{cwd}"'])

            subprocess.run(cmd, check=False, capture_output=True)
            logging.info(f"Notification sent: {title} - {subtitle}")
        except FileNotFoundError:
            logging.warning("terminal-notifier not found, notification skipped")
        except Exception as e:
            logging.error(f"Error sending notification: {e}")


def validate_input_data(data, expected_event_name):
    """Validate input data matches design specification"""
    required_fields = {
        "UserPromptSubmit": ["session_id", "prompt", "cwd", "hook_event_name"],
        "Stop": ["session_id", "hook_event_name"],
        "Notification": ["session_id", "message", "hook_event_name"],
    }

    if expected_event_name not in required_fields:
        raise ValueError(f"Unknown event type: {expected_event_name}")

    # Check hook_event_name matches expected
    if data.get("hook_event_name") != expected_event_name:
        raise ValueError(
            f"Event name mismatch: expected {expected_event_name}, got {data.get('hook_event_name')}"
        )

    # Check required fields
    missing_fields = []
    for field in required_fields[expected_event_name]:
        if field not in data or data[field] is None:
            missing_fields.append(field)

    if missing_fields:
        raise ValueError(
            f"Missing required fields for {expected_event_name}: {missing_fields}"
        )

    return True


def main():
    """Main entry point - read JSON from stdin and process event"""
    try:
        # Check if hook type is provided as command line argument
        if len(sys.argv) < 2:
            print("ok")
            return

        expected_event_name = sys.argv[1]
        valid_events = ["UserPromptSubmit", "Stop", "Notification"]

        if expected_event_name not in valid_events:
            logging.error(f"Invalid hook type: {expected_event_name}")
            logging.error(f"Valid hook types: {', '.join(valid_events)}")
            sys.exit(1)

        # Read JSON data from stdin
        input_data = sys.stdin.read().strip()
        if not input_data:
            logging.warning("No input data received")
            return

        data = json.loads(input_data)

        # Validate input data
        validate_input_data(data, expected_event_name)

        tracker = ClaudePromptTracker()

        if expected_event_name == "UserPromptSubmit":
            tracker.handle_user_prompt_submit(data)
        elif expected_event_name == "Stop":
            tracker.handle_stop(data)
        elif expected_event_name == "Notification":
            tracker.handle_notification(data)

    except json.JSONDecodeError as e:
        logging.error(f"JSON decode error: {e}")
        sys.exit(1)
    except ValueError as e:
        logging.error(f"Validation error: {e}")
        sys.exit(1)
    except Exception as e:
        logging.error(f"Unexpected error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()