拉取阿里云近一小时新订单并发送到钉钉

#1. python3代码

#!/usr/bin/env python
# coding=utf-8
# 拉取阿里近一小时的新订单详情
import json
import time
import requests
from datetime import datetime, date
from datetime import timedelta
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkbssopenapi.request.v20171214.QueryOrdersRequest import QueryOrdersRequest


# 获取 n 天前时间，并格式化为api指定格式时间戳
def get_date(days=0.0):
    return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:00Z")


# 钉钉
def postdingding(acount_name, id, amount, code, paytime):
    HEADERS = {"Content-Type": "application/json ;charset=utf-8"}
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "%s新订单 " % acount_name,
            "text": "%s新订单\n\n  订单编号： %s\n\n 订单金额:  %s\n\n 产品类型:  %s\n\n 支付时间:  %s" % (
                acount_name, id, amount, code, paytime)
        },
        "at": {
            "atMobiles": [
                ""
            ],
            "isAtAll": "true"
        }
    }
    String_textMsg = json.dumps(data)
    print(data)
    # 钉钉地址
    dingdingurl = 'https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxxxxxxxxxxxxxxxxxx'
    res = requests.post(dingdingurl, data=String_textMsg, headers=HEADERS)
    print(res.text)


def req(account_name, access_key_id, access_secret):
    client = AcsClient(access_key_id, access_secret, 'cn-hangzhou')
    request = QueryOrdersRequest()
    request.set_accept_format('json')
    # 阿里云api使用utc时间，本地时间为utc+8, 故往前推8小时，即0.333
    request.set_CreateTimeEnd(get_date(0.333))
    request.set_PageNum(1)
    request.set_PageSize(100)
    request.set_OrderType("New")
    # 前推9小时
    request.set_CreateTimeStart(get_date(0.375))

    response = client.do_action_with_exception(request)
    res = str(response, encoding='utf-8')
    data = json.loads(res)

    for order in data["Data"]["OrderList"]["Order"]:
        ProductCode = order["ProductCode"]
        PretaxAmount = order['PretaxAmount']
        SubscriptionType = order['SubscriptionType']
        PaymentTime = order['PaymentTime']
        timeArray = time.strptime(PaymentTime, "%Y-%m-%dT%H:%M:%SZ")
        timeStamp = int(time.mktime(timeArray))
        # 换回utc+8 时间
        timeStamp += 28800
        time_local = time.localtime(timeStamp)
        PaymentTime = time.strftime("%Y-%m-%d %H:%M:%S", time_local)
        OrderId = order['OrderId']
        PaymentStatus = order['PaymentStatus']
        if SubscriptionType == 'Subscription' and PaymentStatus == 'Paid':
            postdingding(account_name, OrderId, PretaxAmount, ProductCode, PaymentTime)


def main():
    # key.txt 为本地保存的格式为  [账户名] [access_key_id]  [access_secret]  的文件（每行三个参数用任意个空格隔开,如: name id key )
    with open('key.txt', 'r', encoding='utf-8') as f:
        data = f.readlines()
        for str in data:
            i = [n for n in str.split()]
            req(i[0], i[1], i[2])


if __name__ == "__main__":
    main()

#2. 定时脚本

每小时的59分执行脚本

#crontab -l
59 * * * * /root/daily_monitor/bin/python3 /root/daily_monitor/ali_new_order.py