フォームで取得したフォルダからファイルをダウンロードする

downloadFilesFromFolder.gs

#!/usr/bin/env python3

from __future__ import annotations

import re
import subprocess
import sys
from pathlib import Path

import gspread
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow


# =========================
# 設定
# =========================

SPREADSHEET_URL = "ここにGoogleシートのURLを入力"
WORKSHEET_NAME = "シート1"

RCLONE_REMOTE = "gdrive"

DOWNLOAD_ROOT = (
    Path.home()
    / "Downloads"
    / "lecture_downloads"
)

# OAuthクライアント認証情報
SCRIPT_DIR = Path(__file__).resolve().parent
CREDENTIALS_FILE = SCRIPT_DIR / "credentials.json"
TOKEN_FILE = SCRIPT_DIR / "token.json"

SCOPES = [
    "https://www.googleapis.com/auth/spreadsheets.readonly",
]


def get_gspread_client() -> gspread.Client:
    """OAuth認証を行い、gspreadクライアントを返す。"""

    credentials = None

    if TOKEN_FILE.exists():
        credentials = Credentials.from_authorized_user_file(
            str(TOKEN_FILE),
            SCOPES,
        )

    if (
        credentials
        and credentials.expired
        and credentials.refresh_token
    ):
        credentials.refresh(Request())

    if not credentials or not credentials.valid:
        if not CREDENTIALS_FILE.exists():
            raise FileNotFoundError(
                f"credentials.jsonが見つかりません: "
                f"{CREDENTIALS_FILE}"
            )

        flow = InstalledAppFlow.from_client_secrets_file(
            str(CREDENTIALS_FILE),
            SCOPES,
        )

        credentials = flow.run_local_server(port=0)

        TOKEN_FILE.write_text(
            credentials.to_json(),
            encoding="utf-8",
        )

    return gspread.authorize(credentials)


def sanitize_name(name: str, fallback: str) -> str:
    """ローカルフォルダ名に使えない文字を置換する。"""

    name = name.strip()

    if not name:
        return fallback

    name = re.sub(r'[\\/:*?"<>|]', "_", name)
    name = name.rstrip(".")

    return name or fallback


def extract_folder_id(url: str) -> str | None:
    """Google DriveフォルダURLからフォルダIDを抽出する。"""

    text = str(url).strip()

    if not text:
        return None

    patterns = [
        r"/folders/([A-Za-z0-9_-]+)",
        r"[?&]id=([A-Za-z0-9_-]+)",
        r"^([A-Za-z0-9_-]{20,})$",
    ]

    for pattern in patterns:
        match = re.search(pattern, text)

        if match:
            return match.group(1)

    return None


def read_sheet_rows() -> list[list[str]]:
    """
    GoogleシートからA〜D列を取得する。

    A列: 講義回名
    B列: 未使用
    C列: PDFフォルダURL
    D列: 動画フォルダURL
    """

    client = get_gspread_client()
    spreadsheet = client.open_by_url(SPREADSHEET_URL)

    worksheet = spreadsheet.worksheet(
        WORKSHEET_NAME
    )

    return worksheet.get("A2:D")


def run_rclone(
    folder_id: str,
    destination: Path,
    label: str,
) -> bool:
    """Google Driveフォルダ内をrcloneでコピーする。"""

    destination.mkdir(
        parents=True,
        exist_ok=True,
    )

    source = (
        f"{RCLONE_REMOTE},"
        f"root_folder_id={folder_id}:"
    )

    command = [
        "rclone",
        "copy",
        source,
        str(destination),
        "--progress",
        "--transfers",
        "4",
        "--checkers",
        "8",
        "--create-empty-src-dirs",
        "--drive-export-formats",
        "pdf",
    ]

    print(f"\n{label}")
    print(f"  保存先: {destination}")

    result = subprocess.run(
        command,
        check=False,
    )

    if result.returncode != 0:
        print(
            f"  rcloneエラー: "
            f"終了コード {result.returncode}",
            file=sys.stderr,
        )
        return False

    return True


def main() -> None:
    rows = read_sheet_rows()

    DOWNLOAD_ROOT.mkdir(
        parents=True,
        exist_ok=True,
    )

    failures: list[str] = []

    for sheet_row, row in enumerate(
        rows,
        start=2,
    ):
        # A列
        lecture_name = (
            row[0].strip()
            if len(row) >= 1
            else ""
        )

        # C列
        pdf_folder_url = (
            row[2].strip()
            if len(row) >= 3
            else ""
        )

        # D列
        video_folder_url = (
            row[3].strip()
            if len(row) >= 4
            else ""
        )

        if (
            not lecture_name
            and not pdf_folder_url
            and not video_folder_url
        ):
            continue

        lecture_directory_name = sanitize_name(
            lecture_name,
            f"lecture_{sheet_row}",
        )

        lecture_directory = (
            DOWNLOAD_ROOT
            / lecture_directory_name
        )

        print(
            f"\n{'=' * 60}\n"
            f"{sheet_row}行目: "
            f"{lecture_directory_name}\n"
            f"{'=' * 60}"
        )

        pdf_folder_id = extract_folder_id(
            pdf_folder_url
        )

        if pdf_folder_url and not pdf_folder_id:
            message = (
                f"{lecture_directory_name}: "
                "PDFフォルダURL解析失敗"
            )
            print(message, file=sys.stderr)
            failures.append(message)

        elif pdf_folder_id:
            success = run_rclone(
                pdf_folder_id,
                lecture_directory / "pdf",
                "PDFフォルダを取得",
            )

            if not success:
                failures.append(
                    f"{lecture_directory_name}: "
                    "PDF取得失敗"
                )

        video_folder_id = extract_folder_id(
            video_folder_url
        )

        if (
            video_folder_url
            and not video_folder_id
        ):
            message = (
                f"{lecture_directory_name}: "
                "動画フォルダURL解析失敗"
            )
            print(message, file=sys.stderr)
            failures.append(message)

        elif video_folder_id:
            success = run_rclone(
                video_folder_id,
                lecture_directory / "video",
                "動画フォルダを取得",
            )

            if not success:
                failures.append(
                    f"{lecture_directory_name}: "
                    "動画取得失敗"
                )

    print(f"\n保存先: {DOWNLOAD_ROOT}")

    if failures:
        print("\n失敗した処理:")

        for failure in failures:
            print(f"  - {failure}")

        sys.exit(1)

    print("\nすべての処理が完了しました。")


if __name__ == "__main__":
    main()