Google Drive の画像を linebot へ送信する機能のモジュール化

Google Drive の画像を linebot へ送信する機能のモジュール化

import requests
import random
import time
from googleapiclient.discovery import build
from google.oauth2 import service_account

# ① Google Drive API の設定
SCOPES = ["https://www.googleapis.com/auth/drive"]
SERVICE_ACCOUNT_FILE = "service_account.json"

creds = service_account.Credentials.from_service_account_file(SERVICE_ACCOUNT_FILE, scopes=SCOPES)
drive_service = build("drive", "v3", credentials=creds)

# ② 送信先の LINE API 設定
LINE_PUSH_URL = "https://api.line.me/v2/bot/message/push"
LINE_CHANNEL_ACCESS_TOKEN = ""

USER_ID = ""


# ③ Google Drive の「temp」フォルダID
FOLDER_ID = ""

def get_drive_images():
    """
    Google Drive の temp フォルダ内の画像リストを取得し、画像の `id` をリストで返す
    """
    query = f"'{FOLDER_ID}' in parents and mimeType contains 'image/' and trashed=false"
    results = drive_service.files().list(q=query, fields="files(id, name)").execute()
    files = results.get("files", [])

    if not files:
        print("No images found in the Drive folder.")
        return None

    return [file["id"] for file in files]

def get_drive_image_url(file_id):
    """
    Google Drive のファイルIDを公開URLに変換する
    """
    try:
        # ファイルのアクセス権限を「公開」に設定
        drive_service.permissions().create(
            fileId=file_id,
            body={"role": "reader", "type": "anyone"},
        ).execute()

        # 画像のダウンロードURLを取得
        return f"https://drive.google.com/thumbnail?id={file_id}&sz=w1000"
    except Exception as e:
        print(f"Error making image public: {e}")
        return None

def send_image(image_url):
    """
    LINE API を使い、取得した画像URLを LINE ユーザーに送信する
    """
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}"
    }

    payload = {
        "to": USER_ID,
        "messages": [
            {
                "type": "image",
                "originalContentUrl": image_url,
                "previewImageUrl": image_url
            }
        ]
    }

    response = requests.post(LINE_PUSH_URL, headers=headers, json=payload)
    print(response.status_code, response.text)

def main():
    """
    画像リストを取得し、ランダムな画像を選んでLINEに送信
    """
    image_ids = get_drive_images()
    if not image_ids:
        print("No images found, skipping LINE push.")
        return

    # 画像をランダムに選択
    random_image_id = random.choice(image_ids)
    image_url = get_drive_image_url(random_image_id)

    if image_url:
        # LINE に画像を送信
        send_image(image_url)
    else:
        print("Failed to get a valid image URL.")

# スケジュール実行（30分ごと）
if __name__ == "__main__":
    while True:
        main()
        time.sleep(1800)  # 30分ごとに実行

をモジュールにして文字送信以外に画像も送信可能にする

 vim line_image_sender.py

内容は

import requests
import random
from googleapiclient.discovery import build
from google.oauth2 import service_account

# 設定
SCOPES = ["https://www.googleapis.com/auth/drive"]
SERVICE_ACCOUNT_FILE = "service_account.json"

LINE_PUSH_URL = "https://api.line.me/v2/bot/message/push"
LINE_CHANNEL_ACCESS_TOKEN = "YOUR_LINE_CHANNEL_ACCESS_TOKEN"
USER_ID = "YOUR_LINE_USER_ID"
FOLDER_ID = "YOUR_GOOGLE_DRIVE_FOLDER_ID"

# 認証とサービスの初期化
creds = service_account.Credentials.from_service_account_file(SERVICE_ACCOUNT_FILE, scopes=SCOPES)
drive_service = build("drive", "v3", credentials=creds)

def get_drive_images(folder_id=FOLDER_ID):
    query = f"'{folder_id}' in parents and mimeType contains 'image/' and trashed=false"
    results = drive_service.files().list(q=query, fields="files(id, name)").execute()
    files = results.get("files", [])
    return [file["id"] for file in files] if files else []

def get_drive_image_url(file_id):
    try:
        drive_service.permissions().create(
            fileId=file_id,
            body={"role": "reader", "type": "anyone"},
        ).execute()
        return f"https://drive.google.com/thumbnail?id={file_id}&sz=w1000"
    except Exception as e:
        print(f"Error making image public: {e}")
        return None

def send_image(image_url, user_id=USER_ID, token=LINE_CHANNEL_ACCESS_TOKEN):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    payload = {
        "to": user_id,
        "messages": [
            {
                "type": "image",
                "originalContentUrl": image_url,
                "previewImageUrl": image_url
            }
        ]
    }
    response = requests.post(LINE_PUSH_URL, headers=headers, json=payload)
    print(response.status_code, response.text)

def send_random_image():
    image_ids = get_drive_images()
    if not image_ids:
        print("No images found.")
        return
    random_image_id = random.choice(image_ids)
    image_url = get_drive_image_url(random_image_id)
    if image_url:
        send_image(image_url)
    else:
        print("Failed to get valid image URL.")

# スクリプトとして直接実行された場合の処理（30分おきに実行）
if __name__ == "__main__":
    import time
    while True:
        send_random_image()
        time.sleep(1800)

これを config.jsonから読み込むようにする

cp ../gas_bot/config.json .

でコピー

{
  "token": "",
  "ollama_model": "elyza:jp8b",
  "line_bot_channel_access_token": "",
  "channel_secret": "",
  "line_bot_user_id": "",
  "google_drive_folder_id": ""
}

というように
Google drive のIDを追加する

import requests
import random
import json
from googleapiclient.discovery import build
from google.oauth2 import service_account

# 設定ファイルの読み込み
with open("config.json", "r") as f:
    config = json.load(f)

LINE_PUSH_URL = "https://api.line.me/v2/bot/message/push"
LINE_CHANNEL_ACCESS_TOKEN = config["line_bot_channel_access_token"]
USER_ID = config["line_bot_user_id"]
FOLDER_ID = config["google_drive_folder_id"]

# Google Drive API の設定
SCOPES = ["https://www.googleapis.com/auth/drive"]
SERVICE_ACCOUNT_FILE = "service_account.json"

creds = service_account.Credentials.from_service_account_file(SERVICE_ACCOUNT_FILE, scopes=SCOPES)
drive_service = build("drive", "v3", credentials=creds)

def get_drive_images(folder_id=FOLDER_ID):
    query = f"'{folder_id}' in parents and mimeType contains 'image/' and trashed=false"
    results = drive_service.files().list(q=query, fields="files(id, name)").execute()
    files = results.get("files", [])
    return [file["id"] for file in files] if files else []

def get_drive_image_url(file_id):
    try:
        drive_service.permissions().create(
            fileId=file_id,
            body={"role": "reader", "type": "anyone"},
        ).execute()
        return f"https://drive.google.com/thumbnail?id={file_id}&sz=w1000"
    except Exception as e:
        print(f"Error making image public: {e}")
        return None

def send_image(image_url, user_id=USER_ID, token=LINE_CHANNEL_ACCESS_TOKEN):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}"
    }
    payload = {
        "to": user_id,
        "messages": [
            {
                "type": "image",
                "originalContentUrl": image_url,
                "previewImageUrl": image_url
            }
        ]
    }
    response = requests.post(LINE_PUSH_URL, headers=headers, json=payload)
    print(response.status_code, response.text)

def send_random_image():
    image_ids = get_drive_images()
    if not image_ids:
        print("No images found.")
        return
    random_image_id = random.choice(image_ids)
    image_url = get_drive_image_url(random_image_id)
    if image_url:
        send_image(image_url)
    else:
        print("Failed to get valid image URL.")

# スクリプトとして実行されたとき、30分おきに画像送信
if __name__ == "__main__":
    import time
    while True:
        send_random_image()
        time.sleep(1800)

というようにコード変更
これで設定ファイルから読み込むようになる

あとは linebot の画像の管理

これだとどんどん画像が増えていく