フォロワーを管理するためだけのWebアプリを作った

- はじめに -

フォロワーも増えてきて大体毎月n*100単位でフォロワーが変動するようになってきて、巷のフォロー管理ツールじゃ全然要求を満たせないので、自作する体でGWに友人と1日ハッカソンみたいなのを開いて勢いだけで作った話。

「100万人とかフォロワーが居る訳じゃないけど1000から1万くらいの "小さな界隈のアルファ" 」は多く居て、皆どうやってTwitterを見てるのか知りたいのでまず自分から。

自分がTwitterを見ている方法や、フォロバに関する個人的な意見など、偏見を含む記事。ほぼポエム

- 作ったもの -

自身の要件を満たす物を一旦作った。

• 裏側はpandasが走る事でquery検索やsort、別途の分析、拡張が容易
• 画面内でqueryの調整、閲覧、フォロー、アンフォロー、リスト管理ができる
• tableに最低限の情報が羅列されている
• queryを保存したりDB側にも更新機構がある

つくったもの

formにpandasのquery条件やsort条件が入れられるようになっていて、その場でuserテーブルを見るのはもちろん、よしなにqueryを作ってURLパラメータで保存しておけば、同条件でいろいろ見られる。

例えば

# 直近のフォロワー100件をフォローされた順に表示
# http://127.0.0.1:5000?nq=followed&tcrq=yy-mm-dd&tltq=yy-mm-dd&fcrq=yy-mm-dd&fltq=yy-mm-dd&sorq=follower_number&asq=True&samq=100

# 片思いでフォローしてるけど最新のつぶやきが1年以上前の人をアカウント生成日順に表示
# http://127.0.0.1:5000?nq=~followed and following&tcrq=yy-mm-dd&tltq=2018-05-01&fcrq=yy-mm-dd&fltq=yy-mm-dd&sorq=created_at&asq=True&samq=100

# bioに機械学習と書いてあり2018年以降に作られたアカウントでフォロワーが1000以上の人をフォロワー数順に
# http://127.0.0.1:5000?nq=description.str.contains("機械学習") and followers_count > 1000&tcrq=yy-mm-dd&tltq=yy-mm-dd&fcrq=2018-01-01&fltq=yy-mm-dd&sorq=followers_count&asq=True&samq=100

これだけ色んな情報を持っておいていつでも見れるようにしておけば、完全にフィードアウトしてしまって定期botと化してしまったアカウントとか、フォロバ即フォロー解除でフォロワー増強するクソ野郎だとかを定期的に見つけてフォロー外していったりできると思う。ランダムサンプリングする事もできるので、定期的に見てれば長期的にフォロワーが整理されていくだろうという仕組み。

- フォロバ作業 -

実際にこのツールをVivaldiのWebパネルに登録して、擬似的な2画面でフォロバ作業をやってみている様子。

なんか変なアカウントをフォロバして外している様子

ここからはかなり偏見が混じるが、私のフォローに関する話とかを書いておく。
多分これを書かないと、管理ツールの思想が理解しづらいと思う。

リスト管理とフォロー

たまに「めっちゃフォローしてますよね」と言われるが、私はただTwitterとリスト管理が好きなだけ。
エゴサは出来る限りの検索queryと正規表現を作ってあるし、リストは34つある。
フォロバすると、「ここ半年」という名前の「フォローしてから半年経ってない人が入るリスト」におうちサーバが自動的に入れてくれる。
そのリストは割と見ていて、半年後にはそのリスト内の様子から大体どこかのリストには振り分けられている仕組みでなんとかやっている。
ただそれを色んな方法で見ているだけ。

最低限プライバシーを考慮した17columnの様子です pic.twitter.com/ZkNTkkB5O3

— ばんくし (@vaaaaanquish) June 25, 2018

　
リスト分けは「つぶやく頻度」なんかがわかりやすくて、週一だとか月一ペースのフォロワーは「たまに見る」といったリストに入っている。

エンジニアは分かりやすく「フロントエンド」「インフラ」「アプリ」といった分野でも分けているし、「機械学習」のような自分の専門であれば興味度合いを大体で3段階に分けて「げきつよ」「つよい」「興味ある」みたいに分けている(もう2段階くらい必要だなと最近は思っている)。もちろん技術全体的にすごい人は「高レベル人材」「好きな神」なるリストがあるし、「CTO」とか「経営」とか「採用人事公式アカウント」とかにも分けてある。「オタク」とか「あったことある」とか「仕事したことある」とかもあって、それぞれ複数のリストにまたがって入っている人が居る形になっている。

幸せになってほしい人を入れる「幸せになってほしい人リスト」もあり、こちらは公開している。
昔いろいろ言われたリストだが幸せになってほしい。

今回作ったものを奥さんに見せたら「そんなアプリ作ったの…？暇じゃん…」と言われたし、自分でもアホだと思うけど、検索queryとか考えたり特定界隈のバーストを見るのが楽しいので仕方ない。
月に数百人にフォローされても返せるのには、その辺りが好きなのが起因してると思う。

どんな人をフォローしているか

リスト管理が好きなので、フォロバの基準はアホみたいに低い。
最近はリスト内で見れないものがあると悔しいのでブロックもミュートも0人という、一部から見たら異常な運用だと思う。

　
実際フォロバする作業をしていくと、低いとはいえ「フォロバしない人」の基準自体はある。
例えば「留学、英会話をマスター、世界一周、恋愛テクニック、FX、FXツール、独立、投資、Wワーク、小遣い稼ぎ、フリーランス、自由、起業、複業、未経験、相互フォロー、エロ、〜〜と繋がりたい、DM、LINE@」辺りの単語がbioに入ってると、注意視度合いが上がる。大体フォロバした瞬間にDMでスパムやアフィリエイトリンク、情報商材のリンクを自動で送ってくる経験則から来てると思う。数撃ちゃ当たると思いやがってと思いつつ、フォロバ即商材DMを貰うと「俺もまだまだだな…」ってなる。

中でもエロだとか「人気のツイート格言まとめ」「話題の画像、人気画像まとめ」といった転載、パクツイbotはそもそも信念としてフォローしていないが、仮にTwitter公式認証のついたアカウントでもパクツイや商材DMと同じことやってくるアカウントはあるので疑念が強い。

すぐDM送ってくるタイプの

　
他にも「月間100万PVみたいなブログアピールがすごい」とか「〜〜の中の人です」系は、実際直近のツイートを見に行くと完全な宣伝botと化していて、見る必要性がないなと思ってフォローしないパターンが多い。その点、最新Tweetの投稿アプリがTweetbotだったりすると、ツイートを舐めなくても直感的に分かりやすい。

amebloまで見に行くとわかる情報商材屋

bioという意味では「エンジニア」も怪しい単語に最近入りつつある。
「アプリ作ってます」と書いてあるのでフォロバしてリストに入れようかなと思うと一生自アプリを宣伝するbotだったり、「初心者エンジニアがプログラミング教室通って独立して月50万」みたいなリンクを送ってくるパターンは、体感でかなり増えている。後は、捨て垢で質問だけするためにフォローとか。QiitaとかTeratailとかQuoraとかでやると良いと思う。

酷い乱数IDだとか、ツイートが0~10、アイコンもヘッダーもプロフィールもなしとか、情報出さずに鍵垢とかは普通に判断に困る。
IDでは、「(@hoge)の移行垢、サブ垢」とかも危険で、実際はサブでも何でもないパターンがあってこれも判断に困る。
実際「ばんくし」で検索した時に出てくるアカウントはvaaaaanquish以外私ではないし…

「興味あり、憧れて、勉強中、初心者」辺りもレベル差が激しいだけでなく、「同じ初心者を釣って相互フォローになって商材を送るアカウント」である場合があるので、本当に勉強してるかとかは見ないと大体「俺もまだまだだな…」ってなる。

　
ただ、これらは私が経験的に警戒しているというだけで、「こうしているとフォロバされない」とかでもなければ「こうした方が良い」とか「こうあるべき」という話ではない。もちろん自動DMのアカウントを作るなんて、プログラミングを2日くらいやれば誰でも出来るだろうし、そういうTwitterアカウント運用ツールが存在する事も知っていて、それ自体が悪いとは一切思ってない。
どれだけ謎めいたアカウントでも、アカウント作成日が2009年とか、最新ツイートが完全に人に寄っていたりだとか、明らかに転載でない画像投稿があると「ん〜、まあいいか」となる場合もあるし、知ってる人だと何もなくてもフォローしたりする。警戒する単語があったからどうするというより、総合的に見て私がモデルになって判断するというだけである。

　
積極的な方で言うと、最近は「オタク全面出しアカウント」がむしろめちゃくちゃ信頼できる。
2010年〜2015年くらいのTwitterは、アニメアイコンのアカウントが溢れかえっていて色んな所で争っていたり騒動を起こしていた事もあって「え、あの人フォローしてるん？」みたいな事を言われる機会も多かったのでフォローを厳選する必要は正直あったけど、良くも悪くも年齢層の変化か頓珍漢な人が減ったか自身のアンテナの変化か、〇〇好きをアピールしてくれている人の方がリスト分けもしやすいし、まだ見ていられるので閾値がグッと下がる。政治や意識だけ高い人より好きな事喋ってる人は良いよね…という事かもしれない。

全部車！ガジェット！みたいな雰囲気だったり、アイコンとヘッダーとbioが統一的な（概念的に近いような）アカウントだったり、得意な事が統一的に書いてあったり、〇〇株式会社のエンジニアみたいな人もフォロバする時の安心感がすごいし、何よりリストに入れやすい。

安心感的なところでいうと、はてな等のブログ、プロフィールサイト、Amazon wishlist、Qiita、Kaggle、競プロ、…他サービスと共通のIDが見えているとスパムアカウントではないだろうなと思うし、IDと同じドメインを持って運用しているアカウントだと即フォロバできるくらいの勢いがある(最悪whoisできる)。画像投稿もあると人柄が見えて実は嬉しい。

安心感があるアカウント

後は「bioが単発のダジャレで面白い」とか「幾何学模様アイコン」「数式、文字アイコン」「猫アイコン」とか「bioにvimと書いてある」とか感覚的に良いと思っている指標はいくつかあるけど、大体その辺が見えていれば一旦信頼できると思っている。

最悪今回作ったツールみたいなやつでフォロー解除していけばいいだけなので、まずはフォローから。

　
こんな感じの概念を元に、フォロワーを見るためだけに作ったのが今回のツールである。

- できたもの -

GitHubにコードを置いた。

github.com

近年では、TwitterのAPI規制が激しく、こんなしょうもないアプリを公開してペイできるほど甘い世界じゃなくなりつつある。API有料化自体は賛成しているが、ある程度定期的に使われて、収入源があるAppじゃないと厳しいという現実はある。

自分用に作ったTwitterのAPIを使ったアプリ、調べれば調べるほど公開するコスパが悪すぎる。個人でTwitterクライアントとか作ってる人、もはや狂人か石油王だけでは。

— ばんくし (@vaaaaanquish) May 4, 2019

なので、TwitterAPI keyとClient Secretを自身で取ってこれて、Python実行環境が作れる人に絞って一旦公開する運びとなった。
アホみたいなTwitter連携アプリがガンガンリリースされていたあの頃に戻りたいという気持ちもある。

概念

概念として3つ大事な要素を考えて作った。

• バックエンド何も考えたくないのでdata_managerなる神を1つ作ってAPIやファイル操作、pandas I/Oを一任したい
• 主目的であるWebAppでフォロバ作業はしたい
• 先述のように自分がどんな基準でフォローするかを考慮しながら調整、分析のPDCAを回したい
  • ipynbで分析できると嬉しい
  • MLモデル作れるようになりたい
  • 更新、拡張可能な形式だと嬉しい

ハッカソンで出した概念図的なやつ

pandasで管理されてるので、そのファイルさえ読み込めば直接ipynbで分析したりできるのが今回の目的に沿ってると思う。
私のfollow情報を用いた簡易分析例のipynbはこんな感じ
twitter_manager/簡単に見ていくやつ.ipynb at master · vaaaaanquish/twitter_manager · GitHub

こういうのがやりたかったんだよな

フォロワーbioとtwから生成したワードクラウド

jsonファイルの管理が適当なのでDB建てるなり徐々に綺麗になっていくと思う。

- おわりに -

書き始めてから気付いたけど、技術的な事はハッカソンでやり切ってるので特に目立った工夫もないし、技術記事にする必要はなかったかも…

邪念な文章な気がするけど、はてなブログは大体これで良いと思う。

2時間くらいフォロバ作業してたらソッコーAPI枯れたので悲しい図

"You are unable to follow more people at this time." でた

— ばんくし (@vaaaaanquish) May 18, 2019

Pytorchでpretrainモデルを用いた画像向けSiameseNetworkのメモ

- はじめに -

NIPS 2016のSiamese Neural Networks for One-shot Image Recognitionを参考に、画像の距離学習を行う。

Siamese Networkは、各クラスの画像量にバラつきがあり、一部クラスが数枚しかない学習データでも上手く学習させられるネットワークである。

「特徴量同士の距離が近い画像」を探す事で、分類や検索といった問題を解くために利用できる。
やりたい事としては以下のような感じ。

sample (food-101 datasetより)

本記事は、以下の記事で利用したcnn_finetuneライブラリを用いて、pnasnet5largeをpretrainとした、画像の距離学習のtrain, testを行うメモである。

vaaaaaanquish.hatenablog.com

また、最後にfood-101データセットを用いた結果の例を示す。

- train -

距離関数とSiamese Networkを定義し、trainする。

最も簡単な構成で、pre trainから実現する。

距離関数の設計

距離関数には、古典的なContrastiveLossを利用する。
より一般的には、この距離関数の設定がtest時の精度に大きく影響するため、対象となるデータやモデルの大きさに応じて、変更すると良い。

import torch

class ContrastiveLoss(torch.nn.Module):
    def __init__(self, margin=1.0):
        super(ContrastiveLoss, self).__init__()
        self.margin = margin

    def forward(self, x0, x1, y):
        diff = x0 - x1
        dist_sq = torch.sum(torch.pow(diff, 2), 1)
        dist = torch.sqrt(dist_sq)
        mdist = self.margin - dist
        dist = torch.clamp(mdist, min=0.0)
        loss = y * dist_sq + (1 - y) * torch.pow(dist, 2)
        loss = torch.sum(loss) / 2.0 / x0.size()[0]
        return loss

参考: Chopra, Sumit, Raia Hadsell, and Yann LeCun. ”Learning a similarity metric discriminatively, with application to face verification.”, CVPR 2005.

モデルの定義

前述した通り、cnn_finetuneライブラリを用いてpnasnet5largeのimagenet pretrainモデルを導入し、そのネットワークの中間層出力を用いてSiamese Networkを構築する。

Siamese Neural Networks for One-shot Image Recognitionより

上図のInput, hidden layerをpnasnetの特徴量抽出部分に変更し、後段の層も深くしてみる。

from cnn_finetune import make_model
import torch.nn as nn

resize = (256, 256)  # 入力画像サイズ

class Identity(nn.Module):
    def __init__(self):
        super(Identity, self).__init__()        

    def forward(self, x):
        return x


def make_pnas():
    model = make_model('pnasnet5large', pretrained=True, input_size=resize)
    model.module._classifier = Identity()
    return model


class SiameseNetwork(nn.Module):
    def __init__(self):
        super(SiameseNetwork, self).__init__()
        self.cnn = nn.Sequential(
                make_pnas(),
                nn.Linear(4320,500),
                nn.ReLU(inplace=True),
                nn.Linear(500, 10),
                nn.Linear(10, 2))

    def forward(self, input1, input2):
        output1 = self.cnn(input1)
        output2 = self.cnn(input2)
        return output1, output2

SiameseNetworkは別のネットワークから2つの出力を行う形となる。

データ読み込み部の設計

学習のためのデータセットとしては、前回記事同様、以下のようなheader付きのhogehoge.csvなる「学習画像の名前(ImageName)」と「学習画像に対するラベル(ImageLabel)」がある想定。

ImageName,ImageLabel
0001.jpg,1
0002.jpg,3
0003.jpg,0
0004.jpg,3

画像の変形によるアップサンプリングを行いながら、等確率で「同じラベルの画像」「別ラベルの画像」を返す実装を以下に示す。
transformには、pretrainで利用されているImageNet画像の平均、分散値を利用する。

import torch
import os
import pandas as pd
import pickle
import torchvision.transforms as transforms
from torch.utils.data import Dataset
from PIL import Image


resize = (256, 256)  # 入力画像サイズ
trans= [transforms.Resize(resize),
            transforms.ColorJitter(brightness=0.3, contrast=0.3, saturation=0.3, hue=0.3),
            transforms.RandomHorizontalFlip(),
            transforms.RandomAffine(0.3, shear=0.3),
            transforms.RandomRotation(degrees=30),
            transforms.ToTensor(),
            transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225))]


class MyDataSet(Dataset):
    def __init__(self, root_dir):
        self.transform = transforms.Compose(trans)
        self.train_df = pd.read_csv('hogehoge.csv')
        self.root_dir = './train'
        self.images = list(self.train_df.ImageName.unique())
        self.labels = list(self.train_df.ImageLabel.unique())
      
    def __len__(self):
        return len(self.images)
    
    def image_open(self,t):
        image = Image.open( os.path.join(self.root_dir, t) )  
        return image.convert('RGB')

    def __getitem__(self, idx):
        # labelに対して画像を選択
        source_label = self.labels[idx]
        source_image_name = self.train_df.query('ImageLabel=="{}"'.format(source_label)).sample(1)['Image'].iloc[0]
        
        # labelに対して同じラベル、違うラベルをそれぞれ50%で返す
        if random.randint(0,100)<50:
            target_image_name = self.train_df.query('ImageLabel=="{}"'.format(source_label)).sample(1)['Image'].iloc[0]
            label = 1
        else:
            target_image_name = self.train_df.query('ImageLabel!="{}"'.format(source_label)).sample(1)['Image'].iloc[0]
            label = 0
        
        # 画像ロード
        source_image = self.image_open(source_image_name)
        target_image = self.image_open(target_image_name)
        
        return self.transform(target_image), self.transform(image), label

kwargs = {'num_workers': 1, 'pin_memory': True}    
train_set = MyDataSet()
train_loader = torch.utils.data.DataLoader(
    train_set, batch_size=32, shuffle=True, **kwargs)

実際は同じ画像を学習しないだとか、少ないラベルをなるべくサンプリングするだとか、現在のモデルの精度によって当たらないラベルを多めにサンプリングするなどすると良い。

optimizer

前回記事あまり考えずSGDを選択する。criterionには前述したContrastiveLossを利用する。

import torch.nn as nn
import torch.optim as optim
criterion = ContrastiveLoss()
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9)

train run

学習を回す。

from torch.autograd import Variable

def train(epoch):
    model.train()
    for batch_idx, (x0, x1, labels) in enumerate(train_loader):
        labels = labels.float()
        # x0, x1, labels = x0.cuda(), x1.cuda(), labels.cuda()
        x0, x1, labels = Variable(x0), Variable(x1), Variable(labels)
        output1, output2 = model(x0, x1)
        loss = criterion(output1, output2, labels)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    torch.save(model.state_dict(), '../model/model-epoch-%s.pth' % epoch)

model = SiameseNetwork()
# model.cuda()
for epoch in range(1, 100):
    train(epoch)

- test -

データが少なければ全てのデータを特徴量にして、全てのtrain, testデータに対して総当たりで距離を計算すれば良い。

高次元ベクトル検索ライブラリを利用して検索する。

もし保存したtrainのモデルを利用したい場合は、前回記事を参照。

データ読み込み部の設計

test用にデータを読み込むだけのクラスを定義する。

import os
import pandas as pd
import torchvision.transforms as transforms
from torch.utils.data import Dataset
from PIL import Image

tran = [transforms.Resize(resize),
            transforms.ToTensor(),
            transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225))]

class MyDataSet(Dataset):
    def __init__(self, root_path, csv, pg=False):
        self.df = pd.read_csv(csv)
        self.root_path = root_path
        self.images = list(self.df.ImageName.unique())
        self.transform = transforms.Compose(tran)
        
    def __len__(self):
        return len(self.images)
    
    def __getitem__(self, idx):
        image_name = self.images[idx]
        image = Image.open( os.path.join(self.root_path, image_name) )
        return self.transform(image), image_name

test_set = MyDataSet('./test/', './test.csv')
test_loader = torch.utils.data.DataLoader(test_set,batch_size=1, shuffle=False)
train_set = MyDataSet('./train/', './train.csv')
train_loader = torch.utils.data.DataLoader(train_set,batch_size=1, shuffle=False)

trainとtestどちらも特徴量化して、testの1画像をqueryとして、train内から近いものを探す形で、分類問題や検索問題を解く。
そのため、trainは全て特徴量にしておく。

test run

画像を特徴量にする。
もしデータが大量の場合は、DBに保存したり、後述する高次元ベクトル検索ライブラリを利用する。

import numpy as np
import torch.nn as nn

model = model.eval()
aap2d = nn.AdaptiveAvgPool2d(output_size=1)
# aap2d.cuda()

# - test images -
test_output = []
test_output_name = []
for batch_idx, (x,name) in enumerate(test_loader):
    # x = x.cuda()
    output = model.cnn[0].module.features(x)
    output = aap2d(output).squeeze()
    test_output.append(np.array(output.cpu().tolist()))
    test_output_name.append(name)

# - train images -
train_output = []
train_output_name = []
for batch_idx, (x,name) in enumerate(train_loader):
    # x = x.cuda()
    output = model.cnn[0].module.features(x)
    output = aap2d(output).squeeze()
    train_output.append(np.array(output.cpu().tolist()))
    train_output_name.append(name)

全てのデータから探索する

全てのデータ同士のユークリッド距離を計算してやるやつのサンプル。
kaggleなど全データが少ない時に使える。

for tx,tname in zip(test_output, test_output_name):
    dists = []
    for y in train_output:
        dists.append(np.linalg.norm(tx-y))
    # 小さい順に
    j = sorted(list(zip(dists, train_output_name)), key=lambda x: x[0])
    print(tname[0], j[:10])
    break

検索ライブラリに突っ込む

全部探索してたら時間がすごいかかるので、高次元ベクトル検索ライブラリを利用する方法がある。

検索では私は大体何も考えずにnmslibに突っ込む（導入が最も簡単）。

# pip install nmslib
# https://github.com/nmslib/nmslib/tree/master/python_bindings

import nmslib

index = nmslib.init(method='hnsw', space='cosinesimil')
index.addDataPointBatch(train_output)
index.createIndex({'post': 2}, print_progress=True)

ids, distances = index.knnQuery(test_output[0], k=19)
print(ids)
print(distances)

検索ライブラリの比較：qiita.com

その他、Yahoo! JAPAN社もNGTなるライブラリ出してるので要検討。

GitHub - yahoojapan/NGT: Neighborhood Graph and Tree for Indexing High-dimensional Data

food-101による例

一応例としてfood-101データセットを学習して、queryに対して近い画像を出したものを示す。

www.vision.ee.ethz.ch

記事上部のドーナッツは上手くいった例で実際はこんな感じになる。

実例 (food-101データセットより)

左がquery、近い順に左から画像が並んでいる。ドーナッツに関してはドーナッツが収集できており、似た画像が拾ってきていると言える。
2行目のアイスクリームをqueryにした場合、2番目の画像についたラベルはアップルパイであった。視覚的には似てるけど。
3行目はスパゲッティで検索しているが、実際に出てくる画像は「オムレツ、パエリア、リゾット、パスタ、ハンバーガー(イカリング)」となっている。気持ちはわかるけど。
まあ大体3行目のような感じで間違えるので、後処理なんかで色々やってやると良さそう。

　
food-101 datasetは画像の明暗が激しかったり、人がインスタに載せるような角度だったり、以下のような画像が混ざっていて厳しい所もあるので前処理も重要そうだという知見が得られた。

difficult images

上記は、左から「アイスクリーム」「アイスクリーム」「ハンバーガー」「ピザ」「ブレッドプディング」である。わからん。

　
パスタやステーキ、餃子、寿司、ティラミス、マカロン、…といった他物体や人が映る可能性の低いものもあるので、タスクに応じて、カテゴリを絞ったり細分化されている所をマージするなどすると良さそう。

- おわりに -

Pytorchのpretrainモデルを利用したSiamese Networkを構築した。

verification modelの拡張としてtripret lossを利用したり、partなモデルに拡張してより細かな物体同士の距離を用いたりできるので、いつか記事として書く。GitHubにも上げる。

特に他意はない。

「機械学習のための特徴量エンジニアリング」が良かったので訳者に媚を売る

- はじめに -

本ブログでは恒例になりつつある、献本されたので媚を売るシリーズです。

「機械学習のための特徴量エンジニアリング」は2/23に発売される、機械学習エンジニアのための書籍です。

本記事は、筆者に媚びを売りつつ、どういった内容の書籍か、どういう人が読むと良さそうか、私がどう感じたかをつらつら書いていくもでのす。

機械学習のための特徴量エンジニアリング ―その原理とPythonによる実践

- 書籍の概要 -

「機械学習のための特徴量エンジニアリング」は、謎のデータサイエンス集団「株式会社ホクソエム」によって翻訳された、オライリー出版の「Feature Engineering for Machine Learning」日本語版にあたります。元の書籍を書いたAlice ZhengはMSRやAmazonで機械学習の研究開発に従事している方で、共著者のAmanda Casariは現在Googleで働いているようです。業界のトップといっても過言ではないですね…。

Feature Engineering for Machine Learning: Principles and Techniques for Data Scientists

私はこちらの英語版については会社で目を通した事があり、ふんわりとした理解で「幅広くて良い本っぽいなー」くらいに思っていましたが、日本語翻訳が出るとの事で、Twitterで情報を見た瞬間に1冊予約しました。後日ホクソエムの中の人から献本されたので、一冊は妻にでもあげようと思っています。

　
　
翻訳の株式会社ホクソエムについて「謎の集団」とは言いましたが、データサイエンスの業界においてはR言語界隈での活動やブログが非常に有名で、中の人達も寄りすぐりのエンジニアが揃った良い会社であると噂されています。

blog.hoxo-m.com

ホクソエムさんのより有名な実績として、昨年度に出版された「前処理大全」があります。

前処理大全[データ分析のためのSQL/R/Python実践テクニック]

　
こちらもまた「ITエンジニア本大賞 2019」の技術書部門TOP10入も果たした良書です。
www.shoeisha.co.jp

　
　
そして「機械学習のための特徴量エンジニアリング」の実際の書籍ですが、機械学習コンペティションの最高位となるKaggle Grand Masterである@smlyさんの推薦帯もついており、書店で目立つ事間違いなし。書店に入ってから迷わず書籍に向かい購入に至る事ができると思います。

Kaggle Grand Master推薦帯

　
上記の通り、最強に最強を重ねた布陣でオライリーから出版された書籍ということになります。

- どんな層に向けた書籍か -

前項で媚を売りきったので、どんな層が読むと良さそうか個人的な感想をまとめます。
端的にまとめると以下のようになると思います。

• 大学学部レベルの知識を持ち、今後機械学習エンジニアとしてやっていきたい人
• 機械学習のモデル構築を幅広く学びたい人
• Kaggle Expert前後くらいの人

まず、前提知識はそれなりに必要な書籍ではあると思います。「1章・機械学習パイプライン」「2章・数値データの取り扱い(前半)」では、導入のため数行でベクトル空間や分布といったワードが解説されますが、数行です。その後の2章中盤からは「線形回帰モデルによってオープンデータを学習させクロスバリデーションを行いR^2スコアの信頼区間を見ると…」「pairwiseな特徴を使ってモデリングしてみると…」という話に一気に飛躍し、コードが紹介されていきます。最低限の知識としてサンプル程度の機械学習モデルを構築した事があり、その前提となるモデルや評価に関する情報は得る、もしくは調べながら状態で読むのが良さそうです。
（翻訳者は"まえがき"において他オススメの書籍を紹介しているため、参考にすると良いと思います）

　
機械学習モデル構築を幅広く学びたい人にもオススメできる書籍です。例えば、「新卒で業務で今から本格的にモデリングするんだけど…」とか「Pythonで機械学習のサンプルスクリプトはQiitaや本見ながら動かしたけど次何すれば良いのか…」という人は読んでおくと良いと思います。理由として、この書籍は以下のような構成になっているからです。

• 1章・機械学習パイプライン
• 2章・数値データの取り扱い
• 3章・テキストデータの取り扱い
• 4章・特徴量スケーリングによる効果
• 5章・カテゴリ変数の取り扱い
• 6章・次元削減
• 7章・非線形特徴量の生成
• 8章・特徴量生成の自動化(画像特徴と深層学習)
• 9章・論文レコメンドシステム構築を模したトレーニング

数値、テキスト、画像の特徴に関する情報が万遍なく入っており、モデリングにおいていつ何時も重要な特徴量の取り扱いや生成について深く触れられています。「テキストの処理どうするんだっけ」と困ったらこの書籍と前処理大全を開けば、ひとまず正しいモデリングができるようになっていると感じました。たとえテキスト分析やったことなくても、ベースラインになるちゃんとしたモデル作れるようなコードになってるなと思います。
前述したように、事前知識が必要となる箇所が少しずつ見受けられるため、「ん？この表現よくわからないな」と思ったら調べる根気もあると良いでしょう。
　
　
Kaggle Expertくらいだと「機械学習を初めて成果が出たけど、Kernelの手法ばかり試していて、古典的な特徴量エンジニアリングのベース知識が欲しい…！」という人も多くいるでしょう。この書籍に書いてある事くらいはスラスラ出てくるくらいにしておくと良いと思います。まえがきもですが、途中でもKaggleの話出てきますよ(leakageやstackingなど)。

- 感想とか -

前述の通り、前提知識は普通に必要です。その分参考文献を多く記載しており、そちらまで読みに行ったり、自分で調べたりする根気は少なからず必要だと思います。これは悪い点というわけではなく、少し大変なので大学で最低限統計と機械学習やったりしてるくらいじゃないと初学者にポンとは渡せない的な意。あまり理論的な深みに行き過ぎると、書籍として重くなるのは間違いないので、あくまで「普段から機械学習のモデリングに触れる機会がある」という中級者を一段引き上げる書籍という立ち位置だと思います。（オライリーの本って大体そんな感じだけど）
　
k-meansで特徴量作ったりだとか特徴量ハッシングだとか、昔からある手法ではあるけど皆意外と知らないから突然はてブで話題になるというタイプのやつ最近増えてきてるし、知識の補強という意味でも、機械学習エンジニアの皆さんは読むと良いと思います。

機械学習でよく出てくる分布やplotの説明もあり、概念的な理解も進むと思います。

図とその見方の説明

個人的には、ソースコードついてるのもGoodで、全てJupyter notebookで可視化までやってくれているので、是非触って遊ぶのも良いでしょう。
github.com
　
　
強いて気になる所をあげるとするならば、訳書という事もあり、少し情報としては古いものが多めかなという気持ちにはなりました。

例えば、「深層学習ではZCAを前処理ステップとして使用する」という文言がありますが、これは2009年のLearning Multiple Layers of Features from Tiny Imagesが元になっており、今ではDeep LearningでZCAを前処理として利用するモデルはほぼ聞かないという認識です。画像で出てくるCNNもReLUなAlexNetですし、そっくりそのまま現代で最高精度かと言われたら微妙なところです。私が院生くらいの頃実際にZCAで白色化して〜というの流行ってましたし、時系列知ってれば大丈夫なので、一緒に色々調べてモダンな物についても見てみると良いと思います。他にもEmbeddingやSamplingなどで新しい手法が続々出てきている昨今では、情報を追い続けるというのも機械学習エンジニアとしての1つの力だなと改めて思いました。

(このあと前処理大全を買えというメッセージなのかも知れません…！)

- おわりに -

私個人としては、目新しい発見こそなかったものの、特徴量エンジニアリングの基礎がつまっており、「うんうん、そうだね」と頷きながら読める良書でした。今まで日本の書籍で機械学習の基礎を抑えた書籍こそあったものの、特徴量や前処理を主題にした書籍が出てくるというのは珍しく、非常に良い事だと思います。昔は本当特徴量生成の手法とか全然外に出してくれないような情報だったので、知の高速道路が整備されて良いですね。

本文では「Kaggle Expertくらいの人は買えば」と言った私がまだExpertである事を恥じながら、より勉強に励まねばという表明をして締めにしたいと思います。

知識として色々知っていても出来なきゃ意味ないですからね。がんばってやっていきましょう。
　

機械学習のための特徴量エンジニアリング ―その原理とPythonによる実践

2019/2/23発売 ただいま予約受付中みたいです

horovodを用いたPytorchの分散学習

- はじめに -

近年、分散深層学習の研究、ライブラリ開発が盛んに行われている。

本記事はuber社が公開しているhorovodを利用した分散CNNのメモである。

- 前提 -

horovodとは、バックエンドをOpenMPIとしTensorFlow、Keras、PyTorchを最小限のコード変更で分散学習できるようにするためのパッケージである。

github.com

現状TensorFlowを使って書かれたコードをDistributed TensorFlowに対応させるにはパラメータサーバやマスタサーバの動きを理解した上で多くの変更を要するが、horovodではそれらをncclのall reduceを利用しwrappingしてあるため、最小限のコード変更で分散学習が可能となる。

また、公式によると普通に書くより早いらしい（未検証）

複数ノードで利用する場合、各ノードがOpenMPIを通して疎通できる必要がある。その環境構築については以下に記載している。ChainerMNが動けば、ほぼ変更なくhorovodを動かす事ができる。

vaaaaaanquish.hatenablog.com

OpenMPI周りの設定が終わったらpipでhorovodを導入する。

pip install horovod

もしくは、DockerHubにHorovod-dockerも公開されていため、バックエンドの設定が整えば、こちらを利用する事で分散学習を始められる。
horovod/docker.md at master · uber/horovod · GitHub

　
PyTorchでCNNモデルを簡易に利用する方法は以下に記載している。
以下に記載のpretrain modelを利用したCNNモデルをhorovodで分散学習させる。

vaaaaaanquish.hatenablog.com

- 学習スクリプトの変更 -

学習を行うtrain.pyを以下に示す。
学習スクリプトはpretrainを学習させる記事に詳細を書いてあるので参考に。
読み込むデータのPathやログ出力先はマウントしているディレクトリ等でよしなに。

（※ 以下はPyTorch 0.4.0ですが、バージョンによってDataloader周りとかちょいちょい違いがあるので注意）

import os
import traceback
import datetime
import torch
import torchvision.transforms as transforms
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset
from cnn_finetune import make_model
import pandas as pd
from PIL import Image

# --- 追加 ---
import horovod.torch as hvd
torch.manual_seed(42)
hvd.init()
torch.cuda.set_device(hvd.local_rank())
torch.cuda.manual_seed(42)
# ----------

# 10クラス分類を想定
model = make_model('senet154', num_classes=10, pretrained=True, input_size=(256, 256))
criterion = nn.CrossEntropyLoss()

class MyDataSet(Dataset):
    def __init__(self, csv_path, root_dir):
        self.train_df = pd.read_csv(csv_path)
        self.root_dir = root_dir
        self.images = os.listdir(self.root_dir)
        # normalizeのmean, stdはpretrain modelより
        # https://github.com/Cadene/pretrained-models.pytorch/tree/master/pretrainedmodels/models
        self.transform = transforms.Compose([
            transforms.Resize((256, 256)),
            transforms.ColorJitter(brightness=1, contrast=1, saturation=1, hue=0.5),
            transforms.RandomHorizontalFlip(),
            transforms.ToTensor(),
            transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
        ])
        
    def __len__(self):
        return len(self.images)
    
    def __getitem__(self, idx):
        image_name = self.images[idx]
        image = Image.open( os.path.join(self.root_dir, image_name) )
        image = image.convert('RGB')
        label = self.train_df.query('ImageName=="'+image_name+'"')['ImageLabel'].iloc[0]
        return self.transform(image), int(label)

train_set = MyDataSet('train.csv', './train')

# --- 追加, 変更 ---
train_sampler = torch.utils.data.distributed.DistributedSampler(
    train_set, num_replicas=hvd.size(), rank=hvd.rank())
train_loader = torch.utils.data.DataLoader(
    train_set, batch_size=batch_size, sampler=train_sampler, pin_memory=True)
hvd.broadcast_parameters(model.state_dict(), root_rank=0)
model.cuda()
optimizer = optim.SGD(model.parameters(), lr=0.01 * hvd.size(),
                      momentum=0.9)
optimizer = hvd.DistributedOptimizer(
    optimizer, named_parameters=model.named_parameters())
# ----------

    
def train(epoch):
    total_loss = 0
    total_size = 0
    model.train()
    train_sampler.set_epoch(epoch)

    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.cuda(), target.cuda()
        optimizer.zero_grad()
        output = model(data)
        loss = criterion(output, target)
        total_loss += loss.item()
        total_size += data.size(0)
        loss.backward()
        optimizer.step()

        # --- hvd.rankで出力を絞るよう変更 ----
        if batch_idx % 100 == 0 and hvd.rank() == 0:
            now = datetime.datetime.now()
            with open('/mnt/log.text', 'a') as fa:
                fa.write('[{}] Train Epoch: {} [{}/{} ({:.0f}%)]\tAverage loss: {:.20f}\n'.format(now, epoch, batch_idx * len(data), len(train_loader.dataset), 100. * batch_idx / len(train_loader), total_loss / total_size))


# main
try:
    for epoch in range(1, 100):
        train(epoch)
         # --- hvd.rankでstate_dict保存を絞るよう変更 ---
        if hvd.rank() == 0:
            torch.save(model.state_dict(), '/mnt/senet154_{}.model'.format(epoch))
except Exception as e:
    now = datetime.datetime.now()
    with open('/mnt/log.text', 'a') as fa:
        fa.write('[{}]error: {}\n'.format(now, str(e)))
        fa.write(traceback.format_exc()+'\n')
    raise

変更箇所はコメントの通り少ない。

初回起動時に「cnn_finetune.make_model」でpretrainモデルのダウンロードが走ってしまうため、複数ノードでdockerを利用するなら一回docker内でmake_modelを実行しダウンロードしてモデルファイルをdocker imageに含めるか、lusterfsのような共通で見れるディレクトリをマウントしてそのモデルを参照するようにすると良い。

また、こちらで学習したモデル(state_dict)は、前述したpretrainを学習させる記事内のtestコードで推論できる。

- 実行 -

動作させるどこかしらのノードないしdockerにログインし以下を実行する。

もし各ノードでdockerを利用している場合は、PyTorchではdocker run時に「--ipc=host」を付けなければ「Unable to write to file」となってしまう事に留意する。
Unable to write to file </torch_18692_1954506624> - PyTorch Forums

また、horovodでは/tmpを利用するため、docker run時に「-v /tmp:/tmp」等としtmpもマウントしておく必要がある。

　
mpiexecコマンドを利用し実行する。
hostfileについてはChainerMNの記事参照。

mpiexec --allow-run-as-root \
     --mca btl_tcp_if_include ib0 \
     -mca pml ob1 \
     -mca btl ^openib \
     -x PATH=$PATH -x PYTHONPATH=$PYTHONOATH -x LD_LIBRARY_PATH=$LD_LIBRARY_PATH -x CPATH=$CPATH -x LIBRARY_PATH=$LIBRARY_PATH -x NCCL_ROOT=$NCCL_ROOT \
     -bind-to none \
     -map-by slot \
     --hostfile /mnt/host.txt \
     -np 8 \
     python3 /mnt/train.py

dockerを利用している場合「--allow-run-as-root 」が必須である。
また、ChainerMNの記事に記載のコマンドとの違いとして、以下を設定しTCP通信を強制する必要がある。

• -mca pml ob1
• -mca btl ^openib

これを設定しないと以下のようにsubprocessが次々死んでいき、全体の動作も止まってしまう。

HorovodBroadcast_residual_layer_batch_normalization_moving_variance_0 [missing ranks: 1]

　
OpenMPI 3以降であれば、以下を利用してprocessを単一CPUにバインドさせないようにする。
また、defaultではNUMA設定が単一となってしまうため、map-by slotも利用しておくと良いらしい。

• -bind-to none
• -map-by slot

ログが吐かれ始めれば成功。

参考：https://github.com/uber/horovod/blob/master/docs/running.md

- おわりに -

「horovodならコード変更最小限に分散学習！」とは言うけど、OpenMPIが動く前提があり、正直「何よりOpenMPIが動作する環境を作るのがしんどいんじゃい…」と思う。

OpenMPIのsettingが一通り上手くいってしまえば、後はかなり自由にモデリングできると思う。各ノードにhvd.broadcastで別々のデータを送ったり、hvd.allreduceでなくallgatherを使えばaggregationの方法を追加できたりするので結構柔軟に書けるとも思う。

要は使い分け。

2018年のxonshrc

- はじめに -

この記事はXonsh Advent Calendar 2018 - Qiita最終日の記事です。

私のxonshrc、業務用のコマンドとか含めると大体1500行くらいあるんでgithubで公開するんじゃなくて一部ずつ切り取ってここで紹介しようかなと思いました。

なんか作る時の参考になれば幸いです。

- 基本的な設定 -

以下に過去の大まかな設定が書いてあります。
xonshのPROMPTにdatetimeを表示する - Stimulator

環境変数

基本的な設定は大体以下のようになっています。

$EDITOR = '/usr/local/bin/vim'
$VISUAL = '/usr/local/bin/vim'
$VI_MODE = False
$COMPLETIONS_CONFIRM = True
$IGNOREEOF = True
$INDENT = "    "
$CASE_SENSITIVE_COMPLETIONS = False
$HISTCONTROL = "ignoredups"
$XONSH_AUTOPAIR = False
$AUTO_CD = True
$XONSH_SHOW_TRACEBACK = True
$SUPPRESS_BRANCH_TIMEOUT_MESSAGE = True
$UPDATE_COMPLETIONS_ON_KEYPRESS = True
def get_git_user():
    return '{BLUE}' + $(git config user.name).strip() + ' {INTENSE_GREEN}{hostname}{WHITE} ( {YELLOW}"{cwd}" {WHITE}) {INTENSE_RED}$ '
$PROMPT = get_git_user
$LS_COLORS="di=34:ln=35:so=32:pi=33:ex=31:bd=46;34:cd=43;34:su=41;30:sg=46;30:tw=42;30:ow=43;30"
$XONSH_HISTORY_SIZE = (3000, 'commands')
XONSH_HISTORY_MATCH_ANYWHERE = True
$PTK_STYLE_OVERRIDES={
 "completion-menu": "bg:ansiblack ansiwhite",
 "completion-menu.completion": "bg:ansiblack",
 "completion-menu.completion.current": "bg:ansiwhite ansiblack",
 "scrollbar.background": "bg:ansibrightblack",
 "scrollbar.arrow": "bg:ansiblack ansiwhite bold" ,
 "scrollbar.button": "bg:ansiblack",
 "auto-suggestion": "ansibrightblack",
 "aborting": "ansibrightblack",
 }

前年との違いで言えば以下辺りが大きい気がします。

• get_git_userによってPROMPTにgit user nameを出すようにした
• PTK_STYLE_OVERRIDESによってstyleを上書きした

githubの所持ユーザ名が増えて、関係ないリポジトリにvaaaaanquishでpushしてしまうみたいな事が2回起こったので確認するようにしました。
後述しますが、gitユーザ変える関数とかも作っています。
またPTK_STYLE_OVERRIDESは以下のissueでも発言した通り、目がチカチカするのを抑えるために必須です。
Change auto_completer's color style · Issue #2840 · xonsh/xonsh · GitHub

キー入力ごとに更新されるCOMPLETIONSですが、iTerm2のcolorと合わせて調整しています。

補完の色合いとか

xonshの補完はzsh風にもできるんですが、私はこちらに慣れてしまいました。

aliases

エイリアスは別にgithubでxonshrcで検索して好きなだけ改造すればと思いますが一応省略して外観を。

if platform.system() == 'Darwin':
    aliases["lt"] = "colorls --tree"
    aliases["l"] = "colorls -ltr --sf"
    aliases["la"] = "colorls -la"
else:
    aliases['ls'] = "ls --color=auto"
    aliases["l"] = "ls -lh"
    aliases["la"] = "ls -lha"
    aliases['free'] = "free -h"
    aliases['f'] = 'free -h'
    aliases['wf'] = 'watch free -h'
aliases['ee'] = "exit"
aliases["v"] = "vim"
aliases["vi"] = "vim"
aliases["vx"] = "vim ~/.xonshrc"
aliases["vv"] = "vim ~/.vimrc"
aliases["vs"] = "vim ~/.ssh/config"
...(gitとか色々etc)

MacとLinuxで分けれるように書いてあります。
あと "ee" でexitして、zshrcでもbashrcでもに "x" でxonsh起動するようにしています。xonshrcもすぐ開けるようにしておけば、xonsh本体の開発にコントリビュートする時に便利です。

xontrib

xontribは以下に落ち着いています。zコマンド病みつき。

xontrib load z
xontrib load readable-traceback
$READABLE_TRACE_STRIP_PATH_ENV=True
$READABLE_TRACE_REVERSE=True
xontrib load coreutils

最近はxonshもptkも開発が早く、まだptkのスピードに追いついてないxontribが見受けられます。
まあ書き直せばいいだけですが、自分は今の所こんな感じです。
（readable-tracebackもpython3.7で動かない情報があるので時間見つけて直します…）

他のxontribはこのへんに
Xontribs — xonsh 0.8.8 documentation
xonshにおけるxontribの紹介 - Stimulator

- ライブラリ周り -

ライブラリの管理はpipでやってます。
その他import周りを工夫しておくことでより便利に扱う事ができます。

import, xontrib_load時の自動install

ライブラリの管理について以下記事で書いていますが、import、xontrib load時にライブラリがない場合、自動でpip installするようにしてあります。
Python moduleがない場合に自動でpip installする - Stimulator

以下のようにすることで、新しい環境でひとまずxonshだけ動かしたいという時に、xonshrcを送るだけで対応できます。

import importlib
import builtins
from xonsh.xontribs import xontrib_context, update_context
import xonsh.xontribs

def _import(name, module, ver=None):
    try:
        globals()[name] = importlib.import_module(module)
    except ImportError:
        try:
            import subprocess
            cmd = "sudo pip install -I {}".format(name)
            subprocess.call(cmd, shell=True)
            globals()[name] = importlib.import_module(module)
        except:
            print("can't import: {}".format(module))
_import('pexpect','pexpect')

def _update_context(name, ctx=None):
    if ctx is None:
        ctx = builtins.__xonsh__.ctx
    if not hasattr(update_context, "bad_imports"):
        _update_context.bad_imports = []
    modctx = xontrib_context(name)
    if modctx is None:
        import subprocess
        cmd = "sudo xonsh -c 'xpip install -I xontrib-{}'".format(name)
        subprocess.call(cmd, shell=True)
        remodctx = xontrib_context(name)
        if remodctx is None:
            _update_context.bad_imports.append(name)
            return ctx
        return ctx.update(remodctx)
    return ctx.update(modctx)
xonsh.xontribs.update_context = _update_context

xonshアップグレードでPATHが見えなくなったりした時にも発動するので割と重宝しています。
実際削除

実際にライブラリなくてもインストールされる様子

importの遅延ロード

以下の記事で書いていますが、xonshrcでは起動が遅くなるのでimportしないけど、xonsh実行時に逐一import書く必要がないようにlazy_moduleを利用しています。
Pythonモジュールの遅延import - Stimulator

import importlib
from xonsh.lazyasd import lazyobject

lazy_module_dict = {
    'pd': 'pandas',
    'np': 'numpy',
    'requests': 'requests',
    'sci': 'scipy',
    'plt': 'matplotlib.pyplot',
    'Path': 'pathlib.Path',
        }
for k,v in lazy_module_dict.items():
    t = "@lazyobject\ndef {}():\n    return importlib.import_module('{}')".format(k, v)
    exec(t)

色々追加したり消したりしましたが、今は利用頻度高いけどimportにちょっと時間がかかる系統が残り上記で落ち着いています。

遅延numpy

- 認証周り -

基本的にパスワード入力をpexpectで処理しています。大体以下にも書いています。
xonshにおけるpexpectを利用した対話コマンド自動化 - Stimulator
ssh等は2要素認証等も含めて自動でやるようにしています。

1passwordを利用したpassword, one-time-passwordの取得

私は、ワンタイムパスワードも利用したい事を理由に全てのパスワード管理に1password、それらをshell上で利用するため1password-cliを使っています。
1Password command-line tool: Full documentation

import pexpect

# masterパスワードをstrで取得する（自分で書いて）
masterp=get_master_pasword()

# 1password-cliから1password.appへの認証
def _pass_auth():
    # 1pass auth
    p = pexpect.spawn("op signin my.1password.com --output=raw")
    while 1:
        i=p.expect([
            r'.*(Enter the password for).*',
            pexpect.EOF, pexpect.TIMEOUT], timeout=3)
        if i==0:
            p.sendline(masterp)
            p.sendline('')
        elif i in [1,2]:
            break
    return str(p.before.strip())[2:-1]


# サービスxのパスワードを取得
def _get_p(x):
    # get_pass
    $op_key = _pass_auth()
    p = $[echo $op_key | op get item @(x) | jq '.details.fields[] | select(.designation=="password").value']
    return p
aliases['getp'] = _get_p


# サービスxのワンタイムパスワードを取得
def _get_op(x):
    # get one time password
    print('auth 1password...')
    $op_key = _pass_auth()
    print('get one time pass...')
    p = $(echo $op_key | op get totp -v @(x))
    return p.strip()
aliases['getop'] = _get_op

masterパスワードをstrで取得する部分はファイル読み込みとかで自身で書いて欲しいです。
これでコマンドで、パスワードやワンタイムパスワードを取得できます。

$ getp yahoo
hoge1234

$ getop yahoo
987123

これをpbcopyコマンドとかに送ればクリップボードから直接ペーストできます。
最近の悩みは_pass_authが若干遅い事です。（多分認証キャッシュの機能で解消できるけどやってない）

sshの自動化

上記のパスワード取得の仕組みを利用しながらssh周りも自動化しています。
サーバの接続の際にも踏み台などでパスワード、ワンタイムパスワードの入力が必須な作業環境で便利です。

import pexpect

# masterパスワードをstrで取得する（自分で書いて）
masterp=get_master_pasword()

# 画面サイズ
import curses
curses.setupterm()
term_lines = int(curses.tigetnum("lines"))
term_cols = int(curses.tigetnum("cols"))

# ssh認証
def _ssh_pex(p):
    while 1:
        i = p.expect([
            r".*(Enter passphrase for key).*",
            r".*(Are you sure you want to continue).*",
            r".*(Verification code).*",
            pexpect.EOF, pexpect.TIMEOUT], timeout=3)
        if i==0:
            print('enter passphrase.')
            p.sendline(masterp)
            p.sendline('')
        elif i==1:
            print('continue: yes.')
            p.sendline('yes')
            p.sendline('')
        elif i==2:
            print('[auth]')
            otp = _get_op('yahoo')
            p.sendline(otp)
            p.sendline('')
        elif i in [3,4]:
            break
    p.interact()

こんな感じで書いておけば上手くいくと思います。
ワンタイムパスワードの期限がきれても_get_opリトライしてくれるので便利です。

これを利用してsshやscp周りを拡張しています。

import pexpect

def _ssha(x):
    if '/' in x[0]: x[0]=x[0].replace('/','')
    p = pexpect.spawn("ssh " + x[0]) #, encoding='utf-8')
    p.setwinsize(term_lines,term_cols)
    # p.logfile = sys.stdout
    _ssh_pex(p)
aliases["ssha"] = _ssha

def _scpa(x):
    if '/' in x[0]: x[0]=x[0].replace('/','')
    p = pexpect.spawn("scp -r " + x[0] + ' ' + x[1])
    _ssh_pex(p)
aliases["scpa"] = _scpa

なんかたまに入るゴミのために変な処理が入ってますが、大体これでなんとかなると思います。

これとconfig組み合わせてワンタイムパス必須なサーバへの接続も自動化しています

local $ ssha gpu001
[auth]
auth 1password...
get one time pass...[123456]
bastion server...
fowarding port [8888]

gpu001.server.hoge.co.jp $

この辺柔軟に書けるのは良いところだと思ってます。

ssh-hostの管理

会社でも趣味でもめちゃくちゃ沢山のhostにsshしたりコマンド飛ばしたりします。
そのため~/.ssh/configがぐちゃぐちゃになってしまう問題があったので、xonshコマンドで管理しています。
　
~/.ssh/configを雑にparseして表示、もしくはstrで返すxonsh関数です。
ptkのstyleを利用して、colorで出したり出さなかったりしています。

from prompt_toolkit import print_formatted_text
from prompt_toolkit.styles import Style
inquirer_style = Style.from_dict({
    'qa': '#5F819D',
    'qu': '#FF9D00',
    'dp': '#000'
})
def _get_host(color=False):
    all_text = ''
    text = ''
    for x in $(cat ~/.ssh/config).split('\n'):
        if 'LocalForward' in x:
            text += ', ' + x.strip().split(' ')[1]
        if 'HostName' in x:
            text += ', ' + x.strip().split(' ')[1]
        elif 'Host ' in x:
            if text!='':
                all_text += text + '\n'
            text = x.split(' ')[1]
    all_text += text + '\n'
    if not color:
        all_d = []
        for x in all_text.split('\n'):
            for i,y in enumerate(x.split(', ')):
                if i==0:
                    all_d.append(('class:qu', y))
                if i==1:
                    all_d.append(('', ', '))
                    all_d.append(('class:qa', y))
                    if len(x.split(', '))==2:
                        all_d.append(('','\n'))
                if i==2:
                    all_d.append(('', ', '))
                    all_d.append(('class:qp', y))
                    all_d.append(('','\n'))
        print_formatted_text(FormattedText(all_d),
                style=inquirer_style)
        return
    return all_text
aliases['host']=_get_host

こんな感じでhost一覧出したりgrepしたりpecoしたりしています。

これを利用して~/.ssh/configも読み書きできるようにしておくと、柔軟にお仕事できます。
一部切り抜いて紹介すると以下。

# _get_host()してparse
def _ssh_host_to_dic():
    host = {}
    for x in _get_host(True).split('\n'):
        rows = x.split(', ')
        if len(rows)>=2:
            host[rows[0]]={'s':rows[1], 'p':None}
        if len(rows)==3:
            host[rows[0]]['p']=rows[2]
    return host

# config生成
def _w_host(host):
    # 中身はよしなに
    with open('~/.ssh/config', 'w') as f:
        f.write('HostKeyAlgorithms +ssh-dss\n')
        f.write('AddKeysToAgent yes\n')
        f.write('host *\n')
        f.write('    ForwardAgent yes\n')
        for k,v in host.items():
            f.write('\nHost {}\n'.format(k))
            f.write('    HostName {}\n'.format(v['s']))
            if k != 'hoge':
                f.write('    IdentityFile ~/.ssh/id_rsa\n')
                f.write('    ProxyCommand ssh hoge.co.jp ncat %h %p\n') # springboard
                f.write('    IdentitiesOnly yes\n')
                if v['p'] is not None:
                    f.write('    LocalForward {} localhost:{}\n'.format(v['p'],v['p'])) #LocalForward

import argparse
def _update_host(args):
    host = _ssh_host_to_dic()
    parser = argparse.ArgumentParser()
    parser.add_argument('name')
    parser.add_argument('-s')
    parser.add_argument('-p')
    args = parser.parse_args(args)
    if args.name not in host.keys():
        host[args.name]={'s':None, 'p':None}
    if args.s is not None:
        host[args.name]['s'] = args.s
    if args.p is not None:
        host[args.name]['p'] = args.p
    _w_host(host)
aliases['uh'] = _update_host

ちょっと切り抜きなので雑ですが、こんなん作っとけば以下でconfig更新したりしてサーバが増減しても安心です。

$ cat ~/.ssh/config
HostKeyAlgorithms +ssh-dss
AddKeysToAgent yes

$ uh hogehoge -s hoge.server -p 9920
$
$ cat ~/.ssh/config
HostKeyAlgorithms +ssh-dss
AddKeysToAgent yes
host *
    ForwardAgent yes

Host hoge
    HostName hoge.server
    IdentityFile ~/.ssh/id_rsa
    IdentitiesOnly yes
    LocalForward 9920 localhost:9920

あとは、自身が欲しいままにconfig_generatorを作っておけば安心。
Tipsですが、ちょっと複雑なコマンド作るときはargparseしておくと --help でオレオレdiscription出せるのでいい感じに使えます。

gitアカウントの管理

gheとかgitのcliは充実してますが、自分は以下だけ。
いくつかあるgitアカウント間違えないように変更しやすくしています。

# git chenge
def g_change(account):
    account = account[0]
    if account=='vanquish': account='vaaaaanquish'
    git config --global user.name f"{account}"
    if account=='hoge':
        git config --global user.email hoge@company.jp
        print(f'change:{account}')
    elif account=='vaaaaanquish':
        git config --global user.email 6syun9@gmail.com
        print(f'change:{account}')
    else:
        print(f'check account name:{account}')
aliases['gac']=g_change

gitacコマンドで複数のアカウントを切り替えて、上記「 環境変数」の項目で書いたようなスクリプトでshellのPROMPTに表示しています。

hoge $
hoge $ gac vanquish
shukawai $

多分もっと優勝できると思ってます。

- 移動、操作 -

移動はもっぱらzとpecoを使っています。
GitHub - peco/peco: Simplistic interactive filtering tool
GitHub - rupa/z: z - jump around

xontrib-zを導入しても良いですし、以下のように書いても良い感じに使えると思います。

def z():
    lines = open($DIR_HIST_PATH[0]).read().rstrip("\n").split("\n")
    return("\n".join([p for p, c in Counter(lines).most_common()]))

履歴の取得

以下の記事で紹介されているものを利用しています。
Xonshを快適にするptk(Python Prompt Toolkit) - Qiita
以下の記事と書きましたが、書いたのは元同僚で私にxonshを薦めてきた諸悪の根源です。

import os
import json
from collections import OrderedDict

def get_history(session_history=None, return_list=False):
    hist_dir = __xonsh__.env['XONSH_DATA_DIR']
    files = [ os.path.join(hist_dir,f) for f in os.listdir(hist_dir)
              if f.startswith('xonsh-') and f.endswith('.json') ]
    file_hist = []
    for f in files:
        try:
            file_hist.append(json.load(open(f))['data']['cmds'])
        except:
            pass
    cmds = [ ( c['inp'].replace('\n', ''), c['ts'][0] )
                 for cmds in file_hist for c in cmds if c]
    cmds.sort(key=itemgetter(1))
    cmds = [ c[0] for c in cmds[::-1] ]
    if session_history:
        cmds.extend(session_history)
    # dedupe
    zip_with_dummy = list(zip(cmds, [0] * len(cmds)))[::-1]
    cmds = list(OrderedDict(zip_with_dummy).keys())[::-1]
    cmds = reversed(cmds)
    if return_list:
        return cmds
    else:
        return '\n'.join(cmds)

後述するkeybindとpecoの連携によって、過去のコマンド実行をあいまい検索できるようにしています。

dirの保存

移動したディレクトリは、on_chdirで発火する関数を作成しファイルでざっくり管理しています。

# file save dir
$DIR_HIST_PATH = "~/.dirhist"
@events.on_chdir
def add_to_file(olddir, newdir, **kw):
    with open($DIR_HIST_PATH[0], 'a') as dh:
        print(newdir, file=dh)

これも後述のkeybindとpecoの連携部分で最近行ったdirをpecoで利用したいがためです。
普段はこれをもうちょっと拡張して、特定の作業ディレクトリに行きやすくなったりするようにしています。

keybindings

キーバインドです。
何でもかんでもpecoに流しています。

from prompt_toolkit.keys import Keys
from prompt_toolkit.filters import (Condition, IsMultiline, HasSelection, ViInsertMode)


@events.on_ptk_create
def custom_keybindings(bindings, **kw):

    # ctrl+vで入力中の単一、複数行コマンドをvimで開く
    @bindings.add('c-v')
    def edit_in_editor(event):
        event.current_buffer.tempfile_suffix = '.py'
        event.current_buffer.open_in_editor(event.cli)

    # ctrl+rで過去の実行コマンドをpeco
    @bindings.add('c-r')
    def select_history(event):
        sess_history = $(history).split('\n')
        hist = get_history(sess_history)
        selected = $(echo @(hist) | peco)
        event.current_buffer.insert_text(selected.strip())

    # ctrl+sでssh_config内のhost先をpeco
    @bindings.add('c-s')
    def select_ssh(event):
        hosts = _get_host(True)
        selected = $(echo @(hosts) | peco)
        if selected:
            event.current_buffer.insert_text('ssha ' + selected.strip().split(', ')[0])

    # ctrl+fで今いるディレクトリのfileをpeco
    @bindings.add('c-f')
    def select_file(event):
        r = lambda x: './'+x if os.path.isdir(x) else x
        files = '\n'.join([r(x.split(' ')[-1]) for x in $(ls -l).split('\n')])
        selected = $(echo @(files) | peco)
        event.current_buffer.insert_text(selected.strip())

    # ctrl+dで過去移動したディレクトリをpeco
    @bindings.add('c-d')
    def _z(event):
        selected = $(echo @(z()) | peco)
        cd_cmd = "cd " + selected.strip()
        event.current_buffer.insert_text(cd_cmd)

    # ctrl+tで翻訳コマンドを入力
    @bindings.add('c-t')
    def _engs(event):
        event.current_buffer.insert_text('t ')

    ....

利用してるkeybinding、実際はもっとあるんですが、業務に関わる部分が多かったものは削っています。

sshやcommandをpecoであいまい検索

tコマンドは後述しますがGoogle翻訳としてよく使っているので書いときました。

- other -

大体上記のような環境でやっていってますが、他に結構使ってる便利なお手製コマンドを書いておきます。

google-translation

仕事で英語圏メンバーとMTGを毎日してますが、英語力低いので翻訳が手元に欲しかったりします。
GitHub - soimort/translate-shell: Command-line translator using Google Translate, Bing Translator, Yandex.Translate, etc.

keybindingで ctrl+t でt を入力、あとは日本語か英数字か正規表現で判定してGoogle翻訳かけるというコマンドを作っています。
VivaldiなるブラウザのWebパネルにもGoogle翻訳をセットしていますが、Shellだとそのままpbcopyでクリップボードに流したり、英語か日本語かの判定文自分で雑にかけるので何だかんだ結構こちらを使っています。

import re

jap = re.compile('[あ-んア-ン一-鿐]')
def _eng(x):
    if len(x)==0: return
    x = ' '.join(x)
    if jap.search(x) is None:
        y = $(trans en:ja @(x))
    else:
        y = $(trans ja:en @(x))
    return y
aliases['t'] = _eng

vaaaaanquish $ t 翻訳
翻訳
(Hon'yaku)

translation

翻訳 の定義
[ 日本語 -> English ]

名詞
    translation
        翻訳, 訳書, 翻訳物, トランスレーション
    deciphering
        解読, 翻訳

翻訳
    translation

最初はDMM英会話の「英語でなんていうknow」なるサービスがすごくよくてスクレイピングするコマンドを作って多用していましたが、慣用句以外なら単語さえわかればconversationに盛り込めるくらいにはなってきたので今はこれだけに落ち着いています。

ドキュメントやコード内の謎の英語とか調べるのにも重宝しています。

画像の表示

ディレクトリ内の画像から雑にsampleして、iterm2経由で表示するやつです。
画像処理なんかをやってる時に、「あ〜、このtrainディレクトリって何入ってんだっけ」となるので使ってます。

import matplotlib.pyplot as plt
import xontrib.mplhooks
import numpy as np
import cv2
import random
from mimetypes import guess_extension


def _imgls(path):
    fig, ax = plt.subplots(3, 3, sharex='col', sharey='row')
    fig.set_size_inches(10, 10)
    d=[]
    for x in os.listdir(path):
        if x.split('.')[-1] in ('jpe', 'jpeg', 'png'):
            d.append(path+x)
    if len(d)>9:
        d = random.sample(d,9)
    for i, imgid in enumerate(d):
        col = i % 3
        row = i // 3
        img = cv2.imread(imgid)
        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
        ax[row, col].imshow(img)
    xontrib.mplhooks.display_figure_with_iterm2(fig)
aliases['imgls'] = _imgls

ディレクトリ内の画像をshell上でsample

以下あたりを参考に作っていて、基づいて少し変更すればLinuxやwinでも動きます。
Pythonでyahoo画像検索した結果をimgcatに流して表示してURLをクリップボードにコピーするやつ - Stimulator
Xonshでmatplotlibグラフをコンソールにインライン描画してメモリ状況を観察する - Stimulator

これが結構便利なので割と使います。

おわりに

1つ1つ記事にも出来ましたが、Xonsh Advent Calendar 2018 - Qiitaがまさかの大盛況で助かりました。

一旦業務部分を切り取ってとりあえず書いたので、今後もうちょっとこの記事は更新されると思います。
業務的なところでは、社内サーバの起動から、GPUクラスタのGPUを監視してリソース奪い取るコマンド、勤怠や日報の自動化、社内チャットへの投稿、社内のMTGルームの取得…など全部適当に書いたコマンドで何とかしています。（なかなか切り分けできてないので見せられないですが）

サッと書いたスクリプトがxonshrcに入り常用してしまうという1年だったため、結構rc自体が汚く肥大化しています。

今現在も、xonshrcのリファクタリングとか、コマンドのhistory backendをAWSに乗せるとか、起動の高速化とかをやっていきたいと思ってるので2019年も更にがんばっていきたいと思います。

それではhappy xonsh year。

Python moduleがない場合に自動でpip installする

- はじめに -

この記事は、Xonsh Advent Calendar 2018 - Qiita 10日目の記事です。

複数のサーバを業務で利用おり、それぞれのサーバ環境設定のためpythonパッケージもインストールする作業が必要だが、別サーバからpip freezeするほどでもない(ちょっとした作業サーバなど)という時にjupyter_configや.xonshrc等の設定ファイルに自動インストールするように書いておけば良いのではと思って書いたメモです。

端的にはimportが失敗したらpip moduleを使ってインストールしてやる仕組みを作ります。
また加えて「PyPiに登録されているパッケージのリストを取得し、その中になければ諦めるという仕組み」「xontribの自動インストール」も示します。

- 動的なpip install -

python3にはpipモジュールが含まれているため、以下のような形で動的なインストールを実装します。
as構文を利用したい場合もあるかと思いますので、globalsに入れるように書いておきます。

from pip._internal import main as _main
import importlib

def _import(name, module, ver=None):
    try:
        globals()[name] = importlib.import_module(module)
    except ImportError:
        try:
            if ver is None:
                _main(['install', module])
            else:
                _main(['install', '{}=={}'.format(module, ver)])
            globals()[name] = importlib.import_module(module)
        except:
            print("can't import: {}".format(module))

_import('pd','pandas', '0.22.0')
print(pd)

pandasをアンインストールした状態でもスクリプト実行時にインストールが行われます。

試しに、上記スクリプトをpythonが動くshellであるxonshの設定ファイルに書き込んでみます。
pandas 消してもxonsh起動時になければ新しくインストールされています。

たぶんjupyterもconfigに書けば同じことができると思います。

参考：pip - Installing python module within code - Stack Overflow

- PyPiに登録されているパッケージのリストを取得する -

上記に記載の仕組みに加え「自身が作成しているオリジナルのモジュールもpipで入れたい」などの要望と入り混じった時、PyPiに登録されているライブラリかどうかを判定する必要が稀に出てきます。
PyPiに登録されているライブラリ一覧を取得するスクリプトを加えたものを以下に示します。

if not os.path.exists('~/.pypilist'):
    import xmlrpc.client as xmlrpclib
    client = xmlrpclib.ServerProxy('https://pypi.python.org/pypi')
    packages = client.list_packages()
    with open('~/.pypilist', 'w') as f:
        f.write('\n'.join(packages))
pypilist = open('~/.pypilist', 'r').read().split('\n')
if module in pypilist:
     print('not found pypi')

また、月更新くらいでpypiにあるパッケージやバージョンを見るようにしておけば、rcファイル等に書く時にも便利になるでしょう。

参考：JSON API for PyPi - how to list packages? - Stack Overflow

- xonshrcに書くために -

xonshアドベントカレンダーの記事ですので、xonshに書くためのtipsも示します。

xontribをインストールする

xontribも先程と同様にインストールできますが、xontribの場合、xontrib loadなるコマンドが用意されています。

実装を見ると、xontribs_load関数で読み込めなかった場合、prompt_xontrib_install関数で生成した以下のような文字列を返す実装になっています。

The following xontribs are enabled but not installed:
   z
To install them run
    xpip install xontrib-z

参考：xonsh/xontribs.py at master · xonsh/xonsh · GitHub

文字列返しても仕方ないので、ロード部分となるupdate_contextをoverrideして自動でpipするようにしてみます。

from pip._internal import main as _main
import xonsh.xontribs
from xonsh.xontribs import xontrib_context, update_context

def _update_context(name, ctx=None):
    if ctx is None:
        ctx = builtins.__xonsh__.ctx
    if not hasattr(update_context, "bad_imports"):
        _update_context.bad_imports = []
    modctx = xontrib_context(name)
    if modctx is None:
        _main(['install', 'xontrib-{}'.format(name)])
        remodctx = xontrib_context(name)
        if remodctx is None:
            _update_context.bad_imports.append(name)
            return ctx
        return ctx.update(remodctx)
    return ctx.update(modctx)

xonsh.xontribs.update_context = _update_context

xonshにはxpipなるpip wrapperがあり、xonsh環境配下にpip packageをinstallし、他env環境などから切り離すことが出来ます。
そちらを利用する事でxonshのアップグレードにも対応できます。
そちらはpip._internal.mainからは利用できないため、その場合は以下のように_mainでなくsubprocessなどでインストールしてやれば良いでしょう。

import subprocess

cmd = "xonsh -c 'xpip install xontrib-{}'".format(name)
subprocess.call(cmd, shell=True)

　
これらを組み合わせ、xonshrcに書いた上でxontrib loadコマンドを実行すれば、勝手にpipでパッケージを取ってくることが出来ます。

xontrib-readable-tracebackは以前XonshのException発生時のtracebackを見やすくする - Stimulatorなる記事で私が作成したライブラリです。結果として1/0のException messageが短くキレイに色付けされ表示されています。

コマンドをxonshrc内でインストールする

コマンドの有無も確認し、なければインストールするようにしておくと便利です。
pythonではshutil.whichメソッドがそれらをサポートしてくれているので利用すると良いでしょう。

「コマンドが無ければ〜」というサンプルを以下に示します。

import shutil

if shutil.which( COMMAND_NAME ) is None:
    # install script

xonshではpython以外にshell scriptもほぼ同等に動くわけなので、何も考えず最新版をインストールシェルスクリプトを書くだけです。

例えば私がよく使っているpecoなら以下をそのまま書くだけ。
Linux に最新版の peco をインストールするシェルスクリプト - Qiita

- おわりに -

これでxonsh周りの実行環境作成は、xonshrcだけで完結すると思います。

個人的には以下のlazy load等も組み合わせながら、xonshの起動を高速化しつつ、最低限のラインとして使っています。
vaaaaaanquish.hatenablog.com

xonshアドベントカレンダーのほうはまだ空きがあるみたいなので、よければ是非。
qiita.com

macOSの通知リストをpython経由で取得しShellに流す

- はじめに -

macOSを使っていると、通知(Notifications)に色んな情報が集約される。
メールが来ただのSlackの更新だの、アプリのアップデート等がそれにあたる。

「OS通知を定期的に取得してシェルのbottom barとかに表示してやれば、作業中でも社内チャットやらメールの通知とかを厳選して気付けたりするしサイコーなのでは？」という所を目的にした作業ログを示すものである。

ちなみにこの記事は、Xonsh Advent Calendar 2018 - Qiitaの7日目の記事。

- macOSの通知の取得 -

2014年くらいにHigh Serriaになった時、ユーザ向けにもAPIが公開されたNotifications。それらを叩けるterminal-notifierや、そのPython wrapperとなるpyncなどが出てきて、ユーザが通知欄に任意のメッセージを送る事は難しくなくなった。
GitHub - julienXX/terminal-notifier: Send User Notifications on macOS from the command-line.
GitHub - SeTeM/pync: Python wrapper for Mac OS 10.8 Notification Center

macOSが提供するAPIでは「アプリXから通知を送る」「アプリXから送った通知を消す」「アプリXが送った通知のリストを取得する」が利用できるが、これらを利用しても「全てのアプリから送られた全ての通知を取得する」事はできない（記事執筆時、2018/11/24）。

完全な情報を取得する事はできないが、以下のpatreon記事によると、システムデータベースには通知の断片ログが残っているようである。patreon主はSignalや各所登壇で良く見るあのPatrick Wardle氏である。
www.patreon.com
Macのsystemのpythonに含まれるFoundationクラスを利用する事でこのログを読むことができる。*1

　
上記記事内のdumpNotificationDB.pyをベースに、通知毎のタイトルと内容と日付、アプリ名を取得し表示するスクリプトを以下に示す。

# -*- coding: utf-8 -*-
# /usr/bin/python (macOS system python)
import os
import sys
import sqlite3
import datetime
import tempfile
import Foundation


def parse_req(req):
    # reqをparseしtitleとbodyのみ取得する
    res = {}
    for x in str(req).split(';'):
        if 'body' in x or 'titl' in x:
            d = x.replace('{','').strip().split(' = ')
            res[d[0]] = d[1].replace('"','')
    return res


def get_notif_json():
    # DBファイルにアクセスしparseして返す
    notificationDB = os.path.realpath(
            tempfile.gettempdir() + '/../0/com.apple.notificationcenter/db2/db')
    conn = sqlite3.connect(notificationDB)
    conn.row_factory = sqlite3.Row
    cursor = conn.execute("SELECT data from record");

    res_j = []
    for row in cursor:
        plist, fmt, err = \
                Foundation.NSPropertyListSerialization.propertyListFromData_mutabilityOption_format_errorDescription_(
                buffer(row[0]),
                Foundation.NSPropertyListMutableContainers,
                None, None)
        if err is not None:
            continue

        notif_dic = {}
        for key, value in plist.iteritems() :
            if key == 'date':
                notif_dic['date'] = Foundation.NSDate.alloc().initWithTimeIntervalSinceReferenceDate_(value)
            if key == 'req':
                req = parse_req(value)
                if 'titl' in req.keys():
                    notif_dic['title'] = req['titl']
                notif_dic['body'] = req['body']
            elif key == 'app':
                notif_dic['app'] = value
        res_j.append(notif_dic)
    return res_j

# データ取得
d = get_notif_json()
# 日付順ソート
d = sorted(d, key=lambda x: x['date'])
# 表示
import codecs
for x in d:
    print x['date']
    print x['app']
    print codecs.decode(x.get('title','no title').lower(), 'unicode-escape')
    print codecs.decode(x['body'].lower(), 'unicode-escape')
    print ''

macOSのpython2系を利用するため、codecsを利用してunicodeをdecodeしたり、printがアレだったりするが仕方ない。

　
先述の通り、system-pyhtonのFoundationが必要なため、/usr/bin/pythonを利用して実行する。

$ /usr/bin/python get_notification.py

2018-11-14 14:58:05 +0000
_SYSTEM_CENTER_:com.apple.battery-monitor
バッテリー残量低下
電源コンセントに接続しない場合、お使いのmacは間もなくスリープ状態に入ります。

2018-11-16 23:27:59 +0000
com.adobe.acc.AdobeCreativeCloud
creative cloud
4 個のアプリのアップデートがあります

2018-11-21 10:08:50 +0000
com.tinyspeck.slackmacgap
teamy
miku : 完全に未来人扱いされてますね笑

2018-11-21 10:10:36 +0000
com.tinyspeck.slackmacgap
teamy
miku: オッケです！

あくまで通知時のログなので、既に通知欄から削除した、していないという情報は得られないが、概ね十分そうである。

- xonshを利用してshell上に表示する -

xonshのアドベントカレンダーなのでxonshを利用してshellに流し込む

上記スクリプトの出力部分を一行空白区切りで情報を出力するように修正します。

import codecs
for x in d:
    date = str(x['date']).split(' ')
    print date[0].encode('utf-8'), date[1].encode('utf-8'), x['app'].encode('utf-8'),\
            codecs.decode(x.get('title','no title').lower(), 'unicode-escape').replace(' ','').encode('utf-8'),\
            codecs.decode(x['body'].lower(), 'unicode-escape').replace('\n','').replace(' ', '').encode('utf-8')

そもそもコードも酷いというのもあるが、encodeしないと「UnicodeEncodeError: 'ascii' codec can't encode characters in position 0-8: ordinal not in range(128)」みたいになるためさらに酷くなっているが仕方ない。
　
このpython2のスクリプトにおけるprintを標準出力から受け取り、xonshで表示できるような関数にする。
　
python3からsubprocess経由で/usr/bin/pythonを使い、上記ファイルを実行するスクリプトを以下に示す。

import subprocess

res = subprocess.run(['/usr/bin/python get_notification.py'], stdout=subprocess.PIPE, shell=True)
print(res.stdout)

これをxonshの関数にそのまま採用して情報を厳選して、xonsh toolbarに表示してみる。

　
python2と3を経由するかなり気持ち悪い感じだが出来たコマンドは以下のような感じ。

import subprocess

def get_last_notif():
    # 最後の通知を返す
    res = subprocess.run(['/usr/bin/python /path/to/get_notification.py'], stdout=subprocess.PIPE, shell=True)
    return res.stdout.decode('utf-8', 'replace').strip().split('\n')[-1]

# xonsh bottom toolbarに表示する
$BOTTOM_TOOLBAR = get_last_notif

（クソ…！よこせストレージを…！！）

　
このままだとUPDATE_COMPLETIONS_ON_KEYPRESSをTrueにしていれば、キー入力毎に評価され、DBを読みに行ってparseしてという処理がキー入力毎に発生してしまいますので、一度どこかにキャッシュして一定時間ごとに更新するなどの処理が必要そうである。

まあとりあえず目的は達成したので、キャッシュの話はまた別途という事にする。

- おわりに -

通知をキャッシュして、直近のものを順繰りで表示しておいたりすると便利そう。
ちゃんと書けばxontribみたいにできるかも…？

ただMacの通知取得の方法が若干アレなので、アップデートで使えなくなる可能性もあるし、素直にAppleが相応のAPIを公開してくれたほうが良さそうだなと思う。

　
xonshアドベントカレンダーの方がまだまだ後半ガラ空きという状態なので書いてほしいです。宜しくお願いします。
qiita.com

*1:記事内にもある通りsystem pythonが実行できる権限があれば過去の通知の一部を読めてしまうため脆弱性になり得る要素である