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[mysql]"       # run_mysql（PyMySQL + SQLAlchemy，跨平台）
pip install "xinops[pbi]"         # refresh_powerbi（msal 刷新 Power BI 数据集）
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密码"

# MySQL 连接（run_mysql 使用；可选，不用 run_mysql 就不配）
mysql_host: "你的MySQL地址"
mysql_port: 3306
mysql_user: "你的MySQL用户名"
mysql_password: "你的MySQL密码"
mysql_database: "默认库名"
# mysql_charset: utf8mb4        # 可选，默认 utf8mb4
# mysql_sql_timeout: 1800       # 可选，单条 SQL 超时秒数，默认 1800

# Power BI 刷新（refresh_powerbi 使用；可选，不用就不配）
# 认证走 msal 设备码，首次登录一次后令牌缓存静默续期，代码中不写任何密钥
pbi_tenant_id: "你的Azure租户ID"
pbi_client_id: "你的应用客户端ID"
# pbi_token_cache: "D:/data/pbi_refresh_token_cache.json"   # 可选，令牌缓存路径，默认脚本同目录
# pbi_refresh_timeout: 1200        # 可选，等待刷新完成总超时秒数，默认 1200
# pbi_refresh_poll_interval: 10    # 可选，轮询间隔秒数，默认 10

# 大模型报错分析（可选，不配则自动使用内置规则翻译）
# 兼容所有 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； MySQL 对应 MYSQL_HOST / MYSQL_PORT / MYSQL_USER / MYSQL_PWD / MYSQL_DATABASE）。 配置文件含敏感信息，切勿提交 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")

使用场景速查

run_mysql：查 MySQL 数据到 DataFrame

run_mysql 与 run_sql 一比一对齐——同样的智能参数识别、SQL 清洗、日期变量替换 （$dt_ymd / $dt_ym）、同样的分号处理、失败自动重连重试 3 次、超时保护、 密码脱敏与大模型核心错误分析。仅后端换成 MySQL。使用前先装可选依赖： pip install "xinops[mysql]"，然后在 xinops_secrets.yaml 里配 mysql_*（见上方密钥章节）。

import sys, xinops
from xinops import *  # noqa

def job_executive():
    # 1) 直接传 SQL 内容（默认 dql）
    df = run_mysql(sql_content="SELECT * FROM c1_cost_record WHERE is_void=0 LIMIT 10")
    print(df.shape)

    # 2) 简写：SQL 文件放脚本同目录（path 自动取 self.path）
    df2 = run_mysql("查询.sql")

    # 3) 完整写法 + 指定方法（dml / dmls / dql / dqls 与 run_sql 完全一致）
    run_mysql(path, "etl.sql", "dmls")

    # 4) 临时放宽/收紧超时（单位：秒；底层为 PyMySQL 原生 read_timeout）
    df3 = run_mysql("大表.sql", sql_timeout=600)

if __name__ == "__main__":
    sys.exit(xinops.run_job(job_executive))

与 run_sql 的差异：Hive 用看门狗强拆 transport 兜底卡死，MySQL 走 PyMySQL 原生 connect_timeout / read_timeout / write_timeout，更稳定；心跳线程 仍保留（每 60 秒打印一条进度日志）。返回值同样是 DQL/DQLS→DataFrame、 DML/DMLS→None。

数据依赖检查：wait_for_table / check_table_ready

上游表延迟产出时，下游脚本常在数据还没到位就跑了，导致结果塌回历史归档或空表。 check_table_ready 做一次性判断、wait_for_table 在其上加"轮询等待 + 企微通知"：

• 分区表：传 days_ago（0=今天 / 1=昨天 / 2=前天…），用 SHOW PARTITIONS 走元数据秒级探测目标分区是否存在（分区一旦形成即视为有数据，与数仓约定一致， 不做 COUNT）。
• 普通表 / 非分区表：不传 days_ago，改为校验表的最后更新时间是否为今天 （复用 Get_update_time）。

import sys, xinops
from xinops import *  # noqa

def job_executive():
    # 等昨天的分区就绪：未就绪时每5分钟轮询、最长1小时，就绪后自动继续，
    # 期间通过企微推送"开始等待 / 已就绪 / 超时"通知
    if wait_for_table('dw.dw_marketing_experience_base_info_d', days_ago=1):
        run_sql(path, "operate.sql", "dmls")
    else:
        print("上游延迟超时，跳过 operate.sql")

if __name__ == "__main__":
    sys.exit(xinops.run_job(job_executive))

wait_for_table 默认 interval_min=5、max_wait_min=60、notify_users=("dongyang",)、 raise_on_timeout=False（超时返回 False 交由调用方决定跳过还是终止）。 只需一次性判断、不想等待时可直接用 check_table_ready(...)（返回 bool，绝不抛异常）。

Power BI 刷新：refresh_powerbi

任意脚本结尾按需刷新 Power BI 语义模型（数据集）。工作区名、数据集名作为参数传入， 租户 / 客户端 / 令牌缓存等配置放 xinops_secrets.yaml 的 pbi_*，代码里不写任何密钥。 认证走 msal 设备码——首次登录一次后令牌缓存静默续期。使用前装可选依赖 pip install "xinops[pbi]"。失败、超时一律抛异常（主脚本可观测），成功与失败都会推企微通知。

import sys, xinops
from xinops import *  # noqa

def job_executive():
    # ... 前面的取数/写表逻辑 ...
    # 结尾刷新指定数据集，等刷新完成，成功/失败都通知 dongyang
    refresh_powerbi("我的工作区", "销售车辆线索报表")

if __name__ == "__main__":
    sys.exit(xinops.run_job(job_executive))

wait_for_complete=False 可"触发即返回"不等待；timeout / poll_interval 不传时 取 pbi_refresh_timeout（默认1200）/ pbi_refresh_poll_interval（默认10）。 刷新失败抛 RuntimeError、超时抛 TimeoutError，run_job 会捕获并再发一条失败通知。

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。