分享一個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()