Claude Code × Python ハイブリッド開発パターン集 — トークン50%削減の実装ガイド

# scripts/daily_batch.py
import json
from datetime import datetime, timedelta
import sqlite3
 
def run_daily_batch():
    """
    毎日の定型処理
    - 前日の売上データを集計
    - ユーザー登録数をカウント
    - エラーログを要約
    結果を JSON に出力
    """
    yesterday = (datetime.now() - timedelta(days=1)).date()
    
    # SQLite から前日のデータを集計
    conn = sqlite3.connect('analytics.db')
    cursor = conn.cursor()
    
    # 売上
    cursor.execute(
        'SELECT SUM(amount) FROM transactions WHERE DATE(created_at) = ?',
        (yesterday,)
    )
    revenue = cursor.fetchone()[0] or 0
    
    # ユーザー登録
    cursor.execute(
        'SELECT COUNT(*) FROM users WHERE DATE(created_at) = ?',
        (yesterday,)
    )
    new_users = cursor.fetchone()[0]
    
    # エラー件数
    cursor.execute(
        'SELECT COUNT(*) FROM error_log WHERE DATE(logged_at) = ?',
        (yesterday,)
    )
    error_count = cursor.fetchone()[0]
    
    conn.close()
    
    # 結果を JSON で保存
    result = {
        'date': str(yesterday),
        'revenue': revenue,
        'new_users': new_users,
        'error_count': error_count,
        'generated_at': datetime.now().isoformat()
    }
    
    with open(f'daily_reports/{yesterday}.json', 'w', encoding='utf-8') as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    
    return result
 
if __name__ == '__main__':
    result = run_daily_batch()
    print(f"✓ 日次バッチ完了: 売上 ¥{result['revenue']}, 新規ユーザー {result['new_users']} 件")

# 毎日午前3時に実行
0 3 * * * cd /path/to/project && python3 scripts/daily_batch.py

# tests/test_email_validator.py
import pytest
from src.validators import validate_email
 
class TestEmailValidator:
    """メール形式のバリデータをテスト駆動で実装"""
    
    def test_valid_email(self):
        """正常形式"""
        assert validate_email('user@example.com') is True
    
    def test_invalid_format(self):
        """フォーマット違い"""
        assert validate_email('user@invalid') is False
        assert validate_email('user.invalid.com') is False
    
    def test_subdomain(self):
        """サブドメイン対応"""
        assert validate_email('user@mail.example.co.jp') is True
    
    def test_plus_addressing(self):
        """Gmail の + アドレッシング対応"""
        assert validate_email('user+tag@example.com') is True
    
    def test_length_limit(self):
        """256文字以上は拒否"""
        long_email = 'a' * 250 + '@example.com'
        assert validate_email(long_email) is False

# .github/claude-instructions.md
 
## Claude Code が担当する処理
 
- **新機能の code review**: 複雑な PR は Claude に review をもらう
- **バグ調査**: `[DEBUG]` ラベルの issue は Claude にアサイン
- **設計相談**: `[DESIGN]` ラベルの discussion は Claude で検討
- **リファクタリング**: 複雑度の高いモジュールは Claude に相談
 
## Python / GitHub Actions が担当する処理
 
- **日次レポート生成**: workflow で毎日 3:00 AM に実行
- **テスト実行**: PR ごとに自動実行
- **デプロイ**: `main` ブランチの push で自動デプロイ
- **ログ集計**: 毎週月曜 8:00 AM に集計 JSON を生成
 
## トークン予算管理
 
- 月額予算: 月 $100（おおよそ 250万トークン）
- 目安: PR レビュー 1件あたり 10,000トークン
- 超過した場合: 以降の依頼を一時停止し、優先度を見直し

# .github/workflows/daily-batch.yml
name: Daily Batch
 
on:
  schedule:
    - cron: '0 3 * * *'  # 毎日 03:00 UTC
 
jobs:
  run-batch:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      
      - name: Run daily batch
        run: python3 scripts/daily_batch.py
      
      - name: Commit results
        run: |
          git config user.name "Automated Batch"
          git config user.email "batch@example.com"
          git add daily_reports/
          git commit -m "chore: daily batch $(date +%Y-%m-%d)"
          git push

# scripts/analyze_with_fallback.py
import json
import subprocess
import sys
 
def analyze_with_claude_fallback(data_file):
    """
    Claude で分析を試みる。失敗時は Python の簡易版を実行
    """
    try:
        # Claude Code に分析を依頼（疑似実装）
        result = call_claude_api(data_file, 
            prompt="このデータの異常値を検出してください")
        
        print("✓ Claude による分析完了")
        return result
    
    except (TimeoutError, ConnectionError) as e:
        print(f"⚠ Claude 呼び出し失敗（{e}）。Python 簡易版で処理します")
        
        # フォールバック: Python で統計的異常検出
        return fallback_anomaly_detection(data_file)
 
def fallback_anomaly_detection(data_file):
    """Python による簡易異常検出"""
    import json
    import statistics
    
    with open(data_file, 'r') as f:
        data = json.load(f)
    
    values = data.get('values', [])
    if not values:
        return {'anomalies': []}
    
    mean = statistics.mean(values)
    stdev = statistics.stdev(values) if len(values) > 1 else 0
    
    anomalies = [
        {'value': v, 'z_score': (v - mean) / stdev if stdev else 0}
        for v in values
        if abs(v - mean) > 2 * stdev  # Z スコア > 2
    ]
    
    return {'anomalies': anomalies}
 
def call_claude_api(data_file, prompt):
    """疑似API呼び出し（実装例）"""
    # 本番では MCP 経由や API キーでの呼び出しになる
    raise TimeoutError("Simulated API timeout")
 
if __name__ == '__main__':
    result = analyze_with_claude_fallback('sensor_data.json')
    
    with open('analysis_result.json', 'w') as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    
    print(f"✓ 分析完了: {len(result.get('anomalies', []))} 件の異常を検出")

.PHONY: help ai-review py-batch py-transform py-analyze run-all test ci
 
help:
	@echo "Claude Code × Python Hybrid Automation"
	@echo ""
	@echo "AI tasks:"
	@echo "  make ai-review        - PR レビュー（Claude 依頼）"
	@echo "  make ai-design        - 新機能設計（Claude 相談）"
	@echo ""
	@echo "Python tasks:"
	@echo "  make py-batch         - 日次バッチ実行"
	@echo "  make py-transform     - データ変換"
	@echo "  make py-analyze       - ログ分析"
	@echo ""
	@echo "Automation:"
	@echo "  make run-all          - 全 Python タスク実行"
	@echo "  make test             - テスト実行"
	@echo "  make ci               - CI パイプライン実行"
 
# AI タスク
ai-review:
	@echo "Claude Code にレビュー依頼中..."
	# 実装: Claude API 呼び出し
 
ai-design:
	@echo "新機能設計を Claude に相談中..."
 
# Python タスク
py-batch:
	python3 scripts/daily_batch.py
 
py-transform:
	python3 scripts/transform_user_data.py
 
py-analyze:
	python3 scripts/analyze_logs.py
 
# テスト
test:
	python3 -m pytest tests/ -v
 
# パイプライン実行
run-all: test py-batch py-transform py-analyze
	@echo "✓ 全パイプライン完了"
 
# CI 環境（フルテスト + linting）
ci: test
	python3 -m flake8 src/ scripts/
	@echo "✓ CI パイプライン成功"

# 朝の定型処理
make run-all
 
# 新機能の設計相談
make ai-design
 
# テストのみ
make test

import logging
from datetime import datetime
 
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)
 
try:
    result = run_batch()
except Exception as e:
    logger.error(f"Batch failure at {datetime.now()}: {e}")
    # Slack/Email に通知
    send_alert(f"Batch failed: {e}")
    sys.exit(1)

def get_data_with_cache(cache_file, max_age_hours=24):
    """キャッシュの鮮度チェック"""
    import json
    from datetime import datetime, timedelta
    
    if not os.path.exists(cache_file):
        return None
    
    with open(cache_file, 'r') as f:
        cache = json.load(f)
    
    generated = datetime.fromisoformat(cache.get('generated_at', ''))
    age = datetime.now() - generated
    
    if age > timedelta(hours=max_age_hours):
        print(f"✗ キャッシュ有効期限切れ（{age.total_seconds() / 3600:.1f} 時間経過）")
        return None
    
    print(f"✓ キャッシュを使用（{age.total_seconds() / 3600:.1f} 時間以内）")
    return cache
 
# 使用例
data = get_data_with_cache('cache.json', max_age_hours=24)
if data is None:
    data = generate_fresh_data()