import os
import requests
import json
import math
import time
import sqlite3
import hashlib
from datetime import datetime
from json import dumps, loads
from urllib.parse import urlencode
from k3cloud_webapi_sdk.main import K3CloudApiSdk
# 客如云常量
BASE_URL = "https://openapi.keruyun.com"
APP_KEY = '45c9f979da976e5f63af39f70a24ff00'
APP_SECRET = 'd9871c47632948c11587b4203f496f24'
BRAN_ID = '34812143'
TOKEN = '5c61d3a18a713b008ff123f06e087315'
DB_FILE = r'E:\项目工程\始运-办公自动化\客如云_数据同步\数据文件\kry_sync_data.db'
# 税账
api_sdk_sz = K3CloudApiSdk()
api_sdk_sz.InitConfig('6453905b9418d9',
'demo1',
'272565_ScaN1csJ4nlUW9wK0dXtUYyE5M6b2KkO',
'92d7536dafd14faaad5d26452ccecbda',
'http://8.134.13.55:8090/k3cloud', # 私有云填写
2052, # 私有云填写
100) # 私有云填写
# 经营
api_sdk_jy = K3CloudApiSdk()
api_sdk_jy.InitConfig('627b29a99c6162',
'demo1',
'229496_XZ6N4ZsL4Jk+3VTu53RAUY8t0qxVxBoI',
'81d9155cf1a647f6b62173b14efb2f04',
'http://8.134.13.55:8090/k3cloud', # 私有云填写
2052, # 私有云填写
100)
# 客如云访问处理
def create_sign(appKey, shopIdenty=None, brandId=None, timestamp=None, version='2.0', body=None):
"""创建签名"""
# 准备参数
# if body is None:
params = {
'appKey': appKey,
'brandId': BRAN_ID,
'version': version,
'timestamp': timestamp,
}
# 根据情况添加shopIdenty或brandId
if shopIdenty:
params['shopIdenty'] = str(shopIdenty)
elif brandId:
params['brandId'] = str(brandId)
# 排序并拼接公共参数
sorted_params = ''.join([f"{k}{params[k]}" for k in sorted(params)])
# 拼接body（如果有）
body_str = ''
if body:
body_str = json.dumps(body, separators=(',', ':'))
# 最终拼接字符串
sign_str = f"{sorted_params}body{body_str}{TOKEN}"
else:
sign_str = f"{sorted_params}{body_str}{TOKEN}"
# SHA-256加密
sign = hashlib.sha256(sign_str.encode()).hexdigest()
return sign
def call_api(endpoint, appKey, shopIdenty=None, brandId=None, body=None):
"""调用API"""
timestamp = int(time.time() * 1000)
sign = create_sign(appKey, shopIdenty, brandId, timestamp, body=body)
# 构建请求URL
url = f"{BASE_URL}{endpoint}?appKey={appKey}"
if shopIdenty:
url += f"&shopIdenty={shopIdenty}"
elif brandId:
url += f"&brandId={brandId}"
url += f"&version=2.0×tamp={timestamp}&sign={sign}"
# 发送POST请求
response = requests.post(
url=url,
headers={'Content-Type': 'application/json'},
data=json.dumps(body, separators=(',', ':')) if body else None
)
return response.json()
def create_sign_shop(appKey, shopIdenty=None, token=None, timestamp=None, version='2.0', body=None):
params = {
'appKey': appKey,
'shopIdenty': shopIdenty,
'version': version,
'timestamp': timestamp,
}
# 排序并拼接公共参数
sorted_params = ''.join([f"{k}{params[k]}" for k in sorted(params)])
body_str = ''
if body:
body_str = json.dumps(body, separators=(',', ':'))
sign_str = f"{sorted_params}body{body_str}{token}"
else:
sign_str = f"{sorted_params}{body_str}{token}"
sign = hashlib.sha256(sign_str.encode()).hexdigest()
return sign
def call_api_shop(endpoint, appKey, shopIdenty=None, body=None):
"""调用API"""
timestamp = int(time.time() * 1000)
token = get_token(shopIdenty)
sign = create_sign_shop(appKey, shopIdenty, token, timestamp, body=body)
# 构建请求URL
url = f"{BASE_URL}{endpoint}?appKey={appKey}"
url += f"&shopIdenty={shopIdenty}"
url += f"&version=2.0×tamp={timestamp}&sign={sign}"
# 发送POST请求
response = requests.post(
url=url,
headers={'Content-Type': 'application/json'},
data=json.dumps(body, separators=(',', ':')) if body else None
)
return response.json()
def get_token(shop_id):
appKey = APP_KEY
shopIdenty = shop_id # 门店ID
version = '2.0' # 版本号
timestamp = int(time.time()) # 当前时间戳（单位：秒）
token = APP_SECRET # 令牌
# 拼接字符串
sign_string = f"appKey{appKey}shopIdenty{shopIdenty}timestamp{timestamp}version{version}{token}"
# 计算SHA256哈希值
sign = hashlib.sha256(sign_string.encode()).hexdigest()
# 构建查询字符串
params = {
'appKey': appKey,
'shopIdenty': shopIdenty,
'version': version,
'timestamp': timestamp,
'sign': sign
}
query_string = urlencode(params)
# 完整的请求URL
uri = '/open/v1/token/get'
request_url = f'{BASE_URL}{uri}?{query_string}'
# 发送GET请求
response = requests.get(request_url)
req = loads(response.text)
shop_token = req['result']['token']
return shop_token
def get_pay_info(shop_no, bill_id):
uri = '/open/standard/order/queryDetail'
body = {
"orderId": bill_id
}
pay_types = []
# 获取支付成功、退款成功的数据
try:
response = call_api_shop(uri, APP_KEY, shopIdenty=shop_no, body=body)
payments = response['result']['data']['openPaymentDetailVoList']
if len(payments) == 1:
if 'PAY_SUCCESS' in payments[0]['payDetailStatus']:
pay_types = [payments[0]['payMethodName']]
flag = 0
if 'REFUND_SUCCESS' in payments[0]['payDetailStatus']:
pay_types = [payments[0]['payMethodName']]
flag = 1
else:
for payment in payments:
if 'PAY_SUCCESS' in payment['payDetailStatus']:
pay_type = payment['payMethodName'] + ':' + str(float(payment['actualReceiveAmt']) / 100)
pay_types.append(pay_type)
elif 'REFUND_SUCCESS' in payment['payDetailStatus']:
pay_type = payment['payMethodName'] + ':' + str(-1*float(payment['actualReceiveAmt']) / 100)
pay_types.append(pay_type)
else:
continue
pay_types = [','.join(pay_types)]
flag = 0
except Exception as e:
pay_types = ['无支付方式']
flag = 0
return flag, pay_types[0]
# 消息通知
def wx_notice(message):
wx_url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=2fcb776f-5216-4381-a03a-683fdc8d4270'
data = {
"msgtype": "text",
"text": {
"content": message,
"mentioned_list": [], # 需@人的姓名
"mentioned_mobile_list": [] # 需@人的手机号
}
}
requests.post(url=wx_url, json=data)
# 金蝶单据处理
def add_bill(flag, doctpye, table, save_data):
"""
:param doctpye: 业务名称
:param table: 表名
:param save_data: json数据
:return: 单据内码、单据编号
""" # 保存
if flag == 'SZ':
api_sdk = api_sdk_sz
else:
api_sdk = api_sdk_jy
dataPara = api_sdk.Save(table, save_data)
dataPara = json.loads(dataPara)
retInfo = list(list(dataPara.values())[0].values())[0]
respcode = retInfo['IsSuccess']
if respcode:
billNo = retInfo['SuccessEntitys'][0]["Number"]
ret_info = "E0000-【{}】-单号[{}]创建成功！".format(doctpye, billNo)
# # 暂不提交和审核
# # 提交
# res_data = api_sdk.Submit(table, {"Numbers": [billNo]})
# res_data = json.loads(res_data) # info = list(list(res_data.values())[0].values())[0] # respcode = info['IsSuccess'] # if respcode: # ret_info = "E0000-【{}】-单号[{}]提交成功！".format(doctpye, billNo)
# # 审核
# res_data = api_sdk.Audit(table, {"Numbers": [billNo]})
# res_data = json.loads(res_data) # info = list(list(res_data.values())[0].values())[0] # respcode = info['IsSuccess'] # if respcode: # ret_info = "E0000-【{}】-单号[{}]已审核！".format(doctpye, billNo)
else:
respdesc = retInfo['Errors'][0]['Message']
ret_info = "E0009-同步{}失败，失败原因:{}".format(doctpye, respdesc)
if '唯一' in ret_info:
pass
else:
msg_info = '金蝶云星空：{}'.format(ret_info)
wx_notice(msg_info)
return ret_info
def get_kingdee_supply(flag, org_id, supply_name):
"""
:param org_id: :param cust_name: :return:
""" if flag == 'SZ':
api_sdk = api_sdk_sz
else:
api_sdk = api_sdk_jy
table = "BD_Supplier" # 供应商资料
data = {
"FormId": table,
"FieldKeys": "FNumber, FName",
"FilterString": [{"Left": "(", "FieldName": "FName", "Compare": "=", "Value": supply_name, "Right": ")",
"Logic": "AND"},
{"Left": "(", "FieldName": "FUseOrgId.FNumber", "Compare": "=", "Value": org_id, "Right": ")",
"Logic": ""}],
"OrderString": "",
"TopRowCount": 0,
"StartRow": 0,
"Limit": 10,
"SubSystemId": ""
}
dataPara = api_sdk.ExecuteBillQuery(data)
dataPara = json.loads(dataPara)
if len(dataPara) == 0:
supply_info = {}
else:
supply_info = {
'supply_no': dataPara[0][0],
'supply_name': dataPara[0][1],
}
return supply_info
def get_kingdee_wl(flag, org_id, wl_name):
"""
:param org_id: :param cust_name: :return:
""" if flag == 'SZ':
api_sdk = api_sdk_sz
else:
api_sdk = api_sdk_jy
table = "BD_MATERIAL" # 物料
data = {
"FormId": table,
"FieldKeys": "FNumber, FName",
"FilterString": [{"Left": "(", "FieldName": "FName", "Compare": "=", "Value": wl_name, "Right": ")",
"Logic": "AND"},
{"Left": "(", "FieldName": "FUseOrgId.FNumber", "Compare": "=", "Value": org_id, "Right": ")",
"Logic": ""}],
"OrderString": "",
"TopRowCount": 0,
"StartRow": 0,
"Limit": 10,
"SubSystemId": ""
}
dataPara = api_sdk.ExecuteBillQuery(data)
dataPara = json.loads(dataPara)
if len(dataPara) == 0:
wl_info = {}
else:
wl_info = {
'wl_no': dataPara[0][0],
'wl_name': dataPara[0][1],
}
return wl_info
def save_ysd(flag, org_id, bill_no, bill_date, cust_no, remark, details):
"""
:param bill_no: :param org_id: :param cust_no: :param bill_date: :param remark: :param details: :return:
""" post_data = {
"Model": {
"FBillTypeID": {
"FNUMBER": "YSD01_SYS"
},
"FBillNo": bill_no,
"FDATE": bill_date,
"FCUSTOMERID": {
"FNumber": cust_no # 客户代码
},
"FCURRENCYID": {
"FNumber": "PRE001"
},
"FSETTLEORGID": {
"FNumber": org_id
},
"FPAYORGID": {
"FNumber": org_id
},
"FSALEORGID": {
"FNumber": org_id
},
"F_TLWD_Assistant": {
"FNumber": cust_no
},
"F_TLWD_Text": bill_no,
"FISTAX": True,
"FCancelStatus": "A",
"FSetAccountType": "3",
"FAR_Remark": remark,
"FEntityDetail": details
}
}
# 保存单据
ret_info = add_bill(flag, "应收单", "AR_receivable", post_data)
return ret_info
# 应付单
def save_yfd(flag, org_id, bill_no, bill_date, supply_no, remark, details):
post_data = {
"Model": {
"FID": 0,
"FBillTypeID": {
"FNUMBER": "YFD01_SYS"
},
"FBillNo": bill_no,
"FISINIT": False,
"FDATE": bill_date,
"FDOCUMENTSTATUS": "Z",
"FSUPPLIERID": {
"FNumber": supply_no
},
"FCURRENCYID": {
"FNumber": "PRE001"
},
"FISPRICEEXCLUDETAX": True,
"FBUSINESSTYPE": "CG",
"FISTAX": True,
"FSETTLEORGID": {
"FNumber": org_id
},
"FPAYORGID": {
"FNumber": org_id
},
"FSetAccountType": "1",
"FISTAXINCOST": False,
"FISHookMatch": False,
"FCancelStatus": "A",
"FISBYIV": False,
"FISGENHSADJ": False,
"FISINVOICEARLIER": False,
"FWBOPENQTY": False,
"FIsGeneratePlanByCostItem": False,
"FsubHeadSuppiler": {
"FORDERID": {
"FNumber": supply_no
},
"FTRANSFERID": {
"FNumber": supply_no
},
"FChargeId": {
"FNumber": supply_no
},
},
"FEntityDetail": details
}
}
# 保存单据
ret_info = add_bill(flag, "应付单", "AP_Payable", post_data)
return ret_info
# 数据库处理
def insert_cgrk_data(curr_obj, ins_data):
"""
:param curr_obj: :param ins_data: :return:
""" bill_no = ins_data['bill_no']
bill_status = ins_data['bill_status']
shop_no = ins_data['shop_no']
shop_name = ins_data['shop_name']
org_id = ins_data['org_id']
org_name = ins_data['org_name']
supply_no = ins_data['supply_no']
supply_name = ins_data['supply_name']
bill_type= ins_data['bill_type']
buss_date= ins_data['buss_date']
purchase_date = ins_data['purchase_date']
total_amt= ins_data['total_amt']
total_amt_notax = ins_data['total_amt_notax']
remarks = ins_data['remarks']
created_time = time.strftime('%Y-%m-%d', time.localtime())
updated_time = time.strftime('%Y-%m-%d', time.localtime())
sz_data_status = 'E0000'
sz_data_status_desc = '待同步'
jy_data_status = 'E0000'
jy_data_status_desc = '待同步'
entities = (bill_no,bill_status,shop_no,shop_name,org_id, org_name, supply_no, supply_name,bill_type,buss_date,
purchase_date,total_amt,total_amt_notax,remarks,created_time,
updated_time,sz_data_status,sz_data_status_desc, jy_data_status, jy_data_status_desc)
try:
curr_obj.execute("""INSERT INTO kry_cgdd (bill_no,bill_status,shop_no,shop_name,org_id,org_name,supply_no,supply_name,bill_type,
buss_date,purchase_date,total_amt,total_amt_notax,remarks,created_time,updated_time,sz_data_status,
sz_data_status_desc, jy_data_status, jy_data_status_desc) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", entities)
except BaseException as e:
if 'UNIQUE constraint' in str(e):
print("有重复数据:[{}] {}".format(e, bill_no))
else:
print("登记数据异常:[{}]".format(e))
return False
return True
def modify_cgdd_data(curr_obj, flag, bill_no, order_status, order_status_desc):
"""
:param curr_obj: :param order_no: :param order_status: :param order_status_desc: :return:
""" try:
if flag == 'SZ':
curr_obj.execute("UPDATE kry_cgdd SET sz_data_status=?, sz_data_status_desc=? WHERE bill_no=?",
(order_status, order_status_desc, bill_no))
else:
curr_obj.execute("UPDATE kry_cgdd SET jy_data_status=?, jy_data_status_desc=? WHERE bill_no=?",
(order_status, order_status_desc, bill_no))
except BaseException as e:
print("更新数据失败:{} ".format(e))
return False
return True
def insert_xsdd_data(curr_obj, ins_data):
"""
:param curr_obj: :param ins_data: :return:
""" shop_no = ins_data['shop_no']
shop_name = ins_data['shop_name']
org_id = ins_data['org_id']
org_name = ins_data['org_name']
cust_no = ins_data['cust_no']
order_id = ins_data['order_id']
busi_order_no = ins_data['busi_order_no']
order_status = ins_data['order_status']
order_type = ins_data['order_type']
pay_type = ins_data['pay_type']
order_amt = ins_data['order_amt']
promo_amt = ins_data['promo_amt']
order_received_amt = ins_data['order_received_amt']
open_time = ins_data['open_time']
finish_time = ins_data['finish_time']
third_order_no = ins_data['third_order_no']
serial_no = ins_data['serial_no']
third_serial_no = ins_data['third_serial_no']
created_time = time.strftime('%Y-%m-%d', time.localtime())
updated_time = time.strftime('%Y-%m-%d', time.localtime())
if pay_type == '无支付方式':
sz_data_status = 'E0003'
sz_data_status_desc = '不同步'
jy_data_status = 'E00003'
jy_data_status_desc = '不同步'
else:
sz_data_status = 'E0000'
sz_data_status_desc = '待同步'
jy_data_status = 'E0000'
jy_data_status_desc = '待同步'
entities = (shop_no,shop_name,org_id,org_name,cust_no,order_id,busi_order_no,
order_status,order_type,pay_type, order_amt,promo_amt,order_received_amt,
open_time,finish_time,third_order_no,serial_no,third_serial_no,created_time,updated_time,
sz_data_status, sz_data_status_desc,
jy_data_status, jy_data_status_desc)
try:
curr_obj.execute("""INSERT INTO kry_xsdd (shop_no,shop_name,org_id,org_name,cust_no,order_id,busi_order_no,order_status,
order_type,pay_type, order_amt,promo_amt,order_received_amt,open_time,finish_time, third_order_no,serial_no,third_serial_no,created_time,updated_time, sz_data_status, sz_data_status_desc, jy_data_status, jy_data_status_desc) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", entities)
except BaseException as e:
if 'UNIQUE constraint' in str(e):
print("有重复数据:[{}] {}".format(e, order_id))
else:
print("登记数据异常:[{}]".format(e))
return False
return True
def modify_xsdd_data(curr_obj, flag, bill_no, order_status, order_status_desc):
"""
:param curr_obj: :param order_no: :param order_status: :param order_status_desc: :return:
""" try:
if flag == 'SZ':
curr_obj.execute("UPDATE kry_xsdd SET sz_data_status=?, sz_data_status_desc=? WHERE busi_order_no=?",
(order_status, order_status_desc, bill_no))
else:
curr_obj.execute("UPDATE kry_xsdd SET jy_data_status=?, jy_data_status_desc=? WHERE busi_order_no=?",
(order_status, order_status_desc, bill_no))
except BaseException as e:
print("更新数据失败:{} ".format(e))
return False
return True
def get_shop_info(shop_no):
conn = sqlite3.connect(DB_FILE)
curr = conn.cursor()
query = """ SELECT shop_name, kingdee_org_code, kingdee_org_name FROM shop_para WHERE shop_id = ? """
val = (shop_no,)
curr.execute(query, val)
# 获取查询结果
rows = curr.fetchall()
shop_info = {"shop_name": rows[0][0], "org_id": rows[0][1], "org_name": rows[0][2]}
curr.close()
conn.close()
return shop_info
# 采购入库查询
def get_cgrk_data(shop_no, start_date, end_date):
shop = get_shop_info(shop_no)
shop_name = shop['shop_name']
org_id = shop['org_id']
org_name = shop['org_name']
## 获取总记录数
body = {
"orgType": 2,
"orgId": shop_no,
"page": 1,
"rows": 100,
"billType": 912,
"billStatusList": [
962
],
"updateTimeStart": start_date,
"updateTimeEnd": end_date,
}
uri = '/open/standard/procurement/supply/purchase/open/standard/purchase/in/list'
response = call_api(uri, APP_KEY, brandId=BRAN_ID, body=body)
res_code = response['result']['serverCode']
res_msg = response['result']['messageId']
if res_code != 10000:
msg = f'【客如云[{shop_name}]】：[{start_date}~{end_date}]没有采购入库数据！'
wx_notice(msg)
return
total = response['result']['data']['total']
total_row = math.ceil(total/100)
conn = sqlite3.connect(DB_FILE)
curr_obj = conn.cursor()
# 循环读取数据
page = 1
while page < total_row:
body = {
"orgType": 2,
"orgId": shop_no,
"bizDateStart": start_date,
"bizDateEnd": end_date,
"page": page,
"rows": 100,
"billType": 912,
"billStatusList": [
962
]
}
# 调用API
response = call_api(uri, APP_KEY, brandId=BRAN_ID, body=body)
res_code = response['result']['serverCode']
res_msg = response['result']['messageId']
total = response['result']['data']['total']
total_row = math.ceil(total/100)
if res_code != 10000:
break
page += 1
lists = response['result']['data']['list']
if len(lists) == 0:
break
for list in lists:
bill_no = list['billNo']
bill_status = list['billStatus']
supply_name = list['supplyName']
bill_type = list['billType']
buss_date = list['bussDate']
purchase_date = list['purchaseDate']
total_amt = list['totalAmt']
total_amt_notax = list['totalAmtNotax']
remarks = list['remarks']
ins_data = {
'bill_no': bill_no,
'bill_status': bill_status,
"shop_no": shop_no, # 门店ID
"shop_name": shop_name, # 门店名称
"org_id": org_id, # 组织ID
"org_name": org_name, # 组织名称
"supply_no": '',
'supply_name': supply_name,
'bill_type': bill_type,
'buss_date': buss_date,
'purchase_date': purchase_date,
'total_amt': total_amt,
'total_amt_notax': total_amt_notax,
'remarks': remarks,
}
# 插入数据到数据库
insert_cgrk_data(curr_obj, ins_data)
conn.commit()
conn.commit()
curr_obj.close()
conn.close()
# 通知客户企业微信群
msg = f'【客如云[{shop_name}]】：[{start_date}~{end_date}]采购入库获取完成！'
wx_notice(msg)
def tokd_cgdd_sync(flag, shop_no, start_date, end_date):
"""
:return:
""" conn = sqlite3.connect(DB_FILE)
curr = conn.cursor()
if flag == 'SZ':
query = """ select org_id, bill_no, supply_name, buss_date, total_amt, remarks from kry_cgdd
where shop_no=? and buss_date>=? and buss_date<=? and sz_data_status in ('E0000', 'E0009') """
else:
query = """ select org_id, bill_no, supply_name, buss_date, total_amt, remarks from kry_cgdd
where shop_no=? and buss_date>=? and buss_date<=? and jy_data_status in ('E0000', 'E0009') """
val = (shop_no, start_date, end_date,)
curr.execute(query, val)
# 获取查询结果
rows = curr.fetchall()
# 打印查询结果
cnt = 0
if 0 == len(rows):
wx_notice(f"客如云[{shop}]->金蝶[{flag}]账套：应付处理[{start_date}至{end_date}]-未查询到需要同步的异常数据！！")
curr.close()
conn.close()
return
for row in rows:
org_id = row[0]
bill_no = row[1]
supply_name = row[2]
bill_date = row[3]
total_amt = row[4]
remarks = row[5]
try:
supply_info = get_kingdee_supply(flag, org_id, supply_name)
supply_no = supply_info['supply_no']
except BaseException as e:
supply_no = ""
details = []
detail = {
"FMATERIALID": {
"FNumber": "kry-yf001"
},
"FCOSTID": {
"FNumber": "FYXM46_SYS"
},
"FPRICEUNITID": {
"FNumber": "Pcs"
},
"FPriceQty": 1.0,
"FTaxPrice": total_amt,
"FEntryTaxRate": 0,
'FINCLUDECOST': False,
"FComment": remarks,
"FTAILDIFFFLAG": False,
}
details.append(detail)
try:
res_info = save_yfd(flag, org_id, bill_no, bill_date, supply_no, "", details)
if 'E0000' in res_info:
modify_cgdd_data(curr, flag, bill_no, "E0001", '同步成功')
elif 'E0009' in res_info:
if '唯一' in res_info:
modify_cgdd_data(curr, flag, bill_no, "E0002", '同步成功[已同步过')
else:
modify_cgdd_data(curr, flag, bill_no, "E0009", res_info)
wx_notice(f"客如云[{shop_no}]->金蝶[{flag}]账套：应付处理-数据同步失败[{res_info}]！")
except BaseException as e:
res_info = "订单【{}】, 同步失败：{}".format(bill_no, e)
modify_cgdd_data(curr, flag, bill_no, "E0009", res_info)
wx_notice(f"客如云[{shop_no}]->金蝶[{flag}]账套：应付处理-数据同步失败[{res_info}]！")
cnt += 1
# 关闭游标和连接
conn.commit()
curr.close()
conn.close()
wx_notice(f"客如云[{shop_no}]->金蝶[{flag}]账套：应付处理[{start_date}至{end_date}]-同步完成，记录数[{cnt}]！！")
# 消费订单查询
def get_xsdd_data(shop_no, start_date, end_date):
# 查询店铺名称，对应金蝶组织
shop = get_shop_info(shop_no)
shop_name = shop['shop_name']
org_id = shop['org_id']
org_name = shop['org_name']
body = {
"dateType": "FINISH_TIME",
"startDate": start_date,
"endDate": end_date,
"orderStatusList": [
"SETTLED",
"SUCCESS"
],
"pageBean": {
"pageNum": 1,
"pageSize": 1
},
}
uri = '/open/standard/order/queryList'
response = call_api_shop(uri, APP_KEY, shopIdenty=shop_no, body=body)
res_code = response['result']['success']
if not res_code:
msg = f'【客如云[{shop_name}]】：[{start_date}~{end_date}]没有消费订单数据！'
wx_notice(msg)
return
total_page = int(response['result']['data']['totalPage'])
conn = sqlite3.connect(DB_FILE)
curr_obj = conn.cursor()
page = 1
while page < total_page:
body = {
"dateType": "FINISH_TIME",
"startDate": start_date,
"endDate": end_date,
"orderStatusList": [
"SETTLED",
"SUCCESS"
],
"pageBean": {
"pageNum": page,
"pageSize": 100
},
}
response = call_api_shop(uri, APP_KEY, shopIdenty=shop_no, body=body)
res_code = response['result']['success']
if not res_code:
break
page += 1
lists = response['result']['data']['list']
if len(lists) == 0:
break
for list in lists:
order_id = list['orderId']
busi_order_no = list['busiOrderNo']
order_types = {
"FOR_HERE": "堂食",
"TAKE_OUT": "外带",
"PLATFORM_TAKE_OUT": "平台外卖",
"SELF_TAKE_OUT": "自营外卖",
"OPEN_PLATFORM_TAKE_OUT": "开放平台外卖",
"SELF_TAKE": "自提",
"NO_ORDER_CASHIER": "无单收银",
"MEMBER_STORE": "会员充值",
"MEMBER_CARD_SALE": "会员售卡并储值",
"MEMBER_MANUAL_STORE": "会员补录",
"REPAYMENT_ORDER": "销账订单",
"PRIVATE_PAY_MEMBER_SALE": "权益卡售卡订单",
"COUPON_PACKAGE_SALE": "券包售卖订单",
"ONLINE_PAY_BILL": "在线买单",
"STATIC_CODE": "客如云收款码",
"DEPOSIT": "订金订单",
"EARNEST_MONEY": "定金订单",
"CASH_PLEDGE": "押金订单",
"GIFT_CARD_SALE": "礼品卡订单",
"GIFT_PACKAGE_SALE": "礼包订单",
"GIFT_CARD_STORE": "礼品卡储值订单",
"GROUP_PURCHASING": "拼团订单",
"POINTS_MALL": "积分商城订单",
"BARGAIN": "砍价订单",
"SEC_KILL": "秒杀订单",
"COUPON_SALE": "团购券售卖"
}
order_status = {
"WAIT_PROCESSED": "待处理",
"WAIT_SETTLED": "待结账",
"SETTLED": "已结账",
"REFUND": "已退单",
"INVALID": "已作废",
"CANCELLED": "已取消",
"REJECTED": "已拒绝",
"SUCCESS": "已完成"
}
order_status_desc = order_status.get(list['orderStatus'])
order_status = list['orderStatus'] + '-' + order_status_desc
order_type = order_types.get(list['orderType'])
try:
if "会员充值" in order_type:
org_id = '103'
except:
print(f"orderType=[{list['orderType']}]")
payment_flag, pay_type = get_pay_info(shop_no, order_id)
if payment_flag:
order_amt = -1*float(list['orderAmt'])/100
promo_amt = -1*float(list['promoAmt'])/100
try:
order_received_amt = -1*float(list['orderReceivedAmt'])/100
third_order_no = ''
except:
msg = f"【客如云[{shop_name}]】：[orderReceivedAmt={list['orderReceivedAmt']}]订单数据异常！"
third_order_no = msg
wx_notice(msg)
order_received_amt = 0
else:
order_amt = float(list['orderAmt']) / 100
promo_amt = float(list['promoAmt']) / 100
try:
order_received_amt = float(list['orderReceivedAmt']) / 100
third_order_no = ''
except:
msg = f"【客如云[{shop_name}]】：[orderReceivedAmt={list['orderReceivedAmt']}]订单数据异常！"
third_order_no = msg
wx_notice(msg)
order_received_amt = 0
open_time = list['openTime']
finish_time = list['finishTime']
# third_order_no = list['thirdOrderNo']
serial_no = list['serialNo']
third_serial_no = list['thirdSerialNo']
ins_data = {
"shop_no": shop_no, # 门店ID
"shop_name": shop_name, # 门店名称
"org_id": org_id, # 组织ID
"org_name": org_name, # 组织名称
"cust_no": "KRY-0001", # 客如云
"order_id": order_id, # 订单ID
"busi_order_no": busi_order_no, # 业务订单号
"order_status": order_status, # 订单状态
"order_type": order_type, # 订单类型
"pay_type": pay_type, # 支付方式
"order_amt": order_amt, # 订单金额(单位/分)
"promo_amt": promo_amt, # 优惠金额(单位/分)
"order_received_amt": order_received_amt, # 订单收入/订单实收(单位/分)
"open_time": open_time, # 下单时间
"finish_time": finish_time, # 完结时间
"third_order_no": third_order_no, # 第三方订单号
"serial_no": serial_no, # 订单流水号
"third_serial_no": "" # 第三方流水号
}
# 插入数据到数据库
insert_xsdd_data(curr_obj, ins_data)
conn.commit()
curr_obj.close()
conn.close()
# # 通知客户企业微信群
msg = msg = f'【客如云[{shop_name}]】：[{start_date}~{end_date}]订单数据获取完成！'
wx_notice(msg)
## 同步金蝶处理
def tokd_xsdd_sync(flag, shop_no, start_date, end_date):
"""
:return:
""" conn = sqlite3.connect(DB_FILE)
curr = conn.cursor()
if flag == 'SZ':
query = """ select org_id, cust_no, finish_time, busi_order_no, order_type, pay_type, order_received_amt
from kry_xsdd where shop_no=? and SUBSTR(finish_time,1,10)>=? and SUBSTR(finish_time,1,10)<=?
and sz_data_status in ('E0000', 'E0009') """
else:
query = """ select org_id, cust_no, finish_time, busi_order_no, order_type, pay_type, order_received_amt
from kry_xsdd where shop_no=? and SUBSTR(finish_time,1,10)>=? and SUBSTR(finish_time,1,10)<=?
and jy_data_status in ('E0000', 'E0009') """
val = (shop_no, start_date, end_date,)
curr.execute(query, val)
# 获取查询结果
rows = curr.fetchall()
# 打印查询结果
cnt = 0
if 0 == len(rows):
wx_notice(f"客如云[{shop}]->金蝶[{flag}]账套：应收处理[{start_date}至{end_date}]-未查询到需要同步的异常数据！！")
curr.close()
conn.close()
return
for row in rows:
org_id = row[0]
cust_no = row[1]
bill_date = row[2]
bill_no = row[3]
order_type = row[4]
pay_type = row[5]
ys_amt = row[6]
details = []
if ',' in pay_type:
pay_types = pay_type.split(',')
for pay_info in pay_types:
pay_type = pay_info.split(':')[0]
ys_amt = float(pay_info.split(':')[1])
if ys_amt < 0:
qty = -1
ys_amt = abs(ys_amt)
else:
qty = 1
try:
wl_info = get_kingdee_wl(flag, org_id, pay_type)
wl_code = wl_info['wl_no']
except:
wl_code = ""
detail = {
"FMATERIALID": {
"FNumber": wl_code
},
"FTaxPrice": ys_amt,
# "FEntryTaxRate": 13,
"FPriceQty": qty,
"FComment": "",
}
details.append(detail)
else:
if ":" in pay_type:
pay_type = pay_type.split(':')[0]
try:
wl_info = get_kingdee_wl(flag, org_id, pay_type)
wl_code = wl_info['wl_no']
except:
wl_code = ""
if ys_amt < 0:
qty = -1
else:
qty = 1
detail = {
"FMATERIALID": {
"FNumber": wl_code
},
"FTaxPrice": ys_amt,
# "FEntryTaxRate": 13,
"FPriceQty": qty,
"FComment": "",
}
details.append(detail)
try:
res_info = save_ysd(flag, org_id, bill_no, bill_date, cust_no, order_type, details)
if 'E0000' in res_info:
modify_xsdd_data(curr, flag, bill_no, "E0001", '同步成功')
elif 'E0009' in res_info:
if '唯一' in res_info:
modify_xsdd_data(curr, flag, bill_no, "E0002", '同步成功[已同步过]')
else:
modify_xsdd_data(curr, flag, bill_no, "E0009", res_info)
wx_notice(f"客如云[{shop_no}]->金蝶[{flag}]账套：应收处理-数据同步失败[{res_info}]！")
except BaseException as e:
res_info = "订单【{}】, 同步失败：{}".format(bill_no, e)
modify_xsdd_data(curr, flag, bill_no, "E0009", res_info)
wx_notice(f"客如云[{shop_no}]->金蝶[{flag}]账套：应收处理-数据同步失败[{res_info}]！")
cnt += 1
# 关闭游标和连接
conn.commit()
curr.close()
conn.close()
wx_notice(f"客如云[{shop_no}]->金蝶[{flag}]账套：应收处理[{start_date}至{end_date}]-同步完成，记录数[{cnt}]！！")
def get_shoplist():
uri = '/open/standard/shop/MerchantOrgReadService.queryBrandStores'
# 调用API
response = call_api(uri, APP_KEY, brandId=34812143, body=None)
# 打印响应内容
shops = response['result']['shops']
for shop in shops:
print(shop['shopId'], shop['name'])
if __name__ == '__main__':
# 业务数据
# get_shoplist()
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""
| shop_id | shop_name | kd_code | kd_name | --------------------------------------------------------------------------------------- 35170196 名古屋日料烤肉(新翼广场店) 103.04 佛山市起翼餐饮服务有限公司顺德第一分公司
35114097 啡同凡享(新翼广场店) 103.07 佛山市起翼餐饮服务有限公司顺德第四分公司
35082103 喜翼楼(赤坎店) 106 湛江市喜翼楼餐饮服务有限公司
35172137 喜翼港式茶餐厅(新翼广场店) 103 佛山市起翼餐饮服务有限公司
34994157 喜翼港式茶餐厅（豪庭店） 103.02 佛山市起翼餐饮服务有限公司湛江豪庭分公司
34830154 喜翼港式茶餐厅（鼎盛店） 103.01 佛山市起翼餐饮服务有限公司湛江鼎盛分公司
35142115 对面粥粉面（新翼广场店） 103.06 佛山市起翼餐饮服务有限公司顺德第三分公司
--------------------------------------------------------------------------------------- """"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""" # shops = [34830154]
# 采购数据同步测试
shops = [35170196, 35114097, 35082103, 35172137, 34994157, 34830154, 35142115]
# for shop in shops:
# get_cgrk_data(shop, '2025-01-01 00:00:00', '2025-01-22 00:00:00') # tokd_cgdd_sync('SZ', shop, '2025-01-01', '2025-01-22') # tokd_cgdd_sync('JY', shop, '2025-01-01', '2025-01-21')
# sync_dates = ["2025-01-01", "2025-01-02", "2025-01-03", "2025-01-04", "2025-01-05", # "2025-01-06", "2025-01-07", "2025-01-08", "2025-01-09", "2025-01-10", # "2025-01-11", "2025-01-12", "2025-01-13", "2025-01-14", "2025-01-15", "2025-01-16", # "2025-01-17","2025-01-18","2025-01-19", "2025-01-20", "2025-01-21"] # sync_dates = ["2025-01-22"]
for sync_date in sync_dates:
s_date = sync_date + " 00:00:00"
e_date = sync_date + " 23:59:59"
# for shop in shops:
# # get_cgrk_data(shop, s_date, e_date) # # tokd_cgdd_sync('SZ', shop, sync_date, sync_date) # # tokd_cgdd_sync('JY', shop, sync_date, sync_date)
# 销售数据同步测试
for shop in shops:
get_xsdd_data(shop, s_date, e_date)
# tokd_xsdd_sync('SZ', shop, sync_date, sync_date)
# tokd_xsdd_sync('JY', shop, sync_date, sync_date)
