import os
import requests
import json
import math
import time
import sqlite3
import hashlib
from datetime import datetime
from json im
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)
