q-tools/files/rsync-queue

#!/usr/bin/python
import sys
import os,time,datetime
import sqlite3
import subprocess,shlex
from argparse import ArgumentParser 

SQLFILE=os.path.expandvars('$HOME/.rsync-queue.sqlite')

def setup_options():
    parser=ArgumentParser(description="Maintain a queue for file synchronization")
    parser.add_argument("-r",action="store_true",dest="run",default=False,
                     help="Start rsync")
    parser.add_argument("-R",action="store_true",dest="forever",default=False,
                     help="Start rsync in a loop, never exit")
    parser.add_argument("--rerun",action="store_true",dest="rerun",default=False,
                     help="When running, run even the correctly exited.")
    parser.add_argument("-l",action="store_true",dest="listDB",default=False,
                     help="List DB contents")
    parser.add_argument("-L",action="store_true",dest="listAllDB",default=False,
                     help="List DB contents, even the completed")
    parser.add_argument("-f",action="store",dest="sqlfile",default=SQLFILE,
                      help="SQL file name to use [%(default)s]")
    parser.add_argument("-c",action="store_true",dest="clear",default=False,
                     help="Clear DB of completed entries")
    parser.add_argument("-C",action="store_true",dest="clearAll",default=False,
                     help="Clear DB of all entries")
    parser.add_argument("-o",action="store",dest="options",default="-vaP",
                     help="Options to rsync")
    parser.add_argument('SRC', action="store",default='', nargs='?')
    parser.add_argument('TGT', action="store",default='', nargs='?')

    options=parser.parse_args()
    if options.forever:
        options.run=True
    if options.clearAll:
        options.clear=True
    if options.SRC!='':
        options.SRC=os.path.abspath(options.SRC)
    if options.TGT!='':
        options.TGT=os.path.abspath(options.TGT)
    return options

def createdb(fname):
    conn=sqlite3.connect(fname)
    db=conn.cursor()
    conn.text_factory=str
    db.execute('CREATE TABLE list (id INTEGER PRIMARY KEY AUTOINCREMENT,\
               SRC TEXT,TGT TEXT, exitcode INTEGER)')
    conn.commit() 
    return

def mark_done(options, SRC,TGT, exitcode):
    conn=sqlite3.connect(options.sqlfile)
    conn.text_factory=str
    db=conn.cursor()
    db.execute("UPDATE list SET exitcode=? \
               WHERE SRC=? AND TGT=?",(exitcode,SRC,TGT))
    conn.commit()
    return

def add(options):
    conn=sqlite3.connect(options.sqlfile)
    conn.text_factory=str
    db=conn.cursor()
    db.execute("INSERT INTO list(SRC,TGT,exitcode)\
               VALUES(?,?,?)",(options.SRC,options.TGT,1))
    conn.commit()
    return

def clear(options,everything=False):
    conn=sqlite3.connect(options.sqlfile)
    conn.text_factory=str
    db=conn.cursor()
    db.execute("DELETE FROM list WHERE exitcode == 0")
    if everything:
        db.execute("DELETE FROM list WHERE SRC LIKE '%'")
    conn.commit()
    return

def get_list(options):
    conn=sqlite3.connect(options.sqlfile)
    conn.text_factory=str
    db=conn.cursor()
    if options.rerun:
        db.execute("SELECT SRC,TGT FROM list ORDER BY id")
    else:
        db.execute("SELECT SRC,TGT FROM list WHERE exitcode > 0 ORDER BY id")
    nextEnt=db.fetchall()
    return nextEnt

def list_URLs(options):
    conn=sqlite3.connect(options.sqlfile)
    conn.text_factory=str
    db=conn.cursor()
    if options.listAllDB:
        db.execute("SELECT * FROM list ORDER BY id")
    else:
        db.execute("SELECT * FROM list WHERE exitcode > 0 ORDER BY id")
    print "EC\tSRC\tTGT"
    for row in db:
        print "%s\t%s\t%s" % (row[3],row[1],row[2])
    return

def start_sync(options):
    sync_list=get_list(options)
    for i,sync in enumerate(sync_list):
        (SRC,TGT)=sync
        if not SRC:
            return
        print("Starting: #%d %s -> %s"%(i+1,SRC,TGT))
        syncopts=shlex.split(options.options)
        command=['rsync']
        command.extend(syncopts)
        command.extend([SRC,TGT])
        popen = subprocess.Popen(command, 
                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=8)
        lines_iterator = iter(popen.stdout.readline, b"")
        j=0
        try:
            for line in lines_iterator:
                sys.stdout.write(line)
                j+=1
                if j%100==0:
                    print("#%d %s -> %s"%(i+1,SRC,TGT))
        except KeyboardInterrupt:
            popen.kill()
            sys.exit(1)
        mark_done(options,SRC,TGT,popen.wait())
        if popen.returncode>0:
            lines_iterator = iter(popen.stderr.readline, b"")
            for line in lines_iterator:
                sys.stdout.write(line)
            print("FAILED: EC: %d, %s -> %s"%(popen.returncode,SRC,TGT))
        print("Finished: #%d %s -> %s"%(i+1,SRC,TGT))

def main():
    options=setup_options();

    if not os.path.exists(options.sqlfile):
        createdb(options.sqlfile);
    if options.SRC!='' and options.TGT!='':
        print("Adding: %s -> %s"%(options.SRC,options.TGT))
        add(options)
    if options.clear:
        print("Clearing database")
        clear(options,options.clearAll)
    if options.listDB:
        list_URLs(options)
    if options.listAllDB:
        list_URLs(options)
    if options.run:
        print("Start synchronization. ctrl-c to exit")
        while True:
            start_sync(options)
            if not options.forever:
                break
            else:
                time.sleep(5)
    sys.exit(0)

main()