ローカルとサーバーの自動同期スクリプトをPythonで書きました

Webサイトのプログラムを修正するとき、ローカルの変更を即座にサーバーに反映するのに以前はVisual Studio Codeの拡張機能ftp-simpleを使っていました。

しかし、ローカルでgit操作ができなかったり、ディレクトリの一括検索ができなかったので、
ローカルの変更をサーバーに反映するスクリプトをPythonで作ってしまいました。
(スクリプト & 設定ファイル例は記事の最後)

githubリポジトリ

注意

サーバー上のファイルを自動で作成、編集、削除、移動するので、
スクリプトをよく読んで注意して使ってください。

使い方

python スクリプトファイル名 設定ファイルパス

スクリプト

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

import os
import time

import paramiko
import json
import sys
import scp

import re

default_ignore = [
    "^\.git.*$",
    "^vendor.*$",
    "^.*\.swx$",
    "^.*\.swp$",
    "^.*\.swo$",
]

class ChangeHandler(FileSystemEventHandler):
    def on_created(self, event):
        filepath = event.src_path

        checkpath = filepath[len(config["local_dir"]) + 1 : ]
        for pat in default_ignore:
            if re.match(pat, checkpath):
                return

        remote_path = config["remote_dir"] + filepath[len(config["local_dir"]) : ]

        if os.path.isdir(filepath):
            shell.send(f"mkdir {remote_path} \n")
            print(f"リモートディレクトリ[{remote_path}]を作成")
            return

        with scp.SCPClient(client.get_transport()) as scp_client:
            scp_client.put(filepath, remote_path)
            print(f"リモートファイル[{remote_path}]を作成")

    def on_modified(self, event):
        filepath = event.src_path

        checkpath = filepath[len(config["local_dir"]) + 1 : ]
        for pat in default_ignore:
            if re.match(pat, checkpath):
                return

        if os.path.isdir(filepath):
            return

        remote_path = config["remote_dir"] + filepath[len(config["local_dir"]) : ]

        with scp.SCPClient(client.get_transport()) as scp_client:
            try:
                scp_client.put(filepath, remote_path)
                print(f"リモートファイル[{remote_path}]を変更")
            except:
                print("[[[ERROR]]] 変更後のファイルのアップロードに失敗しました")
                print(f"[Error Info]filepath = {filepath}")
                print(f"[Error Info]remote_path = {remote_path}")

    def on_deleted(self, event):
        filepath = event.src_path

        checkpath = filepath[len(config["local_dir"]) + 1 : ]
        for pat in default_ignore:
            if re.match(pat, checkpath):
                return

        remote_path = config["remote_dir"] + filepath[len(config["local_dir"]) : ]
        shell.send(f"rm -r {remote_path} \n")
        print(f"リモートファイル[{remote_path}]を削除")

    def on_moved(self, event):
        filepath = event.dest_path

        checkpath = filepath[len(config["local_dir"]) + 1 : ]
        for pat in default_ignore:
            if re.match(pat, checkpath):
                return

        remote_path = config["remote_dir"] + filepath[len(config["local_dir"]) : ]

        if os.path.isdir(filepath):
            shell.send(f"mkdir {remote_path} \n")
            print(f"リモートディレクトリ[{remote_path}]を作成")
            return

        with scp.SCPClient(client.get_transport()) as scp_client:
            scp_client.put(filepath, remote_path)

        remote_src_path = config["remote_dir"] + event.src_path[len(config["local_dir"]) : ]
        shell.send(f"rm -r {remote_src_path} \n")
        print(f"リモートファイル[{remote_src_path}]を[{remote_path}]に移動")


if __name__ in '__main__':
    if len(sys.argv) != 2:
        print("Usage: python hiyo_sync.py config_path")
        exit()

    config_path = sys.argv[1]

    if not os.path.exists(config_path):
        print("Error: config file doesn't exist.")
        exit()

    with open(config_path) as f:
        config = json.load(f)

    # print(config)
    
    if re.match("^.*/$", config["local_dir"]) or re.match("^.*/$", config["remote_dir"]):
        print("Error: local_dirまたはremote_dirの末尾に/を付けないでください")
        exit()

    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(
        config["remote_host"], 
        username=config["username"],
        password=config["password"],
        port=22,
        timeout=15.0,
        look_for_keys=False
    )

    shell = client.invoke_shell()
    shell.send(f"cd {config['remote_dir']} \n")

    while 1:
        event_handler = ChangeHandler()
        observer = Observer()
        observer.schedule(event_handler, config["local_dir"], recursive=True)
        observer.start()
        try:
            while True:
                time.sleep(0.1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()

設定ファイル例

{
    "local_dir": "/local/directory/with/no/last/slash",
    "remote_dir": "/remote/directory/with/no/last/slash",
    "password": "yourpassword",
    "remote_host": "your.host.name",
    "username": "yourname"
}