Skip to content

A
Automat
Commit e48d50f4 by Aeolus
Options

update

parent 3e916194
Showing with 158 additions and 859 deletions
models/models.py
......@@ -35,7 +35,7 @@ class Hatch(Base):
hatch_no = Column(TINYINT(3), nullable=False, comment='机柜仓口号')
production_id = Column(INTEGER(10), nullable=False, comment='商品id')
name = Column(String(100, 'utf8mb4_unicode_ci'), nullable=False, comment='商品名称')
titile = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='商品标题')
title = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='商品标题')
brand_id = Column(INTEGER(10), nullable=False, comment='品牌ID')
brand_name = Column(String(100, 'utf8mb4_unicode_ci'), nullable=False, comment='商品名称')
cate_id = Column(INTEGER(10), nullable=False, comment='分类ID')
......@@ -102,7 +102,7 @@ class Production(Base):
id = Column(INTEGER(10), primary_key=True)
name = Column(String(100, 'utf8mb4_unicode_ci'), nullable=False, index=True, comment='商品名称')
titile = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='商品标题')
title = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='商品标题')
brand_id = Column(INTEGER(10), nullable=False, comment='品牌ID')
cate_id = Column(INTEGER(10), nullable=False, comment='分类ID')
price = Column(INTEGER(10), nullable=False, comment='价格')
......@@ -155,7 +155,7 @@ class RentDetail(Base):
production_id = Column(INTEGER(10), nullable=False, comment='商品id')
is_take = Column(TINYINT(3), nullable=False, server_default=text("'0'"), comment='是否取货')
name = Column(String(100, 'utf8mb4_unicode_ci'), nullable=False, comment='商品名称')
titile = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='商品标题')
title = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='商品标题')
brand_id = Column(INTEGER(10), nullable=False, comment='品牌ID')
brand_name = Column(String(100, 'utf8mb4_unicode_ci'), nullable=False, comment='商品名称')
cate_id = Column(INTEGER(10), nullable=False, comment='分类ID')
......@@ -165,7 +165,7 @@ class RentDetail(Base):
tags = Column(String(255, 'utf8mb4_unicode_ci'), comment='商品标签')
content = Column(Text(collation='utf8mb4_unicode_ci'), comment='商品内容')
summary = Column(Text(collation='utf8mb4_unicode_ci'), comment='商品描述')
status = Column(TINYINT(3), nullable=False, server_default=text("'0'"), comment='1正常 2删除')
status = Column(TINYINT(3), nullable=False, server_default=text("'1'"), comment='1正常 -1删除')
created_at = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"))
updated_at = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"))
......@@ -175,7 +175,7 @@ class SalePlan(Base):
id = Column(INTEGER(10), primary_key=True)
name = Column(String(100, 'utf8mb4_unicode_ci'), nullable=False, comment='方案名称')
titile = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='方案标题')
title = Column(String(200, 'utf8mb4_unicode_ci'), nullable=False, comment='方案标题')
status = Column(TINYINT(1), nullable=False, comment='状态: 1整除-1删除')
created_at = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"))
updated_at = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"))
......
myapps/sukang24h/api/hatch_portal.py
......@@ -27,16 +27,19 @@ def get_production_list():
if not machine_info:
return jsonify(MACHINE_NOT_EXIST_ERROR)
hatch_list = Hatch.query.filter_by(machine_no=machine_no).order_by(Hatch.hatch_no.asc()).all()
hatch_list = Hatch.query.filter_by(machine_no=machine_no, status=1).order_by(Hatch.hatch_no.asc()).all()
if not hatch_list:
return jsonify(HATCH_NOT_EXIST_ERROR)
hatch_data = [{
tmp_dict = {}
for i in hatch_list:
if tmp_dict.get(i.production_id, None):
tmp_dict[i.production_id] = {
"machine_no": i.machine_no,
"hatch_no": i.hatch_no,
"production_id": i.production_id,
"name": i.name,
"titile": i.titile,
"title": i.title,
"brand_id": i.brand_id,
"brand_name": i.brand_name,
"cate_id": i.cate_id,
......@@ -48,8 +51,11 @@ def get_production_list():
"content": i.content,
"summary": i.summary,
"status": i.status,
} for i in hatch_list]
"count": 1
}
else:
tmp_dict[i.production_id]["count"] += 1
hatch_data = tmp_dict.values()
return BaseResponse(data=hatch_data)
......@@ -72,7 +78,7 @@ def get_production_info():
"hatch_no": hatch_info.hatch_no,
"production_id": hatch_info.production_id,
"name": hatch_info.name,
"titile": hatch_info.titile,
"title": hatch_info.title,
"brand_id": hatch_info.brand_id,
"brand_name": hatch_info.brand_name,
"cate_id": hatch_info.cate_id,
......
myapps/sukang24h/api/rent_portal.py
......@@ -13,6 +13,7 @@ import time
from functools import reduce
from flask import Blueprint, request, jsonify, g
from sqlalchemy import func
from config.commen_config import USER_RENT_PREPAY_ID, RENT_SALE_LOCK
from config.wechat_config import platform_appid_config_list, pay_config_list
......@@ -33,30 +34,28 @@ rent_route = Blueprint('rent', __name__)
@rent_route.route("rent", methods=["POST"])
def create_rent():
json_data = request.get_json()
# machine_no = json_data["machine_no"]
# hatch_nos = json_data["hatch_nos"]
machine_no = json_data["machine_no"]
productions = json_data["productions"]
user = g.user
machine_no = '1636127865'
hatch_nos = [1]
# user = WxUser.query.filter_by(id=1).first()
# 验证机柜是否存在
machine = Machine.query.filter_by(machine_no=machine_no).first()
if not machine:
return jsonify(MACHINE_NOT_EXIST_ERROR)
# 根绝给的仓号去获取商品信息
hatch_list = Hatch.query.filter(Hatch.machine_no == machine_no, Hatch.hatch_no.in_(hatch_nos),
Hatch.status == 1).order_by(
Hatch.hatch_no.asc()).all()
total_fee = 0
for id, count in productions.items():
# 根据给的仓号去获取商品信息
hatch_list = Hatch.query.filter(Hatch.machine_no == machine_no, Hatch.production_id == id,
Hatch.status == 1).order_by(func.random()).all()
if not hatch_list:
return jsonify(HATCH_NOT_EXIST_ERROR)
if len(hatch_list) != len(hatch_nos):
if len(hatch_list) < int(count):
return jsonify(HATCH_NOT_ALL_EXIST_ERROR)
total_fee = reduce(lambda x, y: x + y, [i.price for i in hatch_list])
total_fee += hatch_list[0].price * int(count)
# 生成订单编号
rent_no = RentService.create_order_no()
......@@ -72,7 +71,7 @@ def create_rent():
'openid': user.openid,
'timeStamp': str(int(time.time())),
'attach': {
"hatch_nos": hatch_nos,
"productions": productions,
"machine_no": machine_no,
"user_id": user.id,
"machine_id": machine.id,
......@@ -86,8 +85,8 @@ def create_rent():
if not pay_info:
return BaseResponse(**WE_MINIAPP_PAY_FAIL)
# 微信下单成功，插入订单数据
redis_client.set(USER_RENT_PREPAY_ID + str(user.id) + rent_no, pay_info["prepay_id"])
# 微信下单成功，redis存prepay_id 5分钟失效
redis_client.set(USER_RENT_PREPAY_ID + str(user.id) + rent_no, pay_info["prepay_id"], 300)
return BaseResponse(data=pay_info)
......@@ -116,8 +115,13 @@ def wx_pay_callback():
logger.info(xml_data)
logger.info(callback_data)
attach = re.sub('\'', '\"', callback_data["attach"])
print(attach)
rent_data = json.loads(attach)
platform = rent_data["platform"]
machine_no = rent_data["machine_no"]
machine_id = rent_data["machine_id"]
productions = rent_data["productions"]
user_id = rent_data["user_id"]
platform = pay_config_list[platform]
wechat_service = WeChatPayService(app_id=callback_data["appid"], config_name=platform)
verify_result = wechat_service.verify_pay_callbak_sign(callback_data)
......@@ -132,7 +136,7 @@ def wx_pay_callback():
if rent:
return xmltodict.unparse({'xml': error_data}, pretty=True), header
machine = Machine.query.filter_by(id=rent_data["machine_id"]).first()
machine = Machine.query.filter_by(machine_no=machine_no).first()
try:
lock_time = 0
......@@ -146,49 +150,71 @@ def wx_pay_callback():
lock_time += 1
time.sleep(1)
rent = RentService.create_order(rent_no, rent_data, machine, platform=int(rent_data["platform"]))
if int(rent.deposit * rent.number) != int(callback_data['total_fee']):
return xmltodict.unparse({'xml': error_data}, pretty=True), header
rent = Rent()
rent.rent_no = rent_no
rent.machine_no = machine.machine_no
rent.user_id = rent_data["user_id"]
rent.place_id = machine.place_id
rent.total = int(callback_data["total_fee"])
rent.add_time = datetime.datetime.now()
rent.is_pay = 1
rent.pay_time = datetime.datetime.strptime(callback_data["time_end"], "%Y%m%d%H%M%S")
db.session.add(rent)
db.session.commit()
rent.rent_type = type
rent.mch_platform = platform
prepay_id = redis_client.get(USER_RENT_PREPAY_ID + str(rent_data["user_id"]) + rent_no)
if prepay_id:
rent.prepay_id = prepay_id
total_fee = 0
for id, count in productions.items():
count = int(count)
id = int(id)
# 根据给的仓号去获取商品信息
hatch_list = Hatch.query.filter(Hatch.machine_no == machine_no, Hatch.production_id == id,
Hatch.status == 1).order_by(func.random()).all()
powers = PowerService.get_machine_powers(rent.machine_id, rent.number, rent.power_type)
if not powers:
# TODO 退款
data = {
'out_trade_no': rent.rent_no,
'out_refund_no': RentService.create_refund_no(),
'total_fee': rent.deposit * rent.number,
'refund_fee': rent.deposit * rent.number,
# 'notify_url': notify_url
}
if not hatch_list:
return jsonify(HATCH_NOT_EXIST_ERROR)
result = wechat_service.do_refund(refund_data=data)
if not result:
if len(hatch_list) < count:
return jsonify(HATCH_NOT_ALL_EXIST_ERROR)
hatchs = hatch_list[:count]
for i in hatchs:
if i.status == 1:
rent_detail = RentDetail()
rent_detail.rent_no = rent_no
rent_detail.user_id = rent.user_id
rent_detail.machine_no = rent.machine_no
rent_detail.hatch_no = i.hatch_no
rent_detail.production_id = i.production_id
rent_detail.name = i.name
rent_detail.title = i.title
rent_detail.brand_id = i.brand_id
rent_detail.brand_name = i.brand_name
rent_detail.cate_id = i.cate_id
rent_detail.cate_name = i.cate_name
rent_detail.price = i.price
rent_detail.img = i.img
rent_detail.tags = i.tags
rent_detail.content = i.content
rent_detail.summary = i.summary
i.status = 2
db.session.add(rent_detail)
db.session.add(i)
total_fee += i.price
if total_fee != int(callback_data['total_fee']):
return xmltodict.unparse({'xml': error_data}, pretty=True), header
RentService.update_rent_refunded(rent, data["out_refund_no"])
RentService.push_rent_fail_message(rent)
# return error_data, header
return xmltodict.unparse({'xml': response_data}, pretty=True), header
# 处理账户金额变动
RentService.change_customer_deposit(1, rent.customer_id, rent.deposit * rent.number)
# 生成充电宝的订单记录，并修改充电宝的租借状态
RentService.create_production_record(rent, powers)
db.session.add(rent)
db.session.commit()
except Exception as e:
redis_client.delete(RENT_SALE_LOCK + str(machine.id))
return xmltodict.unparse({'xml': error_data}, pretty=True), header
redis_client.delete(RENT_SALE_LOCK + str(rent.machine_id))
redis_client.delete(RENT_SALE_LOCK + str(machine.machine_id))
return xmltodict.unparse({'xml': response_data}, pretty=True), header
......@@ -222,7 +248,7 @@ def get_rent_detail():
tmp_data["production_id"] = i.production_id
tmp_data["is_take"] = i.is_take
tmp_data["name"] = i.name
tmp_data["titile"] = i.titile
tmp_data["title"] = i.title
tmp_data["brand_id"] = i.brand_id
tmp_data["brand_name"] = i.brand_name
tmp_data["cate_id"] = i.cate_id
......@@ -241,7 +267,7 @@ def get_rent_detail():
@rent_route.route("/detail_record", methods=["POST"])
def get_rent_detail_record():
def get_user_detail_record():
json_data = request.get_json()
page = int(json_data['page'])
limit = int(json_data['limit'])
......@@ -259,7 +285,7 @@ def get_rent_detail_record():
tmp_data["production_id"] = i.production_id
tmp_data["is_take"] = i.is_take
tmp_data["name"] = i.name
tmp_data["titile"] = i.titile
tmp_data["title"] = i.title
tmp_data["brand_id"] = i.brand_id
tmp_data["brand_name"] = i.brand_name
tmp_data["cate_id"] = i.cate_id
......@@ -274,3 +300,30 @@ def get_rent_detail_record():
detail_data.append(tmp_data)
return BaseResponse(data=detail_data)
@rent_route.route("/rent_hatchs", methods=["POST"])
def get_user_rent_detail():
json_data = request.get_json()
rent_no = json_data['rent_no']
user_id = g.user.id
rent_detail = RentDetail.query.filter(RentDetail.user_id == user_id, RentDetail.rent_no == rent_no,
RentDetail.status != -1).order_by(
RentDetail.id.desc()).all()
if not rent_detail:
rent_detail = []
detail_data = []
for i in rent_detail:
tmp_data = {}
tmp_data["hatch_no"] = i.hatch_no
tmp_data["production_id"] = i.production_id
tmp_data["is_take"] = i.is_take
tmp_data["name"] = i.name
tmp_data["title"] = i.title
tmp_data["price"] = i.price
tmp_data["img"] = i.img
tmp_data["status"] = i.status
detail_data.append(tmp_data)
return BaseResponse(data=detail_data)
mytest.py 0 → 100644
#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
@version:
author:Aeolus
@time: 2021/10/15
@file: mytest.py
@function:
@modify:
"""
a = {'productions': ['3'], 'machine_no': '1636127865', 'user_id': 1, 'machine_id': 1, 'platform': 1}
service/rent_service.py
......@@ -5,6 +5,8 @@ import random
from sqlalchemy.exc import SQLAlchemyError
import logging
from config.commen_config import USER_RENT_PREPAY_ID
from models.base_model import db
from models.models import Rent
from utils.my_redis_cache import redis_client
......@@ -36,8 +38,10 @@ class RentService(object):
try:
model = Rent()
model.rent_no = rent_no
model.machine_id = machine.id
model.customer_id = data["user_id"]
model.machine_no = machine.machine_no
model.user_id = data["user_id"]
model.place_id = machine.place_id
model.total = 0
model.add_time = datetime.datetime.now()
model.spot_id = machine.spot_id
model.business_id = machine.business_id
......@@ -46,11 +50,7 @@ class RentService(object):
prepay_id = redis_client.get(USER_RENT_PREPAY_ID + str(data["user_id"]) + rent_no)
if prepay_id:
model.prepay_id = prepay_id
if data["power_type"]:
model.power_type = data["power_type"]
power_money = PowerMoney.query.filter_by(machine_id=machine.id,
power_type=data["power_type"]).first()
model.one_day_price = machine.one_day_price if not power_money else power_money.total
db.session.add(model)
db.session.commit()
return model
......
utils/error_code.py
#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
@version:
author:Aeolus
@file: error_code.py
"""
### 通用错误相关
Param_Invalid_Error = {
"error_code": "500",
"error_message": "params is invalid, 参数无效"
}
TOKEN_NOT_VALID_ERROR = {
"error_code": "1001",
"error_message": "无效的token"
}
TOKEN_NOT_PROVIDED_ERROR = {
"error_code": "1002",
"error_message": "token未提供"
}
TOKEN_EXPIRE_ERROR = {
"error_code": "1003",
"error_message": "token超时"
}
PHONE_NOT_BINDING_ERROR = {
"error_code": "1004",
"error_message": "未绑定手机号"
}
PHONE_NOT_NULL_ERROR = {
"error_code": "1005",
"error_message": "手机号为空"
}
PHONE_NOT_VALID_ERROR = {
"error_code": "1006",
"error_message": "无效的手机号"
}
USER_ALREADY_REGISTER_ERROR = {
"error_code": "1007",
"error_message": "用户已注册"
}
VERIFICATION_CODE_NULL_ERROR = {
"error_code": "1008",
"error_message": "验证码为空"
}
VERIFICATION_CODE_INVALID_ERROR = {
"error_code": "1009",
"error_message": "验证码已失效"
}
VERIFICATION_CODE_ERROR = {
"error_code": "1010",
"error_message": "验证码错误"
}
PASSWORD_ERROR = {
"error_code": "1011",
"error_message": "账号或密码错误"
}
## 微信登陆相关
WX_LOGIN_DATA_ERROR = {
"error_code": "3001",
"error_message": "微信登录数据错误"
}
WX_LOGIN_CODE_ERROR = {
"error_code": "3002",
"error_message": "微信登录code值错误"
}
WX_OPENID_NOT_GET_ERROR = {
"error_code": "3003",
"error_message": "微信OpenId获取失败，请刷新重试"
}
WX_SESSION_KEY_ERROR = {
"error_code": "3004",
"error_message": "session key error"
}
### 微信支付相关
WE_MINIAPP_PAY_FAIL = {
"error_code": "3101",
"error_message": "小程序下单失败"
}
### 消息推送相关
WXBizMsgCrypt_OK = {
"error_code": "0",
"error_message": "WXBizMsgCrypt_OK"
}
WXBizMsgCrypt_ValidateSignature_Error = {
"error_code": "4001",
"error_message": "验证签名错误"
}
WXBizMsgCrypt_ParseXml_Error = {
"error_code": "4002",
"error_message": "解析xml错误"
}
WXBizMsgCrypt_ComputeSignature_Error = {
"error_code": "4003",
"error_message": "计算签名错误"
}
WXBizMsgCrypt_IllegalAesKey = {
"error_code": "4004",
"error_message": "Aes key非法错误"
}
WXBizMsgCrypt_ValidateAppid_Error = {
"error_code": "4005",
"error_message": "appid错误"
}
WXBizMsgCrypt_EncryptAES_Error = {
"error_code": "4006",
"error_message": "aes加密错误"
}
WXBizMsgCrypt_DecryptAES_Error = {
"error_code": "4007",
"error_message": "aes解密错误"
}
WXBizMsgCrypt_IllegalBuffer = {
"error_code": "4008",
"error_message": "illegal buffer"
}
WXBizMsgCrypt_EncodeBase64_Error = {
"error_code": "4009",
"error_message": "base64加密错误"
}
WXBizMsgCrypt_DecodeBase64_Error = {
"error_code": "4010",
"error_message": "base64解密错误"
}
WXBizMsgCrypt_GenReturnXml_Error = {
"error_code": "4011",
"error_message": "gen return xml error"
}
MACHINE_NOT_EXIST_ERROR = {
"error_code": '5001',
"error_message": "机柜不存在"
}
MACHINE_IS_USE_ERROR = {
"error_code": '5002',
"error_message": "已有他人正在租借中，请稍后"
}
MACHINE_IS_NOT_ONLINE_ERROR = {
"error_code": '5003',
"error_message": "机柜不在线"
}
MACHINE_ADD_ERROR = {
"error_code": '5004',
"error_message": "机柜添加失败"
}
MACHINE_NO_DUPLICATE_ERROR = {
"error_code": '5005',
"error_message": "machine_no duplicate,机柜编号重复"
}
MACHINE_EDIT_ERROR = {
"error_code": '5006',
"error_message": "machine edit error, 机柜修改错误"
}
HATCH_NOT_EXIST_ERROR = {
"error_code": "5007",
"error_message": "no hatch, 没有商品信息"
}
HATCH_NOT_ALL_EXIST_ERROR = {
"error_code": "5008",
"error_message": "no all hatch, 存在已售出商品"
}
### 订单相关
RENT_ORDER_NOT_BACK_ERROR = {
"error_code": '6101',
"error_message": "有未归还的订单"
}
RENT_ORDER_NOT_TAKE_ERROR = {
"error_code": '6102',
"error_message": "有未取货的订单"
}
RENT_ORDER_NUMBER_MAX = {
"error_code": '6103',
"error_message": "订单数量达到上限"
}
TAKE_CODE_NOT_VALID = {
"error_code": '6104',
"error_message": "取货码错误请确认手机号及取货码是否匹配"
}
CODE_CANCEL_ERROR = {
"error_code": '6105',
"error_message": "取货码已取消"
}
CODE_USED_ERROR = {
"error_code": '6108',
"error_message": "取货码已使用"
}
NO_POWER_ERROR = {
"error_code": '6106',
"error_message": "没有可租借设备"
}
NO_RENT_RECORD = {
"error_code": '6107',
"error_message": "订单不存在"
}
CODE_USED_ERROR = {
"error_code": '6108',
"error_message": "取货码已使用"
}
RENT_ORDER_NUMBER_LIMIT = {
"error_code": '6109',
"error_message": "机柜只允许租借一台"
}
REFUND_NOT_RENT_INFO = {
"error_code": "6301",
"error_message": "没有该订单信息"
}
REFUND_BACK_TIME_ERROR = {
"error_code": "6302",
"error_message": "归还时间异常"
}
REFUND_NOT_PRODUCTION_INFO = {
"error_code": "6303",
"error_message": "没有该讲解器信息"
}
REFUND_MONEY_IS_ZERO = {
"error_code": "6304",
"error_message": "退款金额为零"
}
REFUND_NO_DUPLICATE = {
"error_code": "6305",
"error_message": "退款单号重复"
}
#!usr/bin/env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @file: error_code.py """ ### 通用错误相关 Param_Invalid_Error = { "error_code": "500", "error_message": "params is invalid, 参数无效" } TOKEN_NOT_VALID_ERROR = { "error_code": "1001", "error_message": "无效的token" } TOKEN_NOT_PROVIDED_ERROR = { "error_code": "1002", "error_message": "token未提供" } TOKEN_EXPIRE_ERROR = { "error_code": "1003", "error_message": "token超时" } PHONE_NOT_BINDING_ERROR = { "error_code": "1004", "error_message": "未绑定手机号" } PHONE_NOT_NULL_ERROR = { "error_code": "1005", "error_message": "手机号为空" } PHONE_NOT_VALID_ERROR = { "error_code": "1006", "error_message": "无效的手机号" } USER_ALREADY_REGISTER_ERROR = { "error_code": "1007", "error_message": "用户已注册" } VERIFICATION_CODE_NULL_ERROR = { "error_code": "1008", "error_message": "验证码为空" } VERIFICATION_CODE_INVALID_ERROR = { "error_code": "1009", "error_message": "验证码已失效" } VERIFICATION_CODE_ERROR = { "error_code": "1010", "error_message": "验证码错误" } PASSWORD_ERROR = { "error_code": "1011", "error_message": "账号或密码错误" } ## 微信登陆相关 WX_LOGIN_DATA_ERROR = { "error_code": "3001", "error_message": "微信登录数据错误" } WX_LOGIN_CODE_ERROR = { "error_code": "3002", "error_message": "微信登录code值错误" } WX_OPENID_NOT_GET_ERROR = { "error_code": "3003", "error_message": "微信OpenId获取失败，请刷新重试" } WX_SESSION_KEY_ERROR = { "error_code": "3004", "error_message": "session key error" } ### 微信支付相关 WE_MINIAPP_PAY_FAIL = { "error_code": "3101", "error_message": "小程序下单失败" } ### 消息推送相关 WXBizMsgCrypt_OK = { "error_code": "0", "error_message": "WXBizMsgCrypt_OK" } WXBizMsgCrypt_ValidateSignature_Error = { "error_code": "4001", "error_message": "验证签名错误" } WXBizMsgCrypt_ParseXml_Error = { "error_code": "4002", "error_message": "解析xml错误" } WXBizMsgCrypt_ComputeSignature_Error = { "error_code": "4003", "error_message": "计算签名错误" } WXBizMsgCrypt_IllegalAesKey = { "error_code": "4004", "error_message": "Aes key非法错误" } WXBizMsgCrypt_ValidateAppid_Error = { "error_code": "4005", "error_message": "appid错误" } WXBizMsgCrypt_EncryptAES_Error = { "error_code": "4006", "error_message": "aes加密错误" } WXBizMsgCrypt_DecryptAES_Error = { "error_code": "4007", "error_message": "aes解密错误" } WXBizMsgCrypt_IllegalBuffer = { "error_code": "4008", "error_message": "illegal buffer" } WXBizMsgCrypt_EncodeBase64_Error = { "error_code": "4009", "error_message": "base64加密错误" } WXBizMsgCrypt_DecodeBase64_Error = { "error_code": "4010", "error_message": "base64解密错误" } WXBizMsgCrypt_GenReturnXml_Error = { "error_code": "4011", "error_message": "gen return xml error" } MACHINE_NOT_EXIST_ERROR = { "error_code": '5001', "error_message": "机柜不存在" } MACHINE_IS_USE_ERROR = { "error_code": '5002', "error_message": "已有他人正在租借中，请稍后" } MACHINE_IS_NOT_ONLINE_ERROR = { "error_code": '5003', "error_message": "机柜不在线" } MACHINE_ADD_ERROR = { "error_code": '5004', "error_message": "机柜添加失败" } MACHINE_NO_DUPLICATE_ERROR = { "error_code": '5005', "error_message": "machine_no duplicate,机柜编号重复" } MACHINE_EDIT_ERROR = { "error_code": '5006', "error_message": "machine edit error, 机柜修改错误" } HATCH_NOT_EXIST_ERROR = { "error_code": "5007", "error_message": "no hatch, 没有商品信息" } HATCH_NOT_ALL_EXIST_ERROR = { "error_code": "5008", "error_message": "no all hatch, 存在已售出商品" } ### 订单相关 RENT_ORDER_NOT_BACK_ERROR = { "error_code": '6101', "error_message": "有未归还的订单" } RENT_ORDER_NOT_TAKE_ERROR = { "error_code": '6102', "error_message": "有未取货的订单" } RENT_ORDER_NUMBER_MAX = { "error_code": '6103', "error_message": "订单数量达到上限" } TAKE_CODE_NOT_VALID = { "error_code": '6104', "error_message": "取货码错误请确认手机号及取货码是否匹配" } CODE_CANCEL_ERROR = { "error_code": '6105', "error_message": "取货码已取消" } CODE_USED_ERROR = { "error_code": '6108', "error_message": "取货码已使用" } NO_POWER_ERROR = { "error_code": '6106', "error_message": "没有可租借设备" } NO_RENT_RECORD = { "error_code": '6107', "error_message": "订单不存在" } CODE_USED_ERROR = { "error_code": '6108', "error_message": "取货码已使用" } RENT_ORDER_NUMBER_LIMIT = { "error_code": '6109', "error_message": "机柜只允许租借一台" } REFUND_NOT_RENT_INFO = { "error_code": "6301", "error_message": "没有该订单信息" } REFUND_BACK_TIME_ERROR = { "error_code": "6302", "error_message": "归还时间异常" } REFUND_NOT_PRODUCTION_INFO = { "error_code": "6303", "error_message": "没有该讲解器信息" } REFUND_MONEY_IS_ZERO = { "error_code": "6304", "error_message": "退款金额为零" } REFUND_NO_DUPLICATE = { "error_code": "6305", "error_message": "退款单号重复" }
\ No newline at end of file
......
utils/jwt_util.py
#!usr/bin/.env python
# -*- coding:utf-8 _*-
"""
@version:
author:Aeolus
@time: 2021/03/30
@file: jwt_util.py
@function:
@modify:
"""
import jwt
from flask import current_app
def generate_jwt(payload, expiry, secret=None):
"""
生成jwt
:param payload: dict 载荷
:param expiry: datetime 有效期
:param secret: 密钥
:return: jwt
"""
_payload = {'exp': expiry}
_payload.update(payload)
if not secret:
secret = current_app.config['SECRET_KEY']
token = jwt.encode(_payload, secret, algorithm='HS256')
return token
def verify_jwt(token, secret=None):
"""
检验jwt
:param token: jwt
:param secret: 密钥
:return: dict: payload
"""
if not secret:
secret = current_app.config['SECRET_KEY']
try:
payload = jwt.decode(token, secret, algorithms=['HS256'])
except jwt.PyJWTError:
payload = None
return payload
if __name__ == '__main__':
import time
from config.env_path_config import env_path
from dotenv import load_dotenv
load_dotenv(dotenv_path=env_path, verbose=True, override=True)
import os
SECRET_KEY = os.getenv('SECRET_KEY')
token = generate_jwt({"user_id": 1}, time.time() + 6000, SECRET_KEY)
print(token)
# for i in range(10):
# result = verify_jwt(token, 'secret')
# print(result)
# print(time.time())
# time.sleep(1)
#!usr/bin/.env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @time: 2021/03/30 @file: jwt_util.py @function: @modify: """ import jwt from flask import current_app def generate_jwt(payload, expiry, secret=None): """ 生成jwt :param payload: dict 载荷 :param expiry: datetime 有效期 :param secret: 密钥 :return: jwt """ _payload = {'exp': expiry} _payload.update(payload) if not secret: secret = current_app.config['SECRET_KEY'] token = jwt.encode(_payload, secret, algorithm='HS256') return token def verify_jwt(token, secret=None): """ 检验jwt :param token: jwt :param secret: 密钥 :return: dict: payload """ if not secret: secret = current_app.config['SECRET_KEY'] try: payload = jwt.decode(token, secret, algorithms=['HS256']) except jwt.PyJWTError: payload = None return payload if __name__ == '__main__': import time from config.env_path_config import env_path from dotenv import load_dotenv load_dotenv(dotenv_path=env_path, verbose=True, override=True) import os SECRET_KEY = os.getenv('SECRET_KEY') token = generate_jwt({"user_id": 1}, time.time() + 6000, SECRET_KEY) print(token) # for i in range(10): # result = verify_jwt(token, 'secret') # print(result) # print(time.time()) # time.sleep(1)
\ No newline at end of file
......
utils/middlewares.py
#!usr/bin/.env python
# -*- coding:utf-8 _*-
"""
@version:
author:Aeolus
@time: 2021/03/26
@file: middlewares.py
@function:
@modify:
"""
import logging
from flask import g, request, url_for, current_app, make_response, jsonify
from config.wechat_config import platform_config_list
from models.models import WxUser
from utils.error_code import TOKEN_NOT_VALID_ERROR
from utils.my_response import BaseResponse
from utils.jwt_util import verify_jwt
logger = logging.getLogger(__name__)
def log_enter_interface():
"""
日志打印进入接口
:return:
"""
logger.info("######################### 进入 {} 接口 ################################ ".format(request.path))
def log_out_interface(environ):
"""
日志打印退出接口
:return:
"""
logger.info("######################### 退出 {} 接口 ################################\n".format(request.path))
return environ
def close_db_session(environ):
from models.base_model import db
db.session.close()
return environ
"""用户认证机制==>每次请求前获取并校验token"""
"@myapps.before_request 不使@调用装饰器 在 init文件直接装饰"
def jwt_authentication():
"""
1.获取请求头Authorization中的token
2.判断是否以 Bearer开头
3.使用jwt模块进行校验
4.判断校验结果,成功就提取token中的载荷信息,赋值给g对象保存
"""
if current_app.name == "sukang24h":
NO_AUTH_CHECK_URL = [url_for('wx_auth.my_test'),
url_for('wx_auth.mini_login'),
url_for('rent.wx_pay_callback'),
url_for('hatch.get_production_list'),
]
else:
NO_AUTH_CHECK_URL = []
if request.path not in NO_AUTH_CHECK_URL:
token = request.headers.get('Authorization')
# "校验token"
payload = verify_jwt(token)
# "判断token的校验结果"
if payload:
# "获取载荷中的信息赋值给g对象"
user_id = payload.get('user_id')
if not user_id:
return BaseResponse(**TOKEN_NOT_VALID_ERROR)
try:
g.user = WxUser.query.filter_by(id=user_id).first()
return
except Exception as e:
print(e)
else:
return BaseResponse(**TOKEN_NOT_VALID_ERROR)
def get_platform():
"""
:return:
"""
g.platform = request.headers.get('platform', "sukang24h")
def all_options_pass():
"""
:return:
"""
if request.method == "OPTIONS":
headers = {'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'POST',
'Access-Control-Allow-Headers':
'Content-Type, Content-Length, Authorization, Accept, X-Requested-With , platform',
}
return make_response((jsonify({'error_code': 0}), 200, headers))
#!usr/bin/.env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @time: 2021/03/26 @file: middlewares.py @function: @modify: """ import logging from flask import g, request, url_for, current_app, make_response, jsonify from config.wechat_config import platform_config_list from models.models import WxUser from utils.error_code import TOKEN_NOT_VALID_ERROR from utils.my_response import BaseResponse from utils.jwt_util import verify_jwt logger = logging.getLogger(__name__) def log_enter_interface(): """ 日志打印进入接口 :return: """ logger.info("######################### 进入 {} 接口 ################################ ".format(request.path)) def log_out_interface(environ): """ 日志打印退出接口 :return: """ logger.info("######################### 退出 {} 接口 ################################\n".format(request.path)) return environ def close_db_session(environ): from models.base_model import db db.session.close() return environ """用户认证机制==>每次请求前获取并校验token""" "@myapps.before_request 不使@调用装饰器 在 init文件直接装饰" def jwt_authentication(): """ 1.获取请求头Authorization中的token 2.判断是否以 Bearer开头 3.使用jwt模块进行校验 4.判断校验结果,成功就提取token中的载荷信息,赋值给g对象保存 """ if current_app.name == "sukang24h": NO_AUTH_CHECK_URL = [url_for('wx_auth.my_test'), url_for('wx_auth.mini_login'), url_for('rent.wx_pay_callback'), url_for('hatch.get_production_list'), ] else: NO_AUTH_CHECK_URL = [] if request.path not in NO_AUTH_CHECK_URL: token = request.headers.get('Authorization') # "校验token" payload = verify_jwt(token) # "判断token的校验结果" if payload: # "获取载荷中的信息赋值给g对象" user_id = payload.get('user_id') if not user_id: return BaseResponse(**TOKEN_NOT_VALID_ERROR) try: g.user = WxUser.query.filter_by(id=user_id).first() return except Exception as e: print(e) else: return BaseResponse(**TOKEN_NOT_VALID_ERROR) def get_platform(): """ :return: """ g.platform = request.headers.get('platform', "sukang24h") def all_options_pass(): """ :return: """ if request.method == "OPTIONS": headers = {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'POST', 'Access-Control-Allow-Headers': 'Content-Type, Content-Length, Authorization, Accept, X-Requested-With , platform', } return make_response((jsonify({'error_code': 0}), 200, headers))
\ No newline at end of file
......
utils/my_redis_cache.py
#!usr/bin/.env python
# -*- coding:utf-8 _*-
"""
@version:
author:Aeolus
@file: my_redis_cache.py
"""
from flask_redis import FlaskRedis
redis_client = FlaskRedis(config_prefix='TENCENT_REDIS')
#!usr/bin/.env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @file: my_redis_cache.py """ from flask_redis import FlaskRedis redis_client = FlaskRedis(config_prefix='TENCENT_REDIS')
\ No newline at end of file
......
utils/my_response.py
#!usr/bin/.env python
# -*- coding:utf-8 _*-
"""
@version:
author:Aeolus
@time: 2021/04/27
@file: my_response.py
@function:
@modify:
"""
from flask import Response
from flask.json import dumps
class BaseResponse(Response):
def __init__(self, data=None, error_code=0, error_message='Success', *args, **kwargs):
if data is not None:
result = dumps(dict(data=data, error_code=error_code, error_message=error_message, *args, **kwargs))
else:
result = dumps(dict(error_code=error_code, error_message=error_message, *args, **kwargs))
Response.__init__(self, result, mimetype='application/json')
#!usr/bin/.env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @time: 2021/04/27 @file: my_response.py @function: @modify: """ from flask import Response from flask.json import dumps class BaseResponse(Response): def __init__(self, data=None, error_code=0, error_message='Success', *args, **kwargs): if data is not None: result = dumps(dict(data=data, error_code=error_code, error_message=error_message, *args, **kwargs)) else: result = dumps(dict(error_code=error_code, error_message=error_message, *args, **kwargs)) Response.__init__(self, result, mimetype='application/json')
\ No newline at end of file
......
utils/mylogger.py
import logging
from flask_log_request_id import RequestIDLogFilter
def set_logger():
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(
logging.Formatter(
"%(asctime)s - %(filename)s - %(funcName)s -[line:%(lineno)d] - %(levelname)s - request_id=%(request_id)s: %(message)s",
"%Y-%m-%d %H:%M:%S"))
console_handler.addFilter(RequestIDLogFilter())
logging.getLogger().setLevel(logging.DEBUG)
logging.getLogger().addHandler(console_handler)
import logging from flask_log_request_id import RequestIDLogFilter def set_logger(): console_handler = logging.StreamHandler() console_handler.setLevel(logging.DEBUG) console_handler.setFormatter( logging.Formatter( "%(asctime)s - %(filename)s - %(funcName)s -[line:%(lineno)d] - %(levelname)s - request_id=%(request_id)s: %(message)s", "%Y-%m-%d %H:%M:%S")) console_handler.addFilter(RequestIDLogFilter()) logging.getLogger().setLevel(logging.DEBUG) logging.getLogger().addHandler(console_handler)
\ No newline at end of file
......
utils/wechat/WXBizDataCrypt.py
import base64
import json
from Crypto.Cipher import AES
class WXBizDataCrypt:
def __init__(self, appId, sessionKey):
self.appId = appId
self.sessionKey = sessionKey
def decrypt(self, encryptedData, iv):
sessionKey = base64.decodebytes(bytes(self.sessionKey, encoding='utf8'))
encryptedData = base64.decodebytes(bytes(encryptedData, encoding='utf8'))
iv = base64.decodebytes(bytes(iv, encoding='utf8'))
cipher = AES.new(sessionKey, AES.MODE_CBC, iv)
des_str = cipher.decrypt(encryptedData)
print("==================================")
print(des_str)
des_str = self._unpad(des_str)
print(des_str)
des_str = str(des_str,encoding='utf-8')
decrypted = json.loads(des_str)
if decrypted['watermark']['appid'] != self.appId:
raise Exception('Invalid Buffer')
return decrypted
def _unpad(self, s):
return s[:-ord(s[len(s) - 1:])]
if __name__ == '__main__':
appId = 'wx3185fb4a3633beb0'
sessionKey='S7CMDfC6jXJKSaWKanG8oQ=='
encryptedData='E7LZhvK7mOcaYsv9xcAfsBN9eSbzFh9FyMtFJ0zsFB0M62zRJ0cosZWksUujUR5WYUmNoIfIJnTIF8gRskxxbFU3fm5X7z4ChZecMSaFM65aEK1suRUD1U0ubB7mOwBBlY4ftdPT5kRwWgXKVkM4VAkYGN8A4fjWE93yGtjzxXs9dypQkCLSNWs6Kw5USEzjhtDZnptVy+lHF5fTXRuzoCstW2Cto4YI3G9hmnS64QuWjRteSqIgh8GN1zEPN0dROJjaWBjqraBCt/BfMsk4HBeL4PA75K8WdqVgKGfQ7/rnmPFOsNXWfajx9jl7XcrfoPaaPL1DmIJ1BlQne2GuLFtzZ3O4/8cdVQ9Lb0N/3kFAzjgzNFNLSYj2VNctmWyLdWi8hH90yslvrODIhMzIsuux2GIAfp0rQd/iVIVvtd7PXBOCe5iZ7aaqD0b0mLF4CmsuBpl8Eh20ZHkYw2SqO0x9uFrS/gy1vwtkmsTpcDw='
iv = 'DQcmcXyQkU+VKqb2mKmasQ=='
pc = WXBizDataCrypt(appId, sessionKey)
pc.decrypt(encryptedData, iv)
#
import base64import jsonfrom Crypto.Cipher import AES class WXBizDataCrypt: def __init__(self, appId, sessionKey): self.appId = appId self.sessionKey = sessionKey def decrypt(self, encryptedData, iv): sessionKey = base64.decodebytes(bytes(self.sessionKey, encoding='utf8')) encryptedData = base64.decodebytes(bytes(encryptedData, encoding='utf8')) iv = base64.decodebytes(bytes(iv, encoding='utf8')) cipher = AES.new(sessionKey, AES.MODE_CBC, iv) des_str = cipher.decrypt(encryptedData) print("==================================") print(des_str) des_str = self._unpad(des_str) print(des_str) des_str = str(des_str,encoding='utf-8') decrypted = json.loads(des_str) if decrypted['watermark']['appid'] != self.appId: raise Exception('Invalid Buffer') return decrypted def _unpad(self, s): return s[:-ord(s[len(s) - 1:])] if __name__ == '__main__': appId = 'wx3185fb4a3633beb0' sessionKey='S7CMDfC6jXJKSaWKanG8oQ==' encryptedData='E7LZhvK7mOcaYsv9xcAfsBN9eSbzFh9FyMtFJ0zsFB0M62zRJ0cosZWksUujUR5WYUmNoIfIJnTIF8gRskxxbFU3fm5X7z4ChZecMSaFM65aEK1suRUD1U0ubB7mOwBBlY4ftdPT5kRwWgXKVkM4VAkYGN8A4fjWE93yGtjzxXs9dypQkCLSNWs6Kw5USEzjhtDZnptVy+lHF5fTXRuzoCstW2Cto4YI3G9hmnS64QuWjRteSqIgh8GN1zEPN0dROJjaWBjqraBCt/BfMsk4HBeL4PA75K8WdqVgKGfQ7/rnmPFOsNXWfajx9jl7XcrfoPaaPL1DmIJ1BlQne2GuLFtzZ3O4/8cdVQ9Lb0N/3kFAzjgzNFNLSYj2VNctmWyLdWi8hH90yslvrODIhMzIsuux2GIAfp0rQd/iVIVvtd7PXBOCe5iZ7aaqD0b0mLF4CmsuBpl8Eh20ZHkYw2SqO0x9uFrS/gy1vwtkmsTpcDw=' iv = 'DQcmcXyQkU+VKqb2mKmasQ==' pc = WXBizDataCrypt(appId, sessionKey) pc.decrypt(encryptedData, iv)#
\ No newline at end of file
......
utils/wechat/WXBizMsgCrypt.py
# -*- coding: utf-8 -*-
import base64
import hashlib
import random
import socket
import string
import struct
import time
from Crypto.Cipher import AES
import xml.etree.cElementTree as ET
from utils.error_code.wx_error import WXBizMsgCrypt_OK, WXBizMsgCrypt_ParseXml_Error, WXBizMsgCrypt_EncryptAES_Error, \
WXBizMsgCrypt_DecryptAES_Error, WXBizMsgCrypt_IllegalBuffer, WXBizMsgCrypt_ValidateAppid_Error, \
WXBizMsgCrypt_ValidateSignature_Error
class FormatException(Exception):
pass
def throw_exception(message, exception_class=FormatException):
raise exception_class(message)
class SHA1:
"""计算公众平台的消息签名接口"""
def getSHA1(self, token, timestamp, nonce, encrypt):
""" 用SHA1算法生成安全签名
@param token: 票据
@param timestamp: 时间戳
@param encrypt: 密文
@param nonce: 随机字符串
@return: 安全签名
"""
try:
sortlist = [token, timestamp, nonce, encrypt]
sortlist.sort()
sha = hashlib.sha1()
sha.update(("".join(sortlist)).encode('utf-8'))
return 0, sha.hexdigest()
except Exception as e:
print(e)
return -40003, None
class XMLParse:
"""提供提取消息格式中的密文及生成回复消息格式的接口"""
# xml消息模板
AES_TEXT_RESPONSE_TEMPLATE = """<xml>
<Encrypt><![CDATA[%(msg_encrypt)s]]></Encrypt>
<MsgSignature><![CDATA[%(msg_signaturet)s]]></MsgSignature>
<TimeStamp>%(timestamp)s</TimeStamp>
<Nonce><![CDATA[%(nonce)s]]></Nonce>
</xml>"""
def extract(self, xmltext):
""" 提取出xml数据包中的加密消息
@param xmltext: 待提取的xml字符串
@return: 提取出的加密消息字符串
"""
try:
xml_tree = ET.fromstring(xmltext)
encrypt = xml_tree.find("Encrypt")
touser_name = xml_tree.find("ToUserName")
return WXBizMsgCrypt_OK['error_code'], encrypt.text, touser_name.text
except Exception as e:
print(e)
return WXBizMsgCrypt_ParseXml_Error['error_code'], None, None
def generate(self, encrypt, signature, timestamp, nonce):
""" 生成xml消息
@param encrypt: 加密后的消息密文
@param signature: 安全签名
@param timestamp: 时间戳
@param nonce: 随机字符串
@return: 生成的xml字符串
"""
resp_dict = {
'msg_encrypt': encrypt,
'msg_signaturet': signature,
'timestamp': timestamp,
'nonce': nonce,
}
resp_xml = self.AES_TEXT_RESPONSE_TEMPLATE % resp_dict
return resp_xml
class PKCS7Encoder():
"""提供基于PKCS7算法的加解密接口"""
block_size = 32
def encode(self, text):
""" 对需要加密的明文进行填充补位
@param text: 需要进行填充补位操作的明文
@return: 补齐明文字符串
"""
text_length = len(text)
# 计算需要填充的位数
amount_to_pad = self.block_size - (text_length % self.block_size)
if amount_to_pad == 0:
amount_to_pad = self.block_size
# 获得补位所用的字符
pad = chr(amount_to_pad)
return text + pad * amount_to_pad
def decode(self, decrypted):
""" 删除解密后明文的补位字符
@param decrypted: 解密后的明文
@return: 删除补位字符后的明文
"""
pad = ord(decrypted[-1])
if pad < 1 or pad > 32:
pad = 0
return decrypted[:-pad]
class Prpcrypt(object):
"""提供接收和推送给公众平台消息的加解密接口"""
def __init__(self, key):
# self.key = base64.b64decode(key+"=")
self.key = key
# 设置加解密模式为AES的CBC模式
self.mode = AES.MODE_CBC
def encrypt(self, text, appid):
""" 对明文进行加密
@param text: 需要加密的明文
@return: 加密得到的字符串
"""
# 16位随机字符串添加到明文开头
text = self.get_random_str() + str(struct.pack("I", socket.htonl(len(text)))) + text + appid
# 使用自定义的填充方式对明文进行补位填充
pkcs7 = PKCS7Encoder()
text = pkcs7.encode(text)
# 加密
cryptor = AES.new(self.key, self.mode, self.key[:16])
try:
ciphertext = cryptor.encrypt(text)
# 使用BASE64对加密后的字符串进行编码
return WXBizMsgCrypt_OK['error_code'], base64.b64encode(ciphertext)
except Exception as e:
print(e)
return WXBizMsgCrypt_EncryptAES_Error['error_code'], None
def decrypt(self, text, appid):
""" 对解密后的明文进行补位删除
@param text: 密文
@return: 删除填充补位后的明文
"""
try:
cryptor = AES.new(self.key, self.mode, self.key[:16])
# 使用BASE64对密文进行解码，然后AES-CBC解密
plain_text = cryptor.decrypt(base64.b64decode(text))
except Exception as e:
print(e)
return WXBizMsgCrypt_DecryptAES_Error['error_code'], None
try:
pad = ord(plain_text[len(plain_text) - 1:])
# 去掉补位字符串
# pkcs7 = PKCS7Encoder()
# plain_text = pkcs7.encode(plain_text)
# 去除16位随机字符串
content = plain_text[16:-pad]
xml_len = socket.ntohl(struct.unpack("I", content[: 4])[0])
xml_content = str(content[4: xml_len + 4], encoding='utf-8')
from_appid = str(content[xml_len + 4:], encoding='utf-8')
except Exception as e:
print(e)
return WXBizMsgCrypt_IllegalBuffer['error_code'], None
if from_appid != appid:
return WXBizMsgCrypt_ValidateAppid_Error['error_code'], None
return WXBizMsgCrypt_OK['error_code'], xml_content
def get_random_str(self):
""" 随机生成16位字符串
@return: 16位字符串
"""
rule = string.ascii_letters + string.digits
str = random.sample(rule, 16)
return "".join(str)
class WXBizMsgCrypt(object):
# 构造函数
# @param sToken: 公众平台上，开发者设置的Token
# @param sEncodingAESKey: 公众平台上，开发者设置的EncodingAESKey
# @param sAppId: 企业号的AppId
def __init__(self, sToken, sEncodingAESKey, sAppId):
try:
self.key = base64.b64decode(sEncodingAESKey + "=")
assert len(self.key) == 32
except:
throw_exception("[error]: EncodingAESKey unvalid !", FormatException)
self.token = sToken
self.appid = sAppId
def EncryptMsg(self, sReplyMsg, sNonce, timestamp=None):
# 将公众号回复用户的消息加密打包
# @param sReplyMsg: 企业号待回复用户的消息，xml格式的字符串
# @param sTimeStamp: 时间戳，可以自己生成，也可以用URL参数的timestamp,如为None则自动用当前时间
# @param sNonce: 随机串，可以自己生成，也可以用URL参数的nonce
# sEncryptMsg: 加密后的可以直接回复用户的密文，包括msg_signature, timestamp, nonce, encrypt的xml格式的字符串,
# return：成功0，sEncryptMsg,失败返回对应的错误码None
pc = Prpcrypt(self.key)
ret, encrypt = pc.encrypt(sReplyMsg, self.appid)
if ret != 0:
return ret, None
if timestamp is None:
timestamp = str(int(time.time()))
# 生成安全签名
sha1 = SHA1()
ret, signature = sha1.getSHA1(self.token, timestamp, sNonce, encrypt)
if ret != 0:
return ret, None
xmlParse = XMLParse()
return ret, xmlParse.generate(encrypt, signature, timestamp, sNonce)
def DecryptMsg(self, sPostData, sMsgSignature, sTimeStamp, sNonce):
# 检验消息的真实性，并且获取解密后的明文
# @param sMsgSignature: 签名串，对应URL参数的msg_signature
# @param sTimeStamp: 时间戳，对应URL参数的timestamp
# @param sNonce: 随机串，对应URL参数的nonce
# @param sPostData: 密文，对应POST请求的数据
# xml_content: 解密后的原文，当return返回0时有效
# @return: 成功0，失败返回对应的错误码
# 验证安全签名
xmlParse = XMLParse()
encrypt = sPostData["Encrypt"]
touser_name = sPostData["ToUserName"]
sha1 = SHA1()
ret, signature = sha1.getSHA1(self.token, sTimeStamp, sNonce, encrypt)
if ret != 0:
return ret, None
if not signature == sMsgSignature:
return WXBizMsgCrypt_ValidateSignature_Error['error_code'], None
pc = Prpcrypt(self.key)
ret, xml_content = pc.decrypt(encrypt, self.appid)
return ret, xml_content
# -*- coding: utf-8 -*- import base64 import hashlib import random import socket import string import struct import time from Crypto.Cipher import AES import xml.etree.cElementTree as ET from utils.error_code.wx_error import WXBizMsgCrypt_OK, WXBizMsgCrypt_ParseXml_Error, WXBizMsgCrypt_EncryptAES_Error, \ WXBizMsgCrypt_DecryptAES_Error, WXBizMsgCrypt_IllegalBuffer, WXBizMsgCrypt_ValidateAppid_Error, \ WXBizMsgCrypt_ValidateSignature_Error class FormatException(Exception): pass def throw_exception(message, exception_class=FormatException): raise exception_class(message) class SHA1: """计算公众平台的消息签名接口""" def getSHA1(self, token, timestamp, nonce, encrypt): """ 用SHA1算法生成安全签名 @param token: 票据 @param timestamp: 时间戳 @param encrypt: 密文 @param nonce: 随机字符串 @return: 安全签名 """ try: sortlist = [token, timestamp, nonce, encrypt] sortlist.sort() sha = hashlib.sha1() sha.update(("".join(sortlist)).encode('utf-8')) return 0, sha.hexdigest() except Exception as e: print(e) return -40003, None class XMLParse: """提供提取消息格式中的密文及生成回复消息格式的接口""" # xml消息模板 AES_TEXT_RESPONSE_TEMPLATE = """<xml> <Encrypt><![CDATA[%(msg_encrypt)s]]></Encrypt> <MsgSignature><![CDATA[%(msg_signaturet)s]]></MsgSignature> <TimeStamp>%(timestamp)s</TimeStamp> <Nonce><![CDATA[%(nonce)s]]></Nonce> </xml>""" def extract(self, xmltext): """ 提取出xml数据包中的加密消息 @param xmltext: 待提取的xml字符串 @return: 提取出的加密消息字符串 """ try: xml_tree = ET.fromstring(xmltext) encrypt = xml_tree.find("Encrypt") touser_name = xml_tree.find("ToUserName") return WXBizMsgCrypt_OK['error_code'], encrypt.text, touser_name.text except Exception as e: print(e) return WXBizMsgCrypt_ParseXml_Error['error_code'], None, None def generate(self, encrypt, signature, timestamp, nonce): """ 生成xml消息 @param encrypt: 加密后的消息密文 @param signature: 安全签名 @param timestamp: 时间戳 @param nonce: 随机字符串 @return: 生成的xml字符串 """ resp_dict = { 'msg_encrypt': encrypt, 'msg_signaturet': signature, 'timestamp': timestamp, 'nonce': nonce, } resp_xml = self.AES_TEXT_RESPONSE_TEMPLATE % resp_dict return resp_xml class PKCS7Encoder(): """提供基于PKCS7算法的加解密接口""" block_size = 32 def encode(self, text): """ 对需要加密的明文进行填充补位 @param text: 需要进行填充补位操作的明文 @return: 补齐明文字符串 """ text_length = len(text) # 计算需要填充的位数 amount_to_pad = self.block_size - (text_length % self.block_size) if amount_to_pad == 0: amount_to_pad = self.block_size # 获得补位所用的字符 pad = chr(amount_to_pad) return text + pad * amount_to_pad def decode(self, decrypted): """ 删除解密后明文的补位字符 @param decrypted: 解密后的明文 @return: 删除补位字符后的明文 """ pad = ord(decrypted[-1]) if pad < 1 or pad > 32: pad = 0 return decrypted[:-pad] class Prpcrypt(object): """提供接收和推送给公众平台消息的加解密接口""" def __init__(self, key): # self.key = base64.b64decode(key+"=") self.key = key # 设置加解密模式为AES的CBC模式 self.mode = AES.MODE_CBC def encrypt(self, text, appid): """ 对明文进行加密 @param text: 需要加密的明文 @return: 加密得到的字符串 """ # 16位随机字符串添加到明文开头 text = self.get_random_str() + str(struct.pack("I", socket.htonl(len(text)))) + text + appid # 使用自定义的填充方式对明文进行补位填充 pkcs7 = PKCS7Encoder() text = pkcs7.encode(text) # 加密 cryptor = AES.new(self.key, self.mode, self.key[:16]) try: ciphertext = cryptor.encrypt(text) # 使用BASE64对加密后的字符串进行编码 return WXBizMsgCrypt_OK['error_code'], base64.b64encode(ciphertext) except Exception as e: print(e) return WXBizMsgCrypt_EncryptAES_Error['error_code'], None def decrypt(self, text, appid): """ 对解密后的明文进行补位删除 @param text: 密文 @return: 删除填充补位后的明文 """ try: cryptor = AES.new(self.key, self.mode, self.key[:16]) # 使用BASE64对密文进行解码，然后AES-CBC解密 plain_text = cryptor.decrypt(base64.b64decode(text)) except Exception as e: print(e) return WXBizMsgCrypt_DecryptAES_Error['error_code'], None try: pad = ord(plain_text[len(plain_text) - 1:]) # 去掉补位字符串 # pkcs7 = PKCS7Encoder() # plain_text = pkcs7.encode(plain_text) # 去除16位随机字符串 content = plain_text[16:-pad] xml_len = socket.ntohl(struct.unpack("I", content[: 4])[0]) xml_content = str(content[4: xml_len + 4], encoding='utf-8') from_appid = str(content[xml_len + 4:], encoding='utf-8') except Exception as e: print(e) return WXBizMsgCrypt_IllegalBuffer['error_code'], None if from_appid != appid: return WXBizMsgCrypt_ValidateAppid_Error['error_code'], None return WXBizMsgCrypt_OK['error_code'], xml_content def get_random_str(self): """ 随机生成16位字符串 @return: 16位字符串 """ rule = string.ascii_letters + string.digits str = random.sample(rule, 16) return "".join(str) class WXBizMsgCrypt(object): # 构造函数 # @param sToken: 公众平台上，开发者设置的Token # @param sEncodingAESKey: 公众平台上，开发者设置的EncodingAESKey # @param sAppId: 企业号的AppId def __init__(self, sToken, sEncodingAESKey, sAppId): try: self.key = base64.b64decode(sEncodingAESKey + "=") assert len(self.key) == 32 except: throw_exception("[error]: EncodingAESKey unvalid !", FormatException) self.token = sToken self.appid = sAppId def EncryptMsg(self, sReplyMsg, sNonce, timestamp=None): # 将公众号回复用户的消息加密打包 # @param sReplyMsg: 企业号待回复用户的消息，xml格式的字符串 # @param sTimeStamp: 时间戳，可以自己生成，也可以用URL参数的timestamp,如为None则自动用当前时间 # @param sNonce: 随机串，可以自己生成，也可以用URL参数的nonce # sEncryptMsg: 加密后的可以直接回复用户的密文，包括msg_signature, timestamp, nonce, encrypt的xml格式的字符串, # return：成功0，sEncryptMsg,失败返回对应的错误码None pc = Prpcrypt(self.key) ret, encrypt = pc.encrypt(sReplyMsg, self.appid) if ret != 0: return ret, None if timestamp is None: timestamp = str(int(time.time())) # 生成安全签名 sha1 = SHA1() ret, signature = sha1.getSHA1(self.token, timestamp, sNonce, encrypt) if ret != 0: return ret, None xmlParse = XMLParse() return ret, xmlParse.generate(encrypt, signature, timestamp, sNonce) def DecryptMsg(self, sPostData, sMsgSignature, sTimeStamp, sNonce): # 检验消息的真实性，并且获取解密后的明文 # @param sMsgSignature: 签名串，对应URL参数的msg_signature # @param sTimeStamp: 时间戳，对应URL参数的timestamp # @param sNonce: 随机串，对应URL参数的nonce # @param sPostData: 密文，对应POST请求的数据 # xml_content: 解密后的原文，当return返回0时有效 # @return: 成功0，失败返回对应的错误码 # 验证安全签名 xmlParse = XMLParse() encrypt = sPostData["Encrypt"] touser_name = sPostData["ToUserName"] sha1 = SHA1() ret, signature = sha1.getSHA1(self.token, sTimeStamp, sNonce, encrypt) if ret != 0: return ret, None if not signature == sMsgSignature: return WXBizMsgCrypt_ValidateSignature_Error['error_code'], None pc = Prpcrypt(self.key) ret, xml_content = pc.decrypt(encrypt, self.appid) return ret, xml_content
\ No newline at end of file
......
utils/wechat/__init__.py
#!usr/bin/.env python
# -*- coding:utf-8 _*-
"""
@version:
author:Aeolus
@file: __init__.py.py
"""
\ No newline at end of file
#!usr/bin/.env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @file: __init__.py.py """
\ No newline at end of file
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment