PixivのデイリーランキングのイラストをLineに通知+Dropboxにアップロードする

ネットサービスは壮大な可処分時間の奪い合いになっている

今の情報の飽和の時代、膨大な情報がある中から適切に情報を選択し、消化する必要があり、これは、とても24時間という制限ではなかなか十分に達成できないものになってきてしまいました。

特にネットのサービスは、ユーザにコンテンツを届け、楽しませ、評価してもらうには、同じようなサービスとの競合の軸になる際に、金額の他に、使い勝手や単位時間あたりの満足度も大きく関係してくる視点になります。

私が使用しているWebサービスは以下のものがあり、もう、完全に時間がパンク状態です

• Twitter
• YouTube
• Instagram
• FaceBook
• Pixiv(課金)
• NetFlix(課金)
• Amazon Prime Video(課金)
• Line

Pixivのデイリーライキングはイラストとしてとても時間を掛けた、流麗なイラストが評価されるのですが、これを毎回WebBrowserを立ち上げて、見に行くにはめっちゃめんどくさいです。

でも見たいので、見ちゃうのですが、Dropboxのオフラインアクセス(ダウンロードされている状態)にLineに転送してしまうと、だいぶ時間の節約になります。

実際にこの仕組を構築するまでの、各要素ごとの機能と、コードの流れを示します。

1. PixivのデイリーランキングにSeleniumを利用する

Pixivのデイリーランキングのサイトの描画方法を見ていると、JavaScriptで画面のサイズに応じて動的にリクエストして描画しているようです。
そのため、簡単なJavaScriptを解釈できないスクレイピング方法だと十分にランキングの件数を取得する事ができません。
SeleniumとGoogle Chromeを用いると、JavaScriptをシミュレートしながら、要素をパースできるので、使うと便利です。
この時の画面サイズを1024, 10240と縦長にすることで、一回の描画でほぼ全部のランキングイメージをパースすることができました。

ここから各要素の画像のオリジナルリンクをたどり、大きな画像のURLをローカルに保存します。

url = 'https://www.pixiv.net/ranking.php?mode=daily&content=illust'

def getRanking():
  options = Options()
  options.add_argument('--headless')
  options.add_argument("user-agent=Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.186 Safari/537.36")
  driver = webdriver.Chrome(chrome_options=options, executable_path='/usr/local/bin/chromedriver')
  driver.set_window_size(1024,1024*10)
  dt = datetime.now().strftime('%Y-%m-%d_%H_%M')
  ha = hashlib.sha256(bytes(url, 'utf8')).hexdigest()
  driver.get(url)
  action = webdriver.common.action_chains.ActionChains(driver)
  html   = driver.page_source
  soup   = bs4.BeautifulSoup(html, 'lxml')
  absolutes = []
  for div in soup.findAll('div', {'class':'ranking-image-item'}):
    relative = div.find('a').get('href')
    absolute = f'https://www.pixiv.net{relative}'
    absolutes.append( absolute )
  json.dump(absolutes, fp=open(f'images-abosolutes/{dt}.json', 'w'), indent=2 )
  driver.save_screenshot(f'ss/{ha}.png')
  driver.quit() 

if __name__ == '__main__':
  getRanking()

2. 各画像のページの詳細ではSeleniumはいらない

多くのウェブサイトでは、ページごとに独自性があり、Seleniumを使わないでStaticなページとして見たとしても問題なくスクレイピングできたります。

1.で取得したURLを辿り、オリジナルの大きな画像があるページまで行き、htmlを解析して、jpgファイルを得ます。

from pathlib import Path
paths = [ path for path in sorted( Path('images-abosolutes').glob('*') ) ]
path  = paths[-1]

import json
urls = json.load(path.open())

import requests
from bs4 import BeautifulSoup as BS
import os
from hashlib import sha256
from datetime import datetime
import time

def each_image():
  dt = datetime.now().strftime('%Y-%m-%d_%H_%M') 
  try:
    os.mkdir(f'images-pixiv/{dt}')
  except: ...
  for url in urls:
    try: 
      print(url)
      r = requests.get(url,  headers={'referer': 'https://www.pixiv.net/ranking.php?mode=daily&content=illust'})
      print(r.status_code)
      html = r.text
      soup = BS(html)
      src = None
      for img in soup.findAll('img'):
          is_src = img.get('src')
          print(is_src)
          if '600x600' in is_src:
              src = is_src
      if src is None:
        continue
      # c/600x600/ 削除
      src = src.replace('c/600x600', '')
      hash = sha256(bytes(src, 'utf8')).hexdigest()
      r = requests.get(src,  headers={'referer': url})
      cnt = r.content 
      with open(f'images-pixiv/{dt}/{hash}.jpg', 'wb') as fp:
        fp.write( cnt )
      with open(f'images-pixiv/{dt}/{hash}.txt', 'w') as fp:
        fp.write( url )
    except Exception as ex:
      print(ex)

if __name__ == '__main__':
  each_image()

3. ダウンロードした画像をLineに送る

Line Notifyというサービスを使うと、自分のスマホに画像やテキストを送ることができます。

2.で取得した画像をLineで送ると、予め、Line側にデータが転送されているので、スマホで電車やバスの中でのスキマ時間に他のメッセージと並列して消化することができます。

PixivのデイリーランキングからのLineに画像転送 pic.twitter.com/WjSnypEyiW

— A Lie of Lillie (@lie_of_lillie) 2018年6月30日

from pathlib import Path
import requests
def push():
  url     = "https://notify-api.line.me/api/notify"
  token   = open('secret.txt').read().strip()
  headers = {"Authorization" : "Bearer "+ token}
  path = sorted(Path('./images-pixiv').glob('*'))[-1]
  for jpg in path.glob('*.jpg'):
    try:
      files   = { "imageFile": jpg.open('rb') }
      payload = { "message": f'from pixiv' }
      r = requests.post( url, headers=headers, params=payload, files=files)
    except Exception as ex:
      print(ex)
if __name__ == '__main__':
  push()

sercret.txtにはLine Notifyのトークンを含めてください

4. ダウンロードした画像をDropboxに送る

DropBoxもアプリを作成することが可能で、ファイルの読み書きをPython等のスクリプトから行うことが可能です。使いがっては、PythonからS3やGCSのファイルを操作する感覚に近く、わかりやすいです。

DropBoxにアップロードすると、スマホのこのサービスを用いたエコシステムが発達しているため、自動的にオフラインにダウンロードするアプリ（Dropsync）などがあり、いつの間にか、Pixivのランキングの画像が、端末にも保存されています。

こうすることで、電波状況が芳しくない、環境であっても、高画質の画像をサクサクと一気にチェックできるのと、PCのスライドショーをこのフォルダにしておくことで、毎日新しい、ランキングの画像をPush型で受け取ることも可能になります。

pic.twitter.com/SzDfZ4wLNi

— A Lie of Lillie (@lie_of_lillie) 2018年6月30日

import dropbox
import time
import datetime
from pathlib import Path
import re
def dbx():
  dbx = dropbox.Dropbox(Path('token_dbx.txt').open().read().strip())
  mode = dropbox.files.WriteMode.overwrite
  _path = sorted( Path('images-pixiv').glob('*') )[-1]
  #  日時で
  for path in _path.glob('*'):
    data = path.open('rb').read()
    dpath = re.search(r'.*?/(.*?/.*?$)', str(path)).group(1)
    print(dpath)
    res = dbx.files_upload( data, '/' + dpath , mode, mute=True)

  # flatにネストせずに保存
  for path in _path.glob('*'):
    if '.txt' in str(path):
      continue
    data = path.open('rb').read()
    dpath = re.search(r'.*?/.*?/(.*?$)', str(path)).group(1)
    print(dpath)
    res = dbx.files_upload( data, '/flat/' + dpath , mode, mute=True)

if __name__ == '__main__':
    dbx()

token_dbx.txtにはDropBoxの登録したご自身のTokenを含めてください。

GitHubのレポジトリ

このレポジトリのpixiv-daily-ranking-webの中の0-schedule.pyがすべての実行プログラムになっています。

このようなscheduleという定期実行のモジュールをもちいて12時間に一回、スクレイピングが動作するようにしています。

import scraping_ranking
import scraping_each_images
import line_push
import dropbox_upload

import schedule
import time

def job():
  try:
    scraping_ranking.getRanking()
    scraping_each_images.each_image()
    line_push.push()
    dropbox_upload.dbx()
  except Exception as ex:
    print(ex)
schedule.every(12).hours.do(job)

job()
while True:
  schedule.run_pending()
  time.sleep(1)