分享一个python启动文件脚本（django示例）

今天给大家分享一个python启动文件脚本

在日常开发中，我们常常需要运行多条命令来完成“静态收集”“数据库迁移”“启动服务”……如果把这些命令整合到一个脚本里就好了

一、整体流程概览

二、脚本

import argparse
import logging
import os
import sys
import timeimport django
from django.core import management# 获取当前脚本所在的目录
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# 定义应用代码所在的目录为 'apps' 子目录
APP_DIR = os.path.join(BASE_DIR, 'apps')# 将当前工作目录切换到脚本所在的目录
os.chdir(BASE_DIR)
# 将 'apps' 目录添加到 Python 的模块搜索路径中，这样可以方便地导入 'apps' 目录下的模块
sys.path.insert(0, APP_DIR)
# 设置 Django 的 settings 模块。这是 Django 项目的核心配置文件。
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "smartdoc.settings")
# 初始化 Django 环境，只有初始化后才能使用 Django 的各种功能，比如模型、管理命令等。
django.setup()def collect_static():"""收集静态文件到指定目录本项目主要是将前端 vue/dist 的前端项目放到静态目录下面:return:"""logging.info("Collect static files")try:# 调用 Django 的 'collectstatic' 管理命令，用于将各个 app 中的静态文件收集到一个统一的目录中。# '--no-input'：禁止用户交互式输入。# '-c'：清除之前收集的静态文件。# verbosity=0：设置命令输出的详细程度，0 表示不输出额外信息。# interactive=False：进一步确保非交互式执行。management.call_command('collectstatic', '--no-input', '-c', verbosity=0, interactive=False)logging.info("Collect static files done")except:# 如果收集静态文件过程中发生任何异常，则忽略（pass），不影响后续流程。passdef perform_db_migrate():"""初始化数据库表"""logging.info("Check database structure change ...")logging.info("Migrate model change to database ...")try:# 调用 Django 的 'migrate' 管理命令，用于将 Django 模型的变化同步到数据库中，创建或更新数据库表结构。management.call_command('migrate')except Exception as e:# 如果数据库迁移过程中发生异常，记录错误日志并退出程序。logging.error('Perform migrate failed, exit', exc_info=True)sys.exit(11)def start_services():# 从命令行参数中获取要启动的服务列表，如果 'args.services' 是一个列表则直接使用，否则将其包装成一个列表。services = args.services if isinstance(args.services, list) else [args.services]start_args = []# 如果命令行参数中包含 '--daemon'，则将其添加到启动参数列表中，表示以守护进程模式运行。if args.daemon:start_args.append('--daemon')# 如果命令行参数中包含 '--force'，则将其添加到启动参数列表中，可能用于强制执行某些操作。if args.force:start_args.append('--force')# 如果命令行参数中包含 '--worker'，则将其值（worker 数量）添加到启动参数列表中。if args.worker:start_args.extend(['--worker', str(args.worker)])else:# 如果命令行参数中没有指定 worker 数量，则尝试从环境变量 'CORE_WORKER' 中获取。worker = os.environ.get('CORE_WORKER')# 如果环境变量 'CORE_WORKER' 存在且是数字，则将其添加到启动参数列表中。if isinstance(worker, str) and worker.isdigit():start_args.extend(['--worker', worker])try:# 调用 Django 的管理命令来启动指定的服务。这里的 'action' 变量在主程序中根据命令行参数确定。# '*services' 和 '*start_args' 用于将列表中的元素作为独立的参数传递给 'call_command'。management.call_command(action, *services, *start_args)except KeyboardInterrupt:# 如果用户按下 Ctrl+C 中断程序，则记录信息并等待 2 秒后退出。logging.info('Cancel ...')time.sleep(2)except Exception as exc:# 如果启动服务过程中发生其他异常，记录错误日志并等待 2 秒后退出。logging.error("Start service error {}: {}".format(services, exc))time.sleep(2)def dev():# 从命令行参数中获取要运行的服务，与 'start_services' 类似。services = args.services if isinstance(args.services, list) else args.services# 如果要运行的服务包含 'web'，则调用 Django 的 'runserver' 管理命令启动开发服务器，监听 0.0.0.0:8080。if services.__contains__('web'):management.call_command('runserver', "0.0.0.0:8080")# 如果要运行的服务包含 'celery'，则调用 Django 的 'celery' 管理命令启动 Celery worker。elif services.__contains__('celery'):management.call_command('celery', 'celery')# 如果要运行的服务包含 'local_model'，则设置环境变量 'SERVER_NAME' 为 'local_model'，# 并从 'smartdoc.const' 模块的 'CONFIG' 字典中获取本地模型服务的主机和端口，# 然后调用 'runserver' 启动开发服务器监听指定的地址。elif services.__contains__('local_model'):os.environ.setdefault('SERVER_NAME', 'local_model')from smartdoc.const import CONFIGbind = f'{CONFIG.get("LOCAL_MODEL_HOST")}:{CONFIG.get("LOCAL_MODEL_PORT")}'management.call_command('runserver', bind)# 这是 Python 的主程序入口点，当脚本直接运行时会执行这里的代码。
if __name__ == '__main__':# 设置环境变量 'HF_HOME'，这可能与脚本中使用的某个库（如 Hugging Face Transformers）有关，指定其配置文件的存储路径。os.environ['HF_HOME'] = '/opt/maxkb/model/base'# 创建一个 ArgumentParser 对象，用于解析命令行参数。parser = argparse.ArgumentParser(description="""qabot service control tools;Example: \r\n%(prog)s start all -d;""")# 添加一个名为 'action' 的位置参数，用户必须提供这个参数来指定要执行的操作。# 'type=str'：指定参数类型为字符串。# 'choices'：限定了 'action' 参数的可选值，包括 'start'（启动服务）、'dev'（开发模式）、'upgrade_db'（升级数据库）、'collect_static'（收集静态文件）。# 'help'：参数的帮助信息。parser.add_argument('action', type=str,choices=("start", "dev", "upgrade_db", "collect_static"),help="Action to run")# 解析已知的命令行参数，将解析结果存储在 'args' 中，并将未知的参数存储在 'e' 中。args, e = parser.parse_known_args()# 根据不同的 'action' 值，为 'services' 参数设置不同的默认值和可选值。# 如果 'action' 是 'start'，则 'services' 参数的默认值为 'all'，可选值为 'all'、'web'、'task'。# 否则（如果 'action' 是 'dev'），则 'services' 参数的默认值为 'web'，可选值为 'web'、'celery'、'local_model'。# 'nargs="*"'：表示 'services' 参数可以接受零个或多个值，这些值将被存储在一个列表中。parser.add_argument("services", type=str, default='all' if args.action == 'start' else 'web', nargs="*",choices=("all", "web", "task") if args.action == 'start' else ("web", "celery", 'local_model'),help="The service to start",)# 添加可选参数 '-d' 或 '--daemon'，用于指定是否以守护进程模式运行。'nargs="?"' 表示该参数可以有零个或一个值，'const=True' 表示如果只指定了该参数而没有提供值，则其值为 True。parser.add_argument('-d', '--daemon', nargs="?", const=True)# 添加可选参数 '-w' 或 '--worker'，用于指定 worker 的数量。'type=int' 表示参数类型为整数，'nargs="?"' 与 '--daemon' 类似。parser.add_argument('-w', '--worker', type=int, nargs="?")# 添加可选参数 '-f' 或 '--force'，用于指定是否强制执行某些操作，与 '--daemon' 类似。parser.add_argument('-f', '--force', nargs="?", const=True)# 解析所有的命令行参数，将最终的解析结果存储在 'args' 中。args = parser.parse_args()# 将解析得到的 'action' 参数的值赋给变量 'action'。action = args.action# 根据 'action' 的值执行相应的操作。if action == "upgrade_db":# 如果 'action' 是 'upgrade_db'，则调用 'perform_db_migrate' 函数来执行数据库迁移。perform_db_migrate()elif action == "collect_static":# 如果 'action' 是 'collect_static'，则调用 'collect_static' 函数来收集静态文件。collect_static()elif action == 'dev':# 如果 'action' 是 'dev'，则先收集静态文件，然后执行数据库迁移，最后调用 'dev' 函数启动开发服务。collect_static()perform_db_migrate()dev()else:# 如果 'action' 不是以上任何值（通常是 'start'），则先收集静态文件，然后执行数据库迁移，最后调用 'start_services' 函数启动指定的服务。collect_static()perform_db_migrate()start_services()