荣登wakatime排行榜前15|蓝白社区的博客|蓝白社区

微信公众号

站点链接

博文目录

#custom-toc-container

荣登wakatime排行榜前15

BGLB0324 2022年1月8日 00:43

最后发布:2022年1月8日 00:43首发:2022年1月8日 00:43

445 0

博文分类：

程序人生

博文标签：

本文链接：http://blog.bglb.work/blog/blog-detail/79

版权

![wakatime](https://img.bglb.work/img/1641573272579.png?x-oss-process=style/blog_img#pic_center)

## 分享一下个人使用的 django 自动化部署脚本
```python
# -*- coding:utf-8  -*-
# @Time     : 2021-12-06 20:53
# @Author   : BGLB
# @Software : PyCharm
import datetime
import logging
import os
import subprocess
import sys
import time
import traceback

import psutil
from pip._internal import main as pip

self_path = os.path.abspath(__file__)

project_path = os.path.dirname(os.path.dirname(self_path))
log_path = os.path.join(project_path, 'logs', 'auto_update.log')

logger = logging.getLogger('auto_update')
file_handel = logging.FileHandler(filename=log_path, encoding='utf-8')
logger.setLevel('INFO')
file_handel.setFormatter(
    logging.Formatter('[%(asctime)s][%(name)s.%(funcName)s():%(lineno)d] [%(levelname)s] %(message)s'))
logger.addHandler(file_handel)

python_path = ''

def exec_cmd(cmd: str, timeout=10):
    proc = subprocess.Popen(cmd,
                            shell=True,
                            stdout=subprocess.PIPE,
                            stderr=subprocess.STDOUT,
                            encoding=None)
    res = []
    try:
        out, err = proc.communicate(timeout=timeout)
    except subprocess.TimeoutExpired as ex:
        res.append("commend_timeout:{}".format(ex))
        return res
    for line in out.splitlines():
        try:
            line_str = line.decode()
        except Exception:
            line_str = line.decode('GBK')

res.append(line_str)

return res

def exec_django_manage(action: str) -> list:
    command = [python_path, os.path.join(project_path, 'manage.py'), action, '--noinput']
    result = exec_cmd(' '.join(command))
    return result[1:]

def start_auto_update():
    """
        启动自己
    :return:
    """
    logger.info('sys.executable: {}'.format(sys.executable))
    logger.info('sys.path: {}'.format(sys.path))
    python_path = sys.executable if sys.executable.endswith('python.exe') else ''
    sys_path = sys.path
    if not python_path:
        for item in sys_path:
            item = item.lower()
            target_str = ['\\env\\', '\\venv\\']
            for target in target_str:
                if target in item:
                    python_path = item.replace('lib\\site-packages', 'Scripts\\python.exe')
                    break

if not python_path:
        for item in sys_path:
            target_str = ['\\lib\\site-packages']
            item = item.lower()
            for target in target_str:
                if target in item:
                    python_path = item.replace('lib\\site-packages', 'python.exe')
                    break

if not python_path:
        logger.error('没有找到 python 路径')
        return

logger.info('python_path: {}'.format(python_path))
    os.system('start "auto_update" /d {} {} {}'.format(project_path, python_path, self_path))

def git_pull():
 """
 git pull
 :return:
 """
 pull = 'git -C {} pull --force origin master:master'.format(project_path)
 change_file_list = 'git -C {} diff --name-status HEAD@{{1}}..HEAD'.format(project_path)
 message_list = 'git -C {} log HEAD@{{1}}..HEAD --pretty=format:"%s from %cn at %ci"'.format(project_path)
 res_lines = exec_cmd(pull, 20)
 logger.info(res_lines)
 res_lines.insert(0, '1. git pull')
 if 'Already up to date.' != res_lines[-1]:
 return False, res_lines
 change_file = exec_cmd(change_file_list, 20)
 logger.info(change_file)
 change_file.insert(0, '2. change file list')
 res_lines.extend(change_file)
 time.sleep(1)
 messages = exec_cmd(message_list, 20)
 logging.info(messages)
 messages.insert(0, '3. push message')
 res_lines.extend(messages)
 for item in change_file:
 if 'requirments.txt' in item:
 try:
 install_requirments = 'install third package'
 res_lines.append('4. {}'.format(install_requirments))
 logger.info(install_requirments)
 pip(['install', '-r', os.path.join(project_path, 'requirments.txt'), '-i',
 'https://pypi.douban.com/simple'])
 install_requirments_success = '下载 requirments.txt 成功！'
 logger.info(install_requirments_success)
 res_lines.append(install_requirments_success)
 except Exception:
 install_requirments_failed = f"下载 requirments.txt 失败：\n{traceback.format_exc()}"
 logger.error(install_requirments_failed)
 res_lines.append(install_requirments_failed)
 return False, res_lines

if 'static/' in item:
 collect_static_start = 'collect static files'
 logger.info(collect_static_start)
 collect_static = exec_django_manage('collectstatic')
 collect_static.insert(0, '5. {}'.format(collect_static_start))

if 'unmodified' not in collect_static[-1]:
                collect_static.append('**收集静态文件失败！**')
                logger.error('收集静态文件失败！')
                return False, res_lines
            else:
                collect_static.append('收集静态文件成功！')
                logger.info('收集静态文件成功! ')

res_lines.extend(collect_static)

return True, res_lines

def restart_apache():
    """
        重启 apache
    :return:
    """

cmd_line = ['net stop apache2.4', 'net start apache2.4']
    res = []
    for cmd in cmd_line:
        res.extend(exec_cmd(cmd, 60*2))
        time.sleep(1)
    if 'apache2.4 服务已经启动成功。' in res or 'The apache2.4 service was started successfully.' in res:
        return True, res
    return False, res

def restart_nginx():
    """
        重启 nginx
    """
    cmd_line = ['nginx -s reload', 'systemctl status nginx']
    res = []
    for cmd in cmd_line:
        res.extend(exec_cmd(cmd, 60*2))
        time.sleep(1)
    if 'Active: active (running)' in "".join(res):
        return True, res
    return False, res

def get_system_info():
    """
        获取系统信息
    :return:
    """
    sys_os = 'WINDOWS'
    info = "## 系统信息\n\n"
    for item in ['WINDOWS', 'LINUX', 'MACOS']:
        if getattr(psutil, item):
            info += "> - 系统平台: **{}**\n\n".format(item)
            sys_os = item
            break

netcard_info = []
    for k, v in psutil.net_if_addrs().items():
        for item in v:
            if item[0] == 2 and not item[1] == '127.0.0.1':
                netcard_info.append((k, item[1]))
    net_ip = "".join([f"> - {item[0]}: **{item[1]}**\n\n" for item in netcard_info])
    info += net_ip
    memory = psutil.virtual_memory()
    info += "> - 内存详情: **{}GB/{}GB**\n\n".format(round(memory.used/1024/1024/1024, 2),
                                                 round(memory.total/1024/1024/1024, 2))
    disk_info = []
    info += "> - 磁盘详情\n\n"
    disk_total = psutil.disk_partitions() if sys_os == "WINDOWS" else ['/', ]
    for item in disk_total:
        if isinstance(item, str):
            disk_key = item
        else:
            disk_key = item.device

temp_disk_info = psutil.disk_usage(disk_key)
        _info = "{}: **{}GB/{}GB**".format(disk_key, round(temp_disk_info.used/1024/1024/1024, 2),
                                           round(temp_disk_info.total/1024/1024/1024, 2))
        disk_info.append(_info)
    info += "> - {}".format("\n\n > - ".join(disk_info))

return info

def current_version():
    version_path = os.path.join(project_path, 'version')
    with open(version_path, mode='r', encoding='utf-8') as f:
        version = f.readline()
        f.close()
        return f'**current_version**：{version} \n\n\n'

def update():
    """
        更新
    :return:
    """
    from dingtalkchatbot.chatbot import DingtalkChatbot
    DINGDING_ROBOT_URL = ''
    DINGDING_SECRET = ''
    DINGDING_AT_MOBILES = []

env_path = os.path.join(project_path, 'conf', 'env.py')
    with open(env_path, encoding='utf=8', mode='r') as f:
        text = f.readlines()
    for item in text:
        if item and item.startswith('DINGDING'):
            value = item.strip(' ').strip('\n').split('=', 1)[-1]

if 'DINGDING_ROBOT_URL' in item:
                DINGDING_ROBOT_URL = eval(value)

if item and 'DINGDING_SECRET' in item:
                DINGDING_SECRET = eval(value)

if item and 'DINGDING_AT_MOBILES' in item:
                DINGDING_AT_MOBILES = eval(value)

dingding = DingtalkChatbot(DINGDING_ROBOT_URL, secret=DINGDING_SECRET)
    flag, git_res = git_pull()
    markdown = '# 自动化部署日志\n\n - start_time: {}\n\n'.format(datetime.datetime.now().strftime('%Y-%m-%d ''%H:%M:%S'))
    version = current_version()
    markdown += f"\n\n{get_system_info()}\n\n"

markdown += '\n\n**pull update**\n'
 for index, item in enumerate(git_res):
 prefix = '' if item.endswith('') else '> - '
 markdown += '\n{}{}\n'.format(prefix, item.lstrip())

if not flag:
 markdown += '\n\n\n**git pull 失败 快去看看**\n'
 dingding.send_markdown(title='自动化部署失败', text=markdown, at_mobiles=DINGDING_AT_MOBILES)
 return

flag, restart_apache_res = restart_nginx()
    markdown += '\n\n\n**restart apache**\n>\n'
    for item in restart_apache_res:
        if item.strip(' '):
            markdown += '> - {}\n'.format(item.lstrip(' '))

if not flag:
 markdown += '\n\n\n**restart apache 失败 快去看看**\n\n'
 dingding.send_markdown(title='自动化部署失败', text=markdown, at_mobiles=DINGDING_AT_MOBILES)
 return

markdown += '\n\n\n - end_time: {}\n\n\n'.format(datetime.datetime.now().strftime('%Y-%m-%d ''%H:%M:%S'))

markdown += version
    dingding.send_markdown(title='自动化部署成功', text=markdown, at_mobiles=DINGDING_AT_MOBILES)

if __name__ == '__main__':
    try:
        update()
    except Exception:
        logger.error(traceback.format_exc())

```
结合 gitee 的 webhooks 以及 钉钉机器人 就可以实现 自动化部署+pip下载+服务器信息+钉钉通知
最近变懒了 能跑就行 改项目代码头大 懒得抽象了.

暂时没有评论

请登录后评论

半根蓝白个人主页

CSDN

Github

博文目录

荣登wakatime排行榜前15