init

examples/multi_column.py

@@ -0,0 +1,49 @@
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from openx_sdk import OpenXClient
+
+
+# 正式使用时建议从环境变量或配置文件读取，不要硬编码在业务代码中。
+ACCESS_KEY = ""
+SECRET_KEY = ""
+
+BASE_URL = "https://open.jsptax.com"
+PATH = "/invoice/output/list"
+QUERY = {
+      "comId": "100342309",
+      "period": "202604",
+      "invoiceType": [
+        "1"
+      ],
+      "pageNum": "1",
+      "pageSize": "10",
+      "taxRate":1,
+      "autoLabel": [
+        "0"
+      ]
+}
+
+BODY = {
+       "comId": 101379804,
+        "period": "202605",
+     "taxRate": 1,
+       "invoiceType": [
+      ],
+     "pageNum":1,
+     "pageSize":10
+}
+
+
+def call_api() -> None:
+    client = OpenXClient(BASE_URL, ACCESS_KEY, SECRET_KEY)
+    # response = client.get(PATH, params=QUERY)
+    response = client.post(PATH, json_body=BODY)
+    print("status:", response.status_code)
+    print(response.text)
+
+
+if __name__ == "__main__":
+    call_api()

examples/sign_only.py

@@ -0,0 +1,30 @@
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from openx_sdk import OpenXSigner
+
+
+# 只演示签名生成，不依赖 requests，也不会发送网络请求。
+signer = OpenXSigner(
+    access_key="",
+    secret_key="",
+)
+
+# 固定 timestamp 和 nonce 时，输出应与 curl 示例中的 signature 完全一致。
+signed = signer.sign(
+    query={
+    "comId": "100787681",
+    "accountId": "861533101421002919",
+    "startDate": "2025-07-01",
+    "endDate": "2025-07-30",
+    "pageNum": "1",
+    "pageSize": "20"
+    },
+    timestamp="1780889319",
+    nonce="9fb5235a-175c-4d24-adde-aca39ad1e7bc",
+)
+
+print(signed.signature)
+print(signed.headers)

openx_sdk/__init__.py

@@ -0,0 +1,9 @@
+from .client import OpenXClient
+from .signer import OpenXSigner, SignResult, generate_signature
+
+__all__ = [
+    "OpenXClient",
+    "OpenXSigner",
+    "SignResult",
+    "generate_signature",
+]

openx_sdk/client.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Mapping, Optional, Union
+from urllib.error import HTTPError
+from urllib.parse import urlencode, urljoin
+from urllib.request import Request, urlopen
+
+from .signer import OpenXSigner
+
+
+class SimpleResponse:
+    """标准库 urllib fallback 返回的简单响应对象。
+
+    为了和 requests 的常用属性保持一致，这里提供 status_code、text、content、headers。
+    """
+
+    def __init__(self, status_code: int, content: bytes, headers: Mapping[str, str]) -> None:
+        self.status_code = status_code
+        self.content = content
+        self.headers = dict(headers)
+        self.text = content.decode("utf-8", errors="replace")
+
+    def json(self) -> Any:
+        """按 JSON 解析响应体。"""
+
+        return json.loads(self.text)
+
+
+class OpenXClient:
+    """带自动签名能力的 HTTP 客户端。
+
+    这是一个轻量封装。优先使用 requests；如果当前环境未安装 requests，
+    会自动使用 Python 标准库 urllib。每次 request/get/post 时会自动根据
+    params 和 body 生成 signature/timestamp/nonce/accessKey Header。
+    """
+
+    def __init__(self, base_url: str, access_key: str, secret_key: str) -> None:
+        """初始化客户端。
+
+        Args:
+            base_url: 接口根地址，例如 https://open-test.jsptax.com。
+            access_key: 接口分配的 accessKey。
+            secret_key: 接口分配的 secretKey，不会作为 Header 发送。
+        """
+
+        self.base_url = base_url.rstrip("/") + "/"
+        self.signer = OpenXSigner(access_key, secret_key)
+
+    def signed_headers(
+        self,
+        query: Optional[Mapping[str, Any]] = None,
+        body: Any = None,
+        timestamp: Optional[Union[str, int]] = None,
+        nonce: Optional[str] = None,
+    ) -> dict[str, str]:
+        """只生成签名 Header，不发送 HTTP 请求。"""
+
+        return self.signer.sign(
+            query=query,
+            body=body,
+            timestamp=timestamp,
+            nonce=nonce,
+        ).headers
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[Mapping[str, Any]] = None,
+        json_body: Any = None,
+        data: Any = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: int = 30,
+    ):
+        """发送带签名的 HTTP 请求。
+
+        Args:
+            method: HTTP 方法，例如 GET/POST。
+            path: 接口路径，例如 /ledger/multi-column。
+            params: URL Query 参数，这些参数会参与签名。
+            json_body: JSON 请求体。会按紧凑 JSON 序列化后参与签名并发送。
+            data: 原始请求体。适合用户已经准备好字符串或 bytes 的场景。
+            headers: 额外 Header，会和签名 Header 合并。
+            timeout: 请求超时时间，单位秒。
+        """
+
+        # 签名必须使用“最终实际发送”的 body 字节；JSON 场景先序列化再签名。
+        body = _encode_json(json_body) if json_body is not None else data
+        signed_headers = self.signed_headers(query=params, body=body)
+        request_headers = dict(headers or {})
+        request_headers.update(signed_headers)
+
+        if json_body is not None:
+            request_headers.setdefault("Content-Type", "application/json")
+
+        url = urljoin(self.base_url, path.lstrip("/"))
+        try:
+            import requests
+
+            return requests.request(
+                method=method,
+                url=url,
+                params=params,
+                data=body,
+                headers=request_headers,
+                timeout=timeout,
+            )
+        except ModuleNotFoundError:
+            return _urllib_request(
+                method=method,
+                url=url,
+                params=params,
+                body=body,
+                headers=request_headers,
+                timeout=timeout,
+            )
+
+    def get(self, path: str, params: Optional[Mapping[str, Any]] = None, **kwargs):
+        """发送 GET 请求。"""
+        return self.request("GET", path, params=params, **kwargs)
+
+    def post(self, path: str, params: Optional[Mapping[str, Any]] = None, **kwargs):
+        """发送 POST 请求。"""
+        return self.request("POST", path, params=params, **kwargs)
+
+
+def _encode_json(value: Any) -> bytes:
+    """把 JSON 对象转成紧凑 UTF-8 字节，确保签名内容和发送内容一致。"""
+
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+
+
+def _urllib_request(
+    method: str,
+    url: str,
+    params: Optional[Mapping[str, Any]],
+    body: Any,
+    headers: Mapping[str, str],
+    timeout: int,
+) -> SimpleResponse:
+    """使用 Python 标准库发送请求，避免强依赖 requests。"""
+
+    request_url = _append_query(url, params)
+    body_bytes = _to_request_body(body)
+    request = Request(
+        url=request_url,
+        data=body_bytes,
+        headers=dict(headers),
+        method=method.upper(),
+    )
+
+    try:
+        with urlopen(request, timeout=timeout) as response:
+            return SimpleResponse(
+                status_code=response.getcode(),
+                content=response.read(),
+                headers=response.headers,
+            )
+    except HTTPError as error:
+        return SimpleResponse(
+            status_code=error.code,
+            content=error.read(),
+            headers=error.headers,
+        )
+
+
+def _append_query(url: str, params: Optional[Mapping[str, Any]]) -> str:
+    if not params:
+        return url
+
+    separator = "&" if "?" in url else "?"
+    return url + separator + urlencode(params, doseq=True)
+
+
+def _to_request_body(body: Any) -> Optional[bytes]:
+    if body is None:
+        return None
+    if isinstance(body, bytes):
+        return body
+    if isinstance(body, bytearray):
+        return bytes(body)
+    if isinstance(body, str):
+        return body.encode("utf-8")
+    raise TypeError("data must be None, bytes, bytearray, or str")

openx_sdk/signer.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from hashlib import sha256
+import hmac
+import time
+from typing import Any, Mapping, Optional, Union
+from uuid import uuid4
+
+
+QueryValue = Any
+
+
+@dataclass(frozen=True)
+class SignResult:
+    """签名结果。
+
+    signature/timestamp/nonce/accessKey 需要放到请求 Header 中。
+    canonical_string 是实际参与 HMAC 计算的原文，主要用于排查验签失败问题。
+    """
+
+    signature: str
+    timestamp: str
+    nonce: str
+    access_key: str
+    canonical_string: str
+
+    @property
+    def headers(self) -> dict[str, str]:
+        """返回接口要求的鉴权 Header。"""
+        return {
+            "signature": self.signature,
+            "timestamp": self.timestamp,
+            "nonce": self.nonce,
+            "accessKey": self.access_key,
+        }
+
+
+class OpenXSigner:
+    """Open API 签名器。
+
+    该实现严格对齐 Go 源码 `openx.Sign`：
+    - Header 名是 `accessKey`，但参与签名的参数名是 `ak`
+    - body 先做 SHA256，再以 `body=<hash>` 参与签名
+    - 参数按 key 升序排序后直接拼接，不加 `&`，不做 URL 编码
+    - 使用 secret_key 字符串的 UTF-8 字节做 HMAC-SHA256 key
+    """
+
+    def __init__(self, access_key: str, secret_key: str) -> None:
+        if not access_key:
+            raise ValueError("access_key is required")
+        if not secret_key:
+            raise ValueError("secret_key is required")
+
+        self.access_key = access_key
+        self.secret_key = secret_key
+
+    def sign(
+        self,
+        query: Optional[Mapping[str, QueryValue]] = None,
+        body: Any = None,
+        timestamp: Optional[Union[str, int]] = None,
+        nonce: Optional[str] = None,
+    ) -> SignResult:
+        """生成签名。
+
+        Args:
+            query: URL 查询参数，例如 {"comId": "100841795"}。
+                如果某个值是 list/tuple，会像 Go 的 url.Values 一样取第一个值。
+            body: 请求体。GET 请求可不传；POST JSON 建议传最终发送的 JSON 字符串或 bytes。
+            timestamp: 秒级时间戳。不传时自动使用当前时间。
+            nonce: 随机字符串。不传时自动生成 UUID。
+        """
+
+        timestamp_text = str(timestamp if timestamp is not None else int(time.time()))
+        nonce_text = nonce or str(uuid4())
+
+        canonical = build_canonical_string(
+            access_key=self.access_key,
+            query=query,
+            body=body,
+            timestamp=timestamp_text,
+            nonce=nonce_text,
+        )
+
+        # Go 源码使用 hmac.New(sha256.New, []byte(sk))，输出 hex.EncodeToString。
+        signature = hmac.new(
+            self.secret_key.encode("utf-8"),
+            canonical.encode("utf-8"),
+            sha256,
+        ).hexdigest()
+
+        return SignResult(
+            signature=signature,
+            timestamp=timestamp_text,
+            nonce=nonce_text,
+            access_key=self.access_key,
+            canonical_string=canonical,
+        )
+
+
+def generate_signature(
+    access_key: str,
+    secret_key: str,
+    query: Optional[Mapping[str, QueryValue]] = None,
+    body: Any = None,
+    timestamp: Optional[Union[str, int]] = None,
+    nonce: Optional[str] = None,
+) -> SignResult:
+    """便捷函数：无需手动创建 OpenXSigner，也可以直接生成签名。"""
+    return OpenXSigner(access_key, secret_key).sign(
+        query=query,
+        body=body,
+        timestamp=timestamp,
+        nonce=nonce,
+    )
+
+
+def build_canonical_string(
+    access_key: str,
+    query: Optional[Mapping[str, QueryValue]],
+    body: Any,
+    timestamp: str,
+    nonce: str,
+) -> str:
+    """构建签名原文。
+
+    Go 源码中的拼接方式是 `key=valuekey=value`，中间没有 `&`。
+    例如：`ak=xxxbody=yyycomId=100...nonce=...timestamp=...`
+    """
+
+    params = {
+        "ak": access_key,
+        "timestamp": timestamp,
+        "nonce": nonce,
+        "body": sign_body(body),
+    }
+    params.update(_normalize_query(query))
+
+    # sort.Strings(keys) 的 Python 等价写法：sorted(params)。
+    return "".join(f"{key}={params[key]}" for key in sorted(params))
+
+
+def sign_body(body: Any) -> str:
+    """计算 body 的 SHA256 hex。
+
+    Go 里 nil body 会当作空字节切片处理，因此这里 None 等价于 b""。
+    """
+
+    return sha256(_to_body_bytes(body)).hexdigest()
+
+
+def _normalize_query(query: Optional[Mapping[str, QueryValue]]) -> dict[str, str]:
+    """把 query 参数转换成 Go `map[string]string` 等价结构。"""
+
+    if not query:
+        return {}
+
+    result: dict[str, str] = {}
+    for key, value in query.items():
+        # Go 源码对 url.Values 使用 v[0]，所以多值参数只取第一个值。
+        if isinstance(value, (list, tuple)):
+            first_value = value[0] if value else ""
+        else:
+            first_value = value
+        result[str(key)] = "" if first_value is None else str(first_value)
+    return result
+
+
+def _to_body_bytes(body: Any) -> bytes:
+    """把用户传入的 body 转成参与 SHA256 的字节。"""
+
+    if body is None:
+        return b""
+    if isinstance(body, bytes):
+        return body
+    if isinstance(body, bytearray):
+        return bytes(body)
+    if isinstance(body, str):
+        return body.encode("utf-8")
+    raise TypeError("body must be None, bytes, bytearray, or str")

pyproject.toml

@@ -0,0 +1,11 @@
+[project]
+name = "jsp-open-api-signature"
+version = "0.1.0"
+description = "Python SDK for JSP Open API signature generation"
+requires-python = ">=3.9"
+dependencies = [
+    "requests>=2.31.0",
+]
+
+[tool.setuptools.packages.find]
+where = ["."]