scrapbox-sync

同期元projectで特定のタグがついたページを、同期先projectにexport/importする

同期元projectで削除されたページを同期先projectでも削除する

用途

関連

実装

非対応

無料だと10秒の実行時間制限がある

いけそう

登録すると無料枠が増えるみたいです

GAS

トリガー機能を利用して定期的にGCFを実行する

導入方法

以下のGASのコードを適宜変更し貼り付ける

1日1回とか

GCF

GASからのリクエストに応じてページをimport・削除する

ローカル環境作成が楽だったので

セキュリティのためにキーを設定する

リクエストのヘッダにこのキーがついてないとエラーになる

（気休めです、もっといい方法ありそう）

導入方法

クレカ登録が必要

このスクリプトくらいなら無料範囲内のはず

月当たり1円とか2円とかかかってしまっている

Cloud Functionをデプロイする

関数名、リージョンは任意

トリガーをHTTP、未認証の呼び出しを許可にチェック

「変数、ネットワーク、詳細設定」欄

詳細→メモリ1GB、タイムアウト60秒

環境変数→ランタイム環境変数に上記のキーを設定する

以下のGCFのコードを作成・貼り付ける

ランタイム：Python3.8

エントリポイント：main

Cloud Build APIの有効化を求められるので有効化する

デプロイ

以降、GASからのリクエストに応じて動くはず

既知の問題・TODO

page取得の間隔をあけてサーバーの負荷を減らす？

たくさんAPIを叩くわけじゃないし、そんなに気にしなくてもいいと思う

GASのコード

メイン

code:main.gs(js)

function main() {

const configs = getConfigs();

configs.forEach(c => {

const response = UrlFetchApp.fetch(

{

method: 'POST',

headers: {

'Content-Type': 'application/json',

'X-Internal-Key': '***', // GCF側で設定したキー

},

payload: JSON.stringify(c)

}

);

console.log(JSON.stringify(response.getContentText("Shift_JIS"), null, 2));

})

}

設定

top levelに書いた変数・関数は全てglobal scopeに属してしまう

code:config.gs(js)

function getConfigs() {

return [

{

src_project: '同期元プロジェクト名',

dst_project: '同期先プロジェクト名',

public_tag: 'not-private', // このタグが付いているページがimportされる

private_tag: 'private', // このタグが付いているページのpublic_tagは無視される

sid: '***', // connect.sid(取り扱い注意)

},

];

}

GCFのコード

依存パッケージ

code:requirements.txt

requests

aiohttp

PyYAML

pyppeteer

メイン

関数の整理・分割案

scrapbox.io <=> GCFとの通信

scrapbox-sync固有の処理は入っていない

同期先プロジェクトへ対象ページをimportする

特定のscrapbox projectの全ページタイトルを取得する

空ページは除外する

指定したproject内の指定したページを削除する

scrapbox-sync関連

外部との窓口

POSTされたデータをもとに処理を行う

scrapbox.io

同期元プロジェクトから同期対象のページを取得する

やること

同期元プロジェクトで削除されたページを取得する

未分類

code:main.py

import os

import re

import json

import urllib

import requests

import asyncio

import traceback

from pyppeteer import launch

from flask import request, abort, jsonify

from flask import current_app as app

from utils import get_all_titles, get_all

def get_public_pages(src_project, public_tag, private_tag, sid):

'''同期元プロジェクトから同期対象のページを取得'''

tag_page = f'{root}/api/pages/{src_project}/{public_tag}'

tagged_pages = asyncio.run(get_all(tagged_pages_urls, cookies={'connect.sid': sid}))

public_pages = []

while tagged_pages:

page = tagged_pages.pop()

# cancel if private tag exists

if f'#{private_tag}' in texts:

continue

# remove public-tag line

tag_idx = texts.index(f'#{public_tag}') # NOTE: only the first one found

# remove line next to tag line if it's empty

# remove unneccesary attributions

print(f'Found {len(public_pages)} pages to import.')

return public_pages

def import_pages(pages, dst_project, sid):

'''同期先プロジェクトへ対象ページをimportする'''

import_json = json.dumps({'pages': pages})

token = requests.get(

f'{root}/api/users/me',

headers={'Cookie': f'connect.sid={sid}'},

response = requests.post(

f'{root}/api/page-data/import/{dst_project}.json',

headers={'Cookie': f'connect.sid={sid}', 'X-CSRF-TOKEN': token},

files={'import-file': ('import.json', import_json, 'application/json')},

)

return response.status_code

def get_deleted_titles(src_pages, dst_project, whitelist):

'''同期元プロジェクトで削除されたページを取得'''

dst_titles = get_all_titles(dst_project)

to_delete_titles = list(filter(

lambda title: (title not in src_titles) and (title not in whitelist),

dst_titles

))

print(f'Found {len(to_delete_titles)} pages to delete.')

return to_delete_titles

def delete_pages(titles, dst_project, sid):

'''同期先プロジェクトで対象ページを削除'''

cookie = {

'name': 'connect.sid',

'value': sid,

'domain': 'scrapbox.io',

}

async def delete_page(enc_url, browser):

page = await browser.newPage()

await page.setCookie(cookie)

await page.goto(enc_url)

await page.evaluate('() => window.confirm = (nope) => true')

await page.waitForSelector('.quick-launch.layout-list')

return True

async def run():

browser = await launch(

handleSIGINT=False,

handleSIGTERM=False,

handleSIGHUP=False,

)

await browser.close()

return results

results = asyncio.run(run())

succeeded, failed = [], []

for title, res in zip(titles, results):

if res == True:

succeeded.append(f'{dst_project}/{title}')

else:

print(f'{dst_project}/{title} could not deleted due to {res}')

failed.append(f'{dst_project}/{title}')

result = dict(succeeded=succeeded, failed=failed)

print(f'delete_pages: {result}')

return result

def main(request):

if request.method != 'POST':

abort(404, 'not found')

abort(403, 'forbidden')

if content_type != 'application/json':

abort(400, 'bad request')

request_json = request.get_json(silent=True)

if not request_json or any(key not in request_json for key in keys):

abort(400, 'bad request')

to_import_pages = get_public_pages(src_project, public_tag, private_tag, sid)

import_response_code = import_pages(to_import_pages, dst_project, sid)

if import_response_code != 200:

abort(500, 'import error')

to_delete_titles = get_deleted_titles(to_import_pages, dst_project, whitelist)

delete_result = delete_pages(to_delete_titles, dst_project, sid)

return jsonify(number_of_imported_pages=f'{len(to_import_pages)}', delete=delete_result), 200

# debug

import yaml

from flask import Flask

if __name__ == '__main__':

with open('./envs.yaml') as f:

os.environ.update(yaml.load(f, Loader=yaml.FullLoader))

app = Flask(__name__)

def index():

return main(request)

app.run('127.0.0.1', 8000, debug=True)

code:utils.py

import requests

import asyncio

import aiohttp

code:utils.py

def get_all_titles(project):

titles = []

following_id = None

while 1:

params = f'?followingId={following_id}' if following_id else ''

response = requests.get(

if not following_id:

break

return titles

async def get(session, url):

async with session.get(url) as response:

return await response.json()

async def get_all(urls, **session_kwargs):

async with aiohttp.ClientSession(**session_kwargs) as session:

return await asyncio.gather(*tasks)

Qiita.icon

Node.js.icon

Deno.icon