如何将包含多个静态函数的Python脚本转换为Python库

我来一步步帮你把这个脚本改成标准的Python库，这样你可以在其他项目里轻松导入、复用这些功能：

1. 搭建标准库目录结构 #

首先创建一个规范的目录，比如命名为ea_uploader（你可以换成自己喜欢的名字），结构如下：

ea_uploader/
├── __init__.py       # 标记为Python模块，导出核心功能
├── core.py           # 存放所有核心函数（clockPrint、uploadToEA、processDate）
└── config.py         # 迁移原EA_Upload_config的配置内容

2. 重构核心代码 #

把原脚本里的函数移到core.py，同时优化代码结构和可读性：

import pandas as pd
import numpy as np
import datetime
from SalesforceEinsteinAnalytics import EA

def clockPrint(sentence):
    now = datetime.datetime.now()
    date_time = now.strftime("%H:%M:%S")
    print(f"{date_time} : {sentence}")  # 用f-string让代码更简洁

def uploadToEA(df_, ds_api_name, operation_, instance, xmd_=None):
    clockPrint(f"Upload Process Initiated for {instance} instance...")
    
    # 把实例URL用字典管理，避免重复if判断
    env_urls = {
        'commercial': 'https://spglobalratings.my.salesforce.com',
        'analytical': 'https://spglobalratingsae.my.salesforce.com'
    }
    
    # 增加实例合法性校验
    if instance.lower() not in env_urls:
        clockPrint(f"Invalid instance: {instance}. Only 'commercial' or 'analytical' are allowed.")
        return
    
    try:
        EAS = EA.salesforceEinsteinAnalytics(env_url=env_urls[instance.lower()], browser='chrome')
        EAS.load_df_to_EA(df_, dataset_api_name=ds_api_name, operation=operation_, xmd=xmd_, fillna=False)
        clockPrint(f"Upload Process Completed successfully for {instance} instance. Navigate to (Einstein Analytics --> Data Manager --> Monitor) to check progress.")
    except Exception as e:
        clockPrint(f"Upload failed for {instance} instance: {str(e)}")
        raise  # 可选：抛出异常让调用者自行处理错误

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

然后在__init__.py里导出核心函数，方便外部快速导入：

from .core import clockPrint, uploadToEA, processDate

__version__ = "0.1.0"  # 给你的库加个版本号，方便后续迭代

3. 处理配置文件 #

原EA_Upload_config.py的内容可以迁移到config.py里，作为默认配置，同时允许用户在使用时覆盖：

# 默认配置，用户可以根据自身需求修改
FILE_PATH = "your_default_data.csv"
DATE_COLUMNS = []
INSTANCES = ["commercial"]
COM_DATASET_API_NAME = "your_commercial_dataset"
COM_OPERATION = "Upsert"
COM_XMD = None
ANA_DATASET_API_NAME = "your_analytical_dataset"
ANA_OPERATION = "Upsert"
ANA_XMD = None

4. 提取测试/示例代码 #

把原脚本中if __name__ == '__main__':下面的代码单独抽出来，做成一个示例脚本example.py（放在库目录同级），方便用户参考如何使用你的库：

import pandas as pd
import numpy as np
from ea_uploader import clockPrint, uploadToEA, processDate
from ea_uploader.config import (
    FILE_PATH, DATE_COLUMNS, INSTANCES,
    COM_DATASET_API_NAME, COM_OPERATION, COM_XMD,
    ANA_DATASET_API_NAME, ANA_OPERATION, ANA_XMD
)

if __name__ == '__main__':
    df = pd.read_csv(FILE_PATH)
    
    # 处理日期列
    if len(DATE_COLUMNS) != 0:
        for c in DATE_COLUMNS:
            df[c] = df[c].apply(processDate)
    
    # 填充空值
    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(processDate)
            df[c].fillna("", inplace=True)
    df.fillna("", inplace=True)
    
    # 执行上传
    for instance in INSTANCES:
        if instance.lower() == 'commercial':
            uploadToEA(df, COM_DATASET_API_NAME, COM_OPERATION, instance, COM_XMD)
        elif instance.lower() == 'analytical':
            uploadToEA(df, ANA_DATASET_API_NAME, ANA_OPERATION, instance, ANA_XMD)
        else:
            clockPrint("Update INSTANCES variable as ['Commercial'] or ['Analytical'] or ['Commercial','Analytical'].")

5. 打包成可安装库（可选但推荐） #

如果想在多个环境使用或者分享给他人，可以把库打包成pip可安装的包：

1. 在根目录创建pyproject.toml（符合现代Python打包标准）：

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "ea-uploader"
version = "0.1.0"
authors = [
  { name="Your Name", email="your.email@example.com" }
]
description = "A utility library to upload pandas DataFrames to Salesforce Einstein Analytics"
requires-python = ">=3.8"
dependencies = [
  "pandas>=1.0",
  "numpy>=1.18",
  "SalesforceEinsteinAnalytics>=x.x.x"  # 替换成你实际使用的版本
]

2. 执行pip install .即可在本地安装这个库；或者用python -m build生成wheel包，上传到PyPI供他人下载。

6. 使用你的库 #

安装完成后，其他项目里就可以这样快速调用：

from ea_uploader import uploadToEA
import pandas as pd

# 加载并处理你的数据
df = pd.read_csv("your_data.csv")
# ...数据处理逻辑...

# 调用上传函数
uploadToEA(df, "your_dataset_api_name", "Upsert", "commercial")

这样你的脚本就完全转换成一个可复用、易维护的Python库啦！

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