今天给大家分享一个python启动文件脚本
在日常开发中，我们常常需要运行多条命令来完成“静态收集”“数据库迁移”“启动服务”……如果把这些命令整合到一个脚本里就好了
一、整体流程概览
二、脚本
import argparse
import logging
import os
import sys
import time
import 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），不影响后续流程。
pass
def 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 CONFIG
bind = 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()
、收益
运维同学只需记住一条命令。
CI/CD 管道中，只需执行一次脚本即可完成全部准备工作。
后续只要在脚本中新增命令分支，即可支持新的功能。
📌 作者：叫我DPT
📅 日期：2025 年
🔗 原创整理，可自由转载，请注明出处