python fabric auto deployment

fabfile.py

"""
fabfile.py

fabric config file.

It was for simplicity. I personally don't like to extend this script to more complex to config, read, work

usage:

Find `Configurations` part to config by all your needs, maybe code adjustment are also needed.

- `fab --list` to list all supported command
- `fab <command> -h` to see command help seperatedly

Normally, you only need to run:

```shell
fab deploy  # this equals fab deploy -m=dev
fab deploy -m=prod
```

"""
import os
import subprocess
import shutil
import requests
from urllib.parse import quote_plus
from fabric import task, Connection


class Server:
    '''
    A Server properties
    '''
    def __init__(self, host, *, name='Server1', port=22, user='root', password='', 
        user_group='www:www', pem='', deploy_path='/home') -> None:

        self.name = name    # server name
        self.host = host    # domain name or ip
        self.port = port    # 
        self.user = user    # login user
        self.password = password # password login
        self.pem = pem           # pem login
        self.user_group = user_group    # used for change remote directory user/group
        self.deploy_path = deploy_path  # remote deploy directory


class ServerGroup:
    '''
    A batch of servers
    '''
    def __init__(self, *, servers=[], domain='xyz.com', name='', need_notify=False, need_match_branch=False):
        self.servers = servers
        self.domain = domain        # for uniapp HBuilderX
        self.name = name
        self.need_notify = need_notify          # if send telegram message
        self.need_match_branch = need_match_branch  # if check git branch equal current mode
        

''' Configurations '''

# Server targets
MapModeGroup = {
    'dev': ServerGroup(servers=[
        Server(
            '18.2.3.223', 
            pem='/Users/<name>/data/pems/project.pem', 
            user='centos', 
            deploy_path='/home/wwwroot/project'
        ),
    ], domain='x.com'),
}

# Telegram config
TG_TOKEN = ''
TG_CHAT_ID = '-23423423' # test

# Zip filename
FILE_TAR = 'project.zip'
BUILD_DIR = 'dist'
# Remote directory upload to
REMOTE_PATH = '/tmp'
# Current app version
VERSION = '1.1'
''' Configurations '''

VersionString = ''


def getGroup(mode='dev'):
    '''
    Get server group by mode
    '''
    return MapModeGroup.get(mode)


def git_branch_name():
    '''
    Get git branch
    '''
    cmd = ['git', 'rev-parse', '--abbrev-ref', 'HEAD']
    return subprocess.check_output(cmd).decode('ascii').strip()


def git_commit_id(short=True):
    '''
    Get git commit hash
    '''
    cmd = ['git', 'rev-parse', '--short', 'HEAD']
    if not short:
        cmd = ['git', 'rev-parse', 'HEAD']
    return subprocess.check_output(cmd).decode('ascii').strip()


def git_last_n_log(n=3):
    cmd = ['git', 'log', '--oneline', f"-{n}"]
    return subprocess.check_output(cmd).decode('utf8').strip()


def run(c, cmd):
    return c.run(cmd, hide=True)


def sudo(c, cmd):
    return c.sudo(cmd, hide=True)


def update_version():
    '''
    Generate version files
    '''
    commit_id = git_commit_id()
    global VersionString
    with open('./public/version.js', 'wt', encoding='utf8') as fh:
        VersionString = f"{VERSION}.{commit_id}"
        fh.write(f"export const version='{VersionString}'\n")


@task(help={'mode': 'Build mode: "prod" or "dev"'})
def build(c, mode='dev'):
    '''
    Build this project
    '''
    g = getGroup(mode)
    if not g:
        return print('Invalid mode')
    
    update_version()

    subprocess.run(['npm', 'run', 'build'])
    

@task
def tar(c):
    '''
    Compress build target
    '''
    try:
        os.unlink(FILE_TAR)
    except Exception as e:
        print('Remove file warn: {}'.format(e))

    name, _ = os.path.splitext(FILE_TAR)
    print('zip it:', shutil.make_archive(name, 'zip', BUILD_DIR))


@task(help={'mode': 'Upload destination: "prod" or "dev"'})
def upload(c, mode='dev'):
    '''
    Upload compressed file to dest servers, and deploy them.
    '''
    g = getGroup(mode)
    if not g:
        return print('Invalid mode')

    succeed = [] # Successfully deployed server list
    for server in g.servers:
        param = {'key_filename': server.pem, 'passphrase': ''} if server.pem else {'password': server.password}
        with Connection(host=server.host, port=server.port, user=server.user, connect_kwargs=param) as conn:
            with conn.cd(REMOTE_PATH):
                conn.put(FILE_TAR, remote=REMOTE_PATH)
                run(conn, 'sudo -u root unzip -quo {} -d {}'.format(FILE_TAR, server.deploy_path))
                run(conn, f'sudo -u root chown -R {server.user_group} {server.deploy_path}')
                run(conn, 'rm -f {}'.format(FILE_TAR))
            succeed.append(server)
    if g.need_notify:
        ipstr = ' '.join([f'{s.name}[*.*.*.{s.host.split(".")[3]}]' for s in succeed])
        send(c, f"New version {VersionString} published {ipstr}")


@task
def send(c, msg='Hello, this is a test message'):
    '''
    Send a telegram message
    '''
    url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage?chat_id={TG_CHAT_ID}&text={msg}"
    ret = requests.get(url).json()
    if not ret['ok']:
        print('send Telegram msg failed:', ret)


@task(help={'mode': 'Deploy mode: "prod" or "dev"'})
def deploy(c, mode='dev'):
    '''
    One step to build, compress, upload, deploy this project
    '''
    g = getGroup(mode)
    if not g:
        return print('Invalid mode')
    
    if g.need_match_branch:
        branch = git_branch_name()
        if mode != branch:
            return print('Branch is wrong')
            
    #dependencies
    build(c, mode)
    tar(c)
    upload(c, mode)
    print('Done.')