auto_script - 核心自动化框架

概述

auto_script 是期货公司结算部综合业务系统的核心基础设施框架，提供了浏览器自动化、数据库操作、邮件发送、企业微信集成等通用功能，是所有业务模块的基础。

文件结构

auto_script/
├── __init__.py              # 模块初始化，导出logger
├── base_controller.py       # 基础控制器类
├── main_controller.py       # 主控制器类(浏览器管理 + new_page 凭据注入)
├── browser.py               # 浏览器自动化封装
├── browser_config.py        # config.toml 凭据加载(按系统名 + sim/real)
├── page.py                  # 页面操作基类
├── page_cffex.py            # 中金所页面封装
├── page_czce.py             # 郑商所页面封装
├── page_dce.py              # 大商所页面封装
├── page_esunny.py           # 易盛页面封装
├── page_gfex.py             # 广期所页面封装
├── page_shfe.py             # 上期所页面封装
├── page_ine.py              # 上海能源中心页面封装
├── page_xone.py             # XOne页面封装
├── page_crm.py              # CRM 页面封装
├── page_risk.py             # 风控 web 页面封装
├── db.py                    # 数据库操作基类
├── db_calendar.py           # 交易日历
├── db_department.py         # 部门组织
├── db_instrument.py         # 合约信息
├── db_investor_info.py      # 投资者信息
├── db_investor_info_month.py# 投资者月度信息
├── db_investor_contact.py   # 投资者联系方式
├── db_investor_fund.py      # 投资者资金
├── db_fund_account.py       # 资金账户
├── db_trading_code.py       # 交易编码
├── db_trading_statistics.py # 交易统计
├── db_position_summary.py   # 持仓汇总
├── db_quote_daily.py        # 日行情
├── db_stock_fund.py         # 个股资金
├── db_stock_trade.py        # 个股成交
├── db_notice.py             # 交易所公告
├── db_etf_fund.py           # ETF 资金
├── db_etf_investor_fund.py  # ETF 投资者资金
├── db_etf_position_limit.py # ETF 持仓限额
├── db_etf_trade.py          # ETF 成交
├── db_settle_config.py      # 结算配置
├── db_settlement_step.py    # 结算步骤
├── db_task_config.py        # 任务配置
├── db_ukey_config.py        # UKey 配置
├── cfmmc_class.py           # 监控中心数据类定义
├── cfmmc_parser.py          # 监控中心文件解析器
├── cfmmc_parser_etf.py      # ETF监控中心文件解析器
├── settle_txt_parser.py     # 结算单TXT解析器
├── log.py                   # 日志管理
├── log_error_monitor.py     # 日志错误监控
├── utils.py                 # 通用工具函数
├── excel.py                 # Excel操作工具
├── http_download.py         # HTTP下载工具
├── email_utils.py           # 邮件收发工具(SMTP/POP/IMAP)
├── wxwork.py                # 企业微信通知工具
├── monitor_wxwork.py        # 企微群聊监控(OCR)
├── sftp_monitor.py          # SFTP监控模块
├── tts_doubao.py            # 豆包(火山引擎)语音合成
├── api_controller.py        # FastAPI接口控制器
├── js_api.py                # JS API调用封装
├── js_api_v2.py             # JS API调用封装V2
├── js_api_6.7.12.json       # JS API 接口定义(随包数据)
├── rem.py                   # 盛立接口封装
├── project_executor.py      # 项目执行器
├── confirmation.py          # 确认操作模块
├── compare_with_pl.py       # 数据对比工具
├── README.md / README.pdf   # 本文档
└── js_api_v6.7.12.pdf       # JS API 接口文档

---

一、核心控制器模块 (Controller)

1. base_controller.py - 基础控制器

作用：通用主控制器基类，提供按步执行的框架支持

核心类：

类名	说明
StepDefinition	步骤定义类，包含步骤名称、处理函数、描述
BaseController	抽象基类，提供所有业务模块的基础功能

BaseController 核心功能：

功能	说明
项目初始化	项目名称、交易日、仿真/生产环境配置
日志目录	自动设置项目专用日志目录
步骤管理	step_definitions 字典管理所有步骤定义
命令行参数	支持通过命令行参数传入交易日
数据库配置	自动配置 PostgreSQL 连接（仿真/生产环境）
交易所配置	交易所会服文件路径、远程共享路径映射
邮件配置	SMTP/POP 邮件服务器配置

核心属性：

属性	类型	说明
tradingday	str	交易日（YYYYMMDD 格式）
cur_date	str	当前日期（YYYY-MM-DD 格式）
download_dir	Path	下载目录
project_dir	Path	项目根目录
db_url	str	PostgreSQL 连接 URL
cfmmc_dir	Path	监控中心文件目录
exchange_files_dir	Path	交易所会服文件目录
exchange_map	dict	交易所代码映射（中英文）

核心方法：

方法	说明
execute_step(step_num)	执行指定步骤，返回布尔值表示成功/失败
execute_steps(list)	批量执行多个步骤，遇失败则中断
execute_auto_steps()	自动执行所有步骤（排除指定步骤）
run_interactive_mode()	交互式执行模式，支持用户输入步骤编号
get_step_items()	获取所有步骤名称列表
get_step_descriptions()	获取步骤描述字典

交易日期处理：

# 自动从交易日.txt读取交易日列表
tradingday = utils.get_下一交易日(current_day, idx=1)  # 获取下一个交易日
# 命令行参数优先级最高
if len(sys.argv) > 1 and len(sys.argv[1]) == 8:
    tradingday = sys.argv[1]  # python script.py 20250607

数据库 URL 配置：

# 自动根据sim参数选择数据库 (主机/密码见 db.py 与本机 config, 此处脱敏)
db_url = "postgresql://postgres:****@127.0.0.1:5432/postgres"          # 仿真
db_url = "postgresql://postgres:****@<生产DB主机>:5432/postgres"       # 生产

交易所远程路径配置：

exc_remote_path = {
    "GFEX": rf"\\<文件服务器主机>\<共享目录>\广期所\{year}广州\{month}月\{day}gz",
    "DCE":  rf"\\<文件服务器主机>\<共享目录>\大商所\{year}大连\{month}月\{day}dl",
    "CFFEX":rf"\\<文件服务器主机>\<共享目录>\中金所\{year}中金\{month}月\{day}",
    # ...
}

使用示例：

from auto_script.base_controller import BaseController, StepDefinition

class MyController(BaseController):
    def _init_step_definitions(self):
        # 步骤0：自动执行（可被定时任务触发）
        self.step_definitions[0] = StepDefinition(
            name="自动执行",
            handler=self._auto_execute,
            description="15:00后自动执行所有步骤"
        )
        # 步骤1-9：手动执行
        self.step_definitions[1] = StepDefinition(
            name="登录",
            handler=self.fn_login,
            description="登录交易所系统"
        )
        self.step_definitions[2] = StepDefinition(
            name="下载数据",
            handler=self.fn_download,
            description="下载结算数据文件"
        )

    def _auto_execute(self) -> bool:
        # 等待15:00后再执行
        if not utils.wait_for_时分("1500", self.tradingday):
            return False
        return self.execute_auto_steps(exclude_steps=[0])  # 排除自己

    def fn_login(self) -> bool:
        # 实现登录逻辑
        return True

2. main_controller.py - 主控制器

作用：继承自 BaseController，封装浏览器管理功能

核心功能：

功能	说明
浏览器管理器	self.browsers 字典管理多个系统的浏览器实例
延迟初始化	浏览器实例在使用时才创建（按需加载）
资源清理	cleanup() 方法关闭所有浏览器，释放资源
上下文管理	支持 with 语句，自动管理资源

支持的系统（系统名即 config.toml 中的表名，共 21 个）：

系统标识	交易所/系统	说明
CTP	期货 CTP 主席	默认系统(含 sim/real)
CTP股票期权	CTP 个股期权	期权交易
CTP股票	CTP 个股现货	A 股交易
CTP信创	CTP 信创主席	实盘内网
CTPOP	OP 系统	期权交易
CRM	财务 CRM	FMIS(含 sim/real)
XOne	XOne 飞创灾备	灾备系统
Risk	风控 web	风控页面
光证OA	光证 OA	办公系统
ESunny	易盛主机	易盛自动结算
ESunnyMini	易盛 mini	内网地址
ESunnyDX	易盛东迁	内网地址
ESunnyLT	易盛楼潭	内网地址
ESunnyOP05	易盛 OP05	内网地址
ESunnyOP07	易盛 OP07	内网地址
CZCE	郑商所	郑州期货(UKey)
SHFE	上期所	上海期货(TOTP)
INE	能源中心	上海能源(TOTP)
DCE	大商所	大连期货(UKey)
CFFEX	中金所	中国金融期货(UKey)
GFEX	广期所	广州期货(UKey)

使用示例：

from auto_script.main_controller import MainController, StepDefinition

class MyExchangeController(MainController):
    def _init_step_definitions(self):
        self.step_definitions[1] = StepDefinition(
            name="下载结算数据",
            handler=self.fn_download_settlement
        )

    def fn_download_settlement(self) -> bool:
        # 创建新页面: url/账号/密码/UKey/动态口令全部从 config.toml 的 [CFFEX.real] 注入
        with self.new_page("CFFEX") as page:
            page.navigate_menu("结算报表", "打包下载")
            # 执行页面操作...
        return True

资源管理：

# 使用上下文管理器，自动清理
with MyController("项目名称") as controller:
    controller.execute_step(1)
# 退出时自动调用 cleanup() 关闭所有浏览器

---

二、浏览器自动化模块 (Browser/Page)

1. browser.py - 浏览器管理

作用：封装 DrissionPage 的 Chromium 浏览器，支持单例模式和系统隔离

核心特性：

特性	说明
单例模式	每个 sys_name 只创建一个浏览器实例
系统隔离	不同交易所使用独立的浏览器配置
仿真/生产切换	根据 sim 参数选择不同 URL
UKey 认证	支持传入 UKey 信息自动完成认证
下载管理	配置专用下载目录，自动等待文件完成

浏览器实例创建流程：

get_browser("CFFEX")
    ↓
检查 browsers 字典中是否已存在
    ↓
不存在 → 创建新 Browser(sim, download_dir, sys_name, headless)
    ↓
返回 Browser 实例

2. browser_config.py - 凭据配置 (config.toml)

作用：从 jsb 根目录的 config.toml 按 系统名 + 仿真/实盘 读取登录凭据，源码中不出现任何明文账号

配置文件结构（TOML，Python 3.11+ 用标准库 tomllib 解析）：

# [系统名.sim] 仿真段(可省略, 缺失字段自动回退 real 段)
# [系统名.real] 实盘段
[CTP.sim]
url = "https://<仿真CTP主机>:50003/..."
username = "<账号>"
password = "<密码>"
broker = "<经纪商代码>"

[CZCE.real]
url = "https://ms.czce.com.cn/index#/un-main"   # 交易所公网域名仅为示例
username = "<会服账号>"
password = "<密码>"
broker = ""
attach = "<UKey名称>|<UKey密码>"   # CZCE/CFFEX/GFEX/DCE
# SHFE/INE 的 attach 为动态口令种子(TOTP)

核心 API：

函数	说明
get_browser_config(sys_name, sim)	返回 {"url","username","password","broker","attach"}

• 配置文件定位：开发环境取 jsb 根目录，PyInstaller 打包后取 exe 同目录
• config.toml 含敏感信息，已在 .gitignore 中忽略，随机器分发
• 未知系统名抛 KeyError，缺少 url 抛 ValueError

3. page.py - 页面操作基类

作用：封装通用的浏览器页面操作方法，简化元素定位、点击、输入等操作

核心配置类：

类名	用途
SettlementSections	结算单组成配置（平仓明细、交割明细、持仓明细等）
PrintConfig	打印结算单配置（报告类型、日期格式、文件格式）
DailyReportConfig	日结算单导出配置（经纪人组、导出格式）

核心操作方法：

方法	说明
navigate_menu()	多级菜单导航（支持 4 级菜单）
button_click()	按钮点击（支持索引定位）
button_export()	导出按钮（自动等待文件下载完成）
button_upload()	上传按钮（支持批量上传）
input_text()	文本输入（支持清空后输入）
select_dropdown()	下拉框选择
wait_loading()	等待页面加载完成
get_table_data()	读取表格数据为 CSV

菜单导航实现逻辑：

def navigate_menu(menu_name, sub_menu_name, sub_sub_menu_name=None, grand_sun_menu_name=None):
    # 1. 点击主菜单
    main_menu.click()
    # 2. 判断子菜单是否需要展开
    if prev_class contains "collapse-button":
        sub_menu.prev().click()  # 展开子菜单
    # 3. 点击孙菜单（如有）
    # 4. 点击曾孙菜单（如有）
    # 5. 等待 iframe 加载完成
    # 6. 返回 iframe 对象供后续操作

文件下载流程：

def button_export(button_name, file_name):
    # 1. 检查文件是否存在，如存在则删除
    if os.path.exists(file_path):
        os.remove(file_path)
    # 2. 设置下载目录
    os.makedirs(self._page.download_path, exist_ok=True)
    # 3. 等待页面数据加载（避免666）
    self._page.wait(3)
    # 4. 点击按钮并触发下载
    btn.click.to_download(download_path, file_name)
    # 5. 等待加载动画消失
    self.wait_loading()
    # 6. 等待文件下载完成（使用 watchdog 监控）
    return utils.wait_for_download_file(file_path, timeout=30)

页面元素定位语法：

# DrissionPage 特有的定位语法
"t=div@class=name"           # 标签+属性定位
"t=span@text()=登录"         # 文本内容定位（精确匹配）
"t=span@text():登录"         # 文本内容定位（模糊匹配）
"t=input@value=查询"         # 输入框值定位
"t=a@href$=.pdf"            # href正则匹配

4. page_*.py（各系统页面封装）

文件	系统	特有功能
page_cffex.py	中金所	现货收盘价查询、出入金明细
page_czce.py	郑商所	滚动交割响应、监管数据
page_dce.py	大商所	大商所特有界面操作
page_esunny.py	易盛	易盛系统特有界面操作
page_gfex.py	广期所	划款明细、仓储费查询
page_shfe.py	上期所	上期所特有界面操作
page_ine.py	能源中心	能源中心特有界面操作
page_xone.py	XOne	XOne 灾备系统操作
page_crm.py	CRM	FMIS 财务系统页面
page_risk.py	风控 web	风控系统页面

---

三、数据库操作模块 (DB)

db.py - 数据库操作基类

作用：基于 SQLAlchemy ORM 提供通用的数据库操作封装

核心类继承关系：

Base (declarative_base)
    └── JsbTable (抽象基类，所有业务表的父类)
            ├── Calendar (交易日历)
            ├── InvestorFund (投资者资金)
            ├── SettlementStep (结算步骤)
            └── ... (其他业务表)

核心方法分类：

方法类别	方法名	说明
连接管理	init_db()	初始化数据库连接（支持 schema 配置）
	get_session()	获取数据库会话（上下文管理器）
	get_qry_session()	获取只读会话（自动提交）
表操作	create_table()	创建数据表
	drop_table()	删除数据表
	truncate_table()	清空数据表
查询	query()	条件查询
	query_first()	主键查询
	get_all()	获取所有记录
	count()	统计记录数
	exists()	检查记录是否存在
增删改	save()	保存对象（insert+update，merge）
	add()	添加对象（insert）
	bulk_save()	批量保存
	delete()	删除记录
转换	to_dict()	对象转字典
	to_json()	对象转 JSON
	from_dict()	字典创建对象
	list_to_json()	对象列表转 JSON
导入导出	import_csv()	CSV 文件导入数据库（高性能 COPY FROM）

数据库初始化流程：

# 1. 创建数据库引擎（指定schema）
JsbTable.init_db()

# 2. 创建表
class InvestorFund(JsbTable):
    __tablename__ = 'tbl_investor_fund'
    __schema__ = 'base_data'  # 可选，默认 base_data

    investor_id = Column(String, primary_key=True, comment="投资者代码")
    fund = Column(String, comment="资金")

InvestorFund.create_table()

会话管理：

# 推荐使用上下文管理器，自动处理提交/回滚
with InvestorFund.get_session() as session:
    # 查询
    results = session.query(InvestorFund).filter_by(investor_id="001").all()
    # 保存
    obj = InvestorFund(investor_id="002", fund=10000)
    session.merge(obj)
# 自动 commit，异常自动 rollback

# 只读会话（用于查询，自动提交）
with InvestorFund.get_qry_session() as session:
    results = session.query(InvestorFund).all()

批量操作：

# 批量保存（自动处理 update_time）
obj_list = [InvestorFund(investor_id=f"00{i}", fund=1000) for i in range(10)]
InvestorFund.bulk_save(obj_list)

# CSV导入（高性能，使用 COPY FROM）
# 自动处理编码转换、字段映射、空值处理
InvestorFund.import_csv(Path("data.csv"))

字段映射：

# 数据库字段 → 中文注释 的映射
class InvestorFund(JsbTable):
    investor_id = Column(String, comment="投资者代码")
    fund = Column(String, comment="资金")

# 获取映射
field_map = InvestorFund.get_field_mapping()
# {'投资者代码': 'investor_id', '资金': 'fund'}

# CSV导入时自动匹配
columns_csv = ['投资者代码', '资金']  # CSV表头
columns_db = [field_map[h] for h in columns_csv]  # 转换为字段名

数据模型文件 (db_*.py)

文件	ORM 类	表名	说明
db_calendar.py	Calendar	calendar	交易日历
db_department.py	Department	department	部门组织
db_instrument.py	Instrument	instrument	合约信息
db_investor_info.py	InvestorInfo	investor_info	投资者信息
db_investor_info_month.py	InvestorInfoMonth	investor_info_month	投资者月度信息快照
db_investor_contact.py	InvestorContact	investor_contact	投资者联系方式
db_investor_fund.py	InvestorFund	investor_fund	投资者资金
db_fund_account.py	FundAccount	fund_account	资金账户
db_trading_code.py	TradingCode	trading_code	交易编码
db_trading_statistics.py	TradingStatistics	trading_statistics	交易统计
db_position_summary.py	PositionSummary	position_summary	持仓汇总
db_quote_daily.py	QuoteDaily	quote_daily	每日行情
db_stock_fund.py	StockFund	stock_fund	个股资金
db_stock_trade.py	StockTrade	stock_trade	个股成交
db_notice.py	Notice	exchange_notice	交易所公告
db_etf_fund.py	ETFFund	etf_fund	ETF 资金
db_etf_investor_fund.py	ETFInvestorFund	etf_investor_fund	ETF 投资者资金
db_etf_position_limit.py	ETFPositionLimit	etf_position_limit	ETF 持仓限额
db_etf_trade.py	ETFTrade	etf_trade	ETF 成交
db_settle_config.py	SettleConfig	settle_config	结算配置
db_settlement_step.py	SettlementStep	settlement_step	结算步骤
db_task_config.py	TaskConfig	task_config	任务配置
db_ukey_config.py	UKeyConfig	ukey_config	UKey 配置

---

四、Excel 操作模块 (Excel)

excel.py - Excel 处理工具

作用：封装 Excel 操作，支持公式计算和数据读取

核心功能：

功能	说明
公式计算	使用 xlwings 调用 Excel 应用程序计算公式
文件读取	使用 openpyxl 读取数据（read_only, data_only）
编码处理	自动处理 GBK/UTF-8 编码

使用示例：

from auto_script.excel import read_excel_with_calc

# 读取Excel文件并触发公式计算
# 注意：会启动隐藏的 Excel 应用程序
wb = read_excel_with_calc("report.xlsx")

# 读取数据
for sheet in wb.sheetnames:
    ws = wb[sheet]
    for row in ws.iter_rows(values_only=True):
        print(row)

技术实现：

def read_excel_with_calc(file_path):
    # 1. 启动后台 Excel 应用
    app = xw.App(visible=False)
    try:
        # 2. 打开文件并保存（触发公式重算）
        wb = app.books.open(file_path)
        wb.save()
        wb.close()
    finally:
        app.quit()
    # 3. 使用 openpyxl 读取计算后的数据
    return load_workbook(file_path, read_only=True, data_only=True)

---

五、工具函数模块 (Utils)

utils.py - 通用工具函数

作用：提供日常开发中常用的工具函数

主要功能分类：

1. 路径和目录操作

函数	说明
get_base_dir()	获取脚本基础目录（支持打包后的 exe）
extract_resources_file_to_app()	解压资源文件到应用程序目录
extract_all_resources_to_app_dir()	递归解压所有资源文件

2. 交易日处理

函数	说明
get_下一交易日()	根据交易日.txt 获取下一交易日（二分查找）
get_last_month_range()	获取上个月的范围（开始/结束日期）
wait_for_时分()	等待指定时间点（HHMM 格式）
wait_for_下1日()	等待下一交易日 01:00
wait_for_会服()	等待交易所会服文件全部下载完成
wait_for_特法可抽()	等待特法可抽状态

3. 文件下载监控

函数	说明
wait_for_download_file()	使用 watchdog 监控文件下载完成
convert_to_xlsx()	将 txt/csv 转换为 xlsx

4. 文件编码处理

函数	说明
get_encoding()	自动检测文件编码
convert_file_to_utf8()	批量转换文件编码为 UTF-8

5. 压缩解压

函数	说明
zip_files()	将文件打包为 ZIP
zip_dir()	将目录压缩为 ZIP（支持密码）

6. SFTP 操作

函数	说明
upload_to_sftp()	上传文件到 SFTP 服务器（支持密钥）

7. 脚本执行

函数	说明
run_script()	执行 Python 脚本文件（支持参数传递）

交易日获取实现：

def get_下一交易日(tradingday: str, idx: int = 1) -> str:
    # 1. 读取交易日.txt文件
    with open("交易日.txt", encoding="utf-8") as f:
        tradingdays = [d.strip() for d in f.readlines()]
    # 2. 二分查找定位（高效）
    i = bisect.bisect_left(tradingdays, tradingday)
    # 3. 返回指定偏移的交易日
    return tradingdays[i + idx]

时间等待实现：

def wait_for_时分(target_hhmm: str, tradingday: str) -> bool:
    # 组合目标时间
    target = tradingday + target_hhmm  # "202506071500"
    # 轮询检查
    while time.strftime("%Y%m%d%H%M") < target:
        time.sleep(1)
    return True

文件下载监控实现：

def wait_for_download_file(file_path, timeout=300):
    # 1. 创建事件处理器
    handler = DownloadCompleteHandler(file_path)
    # 2. 启动 watchdog 观察者
    observer = Observer()
    observer.schedule(handler, os.path.dirname(file_path))
    observer.start()
    try:
        # 3. 等待文件完成或超时
        while not os.path.exists(file_path) and not handler.file_completed:
            time.sleep(0.5)
        return os.path.exists(file_path) or handler.file_completed
    finally:
        observer.stop()
        observer.join()

---

六、消息与通知模块 (WxWork / Email / TTS)

wxwork.py - 企业微信机器人通知

作用：封装企业微信机器人 API，支持发送文本、图片、文件、Markdown 消息

支持的机器人：

机器人名称	Webhook Key	用途
test	****（已脱敏，见本机配置）	测试用
结算	****	结算组通知
风控	****	结算风控部通知
数据	****	数据汇报
资金	****	资金组通知
酱油群	****	日常交流
客服	****	证件补正

消息类型：

函数	消息类型	说明
wxwork_send_msg()	text	纯文本消息
wxwork_send_msg_markdown()	markdown	Markdown 格式（支持颜色）
wxwork_send_msg_markdown_v2()	markdown_v2	V2 版本 Markdown
wxwork_send_file()	file	文件消息（需先上传获取 media_id）
wxwork_send_image()	image	图片消息（Base64 编码+MD5 校验）

使用示例：

from auto_script.wxwork import wxwork_send_msg, wxwork_send_msg_markdown, wxwork_send_file

# 发送文本消息
wxwork_send_msg("结算任务已完成", "结算组")

# 发送Markdown消息（带颜色）
wxwork_send_msg_markdown(
    item="007_广期所",
    result="成功",  # 成功/完成/结束 → 绿色
    bot_name="结算组"
)

# 发送文件
wxwork_send_file("report.xlsx", "数据汇报")

# 发送图片
wxwork_send_image("screenshot.jpg", "结算组")

Markdown 消息格式：

# 自动根据关键词着色
if "成功" in result or "完成" in result or "结束" in result:
    colored_result = '<font color="info">成功</font>'  # 绿色
else:
    colored_result = '<font color="warning">处理中</font>'  # 橙色

# 消息内容
markdown_content = f"""
**报告时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}
**项目**: {item}
**结果**: {colored_result}
"""

图片发送流程：

def wxwork_send_image(image_path, bot_name):
    # 1. 读取图片并 Base64 编码
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
    # 2. 计算 MD5 哈希
    md5_hash = hashlib.md5(f.read()).hexdigest()
    # 3. 发送消息
    data = {
        "msgtype": "image",
        "image": {"base64": image_data, "md5": md5_hash}
    }
    response = requests.post(webhook_url, data=json.dumps(data))

文件发送流程：

def wxwork_send_file(file_path, bot_name):
    # 1. 上传文件获取 media_id
    upload_url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key={key}&type=file"
    with open(file_path, "rb") as f:
        files = {"file": (os.path.basename(file_path), f)}
        response = requests.post(upload_url, files=files)
    media_id = json.loads(response.text)["media_id"]

    # 2. 发送文件消息
    data = {"msgtype": "file", "file": {"media_id": media_id}}
    response = requests.post(webhook_url, data=json.dumps(data))

email_utils.py - 邮件收发

作用：SMTP 发送（带附件）、POP3/IMAP 收取并按条件过滤邮件，自动检测编码

核心类/函数：

名称	说明
EmailData	邮件数据类（主题、正文、附件、发件人等）
SendMailConfig	发送配置类（服务器、端口、账号、收件规则）
send_config_parser_file()	解析邮件发送配置文件
send_mail_with_attachment()	SMTP 发送带附件邮件
send_email_with_check()	发送并校验结果
receive_mail_with_filter_by_pop	POP3 收信并过滤
receive_mail_with_filter_by_imap	IMAP 收信并过滤

from auto_script.email_utils import send_mail_with_attachment, receive_mail_with_filter_by_imap

# 发送带附件的邮件
send_mail_with_attachment(
    subject="结算完成通知",
    body="详见附件",
    attachments=[r"./report.xlsx"],
    receivers=["<收件人邮箱>"],
)

# IMAP 收取并过滤(未读 + 主题关键字 + 附件)
mails = receive_mail_with_filter_by_imap(subject_keyword="结算单", unread_only=True)

tts_doubao.py - 豆包语音合成

作用：豆包（火山引擎）单向流式 TTS HTTP 客户端，文本合成为 mp3/pcm/ogg/wav 音频

类/方法	说明
DoubaoTTS(app_key)	凭火山引擎应用 ID 初始化（建议从环境变量 VOLCENGINE_APPID 读取）
synthesize(text, ...)	文本→音频字节，支持 speaker 音色、speed_ratio 语速(0.2~3.0)

from auto_script.tts_doubao import DoubaoTTS

tts = DoubaoTTS(app_key)
audio = tts.synthesize("结算任务已完成", speaker="zh_female_vv_uranus_bigtts")
if audio:
    Path("notice.mp3").write_bytes(audio)

---

七、HTTP 下载模块 (HTTP Download)

http_download.py - HTTP 文件下载

作用：封装 HTTP 下载功能，支持目录文件列表下载和单文件检查下载

核心功能：

函数	说明
check_and_download_dir()	检查并下载目录下的所有文件
check_and_download_files()	检查并下载指定文件列表

文件服务器 URL 结构：

http://<内部文件服务器>/{dir_name}/
    ├── file1.xlsx
    ├── file2.pdf
    └── subdir/
        └── file3.csv

目录下载流程：

def check_and_download_dir(dir_name, pre_url, download_dir):
    # 1. 获取目录列表页面
    response = requests.get(f"{pre_url}/{dir_name}/")
    # 2. 解析HTML获取所有文件链接
    soup = BeautifulSoup(response.text, 'html.parser')
    links = soup.find_all("a")
    # 3. 过滤出文件链接（排除 . 和 ..）
    file_urls = [link.get("href") for link in links
                 if not link.startswith("?") and not link.endswith("/")]
    # 4. 下载每个文件
    for file_url in file_urls:
        with requests.get(file_url, stream=True) as r:
            with open(os.path.join(download_dir, filename), "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)

文件检查下载流程：

def check_and_download_files(path, files, download_dir):
    for file in files:
        # 1. 检查本地是否已存在
        if os.path.exists(os.path.join(download_dir, file)):
            continue  # 跳过已存在的文件
        # 2. 发送 HEAD 请求检查远程文件是否存在
        url = f"{pre_url}/{path}/{file}"
        response = requests.head(url)
        if response.status_code == 200:
            # 3. 下载文件
            with requests.get(url, stream=True) as r:
                with open(os.path.join(download_dir, file), "wb") as f:
                    for chunk in r.iter_content(chunk_size=8192):
                        f.write(chunk)

URL 编码处理：

# 中文文件名需要 URL 编码
from urllib.parse import quote, unquote

file_name = "结算单_20250607.xlsx"
encoded_name = quote(file_name)  # "%E7%BB%93%E7%AE%97%E5%8D%95_20250607.xlsx"
decoded_name = unquote(encoded_name)  # "结算单_20250607.xlsx"

# 请求时使用编码后的URL
url = f"{pre_url}/{quote(path)}/{quote(file)}"

---

八、盛立接口模块 (REM)

rem.py - 盛立出入金接口封装

作用：封装盛立系统的 UI 自动化操作，支持出入金相关功能

技术实现：使用 uiautomation 库进行 Windows UI 自动化

核心功能：

函数	说明
btn_click()	点击按钮（支持 name 或 automation_id 定位）
tab_active()	激活标签页
menu_select()	选择菜单项
active_多系统统一导入导出()	打开导入导出窗口
save_as()	另存为对话框操作
open_file()	打开文件对话框
lst_rows()	获取列表数据

按钮点击实现：

def btn_click(hdl, name=None, automation_id=None):
    # 1. 根据name或automation_id定位按钮
    if name:
        btn = hdl.ButtonControl(name=name)
    else:
        btn = hdl.ButtonControl(AutomationId=automation_id)

    # 2. 等待按钮可用
    start_time = time.time()
    while not btn.IsEnabled or not btn.IsKeyboardFocusable:
        if time.time() - start_time > timeout:
            return False
        time.sleep(1)

    # 3. 获取焦点并点击
    btn.SetFocus()
    if p := btn.GetInvokePattern():
        return p.DoDefaultAction()
    if p := btn.GetLegacyIAccessiblePattern():
        return p.DoDefaultAction()
    btn.Click()
    return True

菜单导航实现：

def menu_select(hdl, name, item_name=None):
    # 1. 获取菜单栏
    menu_bar = hdl.MenuBarControl(AutomationId="MenuBar")
    if not menu_bar.Exists():
        menu_bar = hdl.MenuBarControl(Name="应用程序")

    # 2. 点击主菜单
    menu = menu_bar.MenuItemControl(Name=name)
    menu.GetInvokePattern().Invoke()

    # 3. 点击子菜单（如有）
    if item_name:
        sub_menu = menu.MenuItemControl(Name=item_name)
        sub_menu.GetInvokePattern().Invoke()

列表数据读取：

def lst_rows(hdl):
    # 1. 获取列表控件
    lst = hdl.ListControl(AutomationId="1050")

    # 2. 遍历所有行
    rows = []
    for child in lst.GetChildren():
        if child.ControlTypeName == "ListItemControl":
            # 3. 提取每列数据
            row = [c.Name for c in child.GetChildren()]
            rows.append(row)
    return rows

---

九、数据解析模块 (Parser)

cfmmc_parser.py - 监控中心文件解析

作用：解析 CFMMC（期货监控中心）格式的数据文件

支持的格式：

• DBF 格式
• TXT 格式
• CSV 格式

技术实现：使用 Polars 进行高性能数据处理

解析流程：

读取文件 → 检测编码 → 检测格式 → 解析数据 → 数据清洗 → 返回DataFrame

settle_txt_parser.py - 结算单 TXT 解析

作用：解析 CTP 系统生成的结算单 TXT 文件

支持格式：

• 中文结算单
• 英文结算单

解析内容：

内容类型	说明
客户信息	投资者代码、名称、账户类型
资金状况	上日结存、当日结存、浮动盈亏
持仓汇总	各合约持仓汇总
成交记录	当日成交明细
平仓明细	平仓盈亏明细
持仓明细	持仓明细（持仓明细/持仓汇总）

---

十、API 模块 (API)

api_controller.py - FastAPI 接口控制器

作用：提供 HTTP API 接口，支持远程调用和 SSE 日志推送

API 端点：

端点	方法	说明
/api/steps	GET	获取所有步骤定义
/api/execute	POST	执行指定步骤
/api/logs	GET	SSE 日志流

请求/响应模型：

class ExecuteRequest(BaseModel):
    step: int = Field(..., description="步骤编号")

class ExecuteResponse(BaseModel):
    success: bool
    message: str
    tradingday: str

class LogEvent(BaseModel):
    level: str  # INFO/WARNING/ERROR
    message: str
    timestamp: str

js_api.py / js_api_v2.py - JS API 调用封装

作用：封装与前端系统的 API 交互

功能：

• 发送 HTTP 请求到 JS API 服务
• 支持模拟/生产环境切换
• 统一响应格式处理

---

十一、监控模块 (Monitor)

log_error_monitor.py - 日志错误监控

作用：监控日志中的错误信息，触发告警

monitor_wxwork.py - 企微监控

作用：监控企业微信群聊内容中的关键字

sftp_monitor.py - SFTP 监控

作用：监控 SFTP 服务器上的文件变化

---

十二、使用示例

创建新的业务控制器

from auto_script.main_controller import MainController, StepDefinition
from auto_script import logger

class SettlementController(MainController):
    """结算业务控制器"""

    def _init_step_definitions(self):
        # 自动执行步骤（定时任务触发）
        self.step_definitions[0] = StepDefinition(
            name="自动执行",
            handler=self._auto_execute,
            description="15:00后自动执行"
        )
        # 手动执行步骤
        self.step_definitions[1] = StepDefinition(
            name="登录交易所",
            handler=self.fn_login
        )
        self.step_definitions[2] = StepDefinition(
            name="下载结算数据",
            handler=self.fn_download
        )
        self.step_definitions[3] = StepDefinition(
            name="数据入库",
            handler=self.fn_save_to_db
        )

    def _auto_execute(self) -> bool:
        # 等待15:00
        if not utils.wait_for_时分("1500", self.tradingday):
            return False
        # 执行所有步骤（排除步骤0）
        return self.execute_auto_steps(exclude_steps=[0])

    def fn_login(self) -> bool:
        logger.info("登录交易所系统...")
        # 凭据(url/账号/密码/UKey)全部从 config.toml 的 [CFFEX.real] 注入
        self.page = self.new_page("CFFEX")
        return True

    def fn_download(self) -> bool:
        logger.info("下载结算数据...")
        # 导航到下载页面
        self.page.navigate_menu("结算报表", "打包下载")
        # 点击下载按钮
        return self.page.button_export("下载", file_name=f"settlement_{self.tradingday}.zip")

    def fn_save_to_db(self) -> bool:
        logger.info("保存到数据库...")
        from auto_script.db_investor_fund import InvestorFund
        InvestorFund.init_db()
        # 读取文件并保存...
        return True

if __name__ == "__main__":
    with SettlementController(project_name="结算项目", sim=False) as controller:
        controller.run_interactive_mode()

发送企业微信通知

from auto_script.wxwork import wxwork_send_msg_markdown, wxwork_send_file

# 发送进度通知
wxwork_send_msg_markdown(
    item="007_广期所",
    result="下载完成，正在处理...",
    bot_name="结算组"
)

# 发送结果通知
if success:
    wxwork_send_msg_markdown(
        item="007_广期所",
        result="✅ 结算完成",
        bot_name="结算组"
    )
    wxwork_send_file("settlement_report.xlsx", "数据汇报")
else:
    wxwork_send_msg_markdown(
        item="007_广期所",
        result="❌ 结算失败，请检查",
        bot_name="结算组"
    )

数据库操作

from auto_script.db_investor_fund import InvestorFund

# 初始化
InvestorFund.init_db()

# 查询
funds = InvestorFund.query(InvestorFund.investor_id.like("00%"))

# 保存
obj = InvestorFund(investor_id="001", fund=10000)
obj.save()

# 批量保存
obj_list = [InvestorFund(investor_id=f"{i:03d}", fund=1000) for i in range(100)]
InvestorFund.bulk_save(obj_list)

# CSV导入
InvestorFund.import_csv(Path("fund_data.csv"))

# 对象转换
json_str = obj.to_json()  # 对象 → JSON
obj2 = InvestorFund.from_json(json_str)  # JSON → 对象

文件下载监控

from auto_script import utils

# 等待文件下载完成
file_path = r"d:\downloads\report.xlsx"
if utils.wait_for_download_file(file_path, timeout=300):
    logger.info("文件下载完成")
else:
    logger.error("文件下载超时")

---

配置说明

凭据配置 config.toml

所有系统的登录凭据统一在 jsb 根目录的 config.toml（PyInstaller 打包后放 exe 同目录），结构见 二.2 browser_config.py。该文件含敏感信息，已加入 .gitignore，随机器分发，不入版本库。

运行参数

参数来源	说明
构造参数 sim	True 仿真 / False 实盘（默认），决定取 config.toml 的 sim/real 段
构造参数 headless	是否无头浏览器模式，默认 False
命令行参数 sys.argv[1]	8 位交易日，如 python script.py 20260930，优先级最高
环境变量 VOLCENGINE_APPID	豆包 TTS 应用 ID（仅 tts_doubao 使用）

其他配置文件

文件	说明
交易日.txt	包含所有交易日列表，用于日期计算
各业务模块配置文件	Excel 格式的业务参数配置

数据库配置

JsbTable.init_db() 默认按 sim 参数自动选择连接（密码见 db.py）：

环境	主机（已脱敏）
实盘	postgresql://postgres:****@<生产DB主机>:5432/postgres
仿真	postgresql://postgres:****@127.0.0.1:5432/postgres

也可显式传入：JsbTable.init_db(db_url="postgresql://...")。

---

技术栈

分类	技术	用途
语言	Python 3.11+	开发语言
浏览器自动化	DrissionPage	网页操作
UI 自动化	uiautomation	Windows 桌面应用
ORM	SQLAlchemy / psycopg2	数据库操作
数据处理	Polars / pl_compare	大数据处理/比对
Excel 操作	openpyxl / xlwings	Excel 读写
HTTP 客户端	requests / urllib3	网络请求
API 框架	FastAPI / uvicorn / fastapi_offline	接口服务
OCR	RapidOCR / pyautogui / pywinauto	企微监控
语音合成	豆包 TTS (火山引擎)	通知播报
动态口令	pyotp	SHFE/INE TOTP
配置解析	tomllib (标准库)	config.toml 凭据
日志	colorlog	彩色日志
SFTP	paramiko	SSH 文件传输

---

安装与发布

本目录已按标准 Python 包组织（根目录提供 setup.py 与 pyproject.toml），可发布到 PyPI / 内部源：

# 从源码安装(开发模式)
pip install -e .

# 构建 sdist + wheel
pip install build twine
python -m build

# 上传到内部 PyPI 源
twine upload --repository-url http://<内部源>/ dist/

包名 auto_script，classifier 含 Private :: Do Not Upload，误传公共 PyPI 会被拒绝。

---

维护信息

• 作者：海风 (结算部)
• 更新日期：2026-10-02