如何将包含__name__ == '__main__'的Python脚本封装为可复用公共包

第一步：调整包目录结构 #

首先给你的包起个名称，示例为ea_data_upload，最终目录结构如下：

ea_data_upload/
├── __init__.py    # 导出包的公共API
├── core.py        # 存放三个核心公共函数
├── cli.py         # 存放原来的主入口逻辑，封装为可调用函数
└── __main__.py    # 可选，支持直接用python -m 方式运行包

第二步：拆分核心代码与入口逻辑 #

1. core.py 内容 #

保留三个原有的自定义函数，移除硬编码的配置导入：

import pandas as pd
import numpy as np
import datetime

def clockPrint(sentence):
    now = datetime.datetime.now()
    date_time = now.strftime("%H:%M:%S")
    print(date_time + " : " + sentence)

def uploadToEA(df_, ds_api_name, operation_, instance, xmd_=None): #Upsert #Overwrite
    import SalesforceEinsteinAnalytics as EA
    clockPrint("Upload Process Initiated for "+instance+" instance...")
    if instance.lower() == 'commercial':
        EAS = EA.salesforceEinsteinAnalytics(env_url='https://spglobalratings.my.salesforce.com', browser='chrome')
    if instance.lower() == 'analytical':
        EAS = EA.salesforceEinsteinAnalytics(env_url='https://spglobalratingsae.my.salesforce.com', browser='chrome')

    EAS.load_df_to_EA(df_, dataset_api_name=ds_api_name, operation=operation_,xmd=xmd_,fillna=False)
    clockPrint("Upload Process Completed successfully for "+instance+" instance. Navigate to (Einstein Analytics --> Data Manager --> Monitor) to check progress.")

def processDate(date):
    if pd.isnull(date):
        return np.nan
    else:
        date = pd.to_datetime(date)
        date = datetime.datetime.strftime(date,"%m/%d/%Y")
        return date

2. cli.py 内容 #

把原来if __name__ == '__main__'里的逻辑封装为可接收配置参数的函数，同时保留直接运行的入口方便调试：

import pandas as pd
from .core import clockPrint, uploadToEA, processDate

def run_upload(cfg):
    """
    接收配置对象，执行完整的上传流程
    :param cfg: 配置对象，包含所有需要的配置字段
    """
    df = pd.read_csv(cfg.FILE_PATH, dtype={"As of Date": str})
    if len(cfg.DATE_COLUMNS) != 0:
        for c in cfg.DATE_COLUMNS:
            df[c] = df[c].apply(lambda x: processDate(x))

    for c in df.columns:
        if df[c].dtype == "O":
            df[c].fillna('', inplace=True)
        elif np.issubdtype(df[c].dtype, np.number):
            df[c].fillna(0, inplace=True)
        elif df[c].dtype == "datetime64[ns]":
            df[c] = df[c].apply(lambda x: processDate(x))
            df[c].fillna("", inplace=True)

    df.fillna("", inplace=True)
    for instance in cfg.INSTANCES:
        if instance.lower() == 'commercial':
            uploadToEA(df, cfg.COM_DATASET_API_NAME, cfg.COM_OPERATION, instance, cfg.COM_XMD)
        elif instance.lower() == 'analytical':
            uploadToEA(df, cfg.ANA_DATASET_API_NAME, cfg.ANA_OPERATION, instance, cfg.ANA_XMD)
        else: 
            clockPrint("Update INSTANCES variable as ['Commercial'] or ['Analytical'] or ['Commercial','Analytical'].")

# 保留直接运行的入口，兼容原有使用习惯
if __name__ == '__main__':
    import EA_Upload_config as cfg
    run_upload(cfg)

3. init.py 内容 #

导出公共方法，方便其他代码导入使用：

from .core import clockPrint, uploadToEA, processDate
from .cli import run_upload

__all__ = ["clockPrint", "uploadToEA", "processDate", "run_upload"]

4. 可选：main.py 内容 #

支持直接通过python -m ea_data_upload命令运行包，自动加载用户本地的配置文件：

import EA_Upload_config as cfg
from .cli import run_upload

if __name__ == "__main__":
    run_upload(cfg)

第三步：两种使用方式 #

• 作为库调用：其他Python代码可以直接导入包中的方法使用，支持自定义配置：

  from ea_data_upload import run_upload
  # 导入自定义配置对象
  import my_custom_config as cfg
  run_upload(cfg)
  

• 作为命令行工具使用：用户在当前工作目录编写好自己的EA_Upload_config.py配置文件后，直接运行python -m ea_data_upload即可执行完整上传流程，和原来直接跑脚本的效果完全一致。

内容的提问来源于stack exchange，提问作者darshika verma