夜御牧さん用 rclone 同期バッチ

do-rclone.exsample.yml

# do-rclone.ymlにリネームして使う

targets:
  - src: foo
    dest: bar
  - src: baz
    dest: qux
    exclude: .tmp/**

# Eメール転送用スクリプト
# https://gist.github.com/YamimakiReru/52745c0632af8c680aea3114c7ad0f93
email_via: https://script.google.com/macros/s/[デプロイID]/exec
email_to: foo@example.com

log_format: '%(levelname)s@%(name)s[%(asctime)s] %(message)s'

do-rclone.py

# coding: utf-8

#
# 夜御牧さん用 rclone 同期バッチ
#
# 使い方: python3 do-rclone.py [--dry-run]
#
# CPU使用率が5%以下になるまで、最長5分待機してから実行.
# rclone 終了後、ログをメールで送付.
#
# # # #
#
# DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
# Version 2, December 2004
# 
# Copyright (C) 2023 YAMIMAKI, Reru
#
# Everyone is permitted to copy and distribute verbatim or modified
# copies of this license document, and changing it is allowed as long
# as the name is changed.
#
# DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
# TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
#
# 0. You just DO WHAT THE FUCK YOU WANT TO.

import os, time
import logging, sys
from logging.handlers import TimedRotatingFileHandler
from traceback import format_exception
import subprocess
import urllib.request
import urllib.parse

import yaml
import psutil

def unhandled_exception_hook(type, value, traceback):
    logging.error(''.join(format_exception(type, value, traceback)))

sys.excepthook = unhandled_exception_hook

class RCloneTask:
    def __init__(self) -> None:
        self.my_dir = os.path.dirname(__file__)
        script_name = os.path.splitext(os.path.basename(__file__))[0]
        self.script_basepath = f'{self.my_dir}/{script_name}'

        with open(f'{self.script_basepath}.yml', 'r') as yaml_file:
            self.config = yaml.safe_load(yaml_file)
        logging.basicConfig(level=logging.INFO, format=self.config['log_format'])

        log_path = f'{self.script_basepath}.log'
        file_handler = TimedRotatingFileHandler(log_path, when='W5', backupCount='15', encoding='utf-8')
        file_handler.setLevel(logging.INFO)
        file_handler.setFormatter(logging.Formatter(self.config['log_format']))
        logging.getLogger('').addHandler(file_handler)

    def run(self, argv: list[str]) -> int:
        dry_run = 2 <= len(argv)

        # CPU使用率が5%以下になるまで待機
        # 待ってる間に5分経ったら諦める
        start_time = time.time()
        while True:
            cpu_usage = psutil.cpu_percent(1.0)
            logging.info(f'CPU使用率: {cpu_usage}')
            if 5.0 > cpu_usage:
                break
            current_time = time.time()
            if 300 < (current_time - start_time):
                logging.warning('CPU使用率が高くバッチを実行できませんでした')
                return -1
            time.sleep(9)
    
        logging.info('==== 処理開始 ====')
        return_code = self.do_rclone(dry_run)
        logging.info('==== 処理終了 ====')
        return return_code

    def do_rclone(self, dry_run: bool) -> int:
        latest_log_file = f'{self.my_dir}/latest.log'
        latest_logger = logging.FileHandler(latest_log_file, mode='w', encoding='utf-8')
        latest_logger.setLevel(logging.INFO)
        # latest_logger.setFormatter(logging.Formatter(self.config['log_format']))
        logger = logging.getLogger('latest')
        logger.addHandler(latest_logger)

        return_code = 0
        try:
            for target in self.config['targets']:
                new_return_code = self.call_rclone(logger, target, dry_run)
                return_code = max(return_code, new_return_code)
        finally:
            with open(latest_log_file, 'r', encoding='utf-8') as f:
                body = f.read()
                email = {'to': self.config['email_to'], 'subject': 'rclone実行結果', 'body': body}
                encd_email = urllib.parse.urlencode(email).encode('utf-8')
                with urllib.request.urlopen(self.config['email_via'], data=encd_email) as response:
                    res_text = response.read().decode("utf-8")
                    logging.info(res_text)
        return return_code

    def call_rclone(self, logger: logging.Logger, target: dict[str, str], dry_run: bool) -> int:
        cmd = ['rclone', 'sync', '-v', target['src'], target['dest']]
        if dry_run:
            cmd.append('--dry-run')
        if 'exclude' in target:
            cmd.append('--exclude')
            cmd.append(target['exclude'])
        logging.info(cmd)
        proc = subprocess.Popen(cmd, shell=False,
                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        while True:
            line = proc.stderr.readline()
            if line:
                logger.warning(line.decode().strip())
                continue
            line = proc.stdout.readline()
            if line:
                logger.info(line.decode().strip())
                continue
            return_code = proc.poll()
            if return_code is not None:
                logger.info(f'終了コード: {return_code}')
                return return_code

if __name__ == '__main__':
    the_task = RCloneTask()
    return_code = the_task.run(sys.argv)
    sys.exit(return_code)

requirements.txt

psutil
pyyaml