gistfile1.txt

import asyncio
import json
import re
import subprocess

import aioamqp

async def rsync(loop):
    """ """
    transport, protocol = await aioamqp.connect()
    channel = await protocol.channel()
    await channel.queue_declare(queue_name='rsync')
    # cmd = 'rsync'
    # cmd = ['rsync', '-r', '--relative', '/Users/billyshambrook/rsynctests/src/small', '/Users/billyshambrook/rsynctests/dest/small']
    cmd = ['rsync', '-r', '--relative', '--progress', '--log-format', '{"transferred_bytes": "%b", "filename": "%n", "datetime": "%t"}', '/Users/billyshambrook/rsynctests/srac/small', '/Users/billyshambrook/rsynctests/dest/smaall']
    # cmd = 'rsync -u --log-format \'{"transferred_bytes": "%b"}\' src/small/* dest/small'
    proc = await asyncio.create_subprocess_exec(*cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, loop=loop)

    while True:
        line = await proc.stdout.readline()
        if not line:
            break

        line = line.decode('ascii').rstrip()

        if 'to-chk=' in line:
            m = re.findall(r'to-chk=(\d+)/(\d+)', line)
            progress = int((100 * (int(m[0][1]) - int(m[0][0]))) / int(m[0][1]))

            line = await proc.stdout.readline()

            message = json.loads(line.decode('ascii').rstrip())
            message['progress'] = progress
            await channel.basic_publish(
                payload=json.dumps(message),
                exchange_name='',
                routing_key='rsync'
            )

    exitcode = await proc.wait()

    if exitcode > 0:
        error = await proc.stderr.read()
        print(error.decode('ascii'))

    await protocol.close()
    transport.close()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('json', nargs='?', required=True)
    args = parser.parse_args()

    print(args)



if __name__ == '__main__':
    main()