forked from erp-dev/erp
371 lines
10 KiB
Python
371 lines
10 KiB
Python
import aiohttp
|
||
from typing import Optional, Dict, Any, List
|
||
from pydantic import BaseModel, Field, computed_field
|
||
from enum import Enum
|
||
import json
|
||
from asgiref.sync import sync_to_async
|
||
|
||
|
||
class Product(BaseModel):
|
||
"""产品模型 - 请在此填入你的字段"""
|
||
uid: str
|
||
rowid: str
|
||
name: str
|
||
unit: str | None
|
||
color: str | None
|
||
width: str | None = None
|
||
detail_str: str | None = Field(exclude=True) # 在序列化时排除此字段
|
||
created_at: str
|
||
|
||
@computed_field
|
||
@property
|
||
def detail(self) -> dict[str, Any]:
|
||
"""将 detail_str 转换为字典"""
|
||
try:
|
||
return json.loads(self.detail_str) if self.detail_str else {}
|
||
except (json.JSONDecodeError, TypeError):
|
||
return {}
|
||
|
||
|
||
class ProductListResponse(BaseModel):
|
||
"""产品列表响应"""
|
||
products: List[Product]
|
||
total: int
|
||
|
||
|
||
mdy_table_map = {
|
||
'product': 'spmx',
|
||
'customer': 'quanbu',
|
||
}
|
||
|
||
|
||
class HTTPMethod(Enum):
|
||
"""HTTP 请求方法枚举"""
|
||
GET = "GET"
|
||
POST = "POST"
|
||
|
||
|
||
class MingDaoYunClient:
|
||
"""明道云 API 客户端"""
|
||
|
||
def __init__(self, app_key: str, sign: str, base_url: str = ""):
|
||
"""
|
||
初始化客户端
|
||
|
||
Args:
|
||
app_key: 应用密钥
|
||
sign: 签名
|
||
base_url: API 基础地址
|
||
"""
|
||
self.app_key = app_key
|
||
self.sign = sign
|
||
self.base_url = base_url
|
||
|
||
async def request(
|
||
self,
|
||
method: HTTPMethod,
|
||
endpoint: str,
|
||
data: Optional[Dict[str, Any]] = None,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
headers: Optional[Dict[str, str]] = None,
|
||
) -> Any:
|
||
"""
|
||
通用异步请求函数
|
||
|
||
Args:
|
||
method: 请求方法 (GET/POST)
|
||
endpoint: API 端点路径
|
||
data: POST 请求体数据（JSON）
|
||
params: URL 查询参数
|
||
headers: 自定义请求头
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
|
||
Raises:
|
||
aiohttp.ClientError: 请求失败时抛出
|
||
"""
|
||
url = f"{self.base_url}{endpoint}"
|
||
|
||
# 构建默认请求头
|
||
default_headers = {
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
if headers:
|
||
default_headers.update(headers)
|
||
|
||
# 添加认证参数
|
||
auth_params = {
|
||
"appKey": self.app_key,
|
||
"sign": self.sign
|
||
}
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
if method == HTTPMethod.GET:
|
||
async with session.get(url, params=params, json=auth_params) as response:
|
||
response.raise_for_status()
|
||
return await response.json()
|
||
|
||
elif method == HTTPMethod.POST:
|
||
if data is None:
|
||
data = {}
|
||
|
||
data.update(auth_params)
|
||
async with session.post(url, json=data) as response:
|
||
response.raise_for_status()
|
||
return await response.json()
|
||
|
||
else:
|
||
raise ValueError(f"Unsupported HTTP method: {method}")
|
||
|
||
async def get(
|
||
self,
|
||
endpoint: str,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
headers: Optional[Dict[str, str]] = None,
|
||
) -> Any:
|
||
"""
|
||
发送 GET 请求
|
||
|
||
Args:
|
||
endpoint: API 端点路径
|
||
params: URL 查询参数
|
||
headers: 自定义请求头
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
"""
|
||
return await self.request(
|
||
method=HTTPMethod.GET,
|
||
endpoint=endpoint,
|
||
params=params,
|
||
headers=headers,
|
||
)
|
||
|
||
async def post(
|
||
self,
|
||
endpoint: str,
|
||
data: Optional[Dict[str, Any]] = None,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
headers: Optional[Dict[str, str]] = None,
|
||
) -> Any:
|
||
"""
|
||
发送 POST 请求
|
||
|
||
Args:
|
||
endpoint: API 端点路径
|
||
data: POST 请求体数据（JSON）
|
||
params: URL 查询参数
|
||
headers: 自定义请求头
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
"""
|
||
return await self.request(
|
||
method=HTTPMethod.POST,
|
||
endpoint=endpoint,
|
||
data=data,
|
||
params=params,
|
||
headers=headers,
|
||
)
|
||
|
||
|
||
product_type_map = {
|
||
'name': '668caa5eb80969563ecaee7d',
|
||
'color': '66ed3b9ae01d5599bdb45f6d',
|
||
'uid': '668caa5eb80969563ecaee7c',
|
||
'unit': '668e2ee2ca758c8cc5d0dc1d',
|
||
'width': '668e2194790c0e04058b4f3c',
|
||
'detail_str': '668caa5eb80969563ecaee7e',
|
||
'created_at': 'ctime',
|
||
'rowid': 'rowid',
|
||
}
|
||
|
||
customer_type_map = {
|
||
'name': '62d52f4b8d2972284492dcf9',
|
||
'area': '62d52f4b8d2972284492dd09',
|
||
'created_at': 'ctime',
|
||
'rowid': 'rowid',
|
||
'uid': '668bb9370207cf7520fe551e',
|
||
}
|
||
|
||
class Customer(BaseModel):
|
||
"""客户模型 - 请在此填入你的字段"""
|
||
uid: str
|
||
rowid: str
|
||
name: str
|
||
area: str | None
|
||
created_at: str
|
||
|
||
|
||
def pick_customer(fields: dict) -> Customer:
|
||
"""从字段字典中提取客户信息"""
|
||
data = {k: fields.get(v, '') for k, v in customer_type_map.items()}
|
||
return Customer(**data)
|
||
|
||
|
||
# json param example:
|
||
# {
|
||
# "appKey": "208e55fea5cea59f",
|
||
# "sign": "MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==",
|
||
# "worksheetId": "quanbu",
|
||
# "listType": 1,
|
||
# "sortId": "ctime",
|
||
# "isAsc": false,
|
||
# "notGetTotal": true
|
||
# }
|
||
|
||
|
||
def pick_product(fields: dict) -> Product:
|
||
"""
|
||
从字段字典中提取产品信息
|
||
"""
|
||
data = {k: fields.get(v, '') for k, v in product_type_map.items()}
|
||
return Product(**data)
|
||
|
||
|
||
async def fetch_products_from_mingdaoyun(page: int = 1, page_size: int = 100) -> tuple[list[Product], int]:
|
||
"""
|
||
从明道云获取产品列表
|
||
|
||
Returns:
|
||
tuple: (产品列表, 总数量)
|
||
"""
|
||
client = MingDaoYunClient(
|
||
app_key='208e55fea5cea59f',
|
||
sign='MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==',
|
||
base_url="https://api.mingdao.com"
|
||
)
|
||
response = await client.post(
|
||
endpoint='/v2/open/worksheet/getFilterRows',
|
||
data={
|
||
'worksheetId': 'spmx',
|
||
'pageIndex': page,
|
||
'pageSize': page_size,
|
||
'sortId': 'ctime',
|
||
'isAsc': True,
|
||
}
|
||
)
|
||
data = response.get('data')
|
||
if data:
|
||
print(f'[product-sync] page={page} size={page_size} rows={len(data.get("rows", []))} total={data.get("total")}')
|
||
else:
|
||
print(f'[product-sync] page={page} size={page_size} received empty data')
|
||
if not data:
|
||
return [], 0
|
||
|
||
products = [pick_product(item) for item in data.get('rows', [])]
|
||
total_count = data.get('total', 0)
|
||
return products, total_count
|
||
|
||
|
||
async def fetch_customers_from_mingdaoyun(page: int = 1, page_size: int = 100) -> tuple[list[Customer], int]:
|
||
"""
|
||
从明道云获取客户列表
|
||
"""
|
||
client = MingDaoYunClient(
|
||
app_key='208e55fea5cea59f',
|
||
sign='MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==',
|
||
base_url="https://api.mingdao.com"
|
||
)
|
||
response = await client.post(
|
||
endpoint='/v2/open/worksheet/getFilterRows',
|
||
data={
|
||
'worksheetId': 'quanbu',
|
||
'pageIndex': page,
|
||
'pageSize': page_size,
|
||
'sortId': 'ctime',
|
||
'isAsc': True,
|
||
}
|
||
)
|
||
data = response.get('data')
|
||
if data:
|
||
print(f'[customer-sync] page={page} size={page_size} rows={len(data.get("rows", []))} total={data.get("total")}')
|
||
else:
|
||
print(f'[customer-sync] page={page} size={page_size} received empty data')
|
||
if not data:
|
||
return [], 0
|
||
|
||
customers = [pick_customer(item) for item in data.get('rows', [])]
|
||
total_count = data.get('total', 0)
|
||
return customers, total_count
|
||
|
||
|
||
fabric_type_map = {
|
||
'name': '62d52f4b8d2972284492de61',
|
||
}
|
||
|
||
|
||
class Fabric(BaseModel):
|
||
"""面料模型 - 请在此填入你的字段"""
|
||
name: str
|
||
|
||
|
||
def pick_fabric(fields: dict) -> Fabric:
|
||
"""
|
||
从字段字典中提取面料信息
|
||
"""
|
||
data = {k: fields.get(v, '') for k, v in fabric_type_map.items()}
|
||
return Fabric(**data)
|
||
|
||
|
||
async def sync_fabric_from_mingdaoyun(page: int = 1, page_size: int = 100) -> int:
|
||
"""
|
||
从明道云同步面料数据
|
||
"""
|
||
client = MingDaoYunClient(
|
||
app_key='208e55fea5cea59f',
|
||
sign='MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==',
|
||
base_url="https://api.mingdao.com"
|
||
)
|
||
response = await client.post(
|
||
endpoint='/v2/open/worksheet/getFilterRows',
|
||
data={
|
||
'worksheetId': '668ba100fb551c850214067d',
|
||
'pageIndex': page,
|
||
'pageSize': page_size,
|
||
'sortId': 'ctime',
|
||
'isAsc': False,
|
||
}
|
||
)
|
||
data = response.get('data')
|
||
if data:
|
||
rows = data.get("rows", [])
|
||
print(f'[fabric-sync] fetch rows={len(rows)} total={data.get("total")}')
|
||
if rows:
|
||
print(f'[fabric-sync] first row keys: {list(rows[0].keys())[:10]}')
|
||
print(f'[fabric-sync] first row sample: {rows[0]}')
|
||
else:
|
||
print(f'[fabric-sync] fetch empty data (page={page}, size={page_size})')
|
||
if not data:
|
||
return [], 0
|
||
fabrics = [pick_fabric(item) for item in data.get('rows', [])]
|
||
total_count = data.get('total', 0)
|
||
await sync_to_async(_create_fabric_quick_inputs)(
|
||
page,
|
||
page_size,
|
||
fabrics
|
||
)
|
||
return total_count
|
||
|
||
|
||
def _create_fabric_quick_inputs(page: int, page_size: int, fabrics: list[Fabric]) -> None:
|
||
from basic_info import models as basic_models
|
||
total = len(fabrics)
|
||
created_count = 0
|
||
sample_names = [fabric.name for fabric in fabrics[:5]]
|
||
print(f'[fabric-sync] page={page} size={page_size} fetched={total}')
|
||
print(f'[fabric-sync] sample names: {sample_names}')
|
||
for fabric in fabrics:
|
||
if not fabric.name:
|
||
continue
|
||
obj, created = basic_models.QuickInput.objects.update_or_create(
|
||
name=fabric.name,
|
||
group='布料名',
|
||
defaults={'value': fabric.name},
|
||
)
|
||
if created:
|
||
created_count += 1
|
||
print(f'[fabric-sync] page={page} new_records={created_count} updated_or_existing={total - created_count}')
|