Allegro Skill开发实战：从零构建高效自动化工作流

1次阅读

共计 3683 个字符，预计需要花费 10 分钟才能阅读完成。

Allegro 作为欧洲领先的电商平台，其订单处理、库存同步等高频操作对自动化需求强烈。但开发者常面临三大痛点：

接口稳定性：官方 API 的速率限制严格（默认 100 次 / 分钟），突发流量易触发 429 错误
业务复杂性：波兰语错误消息解析、时区转换等本地化问题增加开发成本
安全合规：GDPR 要求用户数据必须加密存储，普通脚本难以满足审计要求

现有解决方案如浏览器自动化工具（Selenium）或第三方 SaaS 服务，存在维护成本高、数据主权风险等问题。

REST API（推荐场景）：
适用：订单同步、商品上下架等低频操作
优势：文档完善，Python 生态有成熟 SDK（allegro-api）
注意：需处理分页（limit/offset参数）和 HTTP 缓存头
WebSocket（特殊场景）：
适用：实时价格监控、抢单等毫秒级响应需求
挑战：需自行维护长连接，心跳检测逻辑复杂

import requests
from typing import Tuple, Optional

class AllegroAuth:
    def __init__(self, client_id: str, client_secret: str):
        self.token_url = 'https://allegro.pl/auth/oauth/token'
        self.credentials = (client_id, client_secret)

    def get_token(self) -> Tuple[Optional[str], Optional[int]]:
        """返回 (access_token, expires_in) 或(None, None)"""
        try:
            resp = requests.post(
                self.token_url,
                auth=requests.auth.HTTPBasicAuth(*self.credentials),
                data={'grant_type': 'client_credentials'}
            )
            resp.raise_for_status()
            data = resp.json()
            return data['access_token'], data['expires_in']
        except requests.exceptions.RequestException as e:
            print(f'[ERROR] Auth failed: {str(e)}')
            return None, None

关键点：

使用 requests.auth.HTTPBasicAuth 自动处理 Base64 编码
类型注解明确返回值可能为 None
统一捕获所有 requests 异常避免崩溃

import asyncio
from typing import AsyncGenerator

class OrderWatcher:
    def __init__(self, token: str):
        self.headers = {'Authorization': f'Bearer {token}'}

    async def watch_orders(self) -> AsyncGenerator[dict, None]:
        """异步生成器返回订单状态变更"""
        while True:
            try:
                async with httpx.AsyncClient() as client:
                    resp = await client.get(
                        'https://api.allegro.pl/order/events',
                        headers=self.headers,
                        params={'limit': 50}
                    )
                    resp.raise_for_status()
                    yield resp.json()
                    await asyncio.sleep(60)  # 遵守 API 速率限制
            except Exception as e:
                print(f'[WARN] Watch error: {e}, retrying...')
                await asyncio.sleep(10)

优化技巧：

使用 httpx 替代 aiohttp 更兼容类型检查
AsyncGenerator声明提高代码可读性
60 秒间隔符合 Allegro 事件流推荐轮询频率

from datetime import datetime
import pytz

class BatchProcessor:
    @staticmethod
    def merge_requests(items: list) -> list:
        """将单商品更新合并为批量操作"""
        # 按商品分类 ID 分组，每组最多 50 个（API 上限）return [
            {
                'category': category,
                'items': items[i:i + 50],
                'timestamp': datetime.now(pytz.timezone('Europe/Warsaw'))
            }
            for category, group in itertools.groupby(items, key=lambda x: x['categoryId'])
            for i in range(0, len(list(group)), 50)
        ]

import redis
from pickle import dumps, loads

r = redis.Redis(host='localhost', decode_responses=True)

def cache_order_state(order_id: str, state: dict, ttl: int = 3600):
    """存储订单状态，默认 1 小时过期"""
    r.setex(f'allegro:order:{order_id}',
        ttl,
        dumps(state)  # 序列化复杂对象
    )

def get_cached_state(order_id: str) -> Optional[dict]:
    """获取缓存状态，不存在时返回 None"""
    if data := r.get(f'allegro:order:{order_id}'):
        return loads(data)  # 反序列化
    return None

from cryptography.fernet import Fernet
import os

key = os.getenv('ENCRYPTION_KEY')  # 从环境变量读取
cipher = Fernet(key)

def encrypt_user_data(data: str) -> bytes:
    return cipher.encrypt(data.encode('utf-8'))

def decrypt_user_data(token: bytes) -> str:
    return cipher.decrypt(token).decode('utf-8')

import time
from hashlib import sha256

class NonceValidator:
    def __init__(self):
        self.used_nonces = set()

    def generate_nonce(self) -> str:
        """生成基于时间戳的 nonce"""
        nonce = sha256(str(time.time()).encode()).hexdigest()
        self.used_nonces.add(nonce)
        return nonce

    def validate_nonce(self, nonce: str) -> bool:
        """验证 nonce 是否首次使用"""
        return nonce in self.used_nonces

错误：HTTP 400 Invalid input
根源：波兰本地化时间格式要求（YYYY-MM-DDThh:mm:ssZ）
修复：使用datetime.isoformat() + 手动添加时区
错误：HTTP 403 Forbidden
根源：Scope 权限不足（如订单接口需要order.management）
检查：调用前验证 /scopes 端点
错误：HTTP 503 Service Unavailable
根源：Allegro 维护窗口（每周二 3:00-4:00 CET）
策略：实现自动退避重试机制

建议配置 Prometheus 监控：

- job_name: 'allegro_skill'
  metrics_path: '/metrics'
  static_configs:
    - targets: ['localhost:8000']
  relabel_configs:
    - source_labels: [__address__]
      target_label: instance

关键指标：