step1_prepare.py v0.89
#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Co-developed with ChatGPT (GPT-5.1) # from __future__ import annotations import json import os import time import re import sys import datetime import unicodedata # <--- ★NFKC処理 from pathlib import Path from typing import Any, Dict, List, Optional from dotenv import load_dotenv import requests # ★追加: REST API呼び出し用 # import google.generativeai as genai # ★削除: SDKは使用しない from cost_guard import estimate_cost_and_confirm # === 設定 === BASE_DIR: Path = Path(__file__).resolve().parent INPUT_DIR: Path = BASE_DIR / "input" OUTPUT_DIR: Path = BASE_DIR / "output" SUMMARY_DIR: Path = OUTPUT_DIR / "01_summaries" PLAN_DIR: Path = OUTPUT_DIR / "02_plan" FULL_LOG_PATH: Path = INPUT_DIR / "full_log.txt" PLAN_FILE: Path = PLAN_DIR / "story_plan.json" ARCHITECT_PROMPT_FILE: Path = INPUT_DIR / "instruction_architect.txt" EXTRACT_PROMPT_FILE: Path = INPUT_DIR / "instruction_extract.txt" CHUNK_SIZE: int = 10000 OVERLAP: int = 500 RETRY_MAX: int = 5 SLEEP_BASE: int = 10 REQUEST_INTERVAL: int = 5 load_dotenv() GEMINI_API_KEY: Optional[str] = os.getenv("GEMINI_API_KEY") MODEL_FAST: str = "models/gemini-2.5-flash-lite" MODEL_SMART: str = "models/gemini-2.5-pro" # === 共通関数 === # ★変更: SDK初期化ではなく「キーの存在チェック」のみ def configure_gemini(): if not GEMINI_API_KEY: sys.exit("Error: GEMINI_API_KEY missing.") # REST APIでは追加の初期化は不要 def ensure_dirs(): for d in [INPUT_DIR, OUTPUT_DIR, SUMMARY_DIR, PLAN_DIR]: d.mkdir(parents=True, exist_ok=True) def read_text(path: Path) -> str: text = path.read_text(encoding="utf-8") return unicodedata.normalize("NFKC", text) # <--- ★NFKC処理 def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") def get_now_str() -> str: return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") # ★全面変更: google-generativeai SDK → 素のREST API呼び出しに置き換え def call_gemini(prompt: str, model_name: str) -> str: """ Gemini REST APIを使ってテキストを生成する。 - 事前に estimate_cost_and_confirm でコストチェック - RETRY_MAX 回まで指数バックオフでリトライ """ if not GEMINI_API_KEY: raise RuntimeError("GEMINI_API_KEY is not set.") # モデル名は "models/gemini-2.5-..." 形式を想定 url = ( f"https://generativelanguage.googleapis.com/v1beta/" f"{model_name}:generateContent?key={GEMINI_API_KEY}" ) # ★ generationConfig をモデルに応じて切り替える generation_config = {} if model_name.endswith("-pro"): generation_config = { "maxOutputTokens": 4096, "temperature": 0.2 } # 共通のリクエストペイロード payload = { "contents": [ { "parts": [ {"text": prompt} ] } ], "generationConfig": generation_config } # レート制限用のベーススリープ time.sleep(REQUEST_INTERVAL) for attempt in range(RETRY_MAX): try: print(f" [{get_now_str()}] API Call ({model_name})... ", end="", flush=True) # 事前コスト計算(※リトライ毎に呼ばれるがオーバーヘッドは小さい) estimate_cost_and_confirm(prompt, model_name, safety_limit_yen=100) # REST API呼び出し response = requests.post(url, json=payload, timeout=240) # 考える時間かかりそうなとき response.raise_for_status() data = response.json() # レスポンスから本文を抽出 candidates = data.get("candidates") if not candidates: raise ValueError("No candidates in response") content = candidates[0].get("content", {}) parts = content.get("parts", []) text = "".join(part.get("text", "") for part in parts) if not text: raise ValueError("Empty response text") print("OK") return text except Exception as e: wait = SLEEP_BASE * (2 ** attempt) print(f"\n [Error] {e} -> Wait {wait}s") time.sleep(wait) raise RuntimeError("API Failed after retries") def extract_json_block(text: str) -> str: match = re.search(r"```json(.*?)```", text, re.DOTALL | re.IGNORECASE) return match.group(1).strip() if match else text.strip() # === Phase 1: Summary === def split_text(text: str, chunk_size: int, overlap: int) -> List[str]: if not text: return [] chunks: List[str] = [] start = 0 text_len = len(text) while start < text_len: end = min(start + chunk_size, text_len) if end < text_len: nn = text.find('\n', end) if nn != -1 and nn - end < 500: end = nn + 1 chunks.append(text[start:end]) if end >= text_len: break next_start = end - overlap start = next_start if next_start > start else start + chunk_size return chunks def run_summarization() -> List[Path]: print(f"\n[{get_now_str()}] === Phase 1: Deep Summarization ===") if not EXTRACT_PROMPT_FILE.exists(): sys.exit(f"Error: Template {EXTRACT_PROMPT_FILE} not found.") prompt_template = read_text(EXTRACT_PROMPT_FILE) full_log = read_text(FULL_LOG_PATH) chunks = split_text(full_log, CHUNK_SIZE, OVERLAP) print(f"Log Length: {len(full_log)} chars -> Expected Chunks: {len(chunks)}") files: List[Path] = [] skipped_count = 0 for i, chunk in enumerate(chunks, 1): path = SUMMARY_DIR / f"summary_part_{i:02d}.txt" if path.exists(): files.append(path) skipped_count += 1 continue print(f"Processing Part {i}/{len(chunks)}...") if "{chunk}" not in prompt_template: sys.exit("Error: instruction_extract.txt must contain {chunk} placeholder.") prompt = prompt_template.replace("{chunk}", chunk) text = call_gemini(prompt, MODEL_FAST) write_text(path, text) files.append(path) if skipped_count == len(chunks): print(f"All {skipped_count} summary files exist. Skipping Phase 1 API calls.") else: print(f"Phase 1 Complete. Generated {len(chunks) - skipped_count} new summaries.") return files # === Phase 2: Architect === def run_architect(files: List[Path]) -> None: print(f"\n[{get_now_str()}] === Phase 2: Architecture ===") if PLAN_FILE.exists(): print("Plan file already exists. To regenerate, delete output/02_plan/story_plan.json") return if not ARCHITECT_PROMPT_FILE.exists(): sys.exit(f"Error: Template {ARCHITECT_PROMPT_FILE} not found.") prompt_template = read_text(ARCHITECT_PROMPT_FILE) combined = "" for p in files: combined += f"\n\n--- Source Part {p.stem.split('_')[-1]} ---\n{read_text(p)}" print(f"Constructing plan from {len(combined)} chars of summaries...") # ★前回の修正: .format() ではなく .replace() を使う if "{summaries}" not in prompt_template: sys.exit("Error: instruction_architect.txt must contain {summaries} placeholder.") prompt = prompt_template.replace("{summaries}", combined) # 実行 resp = call_gemini(prompt, MODEL_SMART) try: json_str = extract_json_block(resp) json.loads(json_str) # Check validity write_text(PLAN_FILE, json_str) print("Plan generated successfully.") except Exception as e: print(f"JSON Error: {e}") write_text(PLAN_DIR / "plan_error.txt", resp) def main() -> None: configure_gemini() ensure_dirs() files = run_summarization() run_architect(files) print(f"\n[{get_now_str()}] Step 1 Complete! Please edit: {PLAN_FILE}") if __name__ == "__main__": main()
プログラムの概要
大規模なログファイルをGemini APIで段階的に処理し、最終的な物語設計書（story_plan.json）を生成するPythonスクリプトです。
処理フロー
Phase 1: Deep Summarization（詳細要約）
Phase 2: Architecture（構造化設計）
- Phase 1の全要約を結合
instruction_architect.txtのテンプレートに組み込み- Gemini Pro APIで構造化されたJSONプランを生成
output/02_plan/story_plan.jsonに保存
技術的特徴
- REST API直接呼び出し:
google-generativeaiSDKを使わず、REST APIを直接利用 - コスト管理:
cost_guardモジュールによる事前コスト推定と確認 - 信頼性の確保: 指数バックオフによるリトライ処理（最大5回）
- レート制限対策: リクエスト間隔の制御（
REQUEST_INTERVAL） - 効率的な再実行: 既存ファイルを自動スキップし、途中から再開可能
主要な設定項目
| 項目 | デフォルト値 | 説明 |
|---|---|---|
CHUNK_SIZE |
30,000 | 1チャンクの文字数 |
OVERLAP |
500 | チャンク間の重複文字数（文脈保持用） |
MODEL_FAST |
gemini-2.5-flash |
Phase 1で使用する高速モデル |
MODEL_SMART |
gemini-2.5-pro |
Phase 2で使用する高精度モデル |
REQUEST_INTERVAL |
5秒 | API呼び出し間の待機時間 |
必要な環境
- Python 3.12.12で実績
.envファイルにGEMINI_API_KEYを設定- 必要なパッケージ:
requests,python-dotenv- python環境の整え方はAIに相談するとよいです
実行方法
python step1_prepare.py
実行後、output/02_plan/story_plan.json を確認・編集してください。
免責事項
このプログラムはサンプルとして共有するもので、内容の正確性・完全性を保証するものではありません。 API 仕様・価格は変更される可能性があり、本資料を利用したことによるいかなる損害についても作成者は責任を負いません。 また、詳細は必ず最新の Google 公式ドキュメントをご確認ください。（AIに確認しながら進めるとよいと思います。）
