Skip to content

Python SDK

OmniBox Python SDK 提供了丰富的 API,帮助您快速开发爬虫脚本。

引入 SDK

python
from omnibox_sdk import OmniBox

SDK 通过 PYTHONPATH 环境变量自动解析,无需手动指定路径。

API 参考

HTTP 请求

OmniBox.request(url, options)

发送 HTTP 请求。

参数:

  • url (str): 请求 URL
  • options (dict, 可选): 请求选项
    • method (str): HTTP 方法,默认为 "GET"
    • headers (dict): 请求头
    • body (str|dict): 请求体

返回: dict 包含 statusCode, headers, body

示例:

python
# GET 请求
response = await OmniBox.request("https://api.example.com/data", {
    "method": "GET",
    "headers": {
        "User-Agent": "Mozilla/5.0",
    },
})

# POST 请求
response = await OmniBox.request("https://api.example.com/submit", {
    "method": "POST",
    "headers": {
        "Content-Type": "application/json",
    },
    "body": {"key": "value"},
})

# 解析响应
import json
data = json.loads(response["body"])

日志记录

OmniBox.log(level, message)

记录日志。

参数:

  • level (str): 日志级别,可选值:"info", "warn", "error"
  • message (str): 日志消息

返回: None

示例:

python
await OmniBox.log("info", "脚本执行开始")
await OmniBox.log("warn", "警告信息")
await OmniBox.log("error", f"错误信息: {error}")

环境变量

OmniBox.get_env(name)

获取环境变量值。

参数:

  • name (str): 环境变量名称

返回: str

示例:

python
import os

# 方式1:通过 SDK 获取(异步)
api_key = await OmniBox.get_env("API_KEY")

# 方式2:直接访问(推荐,性能更好)
api_key = os.environ.get("API_KEY", "")

加密工具

OmniBox.aes_encrypt(data, key)

AES 加密。

参数:

  • data (str): 要加密的数据
  • key (str): 加密密钥

返回: str 加密后的字符串

示例:

python
encrypted = await OmniBox.aes_encrypt("data", "key")

OmniBox.aes_decrypt(encrypted, key)

AES 解密。

参数:

  • encrypted (str): 加密的字符串
  • key (str): 解密密钥

返回: str 解密后的数据

示例:

python
decrypted = await OmniBox.aes_decrypt(encrypted, "key")

OmniBox.md5(data)

MD5 哈希。

参数:

  • data (str): 要哈希的数据

返回: str MD5 哈希值

示例:

python
hash_value = await OmniBox.md5("data")

OmniBox.base64_encode(data)

Base64 编码。

参数:

  • data (str): 要编码的数据

返回: str Base64 编码字符串

示例:

python
encoded = await OmniBox.base64_encode("data")

OmniBox.base64_decode(encoded)

Base64 解码。

参数:

  • encoded (str): Base64 编码字符串

返回: str 解码后的数据

示例:

python
decoded = await OmniBox.base64_decode(encoded)

网盘相关 API

OmniBox.get_drive_info_by_share_url(share_url)

根据分享链接获取网盘显示名称和图标。

参数:

  • share_url (str): 网盘分享链接

返回: dict 包含 driveType, displayName, iconPath, iconUrl

示例:

python
info = await OmniBox.get_drive_info_by_share_url("https://...")
# 返回: {"driveType": "uc", "displayName": "UC网盘", "iconPath": "", "iconUrl": ""}

完整示例

python
import asyncio
import json
import os
import sys
from omnibox_sdk import OmniBox

async def home(params):
    try:
        await OmniBox.log("info", "开始获取首页数据")
        
        # 获取环境变量
        api_key = os.environ.get("API_KEY", "")
        
        # 发送 HTTP 请求
        response = await OmniBox.request("https://api.example.com/home", {
            "method": "GET",
            "headers": {
                "Authorization": f"Bearer {api_key}",
                "User-Agent": "Mozilla/5.0",
            },
        })
        
        # 解析响应
        data = json.loads(response["body"])
        
        # 处理数据
        classes = [
            {"type_id": cat["id"], "type_name": cat["name"]}
            for cat in data["categories"]
        ]
        
        list_data = [
            {
                "vod_id": video["id"],
                "vod_name": video["title"],
                "vod_pic": video["poster"],
                "type_id": video["categoryId"],
                "type_name": video["categoryName"],
                "vod_remarks": video["quality"],
                "vod_year": video["year"],
            }
            for video in data["videos"]
        ]
        
        await OmniBox.log("info", f"成功获取 {len(list_data)} 个视频")
        
        return {
            "class": classes,
            "list": list_data,
        }
    except Exception as error:
        await OmniBox.log("error", f"获取首页数据失败: {str(error)}")
        return {
            "class": [],
            "list": [],
        }

def main():
    input_data = sys.stdin.read()
    request = json.loads(input_data)
    method = request.get("method")
    params = request.get("params", {})
    
    if method == "home":
        result = asyncio.run(home(params))
    else:
        result = {"error": f"未知方法: {method}"}
    
    output = {"success": True, "data": result, "error": None}
    print(json.dumps(output, ensure_ascii=False))

if __name__ == "__main__":
    main()

注意事项

  1. 异步函数:所有 SDK 方法都是异步的,需要使用 awaitasyncio.run()
  2. 编码处理:Python 脚本会自动处理 UTF-8 编码,无需手动设置
  3. 错误处理:建议使用 try-except 捕获错误
  4. 日志记录:使用 OmniBox.log() 记录重要信息,便于调试
  5. 环境变量:优先使用 os.environ.get() 直接访问,性能更好
  6. JSON 处理:使用 json.loads()json.dumps() 处理 JSON 数据

下一步

基于 MIT 许可发布