python Django到生产调用文件（#WIP）

Posted 2021-05-09

from invoke import task, run
from paramiko import SSHConfig, SSHClient, SSHException, AutoAddPolicy

import logging
import os


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('Paramiko')

PROJECT_DIR = os.path.dirname(__file__)


def tar_command(tar_file, folder_name='', untar=False, chdir=False):
    """ Returns [un]tar command for given folder/file """
    return 'tar {operation} {tar_file} {chdir} {folder_name}'.format(
        folder_name=folder_name, tar_file=tar_file,
        operation=('xvfz' if untar else 'cvfz'),
        chdir=('-C' if chdir else '')
    )


def rm_command(path):
    return 'rm -rf {}'.format(path)


def connect_to_server(server='asistia_nginx'):
    """ Returns a paramiko.SSHClient connected to specific server """
    # Open configuration file
    with open('config') as config_file:
        ssh_config = SSHConfig()
        ssh_config.parse(config_file)

    server_config = ssh_config.lookup(server)
    user, hostname, port, keys = (
        server_config.get('user'),
        server_config.get('hostname'),
        server_config.get('port'),
        server_config.get('identityfile')
    )

    if not user or not hostname or not port or not keys:
        logger.error('Server name not found inn config file')
        return exit()

    try:
        port = int(port)
    except ValueError as e:
        logger.critical('Invalid port')
        exit()

    ssh_client = SSHClient()
    # Auto add to unknown hosts
    ssh_client.set_missing_host_key_policy(AutoAddPolicy())

    try:
        ssh_client.connect(
            hostname, port,
            username=user, key_filename=keys
        )
    except SSHException as e:
        logger.critical('Could not connect to SSH Server')
        logger.error(e)
        exit()

    return ssh_client


@task
def collectstatic(production=False, folder_name='statics'):
    """Runs Django's collectstatic and send files to server"""

    run('./manage.py collectstatic --noinput')
    ssh_client = connect_to_server('asistia_nginx')

    # Get a valid SFTPClient from SSHClient
    sftp_client = ssh_client.open_sftp()

    PRODUCTION_PATH = '/home/asistia/production/'
    STAGING_PATH = '/home/asistia/staging/'
    REMOTE_DIR = PRODUCTION_PATH if production else STAGING_PATH

    legacy_remote_folder_name = 'static'
    legacy_remote_folder_path = os.path.join(REMOTE_DIR, legacy_remote_folder_name)  # NOQA

    tar_name = '{folder_name}.tar.gz'.format(folder_name=folder_name)

    local_tar = os.path.join(PROJECT_DIR, tar_name)
    remote_tar = os.path.join(REMOTE_DIR, tar_name)

    # What name will have the folder when untar
    remote_folder_name = os.path.join(REMOTE_DIR, folder_name)

    logger.info('Putting local statics folder into a tar file')
    run(tar_command(local_tar, folder_name))

    logger.info('Sending local tar file to remote server... this may take a while')  # NOQA
    sftp_client.put(local_tar, remote_tar)

    logger.info('Removing remote static folder')
    ssh_client.exec_command(rm_command(legacy_remote_folder_path))

    logger.info('Untar remote tar file')
    ssh_client.exec_command(tar_command(remote_tar, REMOTE_DIR, chdir=True, untar=True))  # NOQA

    logger.info('Moving untared* file to correct folder')
    ssh_client.exec_command('mv {} {}'.format(
        remote_folder_name,
        legacy_remote_folder_path,
    ))

    logger.info('Removing remote tar file')
    ssh_client.exec_command(rm_command(remote_tar))

    logger.info('Removing local tar file')
    run(rm_command(local_tar))

    logger.info('Removing local statics folder')
    run(rm_command(folder_name))

    sftp_client.close()
    ssh_client.close()
    logger.info('We are done!')


@task
def deploy(production=False):
    """
    Collect statics
    Push changes to Github
    Connect to staging or production server
    Pull changes
    Restarts supervisor app
    # WIP
    """

    # Collect statics
    collectstatic(production)
    logger.info('We are done!... for now')


if __name__ == '__main__':
    # Just for fun
    logger.info('Hi! Nothing to see here :) \n\n')
    run('python -c "import this"')