import google.generativeai as genai
from pathlib import Path
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.0-flash")
 
# ローカル画像ファイルを読み込む
image_path = Path("screenshot.png")
image_data = image_path.read_bytes()
 
response = model.generate_content([
    {
        "mime_type": "image/png",
        "data": image_data
    },
    "このスクリーンショットのUIを分析し、UXの改善点を5つ挙げてください。"
])
 
print(response.text)

import base64
 
with open("chart.jpg", "rb") as f:
    image_data = base64.b64encode(f.read()).decode("utf-8")
 
response = model.generate_content([
    {
        "inline_data": {
            "mime_type": "image/jpeg",
            "data": image_data
        }
    },
    "このグラフのデータを表形式で抽出し、主要なトレンドを分析してください。"
])

# File API を使った動画のアップロードと処理
import google.generativeai as genai
import time
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
# 動画ファイルをアップロード
print("動画をアップロード中...")
video_file = genai.upload_file("tutorial_video.mp4")
 
# アップロード処理完了を待つ
while video_file.state.name == "PROCESSING":
    print("処理中...")
    time.sleep(5)
    video_file = genai.get_file(video_file.name)
 
if video_file.state.name == "FAILED":
    raise ValueError("動画のアップロードに失敗しました")
 
# 動画に関する質問を送る
model = genai.GenerativeModel("gemini-2.0-flash")
 
response = model.generate_content([
    video_file,
    "この動画の内容を以下の形式で要約してください：\n"
    "1. 動画のメインテーマ\n"
    "2. 主要なポイント（3〜5つ）\n"
    "3. 結論または次のアクション"
])
 
print(response.text)
 
# アップロードしたファイルを削除
genai.delete_file(video_file.name)

# 音声ファイルの文字起こし＋分析
with open("meeting_recording.mp3", "rb") as f:
    audio_data = f.read()
 
response = model.generate_content([
    {
        "inline_data": {
            "mime_type": "audio/mp3",
            "data": base64.b64encode(audio_data).decode("utf-8")
        }
    },
    "この会議録音を文字起こしし、以下を抽出してください：\n"
    "1. 決定事項\n"
    "2. アクションアイテム（担当者・期限付き）\n"
    "3. 未解決の課題"
])

# 2つの画像を比較する
before_image = Path("before.png").read_bytes()
after_image = Path("after.png").read_bytes()
 
response = model.generate_content([
    "ビフォー画像：",
    {"mime_type": "image/png", "data": before_image},
    "アフター画像：",
    {"mime_type": "image/png", "data": after_image},
    "2つの画像の違いを詳細に比較してください。デザインの変更点・追加・削除された要素を全てリストアップしてください。"
])

import json
 
# 商品画像から構造化データを抽出
with open("product.jpg", "rb") as f:
    product_image = base64.b64encode(f.read()).decode("utf-8")
 
response = model.generate_content(
    contents=[
        {
            "inline_data": {
                "mime_type": "image/jpeg",
                "data": product_image
            }
        },
        """この商品画像から以下のJSONフォーマットで情報を抽出してください：
        {
          "product_name": "商品名",
          "category": "カテゴリ",
          "colors": ["色1", "色2"],
          "estimated_price_range": "価格帯（例: ¥1,000〜¥5,000）",
          "features": ["特徴1", "特徴2", "特徴3"],
          "target_audience": "ターゲット顧客",
          "condition": "状態（新品/中古）"
        }
        JSONのみ出力してください。"""
    ],
    generation_config=genai.GenerationConfig(
        response_mime_type="application/json"
    )
)
 
product_data = json.loads(response.text)
print(product_data)

# 通常のAPIでのストリーミング
model = genai.GenerativeModel("gemini-2.0-flash")
 
response = model.generate_content(
    "量子コンピュータの仕組みを詳しく説明してください。",
    stream=True
)
 
for chunk in response:
    if chunk.text:
        print(chunk.text, end="", flush=True)
print()

import asyncio
from google import genai
 
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
async def live_conversation():
    """Live API を使ったリアルタイム対話"""
    config = {
        "response_modalities": ["TEXT"],
        "system_instruction": "あなたは丁寧で役立つアシスタントです。"
    }
 
    async with client.aio.live.connect(
        model="gemini-2.0-flash-live-001",
        config=config
    ) as session:
        print("Live API セッション開始")
 
        # テキストメッセージを送信
        await session.send(input="こんにちは！今日の天気について教えてください。", end_of_turn=True)
 
        # レスポンスを受信
        async for response in session.receive():
            if response.text:
                print(f"Gemini: {response.text}", end="", flush=True)
            if response.server_content and response.server_content.turn_complete:
                print()  # 改行
                break
 
asyncio.run(live_conversation())

import asyncio
import pyaudio
from google import genai
from google.genai import types
 
SAMPLE_RATE = 16000
CHANNELS = 1
CHUNK = 1024
 
async def voice_conversation():
    """音声入出力を使ったリアルタイム対話"""
 
    client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
    config = types.LiveConnectConfig(
        response_modalities=["AUDIO"],
        speech_config=types.SpeechConfig(
            voice_config=types.VoiceConfig(
                prebuilt_voice_config=types.PrebuiltVoiceConfig(
                    voice_name="Kore"  # 音声の種類を選択
                )
            )
        )
    )
 
    audio = pyaudio.PyAudio()
 
    # マイク入力ストリーム
    input_stream = audio.open(
        format=pyaudio.paInt16,
        channels=CHANNELS,
        rate=SAMPLE_RATE,
        input=True,
        frames_per_buffer=CHUNK
    )
 
    # スピーカー出力ストリーム
    output_stream = audio.open(
        format=pyaudio.paInt16,
        channels=CHANNELS,
        rate=24000,  # 出力サンプルレート
        output=True
    )
 
    async with client.aio.live.connect(
        model="gemini-2.0-flash-live-001",
        config=config
    ) as session:
 
        async def send_audio():
            """マイクからの音声を送信"""
            while True:
                data = input_stream.read(CHUNK, exception_on_overflow=False)
                await session.send(input={"data": data, "mime_type": "audio/pcm"})
                await asyncio.sleep(0.01)
 
        async def receive_audio():
            """Geminiからの音声を再生"""
            async for response in session.receive():
                if response.data:
                    output_stream.write(response.data)
 
        # 送受信を並列実行
        await asyncio.gather(
            send_audio(),
            receive_audio()
        )
 
asyncio.run(voice_conversation())

import asyncio
import cv2
import base64
from google import genai
from google.genai import types
 
async def realtime_camera_analysis():
    """カメラ映像をリアルタイムでGeminiに送り分析"""
 
    client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
    config = types.LiveConnectConfig(
        response_modalities=["TEXT"],
        system_instruction="カメラに映っているものを継続的に分析し、"
                          "重要な変化があれば報告してください。"
    )
 
    cap = cv2.VideoCapture(0)  # カメラを開く
 
    async with client.aio.live.connect(
        model="gemini-2.0-flash-live-001",
        config=config
    ) as session:
 
        frame_count = 0
 
        while True:
            ret, frame = cap.read()
            if not ret:
                break
 
            # 1秒ごとにフレームを送信（フレームレート制御）
            if frame_count % 30 == 0:
                _, buffer = cv2.imencode(".jpg", frame)
                image_data = base64.b64encode(buffer).decode("utf-8")
 
                await session.send(
                    input=[
                        {"mime_type": "image/jpeg", "data": image_data},
                        "このフレームで何が見えますか？"
                    ],
                    end_of_turn=True
                )
 
                async for response in session.receive():
                    if response.text:
                        print(f"分析: {response.text}")
                    if response.server_content and response.server_content.turn_complete:
                        break
 
            frame_count += 1
 
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
 
    cap.release()
 
asyncio.run(realtime_camera_analysis())

import asyncio
from google import genai
 
MAX_RETRIES = 3
RETRY_DELAY = 2.0
 
async def robust_live_session(prompt: str):
    """再接続ロジック付きLive APIセッション"""
    client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
    for attempt in range(MAX_RETRIES):
        try:
            async with client.aio.live.connect(
                model="gemini-2.0-flash-live-001",
                config={"response_modalities": ["TEXT"]}
            ) as session:
                await session.send(input=prompt, end_of_turn=True)
 
                async for response in session.receive():
                    if response.text:
                        return response.text
                    if response.server_content and response.server_content.turn_complete:
                        break
 
        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt < MAX_RETRIES - 1:
                await asyncio.sleep(RETRY_DELAY * (2 ** attempt))
            else:
                raise

import google.generativeai as genai
from PIL import Image
import base64
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.0-flash")
 
# 方法1: PIL Image オブジェクト
image = Image.open("product.jpg")
response = model.generate_content([
    "この商品画像を分析して、ECサイト向けの商品説明文を300字で書いてください",
    image
])
print(response.text)
 
# 方法2: バイトデータ（API経由でファイルを受け取る場合）
with open("product.jpg", "rb") as f:
    image_bytes = f.read()
 
response = model.generate_content([
    "この画像に写っている全ての物体をリストアップしてください",
    {
        "mime_type": "image/jpeg",
        "data": base64.b64encode(image_bytes).decode()
    }
])

# 複数画像を一度に処理（マルチモーダルの真価）
before_image = Image.open("before.jpg")
after_image = Image.open("after.jpg")
 
response = model.generate_content([
    "以下の2枚の画像を比較して、変化点を詳細に分析してください。",
    "Before:", before_image,
    "After:", after_image,
    "特に以下の観点で分析をお願いします：\n1. 追加された要素\n2. 削除された要素\n3. 変更された要素"
])

import json
 
# レシートやフォームから構造化データを抽出
receipt_image = Image.open("receipt.jpg")
 
response = model.generate_content([
    """このレシートから以下のJSON形式でデータを抽出してください：
    {
      "store_name": "店舗名",
      "date": "YYYY-MM-DD",
      "items": [
        {"name": "商品名", "quantity": 数量, "price": 価格}
      ],
      "subtotal": 小計,
      "tax": 税額,
      "total": 合計
    }
    JSON以外のテキストは出力しないでください。""",
    receipt_image
])
 
try:
    data = json.loads(response.text)
    print(f"合計金額: ¥{data['total']:,}")
except json.JSONDecodeError:
    # JSONとして解析できない場合のフォールバック
    print(response.text)

import google.generativeai as genai
 
model = genai.GenerativeModel("gemini-2.0-flash")
 
# 音声ファイルのアップロードと処理
audio_file = genai.upload_file("meeting_recording.mp3")
 
response = model.generate_content([
    audio_file,
    """この音声を以下の形式で処理してください：
    1. 完全な文字起こし（話者を区別して [話者A]: [話者B]: の形式で）
    2. 会議の要約（5点以内）
    3. アクションアイテムのリスト（担当者・期限付き）
    4. 次回会議に向けた提案事項"""
])

# ストリーミング処理でリアルタイムに応答を受け取る
async def process_audio_stream(audio_bytes: bytes):
    model = genai.GenerativeModel("gemini-2.0-flash")
 
    response = model.generate_content(
        [
            {"mime_type": "audio/mp3", "data": base64.b64encode(audio_bytes).decode()},
            "この音声の内容を日本語でリアルタイムに文字起こししてください"
        ],
        stream=True
    )
 
    async for chunk in response:
        if chunk.text:
            yield chunk.text  # ストリーミング出力

# 動画ファイルのアップロード（大きなファイルは Files API 使用）
import time
 
video_file = genai.upload_file("product_demo.mp4")
 
# アップロード完了を待機
while video_file.state.name == "PROCESSING":
    time.sleep(5)
    video_file = genai.get_file(video_file.name)
 
if video_file.state.name == "FAILED":
    raise ValueError("動画のアップロードに失敗しました")
 
# 動画の分析
model = genai.GenerativeModel("gemini-1.5-pro")  # 長い動画は Pro モデルを推奨
response = model.generate_content([
    video_file,
    """この製品デモ動画を分析して：
    1. デモされている主要機能の一覧（タイムスタンプ付き）
    2. ユーザーインターフェースの評価
    3. 改善提案（上位3件）
    4. マーケティング素材として使える印象的なシーン（タイムスタンプ付き）"""
])

# 特定のシーンを指定して詳細分析
response = model.generate_content([
    video_file,
    """00:45 あたりのシーンに注目して：
    - 何が起きているか詳細に説明
    - この場面のユーザー感情を推定
    - このシーンを改善するための具体的な提案"""
])

class ProductQualityInspector:
    """製品画像・仕様書・動画を統合した品質検査システム"""
 
    def __init__(self):
        self.model = genai.GenerativeModel("gemini-2.0-flash")
 
    def inspect(
        self,
        product_images: list,
        spec_pdf: str,
        demo_video: str = None
    ) -> dict:
        # 仕様書をアップロード
        spec_file = genai.upload_file(spec_pdf)
 
        inputs = [spec_file, "以下の仕様書に基づいて製品を検査してください。\n\n"]
 
        # 製品画像を追加
        for i, img_path in enumerate(product_images):
            img = Image.open(img_path)
            inputs.extend([f"\n製品画像 {i+1}:", img])
 
        # 動画があれば追加
        if demo_video:
            video_file = genai.upload_file(demo_video)
            inputs.extend(["\n動作デモ動画:", video_file])
 
        inputs.append("""
        以下のJSON形式で検査結果を出力してください：
        {
          "overall_status": "pass|fail|review_required",
          "score": 0-100,
          "defects": [{"description": "説明", "severity": "critical|major|minor", "location": "箇所"}],
          "spec_compliance": {"compliant": true/false, "deviations": ["逸脱項目"]},
          "recommendations": ["推奨事項"]
        }
        """)
 
        response = self.model.generate_content(inputs)
        return json.loads(response.text)

async def generate_content_from_media(
    source_video: str,
    target_formats: list[str]
) -> dict:
    """
    動画から各種コンテンツを自動生成
    - ブログ記事
    - SNS投稿文
    - YouTube概要欄
    - 字幕ファイル（SRT）
    """
    model = genai.GenerativeModel("gemini-1.5-pro")
    video = genai.upload_file(source_video)
 
    results = {}
 
    for format_type in target_formats:
        prompts = {
            "blog": "この動画の内容をもとに1500字程度のブログ記事を書いてください",
            "sns": "この動画の魅力を伝えるSNS投稿文を140字以内で3パターン書いてください",
            "youtube_desc": "YouTube動画の概要欄を書いてください。チャプター付きで500字程度",
            "srt": "この動画の字幕ファイルをSRT形式で生成してください"
        }
 
        response = model.generate_content([video, prompts[format_type]])
        results[format_type] = response.text
 
    return results

# 同じ大きなコンテキストを何度も使う場合はキャッシュが効果的
cache = genai.caching.CachedContent.create(
    model="gemini-1.5-pro",
    contents=[
        # 大きなシステムプロンプト・参考資料など
        "以下はあなたが分析に使う製品仕様書全文です：\n" + open("large_spec.txt").read()
    ],
    ttl=datetime.timedelta(hours=1)  # 1時間キャッシュ
)
 
# キャッシュを使って複数のクエリを効率化
model_with_cache = genai.GenerativeModel.from_cached_content(cache)
for question in questions:
    response = model_with_cache.generate_content(question)
    # キャッシュされたコンテキスト分のトークンは安くなる

# 多数の短い処理はバッチ化するとコスト・速度ともに有利
async def batch_process_images(image_paths: list[str], task: str) -> list[str]:
    """複数画像を並列処理"""
    model = genai.GenerativeModel("gemini-2.0-flash")
 
    async def process_one(path: str) -> str:
        img = Image.open(path)
        response = model.generate_content([task, img])
        return response.text
 
    # 並列実行（最大10並列推奨）
    semaphore = asyncio.Semaphore(10)
    async def limited_process(path):
        async with semaphore:
            return await process_one(path)
 
    results = await asyncio.gather(*[limited_process(p) for p in image_paths])
    return results

import time
from google.api_core import retry, exceptions
 
class RobustGeminiClient:
    def __init__(self, api_key: str, model_name: str = "gemini-2.0-flash"):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel(model_name)
 
    @retry.Retry(
        predicate=retry.if_exception_type(
            exceptions.ResourceExhausted,  # レート制限
            exceptions.ServiceUnavailable,  # 一時的な障害
        ),
        initial=1.0,
        maximum=60.0,
        multiplier=2.0,
        deadline=300.0  # 最大5分リトライ
    )
    def generate(self, inputs: list, **kwargs) -> str:
        response = self.model.generate_content(inputs, **kwargs)
        return response.text
 
    def generate_with_fallback(self, inputs: list, fallback_model: str = "gemini-1.5-flash") -> str:
        try:
            return self.generate(inputs)
        except exceptions.InvalidArgument:
            # メインモデルで失敗した場合はフォールバック
            fallback = genai.GenerativeModel(fallback_model)
            return fallback.generate_content(inputs).text