Metadata-Version: 2.4
Name: lkd_pro
Version: 0.2.0
Summary: 自动注册和认证
Home-page: https://github.com/songhang/lkddb_pro
Author: songhang
Author-email: 1026536804@qq.com
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: pywin32>=311
Requires-Dist: Pillow>=10.0.0
Requires-Dist: requests>=2.32.0
Requires-Dist: cachetools>=5.0.0
Requires-Dist: pandas>=2.0.0
Requires-Dist: openpyxl>=3.1.0
Requires-Dist: psutil>=5.9.0
Requires-Dist: opencv-python>=4.8.0
Requires-Dist: pydirectinput>=1.0.4
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# LKD Python SDK

[![Version](https://img.shields.io/badge/version-0.2.0-blue.svg)](https://pypi.org/project/lkd/)
[![Python](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)

## 📖 项目简介

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

### 核心特性

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

---

## 📦 快速开始

### 1. 安装依赖

```bash
pip install -r requirements.txt
```

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

```bash
pip install lkd
```

### 2. 初始化配置

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

```python
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. 使用示例

#### 用户认证

```python
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")
```

#### 商品上传

```python
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']}")
```

#### 获取任务列表

```python
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']}")
```

#### 错误日志上报

```python
from lkd.logs.err_logs import send_logs

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

#### 发送设备心跳

```python
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` - 设备信息生成器

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

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

#### 延迟导入

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

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

### 代码规范

#### 类型注解

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

```python
from typing import Dict, Any, Optional

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

#### 请求类继承

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

```python
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)
```

### 安全规范

#### 线程安全

共享状态访问必须加锁：

```python
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
```

#### 上下文管理器

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

```python
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
```

### 容错机制

#### 多级降级策略

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

```python
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 块中：

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

### 路径规范

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

```python
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 配置：

```python
@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`)

```python
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`)

```python
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`)

```python
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 配置

```python
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
```

### 路径配置

```python
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")
```

### 自动认证客户端

```python
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: 通过全局变量配置（必须在初始化前设置）：

```python
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` 类，支持并发上传：

```python
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 刷新。手动刷新可使用：

```python
from lkd.api import refresh_token

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

### Q: 设备心跳如何发送？

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

```python
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](CHANGELOG.md) 和 [版本日志.txt](版本日志.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
**作者**: 松涵
