automatically check for updates in GitHub repos; if there is update, pull from upstream and send notification to Gmail

auto_git_pull_to_gmail.py

import subprocess
import os
import time
import smtplib
from email.message import EmailMessage


# put your GA directory here, e.g. '/home/fay/code/GA/DSI'
CWD = ''

# put the path to a .gitignore template here
# e.g. '/home/fay/code/GA/DSI/projects/West-Nile-Virus-Prediction/.gitignore'
GITIGNORE = ''

# put your GitHub repo upstream pattern here
# e.g. 'git@git.generalassemb.ly:DSI-DC-5/'
UPSTREAM = ''

# put your GitHub origin pattern here
# e.g. 'git@git.generalassemb.ly:jingfeicai/'
ORIGIN = ''

# put a list of your other GitHub account patterns here (can be left empty)
# e.g. ['git@github.com:GA-DSI-DC-5-Team-1/', 'git@github.com:Ailuropoda1864/']
OTHER_GITHUB = []

# put your Gmail address here
FROM_ADDR = '@gmail.com'

# put your Gmail password here
GMAIL_PSW = ''


def main():
    git_dir = find_git_dir(CWD)
    reports = []
    for dirpath, repo in git_dir.items():
        # print('current directory: {}'.format(dirpath))
        config_path = os.path.join(dirpath, '.git', 'config')
        config = read_file(config_path)

        if '[remote "upstream"]' in config:
            report = git_pull(dirpath, repo)
            if report:
                reports.append(report)
                print(report)

        else:
            if ORIGIN in config:
                add_remote_upstream(dirpath)
                git_fetch(dirpath)

            elif UPSTREAM in config:
                report = git_pull(dirpath, repo, 'origin HEAD'.split())
                if report:
                    reports.append(report)
                    print(report)

            else:
                for github in OTHER_GITHUB:
                    if github in config:
                        break
                else:
                    print('{!r} does not have remote upstream.'.format(dirpath))

    if reports:
        send_to_gmail('Git Pull Script Report', ''.join(reports))

    print('Script finished at {}\n'.format((time.ctime())))


def find_git_dir(cwd):
    return {dirpath: {'dirnames': dirnames, 'filenames': filenames}
            for dirpath, dirnames, filenames in os.walk(cwd)
            if '.git' in dirnames}


def read_file(file):
    with open(file) as f:
        return f.read()


def git_pull(dirpath, repo, remote_branch=['upstream', 'master']):
    remote_ref_path = os.path.join(
        dirpath, '.git', 'refs', 'remotes', *remote_branch)
    try:
        remote_ref = read_file(remote_ref_path)
    except FileNotFoundError as e:
        print(str(e))
        remote_ref = None

    git_fetch(dirpath, remote_branch)

    # if .git/refs/remotes/upstream/master has changed since fetch
    if remote_ref and remote_ref != read_file(remote_ref_path):

        # add gitignore if not present
        if '.gitignore' not in repo['filenames']:
            add_gitignore(GITIGNORE, dirpath)

        # commit all local changes, then merge with upstream
        merge = git_merge(dirpath)
        return 'Remote branch of {!r} is updated:\n{}\n'.format(dirpath, merge)


def git_fetch(cwd, remote_branch=['upstream', 'master']):
    return subprocess.run(['git', 'fetch'] + remote_branch, cwd=cwd)


def git_merge(cwd):
    os.chdir(cwd)
    subprocess.run('git add .'.split())
    subprocess.run(
        'git commit -m'.split() + ["auto commit before merging with remote"])
    try:
        return subprocess.check_output(
            'git merge -m'.split() + ["auto merge by script", 'FETCH_HEAD'],
            stderr=subprocess.STDOUT,
            universal_newlines=True)
    except subprocess.CalledProcessError:
        return 'Error occurred during merge.'.format(cwd)


def add_gitignore(template, cwd):
    subprocess.run(['cp', template, os.path.join(cwd, '.gitignore')])


def add_remote_upstream(cwd):
    dir = cwd.split('/')[-1]
    subprocess.run(
        'git remote add upstream'.split() + [UPSTREAM + dir + '.git'], cwd=cwd)


def send_to_gmail(subj, body, from_addr=FROM_ADDR, psw=GMAIL_PSW,
                  to_addr=None):
    # compose the email
    if to_addr is None:
        to_addr = from_addr
    msg = EmailMessage()
    msg.set_content(body)
    msg['Subject'] = subj
    msg['From'] = from_addr
    msg['To'] = to_addr

    # send the email
    server = smtplib.SMTP('smtp.gmail.com', 587)
    server.starttls()
    server.login(from_addr, psw)
    server.send_message(msg)
    server.quit()

if __name__ == '__main__':
    main()