Python

バージョン情報 pyenv 2.4.0 Poetry 1.8.2 Python 3.11.9 定義・ディレクトリ構成 説明のため、プロジェクトディレクトリ名my_project、パッケージ名my-project、主要なモジュール名my_projectとします。 以下のようなディレクトリ構成にすることを想定しています。 - my_project/ - pyproject.toml - Dockerfile - my_project/ - __init__.py - __main__.py - cli.py - tests/ - __init__.py - test_my_project.py Python/Poetryのインストール pyenvでPythonをインストールします。 記事作成時点で最新のリビジョン（0.0.x）を記載していますが、適宜新しいバージョンが出ているか確認し、 更新してください。 マイナーバージョン（0.x.0）を変更する場合、依存する予定のライブラリが動作するかなど、プロジェクトの要件と相談してください。 env PYTHON_CONFIGURE_OPTS="--enable-shared" pyenv install 3.11.9 PYTHON_CONFIGURE_OPTS="--enable-shared"は、PyInstallerが動作するようにするために設定しています。 pyenv and PyInstaller — PyInstaller 6.5.0 documentation Poetryをインストールします。 Poetry Installation # Linux, macOS, WSL curl -sSL https://install.python-poetry.org | python3 - # Windows (PowerShell) (Invoke-WebRequest -Uri https://install.python-poetry.org -UseBasicParsing).Content | python - Poetryプロジェクトの作成 Poetryのグローバル設定を変更し、Python仮想環境がプロジェクトのディレクトリ/.venvに作成されるようにします。 これは、VSCode拡張機能のPylanceがPython仮想環境を認識できるようにする、または手動で設定しやすくするための変更です。 ...

Selenium 4.9.0 Chrome 112 ChromeDriver 112.0.5615.49 Python 3.11 import time import json from selenium.webdriver import ( Chrome, DesiredCapabilities, ) desired_capabilities = DesiredCapabilities.CHROME desired_capabilities["goog:loggingPrefs"] = { "performance": "ALL", } driver = Chrome( desired_capabilities=desired_capabilities, ) driver.implicitly_wait(5) driver.get("https://www.google.com/") known_url_set = set() while True: performance_log_entries = driver.get_log("performance") for log_entry in performance_log_entries: log_message = json.loads(log_entry.get("message", "{}")).get("message", {}) method = log_message.get("method") params = log_message.get("params", {}) if method == "Network.responseReceived": response = params.get("response", {}) url = response.get("url") if url in known_url_set: continue known_url_set.add(url) print(url) time.sleep(1) Python+SeleniumでChromeデベロッパーツールのNetworkタブ相当の情報を取得する - Qiita java - Using Selenium how to get network request - Stack Overflow

Selenium 4.9.0 Chrome 112 ChromeDriver 112.0.5615.49 Python 3.11 from selenium.webdriver import ( Chrome, ChromeOptions, ) download_dir = "./downloads" os.makedirs(download_dir, exist_ok=True) options = ChromeOptions() options.add_experimental_option("prefs", { "profile.default_content_settings.popups": 0, "download.default_directory": os.path.realpath(download_dir), "download.prompt_for_download": False, "download.directory_upgrade": True, }) driver = Chrome( options=options, ) python - How to change download directory location path in Selenium using Chrome? - Stack Overflow

MATVToolに組み込むかもしれませんが、いまのところ詳細な動作検証をするほど需要がないので、簡易的にここに置いておきます。 動画ファイルによっては、フレームの欠け、重複が発生したり、変換に失敗するかもしれません。 Python 3.11.3 FFmpeg 4.2.7-0ubuntu0.1 Ubuntu 20.04 (WSL2) python3 main.py input.mp4 output.mp4 # License: CC0-1.0 import os import subprocess import tempfile import re import math def get_duration_seconds(input_file: str) -> float: proc = subprocess.run( [ 'ffmpeg', '-hide_banner', '-i', input_file, ], stderr=subprocess.PIPE, ) lines = proc.stderr.decode(encoding='utf-8').splitlines() # hh:mm:ss.ff duration_string = None for line in lines: m = re.match(r'^\s*Duration:\s(.+?),.*$', line) if m: duration_string = m.group(1) break assert duration_string is not None hours = int(duration_string[0:2]) minutes = int(duration_string[3:5]) seconds = int(duration_string[6:8]) milliseconds = float('0.' + duration_string[9:11]) return hours * 3600 + minutes * 60 + seconds + milliseconds def parse_time(string: str) -> float: """ string: HH:MM:SS.FF """ hours = int(string[0:2]) minutes = int(string[3:5]) seconds = int(string[6:8]) milliseconds = float('0.' + string[9:11]) return hours * 3600 + minutes * 60 + seconds + milliseconds def format_time(seconds: int) -> str: hours_minutes = seconds // 60 hours = hours_minutes // 60 minutes = hours_minutes - hours * 60 local_seconds = seconds - hours_minutes * 60 return f'{hours:02d}:{minutes:02d}:{local_seconds:02d}' def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument('input_file', type=str) parser.add_argument('output_file', type=str) parser.add_argument('--split_duration', type=int, default=10) args = parser.parse_args() input_file = args.input_file output_file = args.output_file split_duration = args.split_duration duration_seconds = math.ceil(get_duration_seconds(input_file=input_file)) count = math.ceil(duration_seconds / split_duration) work_dir_obj = tempfile.TemporaryDirectory() work_dir = work_dir_obj.name part_output_file_list = [] for index in range(count): start = index * split_duration end = start + split_duration start_string = format_time(start) end_string = format_time(end) print(index, start_string, end_string) part_output_file = os.path.join(work_dir, f'output{count-index}.mp4') subprocess.run([ 'ffmpeg', '-hide_banner', '-ss', start_string, '-to', end_string, '-i', 'input.mp4', '-vf', 'reverse', '-af', 'areverse', part_output_file, ]) part_output_file_list.append(part_output_file) list_file = os.path.join(work_dir, 'list.txt') with open(list_file, 'w', encoding='utf-8') as fp: for part_output_file in part_output_file_list: fp.write(f"file '{part_output_file}'\n") subprocess.run([ 'ffmpeg', '-hide_banner', '-f', 'concat', '-safe', '0', '-i', list_file, '-c', 'copy', output_file, ]) if __name__ == '__main__': main() 参考 How to Reverse a Video using FFmpeg - OTTVerse 映像と音声を逆再生にエンコードする | ニコラボ FFMPEGで動画を逆再生して保存する方法 | 技術的特異点 Concatenate – FFmpeg

Python 3.9.13 (pyenv) Ubuntu 20.04 (WSL2) https://github.com/aoirint/python_asyncio_examples なんかこう書くとうまく動くということしかわからん、という実装例です。 https://twitter.com/aoirint/status/1544891079786627074 基本実装 同期関数から非同期関数を同期的に呼び出す asyncio.run Details import asyncio import time def main(): async def func(): await asyncio.sleep(3) print('func exited') # 1 asyncio.run(func()) time.sleep(1) print('main exited') # 2 main() print('exited') # 3 同期関数から非同期関数を非同期的に呼び出す threading.Thread + asyncio.run Details import asyncio from concurrent.futures import ThreadPoolExecutor import threading import time def main(): async def func(): await asyncio.sleep(3) print('func exited') # 3 thread = threading.Thread(target=lambda: asyncio.run(func())) thread.start() time.sleep(1) print('main exited') # 1 main() print('exited') # 2 非同期関数から同期関数を同期的に呼び出す ふつうに呼び出す Details import asyncio import time async def main(): def func(): time.sleep(3) print('func exited') # 1 func() await asyncio.sleep(1) print('main exited') # 2 asyncio.run(main()) print('exited') # 3 非同期関数から同期関数を非同期的に呼び出す asyncio.new_event_loop + ThreadPoolExecutor + EventLoop.run_in_executor Details import asyncio from concurrent.futures import ThreadPoolExecutor import time async def main(): def func(): time.sleep(3) print('func exited') # 3 loop = asyncio.new_event_loop() executor = ThreadPoolExecutor() loop.run_in_executor(executor, func) await asyncio.sleep(1) print('main exited') # 1 asyncio.run(main()) print('exited') # 2 非同期関数から非同期関数を同期的に呼び出す awaitキーワード Details import asyncio async def main(): async def func(): await asyncio.sleep(3) print('func exited') # 1 await func() await asyncio.sleep(1) print('main exited') # 2 asyncio.run(main()) print('exited') # 3 非同期関数から非同期関数を非同期的に呼び出す threading.Thread + asyncio.run Details import asyncio import threading async def main(): async def func(): await asyncio.sleep(3) print('func exited') # 3 thread = threading.Thread(target=lambda: asyncio.run(func())) thread.start() await asyncio.sleep(1) print('main exited') # 1 asyncio.run(main()) print('exited') # 2 非同期関数から同期間数を非同期的に3つずつ呼び出す asyncio.new_event_loop + ThreadPoolExecutor + EventLoop.run_in_executor Details import asyncio from concurrent.futures import ThreadPoolExecutor import time async def main(): def func(): time.sleep(3) print('func exited') # 3 loop = asyncio.new_event_loop() executor = ThreadPoolExecutor(max_workers=3) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) loop.run_in_executor(executor, func) await asyncio.sleep(1) print('main exited') # 1 asyncio.run(main()) print('exited') # 2 よく見るエラー RuntimeError: Event loop is closed TBW ...

想定: 3.xで動いていたフォーム送信が4.xへの更新でCSRF検証に失敗するために動かなくなった settings.pyにCSRF_TRUSTED_ORIGINSを追加すればよい。 https://docs.djangoproject.com/en/4.0/ref/settings/#std-setting-CSRF_TRUSTED_ORIGINS Origin: e.g. https://example.com https://developer.mozilla.org/ja/docs/Glossary/Origin CSRF検証のドキュメントを3.xと比べると、以下の記述が増えている。 CsrfViewMiddleware verifies the Origin header, if provided by the browser, against the current host and the CSRF_TRUSTED_ORIGINS setting. This provides protection against cross-subdomain attacks. 4.x: https://docs.djangoproject.com/en/4.0/ref/csrf/#how-it-works 3.x: https://docs.djangoproject.com/en/3.2/ref/csrf/#how-it-works

https://qiita.com/aoirint/items/09ea153751a65bf4876f#github-release%E4%BD%9C%E6%88%90%E6%99%82%E3%81%ABpypi%E3%81%AB%E3%82%A2%E3%83%83%E3%83%97%E3%83%AD%E3%83%BC%E3%83%89 上の記事のWorkflowテンプレートをちょっと改良した。 GitHub Release作成時にリリースタグをパッケージバージョンにしてpush 構成 mypackage/__init__.pyに以下のように開発用のバージョン情報を記述する。 リリース時にGithub Actionsでリリースタグに置換してからPyPIにpushする。 __VERSION__ = '0.0.0' GitHub Secrets PYPI_API_TOKEN GitHub Workflow .github/workflows/pypi.yml name: Publish a package to PyPI on: release: types: - created env: VERSION: ${{ github.event.release.tag_name != '' && github.event.release.tag_name || '0.0.0' }} jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Setup Python uses: actions/setup-python@v2 with: python-version: 3.x - name: Install Dependencies run: | pip3 install -r requirements.txt pip3 install wheel - name: Replace version run: | sed -i "s/__VERSION__ = '0.0.0'/__VERSION__ = '${{ env.VERSION }}'/" mypackage/__init__.py - name: Build Package run: python3 setup.py sdist bdist_wheel - name: Publish to PyPI uses: pypa/gh-action-pypi-publish@release/v1 with: user: __token__ password: ${{ secrets.PYPI_API_TOKEN }}

pip-compileは、pip-toolsパッケージ（jazzband/pip-tools）に含まれるコマンドで、 Pythonパッケージのリストrequirements.inから、 依存ツリーのライブラリバージョンリストrequirements.txtを出力してくれる便利なツールである。 入力例 requirements.in django requests gunicorn mysqlclient python-dateutil requests-oauthlib schedule pip-compile requirements.in 出力例 requirements.txt # # This file is autogenerated by pip-compile with python 3.8 # To update, run: # # pip-compile requirements.in # asgiref==3.5.2 # via django backports-zoneinfo==0.2.1 # via django certifi==2022.6.15 # via requests charset-normalizer==2.0.12 # via requests django==4.0.5 # via -r requirements.in gunicorn==20.1.0 # via -r requirements.in idna==3.3 # via requests mysqlclient==2.1.1 # via -r requirements.in oauthlib==3.2.0 # via requests-oauthlib python-dateutil==2.8.2 # via -r requirements.in requests==2.28.0 # via # -r requirements.in # requests-oauthlib requests-oauthlib==1.3.1 # via -r requirements.in schedule==1.1.0 # via -r requirements.in six==1.16.0 # via python-dateutil sqlparse==0.4.2 # via django urllib3==1.26.9 # via requests # The following packages are considered to be unsafe in a requirements file: # setuptools https://pypi.org/project/pip-tools/ （現在 pip-tools==6.6.2） https://github.com/jazzband/pip-tools ところでPythonパッケージは、システムパッケージの事前インストールを要求することがある。 以下、システムパッケージ名はDebian/Ubuntuを想定する。 ...

threading https://ja.stackoverflow.com/questions/60539/subprocess-popenのstdoutとstderrをリアルタイムに取得する # python 3.9 import subprocess from threading import Thread def main(): command = [ 'mycommand', '-opt1', ] proc = subprocess.Popen( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) def read_stdout(stdout): for line in stdout: line_text = line.decode('utf-8').strip() print(f'STDOUT: {line_text}', flush=True) print('stdout closed') # closed when process exited def read_stderr(stderr): for line in stderr: line_text = line.decode('utf-8').strip() print(f'STDERR: {line_text}', flush=True) print('stderr closed') # closed when process exited Thread(target=read_stdout, args=(proc.stdout,)).start() Thread(target=read_stderr, args=(proc.stderr,)).start() while True: returncode = proc.poll() if returncode is not None: break time.sleep(0.01) print(f'exited {returncode}') asyncio https://stackoverflow.com/questions/28492103/how-to-combine-python-asyncio-with-threads https://qiita.com/matsui-k20xx/items/4d1c00c4eefd60ba635b import time import asyncio from concurrent.futures import ThreadPoolExecutor async def main(): loop = asyncio.get_event_loop() executor = ThreadPoolExecutor(2) def operation(): time.sleep(1) print('aaa', flush=True) # 2個ずつ実行される（asyncio.runの実行終了まで2秒かかる） loop.run_in_executor(executor, operation) loop.run_in_executor(executor, operation) loop.run_in_executor(executor, operation) loop.run_in_executor(executor, operation) # main関数の実行はブロックされない print('before aaa') asyncio.run(main()) import time import asyncio from asyncio.subprocess import create_subprocess_exec, PIPE from concurrent.futures import ThreadPoolExecutor async def main(): command = [ 'mycommand', '-opt1', ] proc = await create_subprocess_exec( command[0], *command[1:], stdout=PIPE, stderr=PIPE, ) loop = asyncio.get_event_loop() executor = ThreadPoolExecutor() def read_stdout(stdout): while True: line = asyncio.run_coroutine_threadsafe(stdout.readline(), loop).result() if not line: break line_text = line.decode('utf-8').strip() print(f'STDOUT: {line_text}', flush=True) print('stdout closed') # closed when process exited def read_stderr(stderr): while True: line = asyncio.run_coroutine_threadsafe(stderr.readline(), loop).result() if not line: break line_text = line.decode('utf-8').strip() print(f'STDERR: {line_text}', flush=True) print('stderr closed') # closed when process exited loop.run_in_executor(executor, read_stdout, proc.stdout) loop.run_in_executor(executor, read_stderr, proc.stderr) await proc.wait() print(f'exited {proc.returncode}') https://stackoverflow.com/questions/45600579/asyncio-event-loop-is-closed-when-getting-loop 終了時にEvent loop is closedというエラーが出ることがある？ ...

Python 3.7以降の場合、標準モジュールimportlib.resourcesが利用できる。 https://docs.python.org/ja/3/library/importlib.html#module-importlib.resources 以下のようなディレクトリ構造でファイルを同梱する。 setup.py MANIFEST.in README.md LICENSE mymodule __init__.py mymodule.py mydirectory __init__.py myfile1.txt myfile2.bin 注意点 読み込むファイルのあるディレクトリに__init__.pyを作成する MANIFEST.inでファイルがパッケージに同梱されるように記述する include README.md include LICENSE recursive-include mymodule/mydirectory * import importlib.resources as ILR # テキストファイル ILR.read_text('mymodule.mydirectory', 'myfile1.txt', encoding='utf-8') # -> str # バイナリファイル ILR.read_binary('mymodule.mydirectory', 'myfile2.bin') # -> bytes