开发指南¶

from easyths.core import BaseOperation
from easyths.models.operations import EmptyParams, OperationResult
from easyths.operations.results import FundsResult


class MyQueryOperation(BaseOperation[EmptyParams]):
    """我的查询操作"""

    operation_name = "my_query"          # 注册名（对外 API 路径）
    description = "查询我的数据"          # 文档描述
    Params = EmptyParams                 # 参数模型（校验+文档唯一来源）
    Result = FundsResult                 # 结果模型（构造+文档唯一来源）

    def execute(self, params: EmptyParams) -> OperationResult:
        ...

def execute(self, params: MyParams) -> OperationResult:
    start_time = time.time()
    # ... GUI 自动化逻辑 ...

    # 业务拒绝（弹窗提示、资金不足等）→ CLIENT_REJECTED
    if rejected:
        return self._fail(message, ErrorCode.CLIENT_REJECTED)

    # 成功：data 必须是 Result 模型的 model_dump()
    return self._ok(
        data=MyResult(field=value).model_dump(),
        message=f"查询完成，耗时{time.time() - start_time:.2f}秒",
    )

# easyths/operations/params.py 中定义参数模型
class MyQueryParams(StockQueryParams):
    """我的查询参数"""
    top_n: Annotated[int, Field(gt=0, le=100, description="返回条数")] = 10


# easyths/operations/results.py 中定义行模型
class MyRow(ResultModel):
    """我的查询行"""
    stock_code: Text = Field(description="证券代码")
    value: Num = Field(description="数值（无该项业务时为 null）")


# easyths/operations/my_query.py 中实现操作
class MyQueryOperation(BaseOperation[MyQueryParams]):
    operation_name = "my_query"
    description = "查询我的数据"
    Params = MyQueryParams
    Result = MyRow

    def parse_my_row(self, record: dict[str, Any]) -> MyRow:
        """剪贴板记录 → 行模型的显式装配（列名差异在此适配）"""
        return MyRow(stock_code=record["证券代码"], value=record["数值"])

    def execute(self, params: MyQueryParams) -> OperationResult:
        start_time = time.time()
        self.switch_left_menus("查询[F4]", "资金股票")
        # ... 定位表格控件、复制到剪贴板、text2df 解析 ...
        table_df = text2df(self.get_clipboard_data())
        rows = [self.parse_my_row(r).model_dump() for r in df_to_records(table_df)]
        return self._ok(
            data=rows,
            message=f"查询完成，耗时{time.time() - start_time:.2f}秒",
        )

# 获取指定条件的控件
control = self.get_control_with_children(
    parent_control,
    class_name="Edit",           # 控件类名
    control_type="Edit",         # 控件类型
    title="标题",                 # 控件标题（精确匹配）
    title_contains="委托",        # 控件标题（子串包含匹配）
    auto_id="1032"               # 自动化 ID（重要）
)

# 获取主窗口
main_window = self.get_main_window(wrapper_obj=True)

# 获取面板
main_panel = main_window.children(control_type="Pane")[0]

# 获取编辑框并输入内容
edit = self.get_control_with_children(main_panel, control_type="Edit", auto_id="1032")
edit.type_keys("内容")

# 获取按钮并点击
button = self.get_control_with_children(main_panel, control_type="Button", auto_id="2")
button.click()

# 切换左侧菜单
self.switch_left_menus("查询[F4]", "资金股票")

# 获取主窗口
main_window = self.get_main_window(wrapper_obj=True)

# 关闭弹窗
self.close_pop_dialog()

# 等待弹窗
self.wait_for_pop_dialog(timeout=1.0)

# 睡眠
self.sleep(0.1)

# 检查弹窗是否存在
if self.is_exist_pop_dialog():
    # 处理弹窗
    pass