开源免费用!WPS 多维表格 AirScript 影刀指令合集,百万数据查询仅 10 分 35 秒

function main(Context) {    const {         operateType,        sheetName,    // 用表名，不是 ID        sheetId = 0,        records = [],        recordIds = [],        imgbas64 = "",        recordId = "",        imgname = "",        pagefy = "",        filter = null,        upId = "",        maxRecords = undefined    } = Context.argv;    // 根据表名获取正确 sheetId    let realSheetId = sheetId;    if (sheetName) {        const allSheets = Application.Sheet.GetSheets();        for (let i = 0; i < allSheets.length; i++) {            const s = allSheets[i];            if (s.name.trim() === sheetName.trim()) {                realSheetId = s.id;                break;            }        }    }    // 只新增：根据表名获取索引    function getSheetIndexByName(name){        const sheets = Application.Sheet.GetSheets();        for(let i=0;i<sheets.length;i++){            if(sheets[i].name.trim() === name.trim()){                return i+1; // 索引从1开始            }        }        return 1;    }    if (operateType === "query") {    function fetchAllRecords() {        let allRecords = [];        let offset = pagefy ? pagefy : null;        let pageCount = 0; // 计数，控制每次只查3页        // 最多查2页就停止，防止超时        while ((allRecords.length === 0 || offset) && pageCount < 2) {            const queryParams = {                SheetId: realSheetId,                Offset: offset,                PageSize: 1000,    // 严格遵守最大1000，不动！                MaxRecords: maxRecords            };            if (filter) queryParams.Filter = filter;            if (!maxRecords) delete queryParams.MaxRecords;            const recordsResult = Application.Record.GetRecords(queryParams);            offset = recordsResult.offset;            allRecords = allRecords.concat(recordsResult.records);            pageCount++; // 页数+1            if (!offset) break;        }        // 返回当前查到的所有数据 + 下一页offset，给Python继续拉        return [offset, allRecords];    }    return fetchAllRecords();}    // 上传图片（只修改这里：使用索引）其他完全不动    if (operateType === "ym") {        return Application.Sheets(getSheetIndexByName(sheetName)).Views(1).RecordRange(recordId, `@${imgname}`).Value = Application.DBCellValue([{            fileData: imgbas64,            fileName: `${recordId}.png`        }]);    }    // 获取图片下载地址    if (operateType === "xz") {        return Application.Record.GetAttachmentURL({            UploadId: upId,            Source: "upload_ks3"        });    }    // 新建    if (operateType === "create") {        return Application.Record.CreateRecords({            SheetId: realSheetId,            Records: records        });    }    // 删除    if (operateType === "delete") {        return Application.Record.DeleteRecords({            SheetId: realSheetId,            RecordIds: recordIds        });    }    // 更新    if (operateType === "update") {        return Application.Record.UpdateRecords({            SheetId: realSheetId,            Records: [{                id: recordId,                fields: records[0].fields            }]        });    }    throw new Error("不支持的操作类型：" + operateType);}return main(Context);

import xbotfrom xbot import print, sleepfrom . import packagefrom .package import variables as glvimport requestsimport base64import asyncioimport aiohttp# ======================== 中文条件映射（官方标准） ========================CONDITION_MAP = {    "等于": "Equals",    "不等于": "NotEqu",    "大于": "Greater",    "大于等于": "GreaterEqu",    "小于": "Less",    "小于等于": "LessEqu",    "介于取等": "GreaterEquAndLessEqu",    "介于不等": "LessOrGreater",    "开头是": "BeginWith",    "结尾是": "EndWith",    "包含": "Contains",    "不包含": "NotContains",    "指定值": "Intersected",    "为空": "Empty",    "不为空": "NotEmpty"}# ======================== 同步统一请求 ========================def request_api(api_url, api_token, op_type, argv_data):    HEADERS = {        "Content-Type": "application/json",        "AirScript-Token": api_token    }    res = requests.post(        api_url,        json={"Context": {"argv": {"operateType": op_type, **argv_data}}},        headers=HEADERS    )    result = res.json()    data = result.get("data", {})    err = result.get("error", "").strip()    res_data = data.get("result")    if res_data is None or err:        if err != "运行超时":            if err:                raise Exception(err)    return res_data# ======================== 异步高性能请求（大数据提速专用） ========================async def _async_request_api(session, api_url, api_token, op_type, argv_data):    HEADERS = {        "Content-Type": "application/json",        "AirScript-Token": api_token    }    async with session.post(        api_url,        json={"Context": {"argv": {"operateType": op_type, **argv_data}}},        headers=HEADERS    ) as resp:        result = await resp.json()        data = result.get("data", {})        err = result.get("error", "").strip()        res_data = data.get("result")        if res_data is None or err:            if err != "运行超时":                if err:                    raise Exception(err)        return res_data# ======================== 全表数据查询（自动分页循环） ========================def query_all(api_url, api_token, sheet_name):    async def run():        all_data = []        page = ""        async with aiohttp.ClientSession() as session:            while True:                res = await _async_request_api(                    session, api_url, api_token, "query",                     {"sheetName": sheet_name, "pagefy": page}                )                offset, records = res                all_data += records                if not offset:                    break                page = offset        return all_data    return asyncio.run(run())# ======================== 单条件筛选查询 ========================def query_filter(api_url, api_token, sheet_name, field, cn_op, value):    op_en = CONDITION_MAP[cn_op]    filters = {        "mode": "AND",        "criteria": [{"field": field, "op": op_en, "values": [value]}]    }    async def run():        all_data = []        page = ""        async with aiohttp.ClientSession() as session:            while True:                res = await _async_request_api(                    session, api_url, api_token, "query", {                        "sheetName": sheet_name,                        "pagefy": page,                        "filter": filters                    }                )                offset, records = res                all_data += records                if not offset:                    break                page = offset        return all_data    return asyncio.run(run())# ======================== 多条件组合筛选查询 ========================def query_filter_multi(api_url, api_token, sheet_name, conditions, mode="AND"):    criteria = []    filters = {}    if conditions:        if len(conditions) != 0:            for field, cn_op, value in conditions:                op_en = CONDITION_MAP[cn_op]                criteria.append({"field": field, "op": op_en, "values": [value]})            filters = {"mode": mode, "criteria": criteria}    async def run():        all_data = []        page = ""        async with aiohttp.ClientSession() as session:            while True:                res = await _async_request_api(                    session, api_url, api_token, "query", {                        "sheetName": sheet_name,                        "pagefy": page,                        "filter": filters                    }                )                offset, records = res                all_data += records                if not offset:                    break                page = offset        return all_data    return asyncio.run(run())# 单条新增数据def create_data(api_url, api_token, sheet_name, fields):    return request_api(api_url, api_token, "create", {        "sheetName": sheet_name,        "records": [{"fields": fields}]    })# 批量新增海量数据def batch_create(api_url, api_token, sheet_name, records_list):    formatted_records = [{"fields": item} for item in records_list]    return request_api(api_url, api_token, "create", {        "sheetName": sheet_name,        "records": formatted_records    })# 更新单行数据def update_data(api_url, api_token, sheet_name, record_id, fields):    return request_api(api_url, api_token, "update", {        "sheetName": sheet_name,        "recordId": record_id,        "records": [{"fields": fields}]    })# 单条删除def delete_data(api_url, api_token, sheet_name, record_id):    return request_api(api_url, api_token, "delete", {"sheetName": sheet_name, "recordIds": [record_id]})# 批量删除多行数据def batch_delete(api_url, api_token, sheet_name, record_ids):    return request_api(api_url, api_token, "delete", {        "sheetName": sheet_name,        "recordIds": record_ids    })# 上传本地图片至多维表格def upload_img(api_url, api_token, sheet_name, record_id, field_name, img_path):    with open(img_path, "rb") as f:        b64 = base64.b64encode(f.read()).decode()    return request_api(api_url, api_token, "ym", {        "sheetName": sheet_name,        "recordId": record_id,        "imgname": field_name,        "imgbas64": f"data:image/png;base64,{b64}"    })# 获取图片文件下载地址def get_image_url(api_url, api_token, sheet_name, upload_id, source="upload_ks3"):    return request_api(api_url, api_token, "xz", {        "sheetName": sheet_name,        "upId": upload_id,        "source": source    })# 新增数据同时附带多张图片def create_with_imgs(api_url, api_token, sheet_name, fields, img_dict):    rid = create_data(api_url, api_token, sheet_name, fields)[0]["id"]    for field, path in img_dict.items():        for i in range(3):            try:                upload_img(api_url, api_token, sheet_name, rid, field, path)                break            except Exception as e:                pass# 更新数据同时替换多张附件图片def update_with_imgs(api_url, api_token, sheet_name, rid, fields, img_dict):    update_data(api_url, api_token, sheet_name, rid, fields)    for field, path in img_dict.items():        for i in range(3):            try:                upload_img(api_url, api_token, sheet_name, rid, field, path)                break            except Exception as e:                pass# ===================== 实战调用示例 =====================def main(args):    # 读取影刀全局变量配置    API_URL = glv["hook"]    API_TOKEN = glv["密钥"]    # 多条件筛选：查询所有带有图片附件的数据    conditions = [        ("图片和附件", "不等于", ""),    ]    data = query_filter_multi(API_URL, API_TOKEN, "测试", conditions, mode="AND")    print([i["fields"]["图片和附件"][0]["uploadId"] for i in data])    print(get_image_url(API_URL, API_TOKEN, "测试", [i["fields"]["图片和附件"][0]["uploadId"] for i in data][1]))