xinops

营销数据运维一体化工具库，Windows / Mac 通用。

一个库搞定所有事：原 Functions_d、WeComMsg、HiveClient、feishu_to_excel 已全部合并进 xinops， 只需 pip install xinops 一个包，不用再维护多个独立库。

与旧库的关系

已合并的独立库

以下独立库已全部合并进 xinops，无需再单独安装、单独维护：

对现有脚本的影响

• 使用 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

查找顺序（找到即用）：

1. 环境变量 XINOPS_CONFIG（或旧名 PYPI_KU_CONFIG）指定的路径
2. 业务脚本所在目录及其上层目录（最多向上 3 层）——推荐放在脚本目录的上层，多台机器迁移时随脚本目录一起拷贝
3. 用户主目录

所有配置项也支持同名环境变量覆盖（如 WECOM_CORPSECRET、SMTP_PWD、HIVE_PWD）。 配置文件含敏感信息，切勿提交 git、切勿放进本仓库目录（仓库 .gitignore 已加双保险）。

模块结构

使用方式

方式一：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 参数说明

💡 notify_users=("dongyang",) 是默认值，写不写效果一样。 以下两种写法完全等价：

sys.exit(xinops.run_job(job))
sys.exit(xinops.run_job(job_executive, notify_users=("dongyang",)))

通知机制

企微失败通知（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")

使用场景速查

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。