LoginSignup
7
4

More than 1 year has passed since last update.

PythonとSeleniumとWebDriverとChrome拡張機能でDownload完了判定と待機

Last updated at Posted at 2022-12-04

はじめに

downloads.Stateを利用してのダウンロードの完了判定と待機が出来ないか検討してみた話です。

前半部分はChrome拡張機能でダウンロードに関する情報をPython側に伝達をする処理を構築していきます。
後半部分はPythonとSeleniumとWebDriverでダウンロードの完了判定と待機をする処理を構築していきます。
最終的なコードはこちらを参照くださいませ。

Chrome拡張機能の詳細な解説はしませんので、
Chrome拡張機能の詳細な解説は他のサイトなどで確認くださいませ。

本投稿は備忘録的な意味合いもあるので、
無駄な記載も含まれています。
ご了承ください。

環境

PythonとSeleniumは導入済み、および、
ChromeとChromeのWebDriverは導入済みである、
Windows11の環境として本投稿は記載します。

情報
Webdriver Manager for Python などを導入している場合は、
Webdriver初期化時の処理は自身の方法に置き換えてください。

準備

Windows11で、
window.downloads.stateフォルダをデスクトップに新規作成し、
window.downloads.stateフォルダに以下ファイルを新規作成します。

C:\Users\USERNAME\Desktop\window.downloads.state>systeminfo | findstr "OS.名 システムの種類"
OS 名:         Microsoft Windows 11 Home
システムの種類: x64-based PC
C:\Users\USERNAME\Desktop\window.downloads.state>dir /b
background.js
content-script.js
manifest.json
sample.py

情報
sample.pyは本投稿の後半部分で使用します。

Chrome Extension ( manifest version 3 ) hello world

Chrome拡張機能でハローワールドしてみます。

各種ファイルを下記内容にします。

拡張機能を読み込んだ後、
https://chromedriver.chromium.org/downloads
などの適当なサイトを開くと、
上記サイトのDevtoolsのConsoleと、
background.jsのDevtoolsのConsoleに、
Hello, I am xxxxxxx scripts が、
それぞれに出力されます。

background.js
console.log('Hello, I am background scripts');
content-script.js
console.log('Hello, I am content scripts');
manifest.json
{
  "manifest_version": 3,
  "name": "DownloadsState",
  "version": "1.0",
  "background": {
    "service_worker": "background.js"
  },
  "content_scripts": [
    {
      "matches": ["<all_urls>"],
      "js": ["content-script.js"]
    }
  ],
  "permissions": [
    "downloads",
    "storage",
    "tabs"
  ]
}

情報
拡張機能読込方法は右記参照 ⇒ Chrome Extension: Load unpacked extension
Devtools起動方法は右記参照 ⇒ Chrome Extension: Inspect background script

Chrome Extension ( manifest version 3 ) downloads

ダウンロードに関する全ての情報を、
background.jsで取得して出力してみます。

background.jsを下記内容に書き換えます。

拡張機能を読み込んだ後、
background.jsのDevtoolsのConsoleを開き、
https://chromedriver.chromium.org/downloads
などで適当なファイルをダウンロードすると、
background.jsのDevtoolsのConsoleに、
ダウンロードの状況が出力されます。

background.js
console.log('Hello, I am background scripts');

+chrome.downloads.onCreated.addListener(async (item) => {
+  const items = await chrome.downloads.search({});
+  console.log('onCreated');
+  console.log(items);
+});
+chrome.downloads.onChanged.addListener(async (delta) => {
+  const items = await chrome.downloads.search({});
+  console.log('onChanged');
+  console.log(items);
+});

情報
拡張機能読込方法は右記参照 ⇒ Chrome Extension: Load unpacked extension
Devtools起動方法は右記参照 ⇒ Chrome Extension: Inspect background script

Downloads State

ダウンロードに関する全ての情報
ダウンロードの状態から、
downloads.Stateと同様に以下を取得してみます。

in_progress(ダウンロード途中)
interrupted(ダウンロード失敗)
complete (ダウンロード完了)
に追加して、
unknown(上記以外)※通常起こりえない状態です
none(ダウンロードを一度も行なっていない)

ダウンロードが単一のファイルの場合は、
該当のファイルのダウンロードの状態を返します。
ダウンロードが複数のファイルの場合は、
ひとつでもダウンロード途中のものがあればin_progressを、
ひとつでもダウンロード失敗のものがあればinterruptedを、
全ファイルダウンロード完了の状態であればcompleteを返します。

以下参考です。
l1ダウンロードに関する全ての情報です。
l2l1から空のものを除外したものです。
l1に恐らく空のものはないので、
l2l1の内容は同じなはずです。
l3stateだけを取得したものです。

background.jsを下記内容に書き換えます。

拡張機能を読み込んだ後、
background.jsのDevtoolsのConsoleを開き、
https://chromedriver.chromium.org/downloads
などで適当なファイルをダウンロードすると、
background.jsのDevtoolsのConsoleに、
ダウンロードの状況が出力されます。

background.js
console.log('Hello, I am background scripts');

+const getWindowDownloadsState = (items) => {
+  let state;
+  if (items?.length) {
+    const l1 = items;
+    const l2 = l1.filter((e) => e);
+    const l3 = l2.map((e) => e.state);
+    const stateInProgress = l3.some((e) => e === 'in_progress');
+    const stateInterrupted = l3.some((e) => e === 'interrupted');
+    const stateComplete = l3.every((e) => e === 'complete');
+    state = stateInProgress ? 'in_progress' : state;
+    state = stateInterrupted ? 'interrupted' : state;
+    state = stateComplete ? 'complete' : state;
+    state = state || 'unknown';
+    // console.log(l1);
+    // console.log(l2);
+    // console.log(l3);
+    // console.log(stateInProgress);
+    // console.log(stateInterrupted);
+    // console.log(stateComplete);
+    // console.log(state);
+  } else {
+    state = 'none';
+  }
+  return state;
+};

chrome.downloads.onCreated.addListener(async (item) => {
  const items = await chrome.downloads.search({});
+ const state = getWindowDownloadsState(items);
+ console.log(`onCreated: ${state}`);
- console.log('onCreated');
- console.log(items);
+ // console.log(items);
});
chrome.downloads.onChanged.addListener(async (delta) => {
  const items = await chrome.downloads.search({});
+ const state = getWindowDownloadsState(items);
+ console.log(`onChanged: ${state}`);
- console.log('onChanged');
- console.log(items);
+ // console.log(items);
});

情報
拡張機能読込方法は右記参照 ⇒ Chrome Extension: Load unpacked extension
Devtools起動方法は右記参照 ⇒ Chrome Extension: Inspect background script

Chrome Extension ( manifest version 3 ) storage session

background.jsで取得した情報を、
content-script.jsに伝達します。

background.js、および、
content-script.jsを下記内容に書き換えます。

拡張機能を読み込んだ後、
https://chromedriver.chromium.org/downloads
などで適当なファイルをダウンロードすると、
上記サイトのDevtoolsのConsoleに、
ダウンロードの状況が出力されます。

background.js
console.log('Hello, I am background scripts');

const getWindowDownloadsState = (items) => {
  let state;
  // 中略
  return state;
};

chrome.downloads.onCreated.addListener(async (item) => {
  const items = await chrome.downloads.search({});
  const state = getWindowDownloadsState(items);
  console.log(`onCreated: ${state}`);
  // console.log(items);
+ await chrome.storage.session.set({ dataWindowDownloadsState: state });
});
chrome.downloads.onChanged.addListener(async (delta) => {
  const items = await chrome.downloads.search({});
  const state = getWindowDownloadsState(items);
  console.log(`onChanged: ${state}`);
  // console.log(items);
+ await chrome.storage.session.set({ dataWindowDownloadsState: state });
});
content-script.js
console.log('Hello, I am content scripts');

+chrome.storage.onChanged.addListener((objects, strings) => {
+  if (objects?.dataWindowDownloadsState?.newValue) {
+    console.log(`storage.onChanged: ${objects.dataWindowDownloadsState.newValue}`);
+  }
+});

情報
拡張機能読込方法は右記参照 ⇒ Chrome Extension: Load unpacked extension
Devtools起動方法は右記参照 ⇒ Chrome Extension: Inspect background script

Chrome Extension ( manifest version 3 ) message passing

message passing でも伝達してみます。

background.js、および、
content-script.jsを下記内容に書き換えます。

拡張機能を読み込んだ後、
https://chromedriver.chromium.org/downloads
などで適当なファイルをダウンロードすると、
上記サイトのDevtoolsのConsoleに、
ダウンロードの状況が出力されます。

background.js
console.log('Hello, I am background scripts');

const getWindowDownloadsState = (items) => {
  let state;
  // 中略
  return state;
};

+const sendMessage = async (key, message) => {
+  const tabs = await chrome.tabs.query({ url: ['http://*/*', 'https://*/*'] });
+  tabs.forEach((tab) => {
+    chrome.tabs.sendMessage(tab.id, { [key]: message });
+  });
+};

chrome.downloads.onCreated.addListener(async (item) => {
  const items = await chrome.downloads.search({});
  const state = getWindowDownloadsState(items);
  console.log(`onCreated: ${state}`);
  // console.log(items);
+ sendMessage('dataWindowDownloadsState', state);
  await chrome.storage.session.set({ dataWindowDownloadsState: state });
});
chrome.downloads.onChanged.addListener(async (delta) => {
  const items = await chrome.downloads.search({});
  const state = getWindowDownloadsState(items);
  console.log(`onChanged: ${state}`);
  // console.log(items);
+ sendMessage('dataWindowDownloadsState', state);
  await chrome.storage.session.set({ dataWindowDownloadsState: state });
});
content-script.js
console.log('Hello, I am content scripts');

chrome.storage.onChanged.addListener((objects, strings) => {
  if (objects?.dataWindowDownloadsState?.newValue) {
    console.log(`storage.onChanged: ${objects.dataWindowDownloadsState.newValue}`);
  }
});

+chrome.runtime.onMessage.addListener((request, sender, sendResponse) => {
+  if (request?.dataWindowDownloadsState) {
+    console.log(`runtime.onMessage: ${request.dataWindowDownloadsState}`);
+  }
+});

情報
拡張機能読込方法は右記参照 ⇒ Chrome Extension: Load unpacked extension
Devtools起動方法は右記参照 ⇒ Chrome Extension: Inspect background script

Chrome Extension : chrome.tabs.onUpdated.addListener

タブが更新された時に実行される処理を追加しておきます。
主にサイトロード時に実行されます。

background.jsを下記内容に書き換えます。

拡張機能を読み込んだ後、
https://chromedriver.chromium.org/downloads
などの適当なサイトを開くと、
上記サイトのDevtoolsのConsoleに、
上記サイトのロード時にも、
ダウンロードの状況が出力されます。

background.js
console.log('Hello, I am background scripts');

const getWindowDownloadsState = (items) => {
  let state;
  // 中略
  return state;
};

const sendMessage = async (key, message) => {
  const tabs = await chrome.tabs.query({ url: ['http://*/*', 'https://*/*'] });
  tabs.forEach((tab) => {
    chrome.tabs.sendMessage(tab.id, { [key]: message });
  });
};

chrome.downloads.onCreated.addListener(async (item) => {
  const items = await chrome.downloads.search({});
  const state = getWindowDownloadsState(items);
  console.log(`onCreated: ${state}`);
  // console.log(items);
  sendMessage('dataWindowDownloadsState', state);
  await chrome.storage.session.set({ dataWindowDownloadsState: state });
});
chrome.downloads.onChanged.addListener(async (delta) => {
  const items = await chrome.downloads.search({});
  const state = getWindowDownloadsState(items);
  console.log(`onChanged: ${state}`);
  // console.log(items);
  sendMessage('dataWindowDownloadsState', state);
  await chrome.storage.session.set({ dataWindowDownloadsState: state });
});

+chrome.tabs.onUpdated.addListener(async (tabId, changeInfo, tab) => {
+  if (changeInfo.status === 'complete') {
+    const items = await chrome.downloads.search({});
+    const state = getWindowDownloadsState(items);
+    console.log(`onUpdated: ${state}`);
+    // console.log(items);
+    sendMessage('dataWindowDownloadsState', state);
+    await chrome.storage.session.set({ dataWindowDownloadsState: state });
+  }
+});

custom data attributes

background.jsとcontent-script.jsからsample.pyへと直接伝達する方法も無く、
background.jsとcontent-script.jsへとsample.pyから直接伝達する方法も無いので、
ダーティーな方法ですが、
該当サイトの<body>のカスタムデータ属性を利用して伝達します。まずは、
該当サイトの<body>のカスタムデータ属性にcontent-script.jsから書き込みます。

Chrome Extension ( manifest version 3 ) storage session で追記したコード側で行なってもよいし、
Chrome Extension ( manifest version 3 ) message passing で追記したコード側で行なってもよいのですが、本投稿は、
Chrome Extension ( manifest version 3 ) message passing で追記したコード側でカスタムデータ属性を書き込みます。

content-script.jsを下記内容に書き換えます。

拡張機能を読み込んだ後、
https://chromedriver.chromium.org/downloads
などで適当なファイルをダウンロードすると、
上記サイトのDevtoolsのElementsで、
body要素のカスタムデータ属性に、
ダウンロードの状況が出力されているのが確認できます。

content-script.js
console.log('Hello, I am content scripts');

chrome.storage.onChanged.addListener((objects, strings) => {
  if (objects?.dataWindowDownloadsState?.newValue) {
    console.log(`storage.onChanged: ${objects.dataWindowDownloadsState.newValue}`);
+   // document.body.setAttribute('data-window-downloads-state', objects.dataWindowDownloadsState.newValue);
  }
});

chrome.runtime.onMessage.addListener((request, sender, sendResponse) => {
  if (request?.dataWindowDownloadsState) {
    console.log(`runtime.onMessage: ${request.dataWindowDownloadsState}`);
+   document.body.setAttribute('data-window-downloads-state', request.dataWindowDownloadsState);
  }
});

PythonとSeleniumとWebDriver

sample.pyを書いていきます。まずは、
PythonとSeleniumとWebDriverでハローワールドなコードです。
https://chromedriver.chromium.org/downloadsを開くだけです。

sample.py
import datetime
import os
import time

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait


def main():

    print("PRESS ENTER KEY TO BOOT")
    input()

    driver = None

    try:
        driver = None
        driver = webdriver.Chrome()
        driver.get("https://chromedriver.chromium.org/downloads")
        time.sleep(1)
    except Exception as e:
        print(e)
    finally:
        driver = driver.quit() if driver is not None else None

    print()
    print("PRESS ENTER KEY TO EXIT")
    input()


if __name__ == "__main__":
    main()

情報
Webdriver Manager for Python などを導入している場合は、
Webdriver初期化時の処理は自身の方法に置き換えてください。

PythonとSeleniumとWebDriverでDownload

PythonとSeleniumとWebDriverで3個ファイルをダウンロードしてみます。
ただし、すぐにquit()をしてしまうので、恐らくダウンロードが完了しません。

sample.py
import datetime
import os
import time

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait


def main():

    print("PRESS ENTER KEY TO BOOT")
    input()

    driver = None

    try:
        driver = None
        driver = webdriver.Chrome()
        driver.get("https://chromedriver.chromium.org/downloads")
+       WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver.storage.googleapis.com')]")))
+       driver.execute_script("""document.querySelector("a[href*='chromedriver.storage.googleapis.com']").setAttribute("target","_self")""")
+       driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver.storage.googleapis.com')]").click()
+       WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_linux64.zip')]")))
+       WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_mac64.zip')]")))
+       WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_win32.zip')]")))
+       driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_linux64.zip')]").click()
+       driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_mac64.zip')]").click()
+       driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_win32.zip')]").click()
        time.sleep(1)
    except Exception as e:
        print(e)
    finally:
        driver = driver.quit() if driver is not None else None

    print()
    print("PRESS ENTER KEY TO EXIT")
    input()


if __name__ == "__main__":
    main()

情報
Webdriver Manager for Python などを導入している場合は、
Webdriver初期化時の処理は自身の方法に置き換えてください。

PythonとSeleniumとWebDriverでChrome拡張機能読込

Chrome拡張機能の読み込みの追加です。

sample.py
import datetime
import os
import time

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait


def main():

    print("PRESS ENTER KEY TO BOOT")
    input()

    driver = None

    try:
+       # extension_location = os.getcwd()
+       extension_location = f"{os.getenv('USERPROFILE')}\\Desktop\\window.downloads.state"
+       options = webdriver.ChromeOptions()
+       options.add_argument(f"load-extension={extension_location}")
        driver = None
-       driver = webdriver.Chrome()
+       driver = webdriver.Chrome(options=options)
        driver.get("https://chromedriver.chromium.org/downloads")
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver.storage.googleapis.com')]")))
        driver.execute_script("""document.querySelector("a[href*='chromedriver.storage.googleapis.com']").setAttribute("target","_self")""")
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver.storage.googleapis.com')]").click()
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_linux64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_mac64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_win32.zip')]")))
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_linux64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_mac64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_win32.zip')]").click()
        time.sleep(1)
    except Exception as e:
        print(e)
    finally:
        driver = driver.quit() if driver is not None else None

    print()
    print("PRESS ENTER KEY TO EXIT")
    input()


if __name__ == "__main__":
    main()

情報
拡張機能読込方法は右記参照 ⇒ Chrome Extension: Installing extensions via ChromeDriver

PythonとSeleniumとWebDriverとChrome拡張機能でDownload完了判定と待機

Download完了判定の処理の追加です。

sample.py
import datetime
import os
import time

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait


+def checking_conditions_download_ready_state_complete(driver):
+    jikan = f"{f'{datetime.datetime.now()}'[:-4]}"
+    state = driver.execute_script("return document.body.getAttribute('data-window-downloads-state')")
+    _ = state == "in_progress" and print(f"{jikan}: download in progress...")
+    _ = state == "interrupted" and print(f"{jikan}: download interrupted...")
+    _ = state == "complete" and print(f"{jikan}: download complete!!")
+    _ = state == "unknown" and print(f"{jikan}: download unknown...")
+    _ = state == "none" and print(f"{jikan}: download none...")
+    _ = state is None and print(f"{jikan}: error...")
+    return bool(state == "complete")


def main():

    print("PRESS ENTER KEY TO BOOT")
    input()

    driver = None

    try:
        # extension_location = os.getcwd()
        extension_location = f"{os.getenv('USERPROFILE')}\\Desktop\\window.downloads.state"
        options = webdriver.ChromeOptions()
        options.add_argument(f"load-extension={extension_location}")
        driver = None
        driver = webdriver.Chrome(options=options)
        driver.get("https://chromedriver.chromium.org/downloads")
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver.storage.googleapis.com')]")))
        driver.execute_script("""document.querySelector("a[href*='chromedriver.storage.googleapis.com']").setAttribute("target","_self")""")
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver.storage.googleapis.com')]").click()
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_linux64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_mac64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_win32.zip')]")))
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_linux64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_mac64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_win32.zip')]").click()
+       for _ in range(10):
+           if checking_conditions_download_ready_state_complete(driver):
+               break
+           time.sleep(1)
        time.sleep(1)
    except Exception as e:
        print(e)
    finally:
        driver = driver.quit() if driver is not None else None

    print()
    print("PRESS ENTER KEY TO EXIT")
    input()


if __name__ == "__main__":
    main()

PythonとSeleniumとWebDriverとChrome拡張機能でDownload完了判定と待機(関数内関数)

関数内関数を使用するように変更します。

sample.py
import datetime
import os
import time

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait


-def checking_conditions_download_ready_state_complete(driver):
+def checking_conditions_download_ready_state_complete():
+   def _predicate(driver):
        jikan = f"{f'{datetime.datetime.now()}'[:-4]}"
        state = driver.execute_script("return document.body.getAttribute('data-window-downloads-state')")
        _ = state == "in_progress" and print(f"{jikan}: download in progress...")
        _ = state == "interrupted" and print(f"{jikan}: download interrupted...")
        _ = state == "complete" and print(f"{jikan}: download complete!!")
        _ = state == "unknown" and print(f"{jikan}: download unknown...")
        _ = state == "none" and print(f"{jikan}: download none...")
        _ = state is None and print(f"{jikan}: error...")
        return bool(state == "complete")

+   return _predicate


def main():

    print("PRESS ENTER KEY TO BOOT")
    input()

    driver = None

    try:
        # extension_location = os.getcwd()
        extension_location = f"{os.getenv('USERPROFILE')}\\Desktop\\window.downloads.state"
        options = webdriver.ChromeOptions()
        options.add_argument(f"load-extension={extension_location}")
        driver = None
        driver = webdriver.Chrome(options=options)
        driver.get("https://chromedriver.chromium.org/downloads")
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver.storage.googleapis.com')]")))
        driver.execute_script("""document.querySelector("a[href*='chromedriver.storage.googleapis.com']").setAttribute("target","_self")""")
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver.storage.googleapis.com')]").click()
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_linux64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_mac64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_win32.zip')]")))
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_linux64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_mac64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_win32.zip')]").click()
        for _ in range(10):
-           if checking_conditions_download_ready_state_complete(driver):
+           if checking_conditions_download_ready_state_complete()(driver):
                break
            time.sleep(1)
        time.sleep(1)
    except Exception as e:
        print(e)
    finally:
        driver = driver.quit() if driver is not None else None

    print()
    print("PRESS ENTER KEY TO EXIT")
    input()


if __name__ == "__main__":
    main()

PythonとSeleniumとWebDriverとChrome拡張機能でDownload完了判定とWebDriverWait().until()による待機

待機の処理をWebDriverWait().until()を使用するように変更します。

sample.py
import datetime
import os
import time

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.wait import WebDriverWait


-def checking_conditions_download_ready_state_complete():
+def expected_conditions_download_ready_state_complete():
    def _predicate(driver):
        jikan = f"{f'{datetime.datetime.now()}'[:-4]}"
        state = driver.execute_script("return document.body.getAttribute('data-window-downloads-state')")
        _ = state == "in_progress" and print(f"{jikan}: download in progress...")
        _ = state == "interrupted" and print(f"{jikan}: download interrupted...")
        _ = state == "complete" and print(f"{jikan}: download complete!!")
        _ = state == "unknown" and print(f"{jikan}: download unknown...")
        _ = state == "none" and print(f"{jikan}: download none...")
        _ = state is None and print(f"{jikan}: error...")
        return bool(state == "complete")

    return _predicate


def main():

    print("PRESS ENTER KEY TO BOOT")
    input()

    driver = None

    try:
        # extension_location = os.getcwd()
        extension_location = f"{os.getenv('USERPROFILE')}\\Desktop\\window.downloads.state"
        options = webdriver.ChromeOptions()
        options.add_argument(f"load-extension={extension_location}")
        driver = None
        driver = webdriver.Chrome(options=options)
        driver.get("https://chromedriver.chromium.org/downloads")
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver.storage.googleapis.com')]")))
        driver.execute_script("""document.querySelector("a[href*='chromedriver.storage.googleapis.com']").setAttribute("target","_self")""")
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver.storage.googleapis.com')]").click()
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_linux64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_mac64.zip')]")))
        WebDriverWait(driver, 10).until(expected_conditions.element_to_be_clickable((By.XPATH, "//a[contains(@href,'chromedriver_win32.zip')]")))
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_linux64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_mac64.zip')]").click()
        driver.find_element(by=By.XPATH, value="//a[contains(@href,'chromedriver_win32.zip')]").click()
-       for _ in range(10):
-           if checking_conditions_download_ready_state_complete()(driver):
-               break
-           time.sleep(1)
+       WebDriverWait(driver, 10).until(expected_conditions_download_ready_state_complete())
        time.sleep(1)
    except Exception as e:
        print(e)
    finally:
        driver = driver.quit() if driver is not None else None

    print()
    print("PRESS ENTER KEY TO EXIT")
    input()


if __name__ == "__main__":
    main()

おわりに

downloads.Stateを利用してのダウンロードの完了判定と待機は上記で完成です。
最終的なコードはこちらを参照くださいませ。

ただ実は、
Chrome拡張機能のダウンロードに関する情報がダウンロード完了(complete)になった瞬間というのは、
Chrome拡張機能の現状の実装ですとダウンロードファイルの終端(EOF)を取得完了した瞬間のようで、
つまり、ダウンロードファイルの拡張子がリネームされる前のようなのです。

sample.py
        WebDriverWait(driver, 10).until(expected_conditions_download_ready_state_complete())
        time.sleep(1)
sample.py
        WebDriverWait(driver, 10).until(expected_conditions_download_ready_state_complete())
        driver.quit()

上段の記載の部分を、
下段の記載に変更して、
ダウンロード完了判定直後にブラウザを閉じるようにすると、
ダウンロードファイルの拡張子がリネームされる前の状態で終了してしまいます。ですので、
ダウンロードファイルの拡張子のリネームをChromeが完了するまでの時間の待機が必要です。
time.sleep(1)はそのためのコードです。
ただし、
ダウンロードファイルが多いとかの場合、
time.sleep(?)の指定した時間内に終わらないかもしれません...

Chrome108?で修正されたのかもしれません。
Chrome108?で修正されたのかもしれませんが、
上記の判定と待機も実施をする方法を投稿しました。

あとがき

ノンプログラマーの素人が記述をしたコードです。
狭い利用範囲と少ない利用頻度での確認ですので、
記載内容に間違いや勘違いがあるかもしれません。
上記内容を参照の際は自己責任でお願い致します。

7
4
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
7
4