Metadata-Version: 2.4
Name: hf_auto_script
Version: 0.1.3
Summary: 期货公司结算部核心自动化框架：浏览器自动化、数据库操作、邮件、企业微信集成
Author-email: 海风 <hubert28@qq.com>
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: DrissionPage>=4.0.0
Requires-Dist: sqlalchemy>=2.0
Requires-Dist: psycopg2-binary>=2.9
Requires-Dist: polars>=0.20
Requires-Dist: pl_compare>=0.10.1
Requires-Dist: pydantic>=2.0
Requires-Dist: requests>=2.31
Requires-Dist: urllib3>=2.0
Requires-Dist: fastapi>=0.100
Requires-Dist: uvicorn>=0.23
Requires-Dist: fastapi_offline>=1.7
Requires-Dist: psutil>=5.9
Requires-Dist: pyautogui>=0.9
Requires-Dist: pywinauto>=0.6
Requires-Dist: rapidocr>=1.4
Requires-Dist: colorlog>=6.7
Requires-Dist: chardet>=5.0
Requires-Dist: python-dateutil>=2.8
Requires-Dist: cryptography>=41.0
Requires-Dist: beautifulsoup4>=4.12
Requires-Dist: typing_extensions>=4.0
Provides-Extra: ocr
Requires-Dist: rapidocr; extra == "ocr"
Provides-Extra: api
Requires-Dist: fastapi; extra == "api"
Requires-Dist: uvicorn; extra == "api"
Requires-Dist: fastapi_offline; extra == "api"
Provides-Extra: all
Requires-Dist: pymsteams; extra == "all"
Requires-Dist: pyotp; extra == "all"
Requires-Dist: openpyxl; extra == "all"
Requires-Dist: pandas; extra == "all"
Dynamic: license-file
Dynamic: requires-python

# 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()` | 获取步骤描述字典                      |

**交易日期处理**：

```python
# 自动从交易日.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 配置**：

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

**交易所远程路径配置**：

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

**使用示例**：

```python
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)        |

**使用示例**：

```python
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
```

**资源管理**：

```python
# 使用上下文管理器，自动清理
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` 解析）：

```toml
# [系统名.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               |

**菜单导航实现逻辑**：

```python
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 对象供后续操作
```

**文件下载流程**：

```python
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)
```

**页面元素定位语法**：

```python
# 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） |

**数据库初始化流程**：

```python
# 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()
```

**会话管理**：

```python
# 推荐使用上下文管理器，自动处理提交/回滚
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()
```

**批量操作**：

```python
# 批量保存（自动处理 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"))
```

**字段映射**：

```python
# 数据库字段 → 中文注释 的映射
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 编码                        |

**使用示例**：

```python
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)
```

**技术实现**：

```python
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 脚本文件（支持参数传递） |

**交易日获取实现**：

```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]
```

**时间等待实现**：

```python
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
```

**文件下载监控实现**：

```python
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 校验）  |

**使用示例**：

```python
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 消息格式**：

```python
# 自动根据关键词着色
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}
"""
```

**图片发送流程**：

```python
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))
```

**文件发送流程**：

```python
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 收信并过滤                            |

```python
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)    |

```python
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
```

**目录下载流程**：

```python
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)
```

**文件检查下载流程**：

```python
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 编码处理**：

```python
# 中文文件名需要 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()`                  | 获取列表数据                                |

**按钮点击实现**：

```python
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
```

**菜单导航实现**：

```python
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()
```

**列表数据读取**：

```python
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 日志流       |

**请求/响应模型**：

```python
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 服务器上的文件变化

---

## 十二、使用示例

### 创建新的业务控制器

```python
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()
```

### 发送企业微信通知

```python
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="结算组"
    )
```

### 数据库操作

```python
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 → 对象
```

### 文件下载监控

```python
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](#2-browser_configpy---凭据配置-configtoml)。该文件含敏感信息，已加入 `.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 / 内部源：

```bash
# 从源码安装(开发模式)
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
