xinops
营销数据运维一体化工具库,Windows / Mac 通用。
一个库搞定所有事:原
Functions_d、WeComMsg、HiveClient、feishu_to_excel已全部合并进 xinops, 只需pip install xinops一个包,不用再维护多个独立库。
| 项目 | 信息 |
|---|---|
| 当前版本 | v1.6.1 |
| 更新日期 | 2026-10-16 |
| Python 版本 | 3.7+ |
| 平台 | Windows / macOS |
与旧库的关系
已合并的独立库
以下独立库已全部合并进 xinops,无需再单独安装、单独维护:
| 原独立库 | 合并后位置 | 状态 |
|---|---|---|
Functions_d(1.46) |
DataProcessingAndMessaging 主类(方法名完全一致) |
✅ 已合并,不再更新 |
WeComMsg(企微消息发送) |
_wecom.py(内置 WeChatWorkSender) |
✅ 已合并,不再更新 |
HiveClient(Hive 连接客户端) |
_hive.py(内置 _HiveClient + 重试机制) |
✅ 已合并,不再更新 |
feishu_to_excel(飞书表格上传) |
_feishu.py(FeishuSheetUploader 类) |
✅ 已合并,不再更新 |
对现有脚本的影响
- 使用 xinops 的脚本:零影响,所有方法签名、行为、返回值完全不变
- 直接使用
WeComMsg/HiveClient的脚本:不受影响,PyPI 上的旧包仍然可用,只是不再更新了 - 迁移建议:新脚本统一用 xinops,老脚本可以继续用旧库,不用着急改
迁移示例(Functions_d → xinops)
# 旧写法
import Functions_d
ux = Functions_d.DataProcessingAndMessaging()
# 新写法(方法名完全一致,只需改 import)
import xinops
ux = xinops.DataProcessingAndMessaging()
安装
pip install xinops # 核心功能(Windows/Mac 通用)
# 可选功能按需加装:
pip install "xinops[excel_win]" # writer_excel_data / screen(xlwings 操作本机 Excel,仅 Windows)
pip install "xinops[browser]" # init_edge_driver(Selenium Edge 自动化)
pip install "xinops[pinyin]" # name_to_pinyin / car_task_transfer
pip install "xinops[all]" # 全部装上(等同旧版 Functions_d 的安装效果)
Mac 用户注意:
writer_excel_data(xlwings)与screen(Excel 截图)依赖 Windows 上的本机 Excel, Mac 上请使用跨平台的writer_excel_data_new(openpyxl 方案,无需安装 Excel)。 xlwings/pythoncom 均为方法内延迟导入,Mac 上未安装不影响库的导入与其他功能。
密钥配置文件(必须)
本库不再在代码中硬编码任何密码。首次使用前,创建 xinops_secrets.yaml:
# 企业微信消息
wecom_corpid: "你的企业微信CorpId"
wecom_corpsecret: "你的企业微信应用Secret"
wecom_agentid: "你的应用AgentId"
# 企业微信文档
wecom_doc_corpid: "你的企业微信CorpId"
wecom_doc_corpsecret: "你的文档应用Secret"
# 邮件 SMTP
smtp_sender: "你的发件邮箱"
smtp_pwd: "你的邮箱密码或授权码"
smtp_host: "你的SMTP服务器地址"
smtp_port: 465
# Hive 连接
hive_host: "你的Hive服务器地址"
hive_port: 10000
hive_user: "你的Hive用户名"
hive_pwd: "你的Hive密码"
# 大模型报错分析(可选,不配则自动使用内置规则翻译)
# 兼容所有 OpenAI Chat Completions 协议:智谱GLM / DeepSeek / Kimi / 豆包 / 通义千问
llm_enabled: true # 可选,默认 true;设 false 关闭大模型分析
llm_api_key: "你的大模型API Key"
llm_base_url: "https://open.bigmodel.cn/api/paas/v4"
llm_model: "glm-4-flash"
llm_reasoning_effort: low # 可选,推理强度 low(最快最省,默认)/medium/high
llm_timeout: 30 # 可选,单次请求超时秒数,默认 30
查找顺序(找到即用):
- 环境变量
XINOPS_CONFIG(或旧名PYPI_KU_CONFIG)指定的路径 - 业务脚本所在目录及其上层目录(最多向上 3 层)——推荐放在脚本目录的上层,多台机器迁移时随脚本目录一起拷贝
- 用户主目录
所有配置项也支持同名环境变量覆盖(如 WECOM_CORPSECRET、SMTP_PWD、HIVE_PWD)。
配置文件含敏感信息,切勿提交 git、切勿放进本仓库目录(仓库 .gitignore 已加双保险)。
模块结构
| 模块 | 功能 |
|---|---|
_core.py |
日志初始化、密钥配置加载、脚本生命周期、日期时间与通用工具 |
_wecom.py |
企业微信消息(内置 WeChatWorkSender)+ 企微智能表格文档 |
_email.py |
邮件发送(旧版 yagmail / 新版 smtplib 带重试) |
_hive.py |
Hive SQL 执行(内置 HiveClient + 重试机制)与错误提取 |
_llm.py |
大模型报错分析(OpenAI 兼容协议,不可用时自动回退规则翻译) |
_excel.py |
Excel 模板写入(xlwings/openpyxl 双方案)、截图、格式化导出 |
_export.py |
Parquet / CSV 导出 |
_misc.py |
Edge 驱动、数据量检查、拼音转换、车辆任务、内网文件服务器上传 |
_feishu.py |
飞书表格上传器 FeishuSheetUploader(原 feishu_to_excel) |
使用方式
方式一:run_job 一键运行(最推荐)
自动完成实例化、日志初始化、方法注入、异常捕获、错误落盘和企微通知。
业务函数内可直接裸调所有方法,无需 ux. 前缀。适合绝大多数业务脚本。
import sys
import xinops
from xinops import * # noqa — 让 IDE 识别 run_sql/uxin_wx 等方法,悬浮有提示
def job_executive():
# path / current_script_name 已自动注入
# 所有方法可直接调用,无需前缀
# run_sql 简写:SQL 文件在脚本同目录时,path 可省略
data = run_sql("xxx.sql")
# 也可以指定方法
data2 = run_sql("etl.sql", "dqls")
# 老写法(path + sql_name)完全不变
data3 = run_sql(path, "xxx.sql")
writer_excel_data_new(path=path, filename="muban.xlsx",
send_file=path + "result.xlsx",
sheet_data=[{"data": data, "sheet_name": "明细"}],
headers=True)
uxin_wx('dongyang', '处理完成')
if __name__ == "__main__":
sys.exit(xinops.run_job(job_executive))
run_job 参数说明
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
job_func |
callable |
None |
业务函数。支持无参(def job():)或带一个参数(def job(ux):)两种写法 |
notify_users |
tuple |
("dongyang",) |
失败时企微通知接收人列表 |
console_log |
bool |
True |
是否同时输出控制台日志(默认开启) |
notify_on_success |
bool |
False |
成功时是否也发企微通知(默认静默成功) |
💡
notify_users=("dongyang",)是默认值,写不写效果一样。 以下两种写法完全等价:sys.exit(xinops.run_job(job)) sys.exit(xinops.run_job(job_executive, notify_users=("dongyang",)))
通知机制
| 场景 | 行为 |
|---|---|
| ✅ 脚本正常运行完成 | 默认静默,不发消息(notify_on_success=True 时发"运行成功") |
| ❌ 脚本报错/抛异常 | 1. 完整堆栈写入 error.log 2. 提取核心错误(优先大模型智能分析,不可用时自动回退 Hive 错误中文翻译表) 3. 给 notify_users 中的每个人发企微通知 |
| ⚠️ 连实例化都失败(配置缺失等) | 兜底:从 xinops_secrets.yaml 直读企微密钥发送;读不到就只落盘 |
企微失败通知(Markdown 格式,v1.6.1 起):
⚠️ 脚本运行失败,请及时处理 ← 标题(加粗大字)
脚本名称:your_script.py ← 橙色高亮
失败时间:2026-10-16 14:30:00
核心错误:
▎ 🤖 大模型分析 ← 引用块样式
▎ 错误原因:SQL 引用了不存在的表 dwd.user_info
▎ 出错对象:表 dwd.user_info
▎ 修复建议:1. 检查表名拼写 ... 2. SHOW TABLES 确认 ...
大模型报错分析(v1.6.0 新增):配置
llm_*后,run_sql失败和run_job失败通知里的"核心错误"会优先由大模型结合出错 SQL 做归因,输出 "错误原因 / 出错对象 / 修复建议"。未配置密钥、额度不足、套餐到期、网络超时等 任何情况下自动回退原有的规则翻译,现有功能不受影响;也可用extract_core_hive_error(err, use_llm=False)强制走旧逻辑。大模型调用失败后有 5 分钟冷却期,避免连续报错时反复等待超时。
更多用法示例
# 多个人接收失败通知
sys.exit(xinops.run_job(job_executive, notify_users=("dongyang", "zhangsan")))
# 成功也发消息(适合重要任务)
sys.exit(xinops.run_job(job_executive, notify_on_success=True))
# 关闭控制台日志(默认开启,不需要时可关闭)
sys.exit(xinops.run_job(job_executive, console_log=False))
# 带 ux 参数的业务函数写法(可以直接用 ux.logger 打日志)
def job_executive(ux):
ux.logger.info("任务开始")
data = run_sql(path, "xxx.sql")
ux.logger.info(f"查到 {len(data)} 条数据")
方式二:类实例化(兼容 Functions_d 写法)
适合需要自己控制生命周期、或从旧版 Functions_d 迁移的场景。
import xinops
ux = xinops.DataProcessingAndMessaging(enable_console_log=True)
ux.Start_Get_filepath_and_filename()
data = ux.run_sql(ux.path, "xxx.sql")
ux.writer_excel_data_new(path=ux.path, filename="muban.xlsx",
send_file=ux.path + "result.xlsx",
sheet_data=[{"data": data, "sheet_name": "明细",
"clear_range": "A2:U1000000", "date_range": "A1"}],
headers=True)
ux.uxin_wx("dongyang", "处理完成")
ux.End_operation()
方式三:只跑一条 SQL / 只发条消息(轻量使用)
如果只需要用到一两个功能,不需要完整的生命周期管理:
import xinops
# 方式 3.1:只查 Hive
ux = xinops.DataProcessingAndMessaging()
ux.Start_Get_filepath_and_filename()
df = ux.run_sql(ux.path, "查询.sql")
print(df.shape)
ux.End_operation()
# 方式 3.2:只发企微消息
ux = xinops.DataProcessingAndMessaging(enable_console_log=True)
ux.uxin_wx("dongyang", "测试消息")
注意:即使只用一个功能,也需要
xinops_secrets.yaml配置文件(提供企微/Hive 的连接凭证)。
方式四:直接用底层类(特殊场景)
极少数情况下需要直接操作底层类,自己传参、自己管理连接:
# 直接发企微(不经过 Mixin)
from xinops._wecom import WeChatWorkSender
sender = WeChatWorkSender("你的CorpId", "你的Secret", "你的AgentId")
result = sender.send_text(["dongyang"], "你好")
print(result)
# 直接查 Hive(不经过 Mixin,不推荐,run_sql 更方便)
from xinops._hive import _HiveClient
client = _HiveClient("你的Hive地址", 10000, "你的用户名", "你的密码")
df = client.pd_dql("SELECT * FROM dwd.test LIMIT 10")
print(df)
client.close()
方式五:飞书表格上传
import xinops
uploader = xinops.FeishuSheetUploader(app_id="cli_xxx", app_secret="xxx")
token = uploader.create_spreadsheet(title="日报")
uploader.upload_to_sheet(token, "汇总", df, start_cell="A1")
使用场景速查
| 场景 | 推荐方式 |
|---|---|
| 日常业务脚本(有日志/有报错通知) | 方式一:run_job |
| 从 Functions_d 迁移过来的老脚本 | 方式二:类实例化 |
| 临时测试,只跑一条 SQL / 发条消息 | 方式三:轻量使用 |
| 需要底层控制(如自定义连接参数) | 方式四:底层类 |
| 飞书表格上传 | 方式五:FeishuSheetUploader |
IDE 悬浮提示
本库自带 __init__.pyi 类型存根文件,PyCharm / VS Code 等 IDE 可识别所有方法:
- 写
xinops.run_sql(...)时,鼠标悬停显示参数说明 - 在
run_job回调中,加from xinops import *后,裸写run_sql(...)也有悬浮提示
完整文档
详细 API 文档(每个方法的参数、返回值、示例)请查看:
docs/index.html
或手动重新生成:
python build_docs.py
文档生成工具会扫描所有模块的 docstring,自动生成 HTML 版完整说明书。
发版时 release.bat 会自动执行文档生成。
发版
release.bat 1.4.0
自动完成:清理 → 生成文档 → 打包 → 安全扫描 → 上传 PyPI。
Metadata
Release files for xinops 1.6.5
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| xinops-1.6.5.tar.gz | 136.2 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| xinops-1.6.5-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 274.7 kB
Release files / xinops-1.6.5.tar.gz
| Download URL | xinops-1.6.5.tar.gz |
|---|---|
| Size | 136.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
4e41e0a80a2e2b1534fb4b9a84d9ed0af817362be70d276eccd9f6d47d260a88
|
|
BLAKE2b-256 checksum How to use checksums |
244a3569bddb8445254720af3a079e6dfba0c5f52f62f08a1d603d433fa8459d
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.11.5
|
Release files / xinops-1.6.5-py3-none-any.whl
| Download URL | xinops-1.6.5-py3-none-any.whl |
|---|---|
| Size | 138.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
5849f6dc1eeeff590b9273d1e5db89294c8e32e6242f4702647adfd705542b37
|
|
BLAKE2b-256 checksum How to use checksums |
9ec5aba79199042d0870736110fa0d177c47275f805e8be56baf8615d7ad237f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.11.5
|
Security for xinops 1.6.5
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.