如何在LibreOffice Python插件中异步获取股票API数据？

要解决API请求阻塞的问题，核心思路是先返回临时"loading..."值，后台线程执行API请求，完成后通过主线程安全的方式更新单元格。由于LibreOffice的UBA（Universal Broker Architecture）对象并非线程安全，不能直接在后台线程修改单元格，需借助com.sun.star.frame.DispatchHelper在主线程执行更新操作。

步骤1：调整IDL定义（支持返回字符串临时值） #

原stizardPrice返回double，无法返回"loading..."字符串，需修改返回类型为any，同时新增异步版本函数：

#include <com/sun/star/uno/XInterface.idl>
#include <com/sun/star/uno/Any.idl>

module com { module stizard { module api {

    interface XStizardApi {
        any stizardPriceAsync( [in] string symbol );
        any stizardAsync( [in] string symbol, [in] string q);
    };

}; }; };

步骤2：修改Python实现（异步逻辑+线程安全更新） #

核心实现要点：

• 函数立即返回"loading..."，同时启动后台线程执行API请求
• 线程完成请求后，通过DispatchHelper在主线程更新单元格
• 用全局字典记录待处理任务，避免单元格重算时重复发起请求

import json
import urllib.request
import threading
from typing import Dict, Tuple

import uno
import unohelper
from com.stizard.api import XStizardApi
from com.sun.star.frame import XDispatchHelper
from com.sun.star.util import URL
from com.sun.star.sheet import XCellRangeAddressable

# 全局字典：存储单元格地址 -> (symbol, query)，防止重复请求
pending_tasks: Dict[str, Tuple[str, str]] = {}
task_lock = threading.Lock()

class StizardImpl(unohelper.Base, XStizardApi):
    def __init__(self, ctx):
        self.ctx = ctx
        # 初始化DispatchHelper，用于主线程执行UI操作
        self.dispatch_helper = self.ctx.ServiceManager.createInstanceWithContext(
            "com.sun.star.frame.DispatchHelper", self.ctx
        )

    def _get_current_cell_addr(self) -> str:
        """获取当前调用函数的单元格地址，格式：SheetName.A1"""
        desktop = self.ctx.ServiceManager.createInstanceWithContext(
            "com.sun.star.frame.Desktop", self.ctx
        )
        active_sheet = desktop.getCurrentComponent().CurrentController.ActiveSheet
        range_addr = XCellRangeAddressable(active_sheet).getRangeAddress()
        # 取单元格起始地址（假设为单个单元格调用）
        col_label = chr(ord('A') + range_addr.StartColumn)
        row_num = range_addr.StartRow + 1
        return f"{active_sheet.Name}.{col_label}{row_num}"

    def _update_cell_value(self, cell_addr: str, value):
        """通过DispatchHelper在主线程更新单元格，保证线程安全"""
        try:
            desktop = self.ctx.ServiceManager.createInstanceWithContext(
                "com.sun.star.frame.Desktop", self.ctx
            )
            model = desktop.getCurrentComponent()
            controller = model.getCurrentController()

            # 1. 定位到目标单元格
            goto_url = URL()
            goto_url.Complete = f".uno:GoToCell?Cell={cell_addr}"
            goto_url = controller.parseURL(goto_url.Complete)
            self.dispatch_helper.executeDispatch(controller, goto_url, "", 0, ())

            # 2. 设置单元格值
            set_url = URL()
            set_args = uno.createUnoStruct("com.sun.star.beans.PropertyValue")
            if isinstance(value, str):
                set_url.Complete = ".uno:SetString"
                set_args.Name = "StringValue"
            else:
                set_url.Complete = ".uno:SetValue"
                set_args.Name = "Value"
            set_args.Value = value
            set_url = controller.parseURL(set_url.Complete)
            self.dispatch_helper.executeDispatch(controller, set_url, "", 0, (set_args,))
        except Exception as e:
            # 错误时更新单元格显示错误信息
            self._update_cell_value(cell_addr, f"请求失败: {str(e)}")

    def _fetch_data_and_update(self, cell_addr: str, symbol: str, query: str = "price"):
        """后台线程执行API请求并触发单元格更新"""
        try:
            api_url = f"https://www.example.com/prices/prices.php?symbol={symbol}&q={query}"
            with urllib.request.urlopen(api_url) as response:
                raw_data = response.read()
                result = json.loads(raw_data)["price"]
                self._update_cell_value(cell_addr, result)
        finally:
            # 任务完成后移除记录
            with task_lock:
                if cell_addr in pending_tasks:
                    del pending_tasks[cell_addr]

    def stizardPriceAsync(self, symbol):
        cell_addr = self._get_current_cell_addr()
        # 检查是否已有待处理任务，避免重复请求
        with task_lock:
            if cell_addr in pending_tasks:
                return "Loading..."
            pending_tasks[cell_addr] = (symbol, "price")
        # 启动后台线程执行请求
        threading.Thread(
            target=self._fetch_data_and_update,
            args=(cell_addr, symbol),
            daemon=True
        ).start()
        return "Loading..."

    def stizardAsync(self, symbol, q = "price"):
        cell_addr = self._get_current_cell_addr()
        with task_lock:
            if cell_addr in pending_tasks:
                return "Loading..."
            pending_tasks[cell_addr] = (symbol, q)
        threading.Thread(
            target=self._fetch_data_and_update,
            args=(cell_addr, symbol, q),
            daemon=True
        ).start()
        return "Loading..."

def createInstance(ctx):
    return StizardImpl(ctx)

g_ImplementationHelper = unohelper.ImplementationHelper()
g_ImplementationHelper.addImplementation(
    createInstance,
    "com.stizard.api.python.StizardImpl",
    ("com.sun.star.sheet.AddIn",)
)

关键注意事项 #

1. 线程安全：所有对文档、单元格的操作必须通过DispatchHelper在主线程执行，禁止直接在后台线程操作UBA对象。
2. 重复请求拦截：通过pending_tasks字典记录正在处理的单元格，防止单元格重算时重复发起API请求。
3. 错误处理：API请求失败时，自动将错误信息填充到对应单元格，提升用户体验。
4. 类型兼容：IDL返回类型改为any，既支持返回字符串（loading/错误信息），也支持返回数值（股票价格）。

使用方式 #

在LibreOffice Calc单元格中调用异步函数：

=STIZARDPRICEASYNC("AAPL")

或

=STIZARDASYNC("AAPL", "volume")

单元格会先显示"Loading..."，待API请求完成后自动更新为实际数据。

内容的提问来源于stack exchange，提问作者Valentin Zagura