菜单
个人主页
(当前)
写文章
浏览博文
    
搜索
登录
微信公众号
站点链接
半根蓝白个人主页
CSDN
Github
友情链接
摘繁华个人博客
博文目录
#custom-toc-container
荣登wakatime排行榜前15
BGLB0324
2022年1月8日 00:43
最后发布:2022年1月8日 00:43
首发:2022年1月8日 00:43
561
0
博文分类:
程序人生
博文标签:
版权声明:本文为博主[BGLB0324]原创文章,遵循
CC 4.0 BY
版权协议,转载请附上原文出处链接和本声明。
本文链接:
http://blog.bglb.work/blog/blog-detail/79
版权
 ## 分享一下个人使用的 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下载+服务器信息+钉钉通知 最近变懒了 能跑就行 改项目代码头大 懒得抽象了.
点赞
0
打赏
暂时没有评论
请
登录
后评论
暂时没有评论