LKD Python SDK

📖 项目简介

LKD 是电商商品数据处理与自动化上传系统。提供 Excel 表格与图片批量匹配、数据清洗、API 上传、错误日志异步上报等一站式解决方案，帮助电商企业高效管理商品数据。

核心特性

• 🔐 自动认证: Token 自动刷新、设备心跳维护
• 💾 本地缓存: 内存 + 文件双重缓存，支持持久化存储

---

📦 快速开始

1. 安装依赖

pip install -r requirements.txt

或使用 pip 安装（如已发布）：

pip install lkd_pro

2. 初始化配置

在使用 LKD 模块前，需要先初始化全局变量和 IP 配置：

from lkd.var.variables import glv
from lkd.settings.initialization import default as init_default

# 设置服务器 IP（必须在最前面）
glv.set("LOCAL_IP", "192.168.2.28:8000")  # 本地服务器
# glv.set("REMOTE_IP", "your.remote.ip")  # 远程服务器（可选）

# 执行完整初始化流程
success = init_default()
if success:
    print("LKD 初始化完成")

3. 使用示例

用户认证

from lkd.api import login, register

# 登录（密码默认使用设备 ID 后 6 位）
result = login("your_password")
if result["success"]:
    print("登录成功")
    access_token = result["data"]["access_token"]
else:
    # 登录失败可尝试注册
    register_result = register("your_password")

商品上传

from lkd.api import send_products

product_data = {
    "title": "商品名称",
    "price": 99.99,
    "stock": 100,
    "shop_id": "shop_001",
}

result = send_products(product_data)
if result["success"]:
    print("商品上传成功")
else:
    print(f"上传失败：{result['error']}")

获取任务列表

from lkd.api import get_task

result = get_task(status="pending", limit=10)
if result["success"]:
    tasks = result["data"]["tasks"]
    for task in tasks:
        print(f"任务 ID: {task['id']}, 状态：{task['status']}")

错误日志上报

from lkd.logs.err_logs import send_logs

send_logs(
    task="task_001",
    error_message="商品上传失败：库存不足",
    screenshot_path="error.png"
)

发送设备心跳

from lkd.api import send_heartbeat
import datetime

start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
result = send_heartbeat(
    start_heartbeat=start_time,
    access_token="your_access_token",
    ref_token="your_refresh_token",
    status=1  # 1=空闲，2=工作中
)

---

📁 目录结构

lkd/
├── api/                  # API 接口与数据上传
│   ├── __init__.py
│   └── api.py            # 所有 API 接口封装
├── cache/                # 缓存管理
│   ├── __init__.py
│   └── local_cache.py    # 本地缓存与文件缓存
├── client/               # HTTP 客户端
│   ├── __init__.py
│   ├── client.py         # 请求客户端 (CPRequest)
│   └── request_*.py      # 请求对象封装 (BaseRequest 子类)
├── env/                  # 环境与设备信息
│   ├── __init__.py
│   └── device_info.py    # 设备唯一标识生成
├── human/                # 人工交互工具
├── logs/                 # 日志系统
│   ├── __init__.py       # 日志配置
│   └── err_logs.py       # 错误日志上报
├── mwin/                 # Windows 自动化
├── settings/             # 配置管理
│   ├── __init__.py
│   ├── settings.py       # 系统配置 (单例模式)
│   └── initialization.py # 初始化流程
├── util/                 # 通用工具
│   ├── file/             # 文件操作工具
│   │   ├── file.py       # 文件管理
│   │   ├── folder.py     # 文件夹管理
│   │   └── path.py       # 路径管理
│   └── up_c.py           # Excel 上传主流程
├── var/                  # 全局变量管理
│   └── variables.py      # GlobalVariables 单例
├── __init__.py           # 包入口
└── test.py               # 测试与初始化工具

---

🔧 架构规范

设计模式

单例模式

核心模块必须使用单例模式，确保全局唯一实例：

• Settings - 系统配置管理器
• GlobalVariables - 全局变量管理器
• CacheManager - 缓存管理器
• DeviceInfo - 设备信息生成器

# Settings 使用示例
from lkd.settings import get_settings

settings = get_settings()  # 始终返回同一实例
host_url = settings.host_url

延迟导入

解决循环依赖问题，在方法内部动态导入：

def _get_api_module():
    """延迟导入 API 模块，避免循环依赖"""
    from lkd.api import api
    return api

代码规范

类型注解

所有公共 API 必须包含完整的类型注解：

from typing import Dict, Any, Optional

def login(password: str) -> Dict[str, Any]:
    """
    用户登录
  
    Args:
        password: 用户密码
  
    Returns:
        登录结果字典
    """
    pass

请求类继承

所有 HTTP 请求类必须继承自 BaseRequest 抽象基类：

from lkd.client.request_base import BaseRequest

class RequestPost(BaseRequest):
    def __init__(self, request_url: str) -> None:
        super().__init__(method="POST", request_url=request_url)

安全规范

线程安全

共享状态访问必须加锁：

import threading

class CacheManager:
    def __init__(self) -> None:
        self._cache: Dict[str, Any] = {}
        self._lock = threading.Lock()
  
    def set(self, key: str, value: Any) -> None:
        with self._lock:
            self._cache[key] = value

上下文管理器

文件/网络操作必须使用上下文管理器：

def read_file(path: str) -> Optional[str]:
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        log_error(f"读取文件失败：{e}")
        return None

容错机制

多级降级策略

密码初始化等关键流程必须实现多级降级：

def _init_password(self) -> None:
    # 优先使用设备 ID 生成的密码
    device_pwd = _get_pwd_from_device()
    if device_pwd:
        self._password = device_pwd
        return
  
    # 尝试从缓存获取
    cached_pwd = CACHE.get("pwd", "")
    if cached_pwd:
        self._password = cached_pwd
        return
  
    # 生成新密码并保存
    self._password = random_pwd()
    CACHE["pwd"] = self._password

异常处理

认证/注册/心跳流程必须包裹在 try-except 块中：

try:
    result = api_client.send_heartbeat(...)
    if result["success"]:
        log_print("心跳发送成功")
except Exception as e:
    log_error(f"心跳发送异常：{e}")

路径规范

文件路径获取必须实现多层降级策略：

def get_documents_path(self) -> str:
    """获取文档路径（带降级策略）"""
    try:
        # 尝试访问 Documents 目录
        path = get_known_folder_path("Documents")
        if path:
            return os.path.join(path, "0_mt")
    except Exception:
        pass
  
    try:
        # 降级到 APPDATA
        path = os.getenv("APPDATA")
        if path:
            return os.path.join(path, "0_mt")
    except Exception:
        pass
  
    # 最终降级到临时目录
    return tempfile.gettempdir()

配置加载规范

Settings.host_url 必须实现动态检查机制，每次访问时自动从 glv 重新读取 IP 配置：

@property
def host_url(self) -> str:
    """API 主机 URL（动态获取）"""
    # 如果当前 host_url 为空，尝试重新从 glv 加载
    if not self._host_url:
        self._local_ip = glv.get("LOCAL_IP", "")
        self._remote_ip = glv.get("REMOTE_IP", "")
        self._host_url = self._build_host_url()
    return self._host_url

---

📚 主要模块说明

API 模块 (lkd.api)

提供完整的后端 API 接口封装：

认证相关

• register(password: str) - 用户注册
• login(password: str) - 用户登录
• refresh_token(ref_token: str, app_name: str) - 刷新 Token

设备管理

• register_device(access_token, ref_token, **kwargs) - 注册或更新设备
• send_heartbeat(start_heartbeat, access_token, ref_token, status, **kwargs) - 发送心跳

任务管理

• get_task(**kwargs) - 获取任务列表
• update_task(id, **kwargs) - 更新任务
• logs(task_id, log_content, screenshot) - 发送任务日志

商品管理

• send_products(data) - 发送商品信息
• get_products(**kwargs) - 查询商品信息

店铺管理

• get_shop(shop_id, **kwargs) - 获取店铺信息
• update_shop(shop_id, **kwargs) - 更新店铺信息
• get_second_categories(**kwargs) - 获取二级类目
• create_or_update_second(**kwargs) - 创建/更新二级类目

采购管理

• get_purchase(**kwargs) - 获取拼多多采购信息
• post_purchase(**kwargs) - 采购 PDD 商品
• put_purchase(**kwargs) - 更新采购信息

绑定关系

• query_pdd_bind_products(**kwargs) - 查询商品绑定关系
• update_pdd_bind_product(id, **kwargs) - 更新绑定关系

缓存模块 (lkd.cache)

from lkd.cache import CACHE, FileCache

# 内存缓存（字典式访问）
CACHE["key"] = "value"
value = CACHE["key"]

# 或使用标准方法
CACHE.set("key", "value")
value = CACHE.get("key")

# 文件缓存（持久化）
file_cache = FileCache("/path/to/cache")
file_cache.start_background_save()  # 启动后台自动保存

日志模块 (lkd.logs)

from lkd.logs import log_print, log_error, log_warning, set_log_dir

# 设置日志目录
set_log_dir("/path/to/logs")

# 记录日志
log_print("普通信息")      # INFO 级别
log_error("错误信息")      # ERROR 级别
log_warning("警告信息")    # WARNING 级别

全局变量模块 (lkd.var)

from lkd.var.variables import glv

# 设置变量
glv.set("LOCAL_IP", "192.168.2.28:8000")
glv.set("gvar_task_path", "/path/to/tasks")

# 获取变量
local_ip = glv.get("LOCAL_IP")
task_path = glv.get("gvar_task_path")

# 检查是否存在
if glv.has("LOCAL_IP"):
    print("IP 已配置")

---

⚙️ 配置说明

IP 配置

from lkd.var.variables import glv

# 本地服务器（优先使用）
glv.set("LOCAL_IP", "192.168.2.28:8000")

# 远程服务器（备用）
glv.set("REMOTE_IP", "your.remote.ip:port")

# Settings 会自动选择 LOCAL_IP（如果存在），否则使用 REMOTE_IP
# 并构建完整的 URL: http://{ip}/api

路径配置

from lkd.var.variables import glv

# 配置关键路径
glv.set("gvar_config_path", "/path/to/config")
glv.set("gvar_task_path", "/path/to/tasks")
glv.set("gvar_img_path", "/path/to/images")
glv.set("gvar_logs_path", "/path/to/logs")
glv.set("gvar_export_path", "/path/to/export")

自动认证客户端

from lkd.env.auto_client import AutoClient

# 创建自动连接客户端
auto_client = AutoClient(app_name="solider")

# 启动心跳
auto_client.run()

# 客户端会自动处理：
# 1. Token 检查和刷新
# 2. 登录/注册
# 3. 设备注册
# 4. 定时心跳发送

---

⚠️ 注意事项

1. 初始化顺序: 必须先设置 IP 配置 (LOCAL_IP/REMOTE_IP)，再执行其他操作
2. Excel 格式规范: 确保 Excel 表格格式规范，列名与配置一致
3. 图片命名: 图片文件夹命名需与 Excel 表格对应
4. 权限要求: Windows 自动化功能需要相应系统权限
5. Python 版本: 需要 Python 3.8 或更高版本
6. 线程安全: 多线程环境下请使用锁机制保护共享状态

---

🔍 常见问题

Q: 如何修改 API 服务器地址？

A: 通过全局变量配置（必须在初始化前设置）：

from lkd.var.variables import glv
glv.set("LOCAL_IP", "new.server.ip:port")

Q: 日志文件在哪里？

A: 默认位于初始化时配置的 gvar_logs_path 目录，可通过 set_log_dir() 自定义。

Q: 如何处理大量商品上传？

A: 使用 ExcelUploader 类，支持并发上传：

from lkd.util.up_c import ExcelUploader

uploader = ExcelUploader(
    excel_path="products.xlsx",
    shop_name="店铺名称",
    max_workers=10  # 并发数
)
uploader.run()

Q: Token 过期了怎么办？

A: 使用 AutoClient 会自动处理 Token 刷新。手动刷新可使用：

from lkd.api import refresh_token

result = refresh_token(
    ref_token="your_refresh_token",
    app_name="solider"
)

Q: 设备心跳如何发送？

A: 推荐使用 AutoClient 自动发送，或手动调用：

from lkd.api import send_heartbeat
import datetime

result = send_heartbeat(
    start_heartbeat=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    access_token="your_access_token",
    ref_token="your_refresh_token",
    status=1
)

---

📝 更新日志

详见 CHANGELOG.md 和 版本日志.txt

v0.2.0 (2026-04-02)

• ✅ 修复 Settings 动态加载 IP 配置问题
• ✅ 优化心跳 URL 拼接逻辑
• ✅ 增强缓存模块线程安全性
• ✅ 改进路径降级策略

---

🤝 贡献指南

欢迎提交 Issue 和 Pull Request！贡献代码请遵循以下规范：

1. 所有公共 API 必须包含完整 docstring
2. 保持类型注解风格一致
3. 关键流程必须包含异常处理
4. 新增功能需补充测试用例

---

📄 许可证

MIT License

---

📧 联系方式

技术支持：请联系项目维护者

---

最后更新: 2026-04-02 版本: 0.2.0 作者: 松涵