新着記事

Viewing posts for the category Python

Raspberry Pi Pico W で Httpサーバ(microdot)とセンサーによるHTTPリクエスト機能を同時に稼働させる

Raspberry Pi Pico W が発表されました。日本ではまだ未発売ですが、技適は取得されたようですので近いうちに国内販売がされそうです。

試しに、Webサーバ ( Microdot )とWebクライアント(urequest) を uasyncio で並列実行するコードを書きましたので、紹介します。

今回作成したコードや動作している動画は、Github で公開しています。

ytyng/rpi-pico-w-webserver-and-client: Raspberry Pi Pico W webserver and client sample code

Raspberry Pi Pico W とは

コストパフォーマンスが高いマイクロコントローラです。カテゴリとしては Arduino 等に近く、今までの Raspberry Pi のように、Linux OS を動作させるようなマシンではありません。

RP2040 というラズベリーパイ財団が開発したチップのデモボードという位置づけとなります。

実際の商品開発では、Raspberry Pi Pico で製品のR&Dを行い、実際は RP2040 を搭載した製品として生産するという流れとなると思いますが、ホビーや SOHOでは Raspberry Pi Pico をそのまま使うことも多いと思います。

実際、当社でも Raspberry Pi Pico を用いてイベント用の機材を作る場合がありますが、
RP2040 を使ったの製品を作るわけではなく、Raspberry Pi Pico をそのままケースに入れて使います。

MicroPython が動作するため、Python に慣れていれば開発は容易にできます。

商品名に W がつかない今までの機種は、ネットワーク機能はありませんでしたが、今回 Raspberry Pi Pico W となって無線チップが搭載され、コストパフォーマンスと使い勝手が最高の IoT デモボードとなりました。

↑ 左が Raspberry PI Pico, 右 が無線LAN チップが搭載された Raspberry Pi Pico W

考えられる用途

Raspberry Pi Pico W の用途で多く使われると考えられる用途は、

  • 接続させているデバイスのセンシング情報を元に、HTTP リクエストを発生させる
  • HTTP サーバを起動し、外部から HTTP リクエストを受け取って、接続されているデバイスを動作させる

この2つが主なものとなると考えられます。

今回は、この2つを Raspberry Pi Pico W の中で同時に実行する方法を書きます。

一通りのチュートリアル

Raspberry PI の公式ページが提供している PDF が充実います。

https://datasheets.raspberrypi.com/picow/connecting-to-the-internet-with-pico-w.pdf

ただ、PDF なので少し読みにくいのと、Thonny に関しては言及されていないため、Mac や Windows を普段使われている方は、この PDF だけでなく、他のサイトで紹介されているような Thonny を使ったセットアップを行うと良いでしょう。

ファームウェアの準備

https://micropython.org/download/rp2-pico-w/

上記 URL で、Raspberry Pi Pico 用の MicroPython ファームウェアの uf2 ファイルが入手できます。

最新版への直リンクはこちらです。 https://micropython.org/download/rp2-pico-w/rp2-pico-w-latest.uf2

ファームウェアのファイルをダウンロードし、 Pico へコピーしてください。

W 対応でない uf2 ファームウェアは別に存在します。そちらを使った場合、Wi-fi の機能が使えませんのでご注意ください。

コピー方法

Pico の BOOTSEL ボタンを押したまま USB で PC に接続すると、PCが Pico をストレージとして認識します。

ダウンロードした uf2 ファイルを Pico にドラッグアンドドロップでコピーすると、自動的にファームウェアがロードされ、 Pico が再起動します。

Wifi に接続する

一番最初に、Wifi に接続する必要があります。
SSID と パスワードが変数化されていれば、後は簡単なコードで接続が行えます。

接続用の関数を作っておくと便利で、他の方を見ても関数化しているようです。

StackOverflow の話題を見ると、Wifi との接続は main.py の中でやらずに boot.py の中でやったほうがいい、というコメントをいくつか見かけましたが、私は 開発のしやすさから main.py の中で行うようにしています。

Wi-fi に接続するコード

https://github.com/ytyng/rpi-pico-w-webserver-and-client/blob/main/network_utils.py

import rp2
import network
import uasyncio
import secrets


async def prepare_wifi():
"""
Prepare Wi-Fi connection.
https://datasheets.raspberrypi.com/picow/connecting-to-the-internet-with-pico-w.pdf # noqa
"""
# Set country code
rp2.country(secrets.COUNTRY)

wlan = network.WLAN(network.STA_IF)
wlan.active(True)

wlan.connect(secrets.WIFI_SSID, secrets.WIFI_PASSWORD)

for i in range(10):
status = wlan.status()
if wlan.status() < 0 or wlan.status() >= network.STAT_GOT_IP:
break
print(f'Waiting for connection... status={status}')
uasyncio.sleep(1)
else:
raise RuntimeError('Wifi connection timed out.')

# CYW43_LINK_DOWN (0)
# CYW43_LINK_JOIN (1)
# CYW43_LINK_NOIP (2)
# CYW43_LINK_UP (3)
# CYW43_LINK_FAIL (-1)
# CYW43_LINK_NONET (-2)
# CYW43_LINK_BADAUTH (-3)

wlan_status = wlan.status()

if wlan_status != network.STAT_GOT_IP:
raise RuntimeError(
'Wi-Fi connection failed. status={}'.format(wlan_status))

print('Wi-fi ready. ifconfig:', wlan.ifconfig())
return wlan

接続させているデバイスのスイッチ(センサー)情報を元に、HTTP リクエストを発生させる

Raspberry Pi Pico W Wi-Fi Doorbell tutorial (HTTP requests & IFTTT) — PiCockpit | Monitor and Control your Raspberry Pi: free for up to 5 Pis!

こちらの方が開発されているドアベルのコードが参考になります。
Youtube 動画もあってわかりやすいです。

urequests

Python でよく使う、requests ライブラリに変わり、MicroPython では似たような使い勝手の urequests ライブラリを使うことができます。

ネットワーク接続が確立されていれば、あとは urequests.get(...) 等で簡単にリクエストが発行できます。

Thonny の tools -> Manage packages からインストールできます。

HTTP サーバとして稼働させる

ソケットをそのまま使って簡易的な HTTP サーバにする

こちらの記事が参考になりました。大変わかりやすく日本語で説明されているので、Pico の初学者にもおすすめします。

Raspberry Pi Pico W で無線Lチカ

リンクされている、mimoroni 社のコードは、 Pico 用の拡張されたファームウェア

https://github.com/pimoroni/pimoroni-pico/releases

や、 Pico W 用の各種ユーティリティコードがあり、開発の参考になります。

このコードの HTTP サーバの部分は、TCP ソケットをそのまま使い、リクエスト本文の中のパス名と文字列一致して判定してい分岐を行っています。

Raspberry Pi 公式のチュートリアルPDFでもその方式で行っていました。

規模が小さいようであれば十分だと思いますが、HTTP ヘッダーを扱いたい場合や、少し規模を拡張したい場合はこの形では難しいでしょう。

Microdot を起動する

Flask や Bottle、fastApi が動けば良いのですが、現状は動作しません。
代わりに、Microdot という ウェブフレームワークがあり、使い勝手としては Flask や Bottle によく似ていて大変勝手が良いです。

こちらを使って ウェブサーバを起動してみます。

Microdot

Thonny で Tools -> Manage packages からインストールすることができます。

ネットワーク接続が確立したら、

app = Microdot()

@app.get('/')
async def _index(request):
return 'Microdot on Raspberry Pi Pico W'

app.run(port=80)

このような親しみやすいコードで HTTP サーバが起動します。

ウェブサーバとセンサーリクエストを同時に使う

Raspberry Pi Pico は、通常シングルスレッド動作です。(ちなみにCPUはデュアルコアです)

一応、 _threading という疑似スレッドができるライブラリはありますが、処理によっては本体が暴走したり固まることが多く、かなりおすすめしません。
暴走すると、最悪、何度もファームウェアのリセットをするこになり、開発体験は良くありません。

代わりに、 asyncio を使ったコルーチン処理を標準で行うことができ、こちらは安定して動作しますので、 Pico で開発する際は、基本的にメソッドはコルーチンで書くのをおすすめします。

Pico は、待機ループで sleep を使うことも多いですし、コルーチンと相性が良いと感じます。

Microdot も非同期対応の起動ができるものが既に開発されています。

Pico 上の MicroPython でのコルーチンは、通常の Python にビルトインされている asyncio を使うのではなく、
uasyncio というパッケージを使います。

Pico 用の uf2 ファームウェアに含まれていますので、別途新たなインストールは必要ありません。

例えば下記のようなコードで、asyncio が有効な処理を開始することができます。

import uasyncio

async def main():
uasyncio.create_task(any_async_method())
await other_async_method()


if __name__ == '__main__':
    uasyncio.run(main())

Pico の起動後、無線 LAN に接続した後は、スイッチ押下待機のループと、Micorodot の起動を
両方ともコルーチンで書くことで、無理なく並列動作をさせることができます。

実際に動作するコードは Github で公開しています。

rpi-pico-w-webserver-and-client/main.py at main · ytyng/rpi-pico-w-webserver-and-client

メインのコードとしてはこのようになります。

"""
Raspberry Pi Pico Web Server with Microdot and Switch Sample Code.
Pin 14 is used for switch input.
"""
import machine
import urequests
import network_utils
from microdot_asyncio import Microdot
import uasyncio


async def switch_loop():
"""
Switch listener loop
Pin 14 is used for switch input.
When press switch, send request to http web server.
"""
print('start switch_loop')
switch_pin = machine.Pin(14, machine.Pin.IN, machine.Pin.PULL_DOWN)

while True:
current_state = switch_pin.value()
if current_state:
# Change the URL to your own server, IFTTT, Slack, etc.
response = urequests.get('https://example.com/')
print(response.content)
response.close()
await uasyncio.sleep(2)
else:
await uasyncio.sleep(0.1)

async def run_web_server():
"""
Start microdot web server
https://microdot.readthedocs.io/en/latest/index.html
"""
app = Microdot()
led_pin = machine.Pin('LED', machine.Pin.OUT)

@app.get('/')
async def _index(request):
return 'Microdot on Raspberry Pi Pico W'

@app.get('/led/<status>')
async def _led(request, status):
"""
/led/on : LED ON
/led/off : LED OFF
"""
if status == 'on':
led_pin.on()
return 'LED turned on'
elif status == 'off':
led_pin.off()
return 'LED turned off'
return 'Invalid status.'

print('microdot run')
app.run(port=80)


async def main():
wlan = await network_utils.prepare_wifi()
print('LED ON: http://{}/led/on'.format(wlan.ifconfig()[0]))

uasyncio.create_task(switch_loop())
await run_web_server()


if __name__ == '__main__':
uasyncio.run(main())

Web ページの動作検証のためのボットスクリプトを Windows 上で作る

この文書は、開発経験の無いチームがウェブアプリケーションの動作検証の責任を持つケースで、検証を簡単なプログラムで行うアプローチについての手法を解説しています。

Webアプリケーションの動作検証の際、手動で実行する以外にプログラムで検証すると便利です。開発者であればは検証コードを書くのは簡単ですが、開発経験の無い方はどこから始めたらいいかわからないと思いますので、比較的用意にスクリプトに入門できるように紹介します。OS は Windows を対象としています。

本記事で紹介しているようなプログラムによるリクエストは悪意の有無にかかわらず、不正アクセス禁止法での不正アクセスとみなされたり、電子計算機損壊等業務妨害罪等に問われる可能性があります。実際にリクエストを行う場合は、自社の管理する、許可されたサーバに対してのみ行うようにしてください。

HTTP の リクエスト・レスポンスの仕組みを知る

まずは、ウェブサーバと通信している HTTP のリクエストがどのようなものか知ることが必要です。この知識があいまいなままだとボットのスクリプトは書けません。

まずは、ウェブブラウザに搭載されている開発者ツールを使ってリクエストやレスポンスを観察するのが、良い勉強になります。

ウェブサーバに対して、

  • どの URL に対して
  • どのHTTP メソッドで (GET, POST, HEAD, PUT 等)
  • どのような HTTP ヘッダーで
    • cookie
    • referer
    • user-agent
  • どのようなリクエスト本文(body)で

以上を意識して、ブラウザの行っているリクエストをスクリプトで再現できれば、どのようなクライアントを使おうが、ウェブサーバからはブラウザでのリクエストと同じようにレスポンスが返ってきます。

Chrome のデベロッパーツールの使い方

  1. Google Chrome を起動してください。
  2. Shift + Ctrl + I を押してください。
    右側にデベロッパーツールが表示されます。
  3. デベローパーツールの上部のタブで、Network を選択してください。
  4. ブラウザの URL 欄に適当にページを打ち込み、ページを表示されてください。
  5. リクエスト一覧が表示されますので、適当なリクエストをクリックして選択してください。
  6. Headers にリクエストヘッダ、レスポンスヘッダ
    Payload にリクエスト本文
    Response にレスポンス本文
    が確認できます。
    特に、Headers のリクエストヘッダでどのようなヘッダを送っているか、確認してください。

Headers の中でも、cookie は特に注目してください。cookie の扱いはボット作成の中で最上位に重要な項目です。不安があれば、他のサイトや書籍を参考に学習してください。

また、 cookie の内容は認証情報を含むため、サイトのログインパスワードと同じぐらい重要な秘密情報です。安易にコピーは行わず、また絶対に他者に教えないようにしてください。ブラウザの cookie を格納している場所は安全ですが、他の場所にコピペすると漏洩リスクとなり、ログイン権限を奪われる危険性があります。

HTTPヘッダとクッキーの学習ができるサイト

HTTP ヘッダ、クッキーについての概要は、とほほ先生のサイト、わわわ先生のサイト、あと Wikipedia を読めば理解できます。

HTTPヘッダ

Cookie

ブラウザを手動で操作する以外での HTTP リクエストを行う方法

HTTP リクエストは、結局は特定の文字をサーバに送るだけですので、様々なクライアントで行うことができますが、よく使われるものを紹介します。

Postman

https://www.postman.com/
フリーの Windows クライアントがあります。GUI でリクエストを構築することができますし、スクリプトによる制御処理も書けます。有用なツールですが、最初は機能が多く複雑なため戸惑うかもしれません。

長所
  • GUI で完結する
  • 大量のテストリクエストの管理がしやすい
  • 署名の計算等、複雑な計算を伴う処理も行える。
短所
  • 複雑なため習得が難しい
  • 単純なリクエストを出すだけであれば冗長

curl

mac や linux を使う方には定番のコマンドラインツールです。Windows 10 からは標準でインストールされています。

長所
  • インストール済みなのですぐに使える
  • 単純なリクエストを1発出すだけなら一番適している
短所
  • 計算を伴う順次リクエストには向かない

Selenium や Puppeteer

プログラムで Chrome などのブラウザを実際に起動し、自動操作するための Selenium や Puppeteer といったツールがあります。

ブラウザを起動するため、ページ内で JavaScript を豊富に扱うページも自動操作し、検証することができます。

最近のウェブサイトは、React や Vue といった JavaScript を用いて表現するサイトが多くなってきており、その場合は Postman, curl, 後述するシンプルなスクリプトでは十分な検証ができない場合がありますので、 ページ内の JavaScript の動作検証が必要になる場合実際のブラウザを実行させる以外に無く、自動操作するには Selenium や Puppeteer を使うしかありません。

扱うには高度なプログラミング知識が必要ですので、今回は言及しません。

長所
  • ブラウザでの JavaScript の実行が必要であればこれ一択
  • 表示レイアウトの確認にも使える
短所
  • 実行環境の構築が難しくて、手間がかかる。
  • 他の、単純なリクエストを出す方式に比べると遅い。
  • 様々な要因により安定させて動作させるのは難しい。業務レベルで使うには高い技術が必要。

Rest Client ( .http)

ドットエイチティーティーピーファイルを作り、その記述したリクエストを簡易に何度も再現させることができます。VSCode や JetBrains エディタの機能として扱うことができ、記述が簡単で読みやすいため私は(当社内でも)かなり使います。

長所
  • 習得が容易
  • スクリプトの構文が容易で書きやすく読みやすい
  • スクリプトのチーム内共有が容易
  • レスポンスへの簡易的なテストを行うことができる
短所
  • 計算を伴う複雑な逐次処理はできない

プログラミング言語でスクリプトを組む

Python, PHP, Javascript, Ruby などで、既にある便利な HTTP クライアントライブラリを使ってスクリプトを組む方法です。
今回の記事ではこちらを今回紹介します。

長所
  • 条件分岐を含む複雑な制御処理が得意
  • 複数セッションによる並列リクエストを再現したい場合は一択
  • 完全無人での自動実行が容易
  • 書いたコードの再利用が容易
  • 処理結果の外部ツールへの連携が柔軟に行える
短所
  • 環境構築が手間
  • 単発のリクエストを検証したい場合は RestClient や Curl, Postman と比べても冗長

Windows に Python 実行環境をインストールする

Microsoft Store からの Python のインストール

Microsoft Store で Python パッケージが提供されるようになり、昔と比べて環境構築が楽になりました。

Microsoft が提供している、初心者向けの Python の開発ガイドが良くできています。この流れにそって進めれば問題なく進められますので、こちらも参考にしてください。

Microsoft Store のアプリを開き、 Python を検索して Python 3.10 をインストールします。

インストール後、念の為再起動を行い、その後コマンドプロンプトを起動して python と打ち込んで、 python が起動するか確かめてください。

Python 公式サイトからの Python のインストール

コマンドプロンプトで、 python と打ち込んで python が起動しないようであれば、Microsoft Store からインストールした Python はアンインストールし、 Python の公式サイトから Windows 版のインストーラパッケージをダウンロードしてインストールしてください。

その際、PATH を追加編集するかのオプションが表示されるので、チェックを入れてください。

https://www.python.org/downloads/

requests ライブラリのインストール

Python の インストールが完了したら、ウェブを操作するボットの作成に必須ともいえる、 requests ライブラリをインストールします。

コマンドプロンプトや PowerShell で、

python -m pip install requests

と入力すると、インストールが完了します。

Visual Studio Code のインストール

エディタは Visual Studio Code を使います。

Microsoft Store から Visual Studio Code を検索してインストールしてください。

プロジェクトフォルダの準備

Windows の、ドキュメントフォルダの下にtest-bot フォルダを作ってください。

VSCode を起動し、 File -> Open Folder で test-bot フォルダを開いてください。

フォルダを開いたら、左側のペインで右クリックし、 New File から first_bot.py というファイルを作ってください。

作成後、右下に「インタープリターを選択」と表示されているようであればクリックしてください。Microsoft Store からインストールした Python が、おすすめに表示されているので選択します。

右下に CRLF と表示されている箇所は、改行コードの設定が表示されています。CRLF は一般的ではないため、クリックして LF に変更しておきます。

Python の機能拡張のインストールがおすすめされると思いますので、Python, Pylance の機能拡張をインストールします。

スクリプトを書く

レスポンス本文を表示するだけのスクリプト

import requests
response = requests.get('https://www.torico-corp.com/')
print(response.text)

これは、 https://www.torico-corp.com/ のレスポンス本文を表示するだけの単純なプログラムです。
書いたら、右上の ▶ ボタンを押して、実行させてください。
出力結果がずらっと表示されます。

requests ライブラリについてのドキュメント

レスポンスの経過時間とステータスコードを表示するスクリプト

import requests
response = requests.get('https://www.torico-corp.com/')
print('経過時間 {}ms'.format(response.elapsed.microseconds / 1000))
print('ステータスコード {}'.format(response.status_code))

このスクリプトは、レスポンスの応答時間とステータスコードをコンソールに表示します。

Python に慣れてきたら、結果をファイルに記録するように改修することで、簡易的な負荷監視などに応用できます。

サイトの検索結果ページからを解析するスクリプト

Webサイトの検索ページにリクエストを行い、結果の HTML をパースしてコンソールに表示するスクリプトです。

HTMLをプログラムで扱えるように解析するために、 BeautifuSoup というライブラリをインストールします。

BeautifulSoup のインストール方法

コマンドプロンプトで

python -m pip install beautifulsoup4

でインストールできます。

BeautifulSoup の解説記事

コード

import requests
from bs4 import BeautifulSoup
response = requests.get('https://tech.torico-corp.com/search/?q=docker')
soup = BeautifulSoup(response.content, features='html.parser')

for h2 in soup.find_all('h2'):
a = h2.find('a')
if not a:
continue
print(a.text)
print(a['href'])

上記スクリプトは、TORICO の技術開発ブログを「docker」で検索し、出てきた記事のタイトルとリンクURL を表示しています。

メールアドレスとパスワードでウェブサイトにログインする

最後に、メールアドレスとパスワードでログインをするスクリプトの雛形を記載します。

requests ライブラリは、クッキー管理を行うことのできる Session というしくみがありますので、それを使います。

解説記事

検証するサイトによりますが、多くの場合は、ログイン時に「CSRFトークンの検証」と「Refererヘッダの検証
User-Agent が悪質なボットでないかの検証」を行っていると思いますので、そこを考慮してスクリプトを作れば、ログインが行えるはずです。

下記のような自動ログインのスクリプトは、必ずご自身が権限を持つサーバにのみ行うようにしてください。他者のサーバに行うと罪に問われる可能性があります。

URL 等は架空のものです。

import requests
from bs4 import BeautifulSoup

# session を作る
s = requests.session()
# User-Agent を設定する場合
s.headers['User-Agent'] = 'Tester Python Bot'

# ログインフォームを取得する
response = s.get('https://example.com/login-form/')

# HTTP のステータスコードに異常が無いか確認
response.raise_for_status()

# ログインフォームをパースする
soup = BeautifulSoup(response.content, features='html.parser')

# パースしたログインフォームから CSRF トークンを取得する
csrf_token = soup.find('input', {'name': 'csrftoken'})['value']

# ユーザー名とパスワードをいれてログインフォームを送信する。
response = s.post(
'https://example.com/login-form/',
data={
'email': 'tester@example.com',
'password': 'MY_AWESOME_PASSWORD',
# 先程取得した CSRF トークンを付与
'csrftoken': csrf_token
},
headers={
# Referer を付与
'Referer': 'https://example.com/login-form/',
})

# HTTP のステータスコードに異常が無いか確認
response.raise_for_status()

# ログイン後の URL が正しいものであるか確認
assert response.url == 'https://example.com/mypage/'

# この時点で、セッション s はログイン済みの状態なので、
# マイページ等をリクエストすることが可能
response = s.get('https://example.com/mypage/myprofile/')

ジュニアエンジニアの業務内容

エンジニアの高津です。
今回はこの1ヶ月でどのような業務を行ったのか紹介していきたいと思います。
TORICOにエンジニアとして入社を検討している人に少しでも参考になれば幸いです。

主な業務内容

自分はコーマス開発部で、漫画全巻ドットコムの開発をメインでやっています。

簡単なバグ(UI)の修正漫画全巻ドットコムのリニューアルの大きく2つに分けられます。

簡単なバグ(UI)の修正

こちらは数時間で終わるような簡単なUIの修正(改修)で入社して3日後にはプルリクエストを出していました。

  • アイコン(font-awesome)が正しく表示出来るようにする
  • 個人情報同意フォームの改修
  • 電話番号記入欄のに数字が4文字入るようにする
    などの業務を行いました。

漫画全巻ドットコムのリニューアル

漫画全巻ドットコムは10年以上続く歴史のあるサービスで最初はPHPで作られていました。
メンテナンス性に問題があったのでDjango+Nuxt.jsにリプレイスしています。
今回自分は電子新着ページ電子割引ページをリニューアルしました。(ブログを書いている時点で実装は終わっていますがレビューが終わってないので本番にはまだ反映されていません)
Nuxt(フロントエンド)は先輩方が作ってくれた雛形を軽く修正して利用出来るのでDjnago(バックエンド)の実装がメインでした。
今回はその中でも難しかったポイントをいくつか列挙したいと思います。

キャッシュを効かす

同じ値を取得して返すだけなのに毎回SQLを叩くのは無駄なのでkye-value型のNoSQLであるredius(メモリ)に一定時間値を保管し、値がキャッシュされていなければSQL等を叩く処理を行います。(keyは引数、valueは返り値で保存)
ページ単位(view)単位でキャッシュする方法とクラスメソッド単位でキャッシュする方法の2パターンあります。
カテゴリー別の作品数を取得する処理はページ間(異なるurl)でも共通したしょりなのでクラスメソッド単位でキャッシュする必要があり、少々手こずりました。

SQLの実行回数(IO)を極力少なくしパフォーマンスをあげる

今回一番苦戦しましたポイントです。
ただ実装するだけであればすぐ終わったのですが、最初の実装ではSQLを12回叩いてしまっていたのでリファクタリングする必要がありました。(俗に言うN+1問題が発生していました)
こちらはやり方を先輩方にご教示頂き、MySQLにだけサポートされているconcat関数をraw_queryで使い1回で取得することに成功しました。
実際には以下のようなSQLに落ち着きました。

select GROUP_CONCAT(sample_id) AS ids, group_type from
dtb_sample WHERE aggregate_type=
'%s' AND product_type = 2
group by group_type ORDER BY sort_key;

テスト

TORICOの開発ではただ動くものを作るだけでなくその後の保守運用のことも考慮して単体テスト、統合テストも書くように徹底されています。
特にDB設計が少々複雑なこともありテストデータを作るところはかなり苦戦しました。

具体的には以下のようなことをテストしました。

  • 各URLにgetして正しい値やstatusが返ってくるか
  • 各メソッドのすべての条件分岐において正しく動作するかどうか
これらのテストを書くことでテストのしずらいメソッドが見つかり、それをリファクタリングすることで保守性の高いコードに改善出来ます。

また、仕様が分からない人がみても理解できるようにWhy「なぜこの処理を書くのか?」を極力書くように意識しました。

まとめ

如何でしたでしょうか?

今現在23卒のエントリーを受け付けています。
少しでも興味を持ってくれた人は是非応募して頂けると幸いです。

実務経験で出会った便利なあれこれ

新卒エンジニアとして半年間働いてきて、現場でさまざまなことを勉強させていただきました。その中でも、もっと早く知っておきたかった便利なツール、Python の書き方など、幅広いあれこれを記事にしたいなと思います。

あれこれ、というざっくりとした括りになってしまっているのでまずは紹介したいものを示しておきたいと思います。

  • git submodule
  • Fabric3
  • Flake8
  • Python
    • 型ヒント
    • 整形文字列 f-string

git submodule

git submodule とは、git のレポジトリをサブモジュール化して複数のプロジェクトで共有することができる機能のことです。これを使うことで一度開発したコードを別のプロジェクトで簡単に再利用できます。例えばTORICOでは、 TORICO-ID のソーシャルログイン機能や決済機能だけでなく、ユーティリティ関数などを Django, Vue, Flutter それぞれのフレームワークでまとめたレポジトリが用意されてあります。

導入も至ってシンプルです。プロジェクト内で

git submodule add <リポジトリURL> <インストールパス>

たったこれだけで、今まで開発してきたコードを再利用することができます。わざわざプロジェクト間でコピペをしてくる必要はありません。さらに、git submodule はバージョン管理もしてくれます。サブモジュールを更新しても影響があるのはそのプロジェクト内だなので安心してサブモジュールの改良をすることができます。

1点注意として、submodule はメインのプロジェクト内で git clone や git pull を行っても自動で更新されることはありません。そのため、

git submodule update (初回のみ -i オプションで初期化)

コマンドを忘れずに実行する必要があります。

Fabric3

こちらは Python のモジュールの一つで、所定のアクションをコマンド一つで呼び出すことができます。例えば、デプロイするときの一連の流れを Python コードにしておくことで、ターミナルに fab deploy と入力するだけでその流れを自動で行ってくれるようになります。流れは以下のように記述します。

env.hosts = ['app1.sample.com', 'app2.sample.com']
def deploy():
    #  通知などを行う
    slack_announce('deploy')
    with cd ('/var/src'):
        run('git checkout master')
        run('git pull origin master')
        run('git submodule update')

このように一連の流れをコードとして残しておくことで、それをコマンド一つで呼び出すことができるようになります。さらに、新規に加入したメンバーでも簡単にデプロイができるというメリットがあります。その環境に慣れていない人が、例えば git submodule などのコマンドを忘れる心配もありません。また、私のような実務経験の無い人でもコードを確認することで流れを理解することができます。

TORICO では上記の git submodule を使ってほぼ全てのプロジェククトに deploy, ssh, flake8(後述), dsh (docker 環境にSSH)のコマンドが用意されています。

Flake8

こちらも Python のモジュールで、自動でコードレビューをしてくれます。導入は pip でインストールするだけです。実行は flake8 コマンドとオプションをターミナルにします。

flake8 --exclude="*migrations/*,venv/*,.venv/*,~* .

flake8 result

すると、このようにコードレビューをして規約に反する部分を教えてくれます。規約コードも教えてくれるため、わからない部分は検索できるようになっています。また、どうしても諸事情で変更できない場合は

from django.contrib import admin  # NOQA: F401

とコメントをつけることで、指定した規約コードの違反を行単位で無視させることもできます。

TORICO では毎回オプションを入力するのは面倒なため、上の Fabric3 を使って実行を簡単にしています。

Python 型ヒント

Python3.5 から導入された機能で型ヒントというものがあります。これは引数や帰り値の型をコードに書くことで可読性を向上させることができる機能です。

def hello(name: str) -> str:
    return f'hello, {name}.'

型ヒントはあくまでも補助的な役割のため、宣言とは異なる型を渡したところでエラーになることはありません。ただ、PyCharmでは型ヒントとは違う型を渡すと警告をしてくれます。可読性が飛躍的にあがるので、ぜひとも書くことを癖にしたいなと思っています。

Python F文字列

上の例でもしれっと使っていましたが、F文字列を使用することで .format 部分を短縮することができます。さらに、f文字列では変数だけでなく式なども使用できます。

print({a} + {b} = {a + b})  # format では a + b はエラーになる

format で記述するとコードが長くなる傾向があるので、スッキリと書けるf文字列は積極的に使用しています。

最後に

今回は現場で知った便利なあれこれを記事にしました。個人で勉強をしているとフレームワークや言語の知識は増やせますが、運用に関わる部分はなかなか知ることができないと思います。便利だなと思ったら、ぜひ積極的に使用してみてください。

新卒エンジニアが今になって就職前にやっておけば良かったと思うこと3選



就社してからは初めてのブログ投稿となります。
お久しぶりです。開発部の鈴木海人です。

株式会社TORICOにエンジニアとして入社して、半年が経ちました。
今回のブログでは、過去の内定をいただいてから入社までの間に対して何もしなかった自分に対して
入社までにやっておいたほうがいいことについてまとめました。
過去の自分のような過ちを他の人が犯さないようにまとめましたのでエンジニア内定をもらって何をすればいいかわからない人はぜひ参考にしてみてください。



私の簡単な経歴はこちら↓
都内私立文系大学卒業
大学3年時の夏に某大手プログラミングスクールに通い、プログラミングの基礎について学ぶ
スクール卒業後、都内のスタートアップの会社で2ヶ月ほどインターン(作業内容は主にLPの作成を行っていました)
大学4年時は就活を行い、株式会社TORICOに内定をいただき、今に至ります。



注意事項
簡単なコーディング知識があることを前提にお話しします。もしプログラミングが全くわからないという人はprogateなどのプログラミングを簡単に学べるサイトでまずは学びましょう。



それでは本題に戻ります。
まず結論からお話しします。以下の3つになります。
  1. タイピング強化

  2. 会社で使用する言語の参考書を1冊読んでおく

  3. ドキュメントで調べる癖をつける





それでは1つずつ説明していきます。

1.タイピング


これはどんな人でも絶対にやっておきましょう。
目安としましてはe-typingで安定してA以上や寿司打で1万円コースクリアでしょうか。
上記のサイトですと日本語入力ですので英単語を打つようなサイトを探してみてもいいかもしれません。
僕は入社してから、過去タイピング練習をしてこなかったことを最も後悔しています。

タイピングを強化しておくメリットには下記が挙げられます。
  1. 仕事スピードが上がる
  2. 成長スピードが上がる
  3. 教えていただいている時の時間を少なくできる


考えてみれば当たり前なのですが、タイピングスピードが2倍になればかけるコードも2倍になり、そのため成長スピードも2倍になります。
逆にタイピングスピードが1/2倍になればかけるコードも1/2倍になり、そのため成長スピードも1/2倍になります。
もはやエンジニアにとって一番重要なのではと思っています。
もちろん最初はコードを書くことよりも調べたり読んだりする時間の方が長いので、タイピングスピードの恩恵をあまり受けられないかもしれません
しかし後々大きく影響してくるので鍛えておきましょう。
後、単純にタイピングで遅くてミスりまくると恥ずかしいです。
1日10分とかでもいいので毎日タイピングの練習をするのがおすすめです。私も練習中です
タイピングに慣れてきたら数字や記号などもしっかりと打ち込めつように練習しましょう。

また少し話はタイピングから話がずれてしまうのですが、
よく使用するショートカットキーの暗記やカーソルの移動スピードmaxなどの使いやすいPC設定も行っておきましょう
こちらもPCを使う上での基礎スキルとなり、使っているか使っていいないかで作業効率が大幅に変わるので意識してみてください。






2.会社で使用する言語の参考書を1冊読んでおく


参考書を読むというのに抵抗感がある人は多いのではないでしょうか?
実際僕もそうでした。ネットなどで調べてみると「ネットに全部載っているのに本を買う必要はない」、「わからないことはその都度ググって調べればいい」など
本に対しては比較的、良い情報が流れていないようなイメージが僕にはあります。
これは私の上司から教えていただいて、確かにとなったのですが、本は体系的(一つ一つのものがある系統に従ってまとまっているさまのことという意味みたい)になっているため正確な情報をしっかりとインプットできるのです。
今までとりあえずわからなくなったらググってを繰り返していたのですが1通り本を読むことにより、もちろん完全暗記はできませんがコードを書いているとき、あれが使えるかなとか、それが出てこなくても
調べて出てきたメソッドなど、そういえばこんなのあったなと思い出せます。
また、おすすめの参考書なのですが
私の上司のおすすめの参考書はとりあえず分厚い本みたいです。。。
残念ながら優しくて短い本では情報量が少なすぎたりであまりお勧めをしていないようです。
参考書を買うときは分厚くて情報量の多い参考書を選びましょう。
こちらもタイピングと同様少ない時間でもいいので移動時間などを活用して少しずつ読み進めましょう。






3.ドキュメントで調べる癖をつける


皆さんはドキュメントで調べ物をしていますか?
僕は基本的にQiitaだったり個人ブログなどを参考にすることが多いです。。。
わかりやすいですよね。。。
なるべく意識はしていますが今でもあまりできていないのが現状です。
ドキュメントで調べる癖をつけた方がいい理由は、正確な情報が手に入れられるからです。
調べ物をしているとき正確な情報じゃないことや、記事が古く参考にならなかったり、バージョン違いで動作しなかったりと
結構クソみたいな記事が上に表示されることはあるあるではないでしょうか
ドキュメントで調べる癖をつけておくと正確な情報を手に入れることができるのはもちろんなのですが
英語で文を読む癖がついたり、英語で調べたりする癖がつくのでためになると思います。
日本語検索とは比べ物にならないほど英語の情報は出てくるので、英語めっちゃできるぜ!って感じを目指さなくてもいいですが
グーグル翻訳を使いながらでも少しずつ調べ物ができるようになると良いです。
プログラミングをしている人にとって英語は切っても切り離せない関係なので
早い段階で慣れておきましょう。





以上3つが私が入社前にやっておけば良かったことになります。
最後にもう1度
  1. タイピング強化

  2. 会社で使用する言語の参考書を1冊読んでおく

  3. ドキュメントで調べる癖をつける



重要順は上から1.2.3となります。
ぜひエンジニアに、これからなる人なりたい人は参考にしてみてください。




新卒エンジニアが最初の半年に任された業務

今年の春に入社しました、情報システム部の清瀬です。

私がTORICOに入社からそろそろ半年が経過しようとしています。このタイミングで、この半年どういった業務を担当してきたのかを記事にしようと思います。TORICOに興味をもっている方にTORICOのエンジニア業務がどんなものなのか、知ってもらえればと思います。

社内アプリへの権限付与の自動化

torico-id のホームページ

TOIRCOには TORICO-ID というログインを共通化できるアプリがあります。マンガ全巻ドットコム や マンガ展 と連携させることでログインやアカウント作成の簡略化ができます。このアプリは10近くある社内アプリにも使われています。社員全員がボタン一つで簡単にログインできるのですが、問題点が一つありました。それは、Admin サイトへの権限の問題です。

TORICO の社員と一般のユーザーを分けるフラグはあります。そのため、社員にのみ権限を与えることは今までもできていました。しかし、全社員に全ての社内アプリのAdmin権限を付与するわけにはいきません。そのため今までは情報システム部が手動で権限の付与をしていました。

そこで部署ごとに権限を付与するサイトを事前に決めておき、権限の付与の自動化を行えるようにしました。作成したモデルは具体的には

  • Website
    • 権限を与えるアプリのモデル
    • 部署とM2Mの関係
    • django-allauth の Application と 1:N の関係
  • ClientGroup
    • ログイン成功時にユーザーに与える、クライアント側の Django の Group のモデ
    • django-allauth の Application と 1:N の関係
  • Department
    • 部署・部門のモデル
    • Website と M2M の関係
    • ClientGroup と M2M の関係
    • User と M2M の関係

の3つです。TORICO では Mezzanin という Django の CMS を使用しているため、Mezzanine の権限も付与できるように Website を Application と 1:N の関係にしています。

あとは Department に Application と ClientGroup を結びつけておくことで、どの部署にどの権限を付与するのかを自動で判別できるようになりました。TORICO は現在拡大期であり毎月のように新しく入社される方がいるため、この機能は非常に役立っていると個人的には思います。また、エンジニアとしても予想外のメリットがありました。それは新しいプロジェクトのローカル環境を作成した際に、いちいち SQLクライアントツールで自分のアカウントにスタッフ権限を付与しなくて済むことです。特に情報システム部は担当するアプリが多いため、その一手間を省けるのは非常に便利です。

古いシステムの Django 化

長年TORICOを支えてきてくれたシステムですが、その中の幾つかを Django に移行するお手伝いもしました。元のコードを読みながら、その機能を Django で再現するという業務でした。元のコードは PHP で書かれているため、PHP と Python を比較するような作業で、個人的にはとても楽しかったです。なにより、古いシステムを新しいシステムに置き換えることで、一つ一つの処理が非常に効率が良くなり、格段にレスポンスを早くすることができました。あらためてフレームワークの凄さを目の当たりできてよかったです。TORICO にはまだ古いシステムが残っているので、来期も移行の業務を積極的にやっていきたいと思っています。

MySQL 5.6 を 5.7 にバージョンアップする

RDS の MySQL5.6 のサポート期限が8月末(延長されて22/3/1まで)までであるということで、MySQL のパージョンアップもさせていただきました。手順としては、

  1. dev 環境のバージョンアップを行い、全ての機能が使えるかテストする
  2. 1 で問題あった機能を修正する
  3. スナップショットでバックアップをとる
  4. 修正したコードをデプロイして、本番のバージョンアップを行う

今回のバージョンアップでは1の部分で問題がありました。対象のアプリでは、ログイン時に MySQL の OLD_PASSWORD というハッシュ関数を使っていたのですが、それがサポートされなくなりました。そのため、代替となる関数をアプリ側で作成しなければなりませんでした。幸い、既に再現をしてくれているコードがあったため、それを拝借するだけで問題はありませんでした。ただ、ログインというアプリの基幹機能の修正だったため、本番へのデプロイは非常に緊張しました。

また、本番のバージョンアップ時にオプションの設定を忘れるというミスをしてしまいました。古いものが引き継がれると勘違いしていたことが原因です。その結果タイムゾーンの設定が狂ってしまい、表示されるべきコンテンツが表示されないという状況になってしまいました。普段からお世話になっている先輩に手助けしていただいたため無事に解決できましたが、もう少しで大惨事となるところでした。今後はもっと慎重に、情報集めからテストまでを行わなければいけないなと反省しています。

最後に

今回は、私がこの半年で任せていただいた主要な業務をいくつか紹介させていただきました。以上に取り上げたもの以外にも、

  • 新機能の追加
    • 5件
  • Django 化
    • 2件
  • DB・ネットワーク
    • 2件
  • セキュリティ(別ブログに記載)
    • 10件
  • コードの軽微な修正
    • 16件

と、様々な業務を体験できました。今回はシステムに関係する業務だけを取り上げましたが、他にも壁にドリルで穴を開けたり社内サーバーを移動させたりと、普段はできない業務も経験させていただきました。TORICO ではフルスタックであることが求められるため、業務も多岐にわたります。そこに魅力を感じた方は、ぜひ TORICO に応募してみてください。

実務経験0 入社一年目のエンジニアが任されたセキュリティの話

2021年春に入社しました、情報システム部の清瀬です。

TORICOの情報システム部では、セキュリティの向上を上半期の目標に掲げておりました。上半期も終わりに近づいてきたということで、新卒で入社した私がどんなセキュリティ向上のために携わってきた業務を記事にしようとおもいます。多くのシステムでは Django を使用しているためコード部分は Django 前提の話になります。

1. インフラレベルでの対応

AWS WAF の導入

  • SQLインジェクション対策
  • クロスサイトスクリプティング
  • その他、包括的な対策

TORICOでは一部の社内用アプリを除き、全てのサービスをAWSでデプロイしています。それら全てのサーバを悪意のあるリクエストから守るために、AWS WAF を導入しました。

AWS WAF は2019年に大幅アップデートされ、今までのものは WAF Classic と名称を変更しました。新しいWAFの最大のメリットは、るAWSによって作られたルールを簡単に導入できる点です。例えば AWSManagedRlesCommonRuleSet では、OWASPに記載された主要な脆弱性・リスク10個をカバーしてくれています。クロスサイトスクリプティングや悪意のあるボットを排除してくれるルールをボタン一つで導入できるため、包括的なセキュリティ対策をすることができます。もちろん、今までのようにオリジナルのルールを作ることができます。そのため、社内からのアクセスには一部のルールを緩和させたりなどもできるようになっています。

余談ですが、WAFの導入の検証を担当したのは、まだ私がインターン1週間くらいの時でした。検証のために簡易的な Django アプリを Ec2 にデプロイしてWAFの導入でクロスサイトスクリプティングやSQLインジェクションがきちんとブロックされるのかを検証しました。インターンなのにこんな重要そうな仕事をさせていただけるのかと驚いたのを今でも覚えています。

過去に使用していた IP アドレスのルーティングを解除

これはセキュリティとは少し違う話かもしれませんが、ドメインの整理などもしました。一部の未使用のドメインが、既に手放してある IP アドレスと結びついたまま放置されていました。もし IP アドレスが悪意のあるサイトに使われていた場合、最悪ユーザーを悪意のあるサイトに誘導してしまうということも起こり得ます。そうならないためにも、不要なドメインを整理しました。

2. アプリレベルでの対応

サービスのAdmin ・社内アプリ を社内からのみ許可する

  • 不正なアクセス対策
  • 個人情報の保護

こちらは Django の Middleware を使って対応しました。サブネットマスクの表現方法などが複数あるため IP を取り扱うのは面倒くさそうだなと思っていましたが、python には ipaddress という便利なモジュールが標準で搭載されています。このモジュールを使うことで簡単に実装することができました。このタスクは上記の WAF でも実装できますが、1つのALBで複数の社内アプリを制御していたためアプリレベルで対応することにしました。アプリレベルのため、各アプリに適した制限を導入することも簡単にできます。

ログインセッションキーを適切な設定にする

  • セッションハイジャック対策
  • CSRF対策

大事なセッションを管理してくれるクッキーにはセキュリティ対策は欠かせません。TORICOの全てのサイトで、その大切なCookieの設定が適切になるように対策をしました。具体的には HttpOnly Secure SameSite=Lax  の3つの設定です。これらの設定を簡単に説明すると、

HttpOnly: JavaScript からセッションにアクセスできないようにする。設定されていなかった場合、JS 経由で セッション情報が漏洩する危険がある。

Secure: https 通信でのみ セッションを送るようにする。設定されていなかった場合、暗号化されない Http 通信でセッションが送られてしまう危険がある。

SameSite: Strict, Lax, None の3つの設定がある。Lax の場合、別オリジンからの POST の時はセッションを送らない。None の場合、例えばECサイトの届け先を変更するフォームのある偽サイトからユーザーのセッションを使ってデータが送信されてしまう。

Django では 2系以上であれば settings で設定することができます。1系の場合は SameSite を設定することができません。クッキーセッションの注意点として、あまりにも厳しい設定にした場合、サービスがうまく動かなくなる可能性があります。特に SameSite は決済部分でエラーがでることがあるそうです。

ファイル出力時に追跡IDとログを残す

  • 情報漏洩元の究明

TORICOではできるだけ業務を自動化するために、いろいろな情報をCSVなどに出力します。万が一そういったデータが社外に流出した時、どのデータが、どこから流出したのかを特定しやすいように追跡IDをそれぞれのファイルに追記するようにしました。ログにはリクエストの情報を全てを保存しているため、いつ・誰が・どの検索ワードで CSV を出力したのかを追うことができます。 Django には DB へのログ出力機能はないため、DBに出力する場合には自作する必要があります。

3. データベースの対応

重要なフィールドの暗号化

  • 個人情報の保護
  • データ漏洩時の影響を最小限にする

重要なフィールドを可逆暗号化することで、万が一情報漏洩した時でもその影響を最小限に留められるようにしました。可逆暗号は Django のオリジナルのフィールドを作ることで簡単に対応することができました。具体的な実装方法についてはこちらの記事で書きました。また、一部の古い社内アプリではアカウントのパスワードがハッシュ化されていなかったため、 bcrypt という方法で不可逆暗号化もしました。

ソルトとペッパーの役割の違いなどが曖昧だったため、とても勉強になったタスクだったなと思っています。

最後に

上半期に行った主要なセキュリティ対策を紹介していきました。セキュリティは深刻な障害の原因になるため、普段のコーディングから気をつけていきたいと思います。

Amazon Pay Checkout v2 API の署名 (RSA-SHA256 (RS256) + RSA PSS Padding) を Python で行う

Amazon Pay の API クライアントを書く際、Amazon のAPIサーバに送信するリクエストに、RSA-SHA256, RSA PSS パディングを使って署名を作り、リクエストに含めて送信する必要があります。

Java や Node はクライアントライブラリがあったので、それを使って簡単に署名できたのですが、弊社 TORICO ではサーバサイドは主に Python を使っており、既存のクライアントライブラリは無かったため、Node のライブラリを参考に署名コードを書きました。

Amazon Pay のAPI

今回は、Amazon Pay の Checkout v2 API を使う必要がありました。

https://developer.amazon.com/ja/docs/amazon-pay/intro.html

リクエストヘッダに含める署名については、
この CV2 (Checkout V2) の 署名リクエストのページに解説があります。

このページを見ていくと、「署名を計算します」の箇所に

署名を計算するには、ステップ2で作成した署名する文字列に秘密鍵を使用して署名します。 SHA256ハッシュとソルト長20のRSASSA-PSSアルゴリズムを使用します。結果をBase64エンコードして、この手順を完了します。 RSASSA-PSSを使用して計算されたすべての署名は、入力が同じであっても一意であることに注意してください。

とありますので、この「 SHA256ハッシュとソルト長20のRSASSA-PSSアルゴリズム」を Pythonでコーディングする必要があります。

ところで、この最後の「 入力が同じであっても一意であることに注意してください」は、「〜一意でない」の間違いじゃないですかね。英語版ページは調べてませんが。

Amazon Pay Scratchpad

https://pay-api.amazon.jp/tools/scratchpad/index.html

検証リクエストを発行できるサイトが用意されていますので、署名ロジックの確認に使うと良さそうです。
私は、今回の開発中は存在を知らなかったので、使っていません。

Node.js の場合

node.js の場合は、クライアントライブラリは
@amazonpay/amazon-pay-api-sdk-nodejs があり、これを使うと署名を含めたAPIリクエストが一発で行えます。

ヘッダの署名を行っているコードは

https://github.com/amzn/amazon-pay-api-sdk-nodejs/blob/master/src/clientHelper.js#L142 このあたりで、
「 SHA256ハッシュとソルト長20のRSASSA-PSSアルゴリズム」のコードは
https://github.com/amzn/amazon-pay-api-sdk-nodejs/blob/master/src/clientHelper.js#L84 ここです。

Python の場合

上記 node.js 相当のコードを書くわけですが、hmac ライブラリには相当のコードはありません。

RSA や パディングの基礎的なロジックは cryptography に入っており、実際の使い方は PyJWT がいい感じになっているので、PyJWT を参考にします。

https://github.com/jpadilla/pyjwt/blob/master/jwt/algorithms.py#L232

PSS パディングはここです。

https://github.com/pyca/cryptography/blob/main/src/cryptography/hazmat/primitives/asymmetric/padding.py#L19

この PSS の第一引数の _mgf ってなんだ、と思いましたが、同モジュール中にある
MGF1(RSAAlgorithm.SHA256())
を入れたら動きました。

署名部分のコード

署名部分のコードはこのようになります。

import base64
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.serialization import load_pem_private_key

private_key = b'''-----BEGIN PRIVATE KEY-----
MIIEvQIB....
....N/Qn4=
-----END PRIVATE KEY-----'''

string_to_sign = 'AMZN-PAY-RSASSA-PSS\nxxxxxxxxxxxxxxxxxxxxxxxx'

key = load_pem_private_key(private_key, password=None)

signature = key.sign(
string_to_sign.encode('utf-8'),
padding.PSS(padding.MGF1(hashes.SHA256()), 20),
hashes.SHA256())

signature = base64.b64encode(signature).decode('utf-8')

後になって気づきましたが、PyJWT に RSAPSSAlgorithm という、今回の用途にぴったりなクラスがあったので、これを使うともうちょっとシンプルなコードになるかもしれません。

署名元の文字列の生成も含めたコード

checkoutSessions を行うコードはこのような感じです。

node.js のコードを参考にした箇所がいくつかあり、それらは実際には使われない、不要なコードになってます。

import base64
import datetime
from hashlib import sha256

import requests
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.serialization import load_pem_private_key

private_key = b'''-----BEGIN PRIVATE KEY-----
MIIEv....
....N/Qn4=
-----END PRIVATE KEY-----'''

config = {
'publicKeyId': 'SANDBOX-AEXXXXXXXXXXXX',
'privateKey': private_key,
'region': 'jp',
'sandbox': True,
}

constants = {
'SDK_VERSION': '2.1.4', 'API_VERSION': 'v2', 'RETRIES': 3,
'API_ENDPOINTS': {'na': 'pay-api.amazon.com', 'eu': 'pay-api.amazon.eu', 'jp': 'pay-api.amazon.jp'},
'REGION_MAP': {'na': 'na', 'us': 'na', 'de': 'eu', 'uk': 'eu', 'eu': 'eu', 'jp': 'jp'},
'AMAZON_SIGNATURE_ALGORITHM': 'AMZN-PAY-RSASSA-PSS',
}

checkoutSessionId = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

options = {
'method': "GET",
'urlFragment': f"/v2/checkoutSessions/{checkoutSessionId}",
'headers': {},
'payload': ''
}

pay_date = datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')

headers = {
'x-amz-pay-region': config['region'],
'x-amz-pay-host': 'pay-api.amazon.jp',
'x-amz-pay-date': pay_date,
'content-type': 'application/json',
'accept': 'application/json',
'user-agent': 'amazon-pay-api-sdk-nodejs/2.1.4 (JS/14.15.1; darwin)',
}

lowercase_sorted_header_keys = list(sorted(headers.keys(), key=lambda x: x.lower()))
signed_headers = ';'.join(lowercase_sorted_header_keys)

canonical_request = [
options['method'],
options['urlFragment'],
'', # GETパラメータだが一旦無し
] + [
f'{h}:{headers[h]}' for h in lowercase_sorted_header_keys
] + [
'', # 空行入れる
signed_headers,
sha256(options['payload'].encode('utf-8')).hexdigest()
]

canonical_request_bytes = ('\n'.join(canonical_request)).encode('utf-8')

string_to_sign = constants['AMAZON_SIGNATURE_ALGORITHM'] + '\n' + sha256(canonical_request_bytes).hexdigest()

key = load_pem_private_key(config['privateKey'], password=None)

signature = key.sign(
string_to_sign.encode('utf-8'),
padding.PSS(padding.MGF1(hashes.SHA256()), 20),
hashes.SHA256())

signature = base64.b64encode(signature).decode('utf-8')

headers['authorization'] = \
f"{constants['AMAZON_SIGNATURE_ALGORITHM']} " \
f"PublicKeyId={config['publicKeyId']}, " \
f"SignedHeaders={signed_headers}, " \
f"Signature={signature}"

response = requests.get(
f"https://pay-api.amazon.jp{options['urlFragment']}",
headers=headers
)
print(response)
print(response.json())

Google Hangouts Chat にプログラムからメッセージを送信する

Google ハングアウトの後継のチャット(インスタントメッセージング)サービス、Chat では、Webhook エンドポイントを使うことでとても簡単にチャットルームへのメッセージの送信ができます。

メッセージの送信に、チャット用の大規模なアプリ開発は不要です。Python でも curl でも JS でも、3行ぐらいでメッセージの送信ができます。

Webhook エンドポイントの作成

1. Chat を開く

https://chat.google.com/

2. チャットルームの作成

左上のメニューから、「チャットルームを作成」を選び、

適当に名前をつける。

3. Webhook エンドポイントの作成

チャットルーム名をクリックするとメニューが開くので、「Webhookを設定」をクリック

+ WEBHOOKを追加 をクリック

適当に名前をつけて、「保存」

Webhook の URL ができる。このURLを記録しておく。

メッセージを送信する

Python

import requests

webhook_url = 'https://chat.googleapis.com/v1/spaces/...%3D'

response = requests.post(
webhook_url,
json={"text": "こんにちは、世界!"}
)

requests を使えば、これだけでメッセージを送信できます。簡単ですね!

curl

curl -X POST "https://chat.googleapis.com/v1/spaces/...%3D" \
--header "Content-Type: application/json; charset=UTF-8" \
--data '{"text": "こんにちは!"}'

これで送信できます。

その他

その他のツールでメッセージを送信するには、Incoming webhook with Python を参考に

HTTPリクエストヘッダ: Content-Type: application/json; charset=UTF-8

リクエストボディはJson: {"text": "Hello from Python script!" }

で送信できます。

400エラーが出たら

テストコードを書いていたら、HTTPステータス400

{
"error": {
"code": 400,
"message": "Request contains an invalid argument.",
"status": "INVALID_ARGUMENT"
}
}

このようなエラーレスポンスが返ってきて困ってたのですが、この原因は単純なURLのコピペミスでした。URLの最後まで正しくコピペできているか、確認してください。

DB(MySQL)をネットワーク越しに簡単にコピーする。mysqldump + パイプで。python subprocess の例も

本番環境のデータベース(MySQL)をネットワーク越しに開発環境にコピーしたい時のプラクティスです。

シェル + パイプ

よくやるのが、bash等 でパイプを使って流し込む方法です。

$ ssh user@production.example.com mysqldump \
--skip-lock-tables \
--host=xxxx.rds.amazonaws.com \
--user=xxxx \
--password=xxxx \
database_name table_name | \
ssh user@dev.example.com mysql \
--host=127.0.0.1 \
--user=xxxx \
--password=xxxx \
--database=xxxx

ssh で本番サーバ user@production.example.com に接続し、mysqldump を実行。その標準出力を SSH 接続を通して手元まで持ってきます。

ssh でもう一つ、開発環境サーバ user@dev.example.com に SSH接続し、mysql を起動。先ほどの本番環境の mysqldump 結果をパイプでそのまま流し込みます。

速度が充分に早く、通信経路も ssh で暗号化されるため安全に、効率良くコピーできます。mysql に余計な穴を空ける必要もありません。

mysqldump のオプションで --where を付けて読み込むデータを絞り込んだりもできます。

Python subprocess を使う

subprocess shell=True で実行

(あまり面白くない。読みにくい。)

import subprocess
subprocess.check_call("上記のコマンド", shell=True)

subprocess で、Popen や check_call などの引数に shell=True を与えることで、シェルコマンドをそのまま実行できます。

subprocess のパイプを使う (おすすめ)

subprocess でシェルのパイプと同様の処理が行えます。

dump_command = [
"ssh",
"user@production.example.com",
"mysqldump",
"--skip-lock-tables",
"--host=xxxx.rds.amazonaws.com",
"--user=xxxx",
"--password=xxxx",
"database_name",
"table_name",
]
dump_process = subprocess.Popen(
dump_command, stdout=subprocess.PIPE)

import_command = [
"ssh",
"user@dev.example.com",
"mysql",
"--host=127.0.0.1",
"--user=xxxx",
"--password=xxxx",
"--database=xxxx",
]
import_process = subprocess.Popen(
import_command, stdout=subprocess.PIPE, stdin=dump_process.stdout)

stdout, stderr = import_process.communicate()

ダンププロセスの stdout を インポートプロセスの stdin に接続して 2 つのプロセスを実行します。

Python のコードにしておけば、再利用性・メンテナンス性を高くでき、使い回しに優れます。

Django のデータベースコネクションを使う場合

from django.db.transaction import get_connection

dump_command = [
"ssh",
"user@production.example.com",
"mysqldump",
"--skip-lock-tables",
"--host=xxxx.rds.amazonaws.com",
"--user=xxxx",
"--password=xxxx",
"database_name",
"table_name",
]
dump_process = subprocess.Popen(
dump_command, stdout=subprocess.PIPE)

connection = get_connection(using='db_alias_name')
cursor = connection.cursor()

cursor.execute(dump_process.stdout.read())

ダンプコマンドの結果を read() して、そのまま Django データベース接続の cursor で流し込むこともできます。

※ダンプ結果を一旦メモリにためるため、データ量が多い場合ちゃんと動くかは不安です。そして他の例より効率は悪そうです。

Amazon マーケットプレイスWebサービス (MWS) APIから注文情報を取得する方法

当社では、Amazon のマーケットプレイスに出店していたり、FBA (フルフィルメントByアマゾン: Amazon社の倉庫に商品を納品し、販売を代行してもらう販売方法) を行っています。

マーケットプレイスにはAPIが用意されており、リクエストすることで受注情報など多くの情報を取得できるのですが、署名の計算に少し躓いたので書いておきます。

Amazon MWS 公式ガイド

アクセスに必要な情報を集める

認証情報 (クレデンシャル)

アクセスに必要な認証情報は、

  1. 出品者ID (マーチャントID, セラーIDと呼ばれることもある)
  2. AWSアクセスキーID
  3. 秘密キー(シークレットキー)

の3つです。もし、アクセスするアカウントがセラーセントラルのアカウントではなく、派生して作られた子アカウントである場合、別途「MWS認証トークン」が必要になります。

これらの情報は、すべてセラーセントラルの「設定」→「ユーザー権限」のページで取得できます。AWSアクセスキーID,秘密キー は、ページ下部の「Amazon MWS 開発者権限」の、「認証情報を表示」をクリックすると表示されます。

マーケットプレイスID

別途、マーケットプレイスID という文字列が必要になる場合があります。

マーケットプレイスIDの一覧はページから確認できます。

Amazon マーケットプレイス Web サービスエンドポイント

例えば、日本なら A1VC38T7YXB528 です。

APIアクセスのテストをする

Amazon社がAPIのテストツール「Scratchpad」を公開しているので、それを使います。後述する HMAC の計算が正しいか確認する意味でも、このツールは必ず使ってみたほうが良いです。

  1. Amazon MWS Scratchpad を開く
  2. 左上「API Selection」を適当に選択。今回は、API セクション: 注文、Operation: ListOrders
  3. Authentication 欄に認証情報を入力、SellerId: には 出品者ID, MWSAuthToken は元アカウントなら空、払い出された子アカウントならトークン文字列を入れる。AWSAccessKeyId、Secret Kye は先ほど取得した文字列を取得。
  4. API必須パラメータの「MarketplaceId.Id.1」には、A1VC38T7YXB528 を入力
  5. API任意パラメータの「LastUpdatedAfter」のみ、2017-05-05 のように入力
  6. 「送信」をクリックすると、結果が表示されます。

HMAC の値を確認しておく

結果が正常に表示された場合、「リクエスト」タブをクリックして開いてみると「署名対象の文字列」というセクションと、その下に計算した HMAC署名 が表示されています。APIを開発する時は、この情報を元に開発するとやりやすいです。「証明対象の文字列」に対して SHA 256 HMAC で計算を行い、その下に書かれている文字列が結果で得られるよう開発をしていきます。

APIライブラリを開発する

Pythonで作ります。

署名方法は、Amazonで「署名バージョン2」と言われる方法です。

公式ドキュメントの署名のロジック説明 (Java のサンプルコードあり)

HMACの計算ロジックの作成

Python には hmac ライブラリがあるので、それを使えばすぐにできます。

Python3での例

import hmac
import hashlib
import base64

secret_key = b"取得した秘密キー"

canonical = b"""
POST
… 「証明対象の文字列」をここにコピペ …
"""

h = hmac.new(secret_key, canonical.strip(), hashlib.sha256)

print(h.hexdigest())
print(base64.b64encode(h.digest()))

これを実行すると、Scratchpad に表示されている「SHA 256 HMAC」「Base64 HMAC」と同じ値が取得できるはずです。

署名対象の文字列の作成

署名対象の文字列は、HTTPメソッド(POST)、ドメイン名(mws.amazonservices.jp)、パス(/Orders/2013-09-01)、それとクエリ文字列を、改行(\n)で連結して作ります。

クエリ文字列の作成

クエリ文字列は、検索パラメータ名をソートさせ、&= で連結して作ります。値は URL エンコードします。

ディクショナリで値を用意してたとすると、

import datetime
import urllib.parse

AMAZON_CREDENTIAL = {
    'SELLER_ID': 'セラーID',
    'ACCESS_KEY_ID': 'AWSアクセスキーID',
    'ACCESS_SECRET': 'アクセスシークレット',
}

data = {
    'AWSAccessKeyId': AMAZON_CREDENTIAL['ACCESS_KEY_ID'],
    'Action': 'ListOrders',
    'MarketplaceId.Id.1': 'A1VC38T7YXB528',
    'SellerId': AMAZON_CREDENTIAL['SELLER_ID'],
    'SignatureMethod': 'HmacSHA256',
    'SignatureVersion': '2',
    'Timestamp': datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'),
    'Version': '2013-09-01',
}

query_string = '&'.join('{}={}'.format(
    n, urllib.parse.quote(v, safe='')) for n, v in sorted(data.items()))

print(query_string)

このようなロジックで作成できます。

sorted メソッドでキーで並び替えを行い、値は urllib.parse.quote で URLエンコードします。safe='' を入れないと / がエンコードされないので、入れます。

後は、改行で連結すれば署名対象文字列になります。

canonical = "{}\n{}\n{}\n{}".format(
    'POST', 'mws.amazonservices.jp', '/Orders/2013-09-01', query_string
)

print(canonical)

署名をつけてリクエストする方法

リクエストメソッドは POST です。ですが、POSTのデータは空で、パラメータはURLのクエリストリングに入れます。

署名は、クエリストリングの末尾に &Signature=署名 という形で付与します。

requests でリクエストしてみる

実際にリクエストするコードを書いてみます。

import base64
import datetime
import hashlib
import hmac
import urllib.parse

import requests
import six

AMAZON_CREDENTIAL = {
    'SELLER_ID': 'セラーID',
    'ACCESS_KEY_ID': 'AWSアクセスキーID',
    'ACCESS_SECRET': 'アクセスシークレット',
}

DOMAIN = 'mws.amazonservices.jp'
ENDPOINT = '/Orders/2013-09-01'


def datetime_encode(dt):
return dt.strftime('%Y-%m-%dT%H:%M:%SZ')


timestamp = datetime_encode(datetime.datetime.utcnow())

last_update_after = datetime_encode(
datetime.datetime.utcnow() - datetime.timedelta(days=1))

data = {
'AWSAccessKeyId': AMAZON_CREDENTIAL['ACCESS_KEY_ID'],
'Action': 'ListOrders',
'MarketplaceId.Id.1': 'A1VC38T7YXB528',
'SellerId': AMAZON_CREDENTIAL['SELLER_ID'],
'SignatureMethod': 'HmacSHA256',
'SignatureVersion': '2',
'Timestamp': timestamp,
'Version': '2013-09-01',
'LastUpdatedAfter': last_update_after,
}

query_string = '&'.join('{}={}'.format(
n, urllib.parse.quote(v, safe='')) for n, v in sorted(data.items()))

canonical = "{}\n{}\n{}\n{}".format(
'POST', DOMAIN, ENDPOINT, query_string
)

h = hmac.new(
six.b(AMAZON_CREDENTIAL['ACCESS_SECRET']),
six.b(canonical), hashlib.sha256)

signature = urllib.parse.quote(base64.b64encode(h.digest()), safe='')

url = 'https://{}{}?{}&Signature={}'.format(
DOMAIN, ENDPOINT, query_string, signature)

response = requests.post(url)

print(response.content.decode())

下品にベターっと書いてますが、これで動きます。

実際にはこれをライブラリ化して肉付けしていくと良いでしょう。

EPUBファイルから画像を抽出する

電子書籍フォーマットとして広く使われている EPUB ファイルから、連番で画像を抽出する方法です。

ツール作りました! pip でインストールできます。https://github.com/ytyng/epub-extract-jpeg

EPUB ファイルの概要

EPUB ファイルとは、平たく言えば ZIP圧縮された XHTML です。 コミックで一般的に使われる形式では、1ページが1つの XHTML ファイルになっており、その中に 1 つの img タグが あり、画像ファイルにリンクされています。

そのため、手順としては

  • EPUB ファイルを解凍
  • 構成情報の XML ファイルを解析し、ページ画像の URL (パス) を取得
  • ページ画像を連番で改名コピー(移動)

となります。

1. EPUBファイルを解凍

unzip で一発です。

$ mkdir /tmp/epub-extract
$ unzip sample.epub -d /tmp/epub-extract

2. 構成情報の XML ファイルを解析し、ページ画像の URL (パス) を取得

まず、展開後のディレクトリにある META-INF/container.xml を開きます。 ここに、rootfile というタグがあるので、その full-path 属性を見ます。 full-path の XML ファイルが、各ページの目次のようなものになります。

full-path が示す XML ファイルで、manifest タグの中に item タグが複数あります。 これらは、EPUB 中の XHTML から使われているファイルです。

3. ページ画像を連番で改名コピー(移動)

この、item タグの中はおそらくページ順になっているので、このファイルをスクリプトで連番で改名コピーしながら収集すれば、ページ画像を抽出できます。

本来であれば、直接画像のパスを読むのではなく、ページの XHTML ファイルを開き、そこからリンクされている画像を収集していくのが正しいのですが、ページの XHTML の順と画像の item タグの順が一致しないケースは稀だと思いますので(EPUB 作成者が意図的に XHTML ファイルと画像ファイルの順番を一致させなかった場合などは、ページ数が正しく取得できません)、item タグの順で処理して基本的には問題無いでしょう。

これで、EPUB ファイルから画像ファイルを抽出する方法は終わりです。 最後に、Python スクリプトにした例を掲載しておきます。

from __future__ import print_function, unicode_literals

import os
import time
import sys
import subprocess
import shutil
from xml.etree import ElementTree

TEMP_DIR = '/tmp/epub-extract-{}'.format(int(time.time()))


def procedure(file_path):
    if not os.path.exists(file_path):
        print("{} is not exist.".format(file_path), file=sys.stderr)
        return

    output_dir, ext = os.path.splitext(file_path)

    if ext != '.epub':
        print("{} is not epub.".format(file_path), file=sys.stderr)
        return

    if os.path.exists(output_dir):
        print("{} is already exists.".format(output_dir), file=sys.stderr)
        return

    os.mkdir(TEMP_DIR)

    subprocess.Popen(
        ('unzip', file_path, "-d", TEMP_DIR),
        stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()

    os.mkdir(output_dir)

    container_xml_path = os.path.join(TEMP_DIR, 'META-INF', 'container.xml')
    etree = ElementTree.parse(container_xml_path)
    rootfile_node = etree.find(
        ".//{urn:oasis:names:tc:opendocument:xmlns:container}rootfile")
    content_opf_path = rootfile_node.attrib['full-path']

    content_xml_path = os.path.join(TEMP_DIR, content_opf_path)
    etree = ElementTree.parse(content_xml_path)
    manifest = etree.find('.//{http://www.idpf.org/2007/opf}manifest')
    items = manifest.findall('.//{http://www.idpf.org/2007/opf}item')

    image_paths = []
    for item in items:
        if item.attrib['media-type'] == 'image/jpeg':
            image_paths.append(item.attrib['href'])

    root_dir = os.path.dirname(content_xml_path)

    for i, image_path in enumerate(image_paths, start=1):
        destination_image_name = '{:03d}.jpg'.format(i)
        source_image_path = os.path.join(root_dir, image_path)
        destination_image_path = os.path.join(
            output_dir, destination_image_name)
        shutil.move(source_image_path, destination_image_path)
        print('{} -> {}'.format(image_path, destination_image_name))

    shutil.rmtree(TEMP_DIR)


def main():
    for arg in sys.argv[1:]:
        procedure(arg)


if __name__ == '__main__':
    main()
追記: Github に上げて、pip でインストールできるようにしました。
Search