一、需求描述

获取收钱吧指定日期的收款金额和明细值

sqb_login 函数：模拟登陆，返回值为可用于验证的 token
get_bill_details 函数：通过token获取传入日期参数的当日收款金额，第一个返回值为总额，第二个返回值为金额明细的列表
wechat_send 函数：发送告警消息到 server酱 微信公众号

使用文档参考官网：https://sct.ftqq.com/

二、源码

import requests
import json
from datetime import datetime 
import hashlib
import argparse
from dateutil.relativedelta import relativedelta  


header = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36',
    'Content-Type': 'application/json'
}

# 模拟登陆收钱吧 网页版 接口，返回可用的token
def sqb_login(username, password):
    # 密码转换为md5值
    md5_hash = hashlib.md5()
    md5_hash.update(password.encode('utf-8'))  
    post_url = "https://web-platforms-msp.shouqianba.com/api/login/ucUser/login"
    password = md5_hash.hexdigest()
    
    payload = json.dumps({
        "username": username,
        "password": password,
        "uc_device": {
            "device_type": 2,
            "default_device": 0,
            "platform": "商户服务平台",
            "device_fingerprint": "d058065e-68e3-4b24-97a2-d714a8cb5a59",
            "device_name": "收钱吧商户平台",
            "device_model": "Windows",
            "device_brand": "Chrome"
        }
    })
    
    response = requests.post(post_url, data=payload, headers=header)
    try:
        if response.status_code == 200:
            results = json.loads(response.text)
            token = results['data']['mchUserTokenInfo']['token']
            return token
        else:
            print('登录失败')
    except Exception as e:
        print(e)

# 获取传参日期的收款金额和明细
def get_bill_details(date_str, token):

    url = f"https://web-platforms-msp.shouqianba.com/api/transaction/findTransactions?token={token}"
    
    date_start_timestamp = int(datetime.strptime(date_str, "%Y-%m-%d").timestamp() * 1000)
    date_end_timestamp = int(datetime.strptime(date_str + ' 23:59:59', "%Y-%m-%d %H:%M:%S").timestamp() * 1000) + 999

    payload = json.dumps({
        # 这里的page和page_size 要根据实际数量自己调整一下，不然收集的数据不全，对应的金额对不上
        "upayQueryType": 0,
        "page": 1,
        "page_size": 200,
        "date_start": date_start_timestamp,
        "date_end": date_end_timestamp
    })
    headers = header

    try:
        response = requests.request("POST", url, headers=headers, data=payload)
        results = json.loads(response.text)
        
        amount_list = []
        for result in results['data']['records']:
            amount_list.append(result['actual_receive_amount'] / 100)
        
        total_amount = sum(amount_list)
        return total_amount
    except Exception as e:
        print(e)

# 获取月度账单
def get_month_bill(start_date, end_date, token):
    url = f"https://mapi.shouqianba.com/v4/account_report/getAccountRecordReportProxy?token={token}&client_version=default_client_version"
    payload = json.dumps({
        "startDate": start_date,
        "endDate": end_date,
        "offsetHour": 0,
        "storeIds": "",
        "reportLoadTypes": "TIME_MONTH_SUMMARY",
        "marketReportLoadTypes": "TIME_MONTH_SUMMARY",
        "source": "msp",
        "size": 500
    })
    
    try:
        response = requests.request("POST", url, headers=header, data=payload)
        results = json.loads(response.text)
        
        result = results['data']['incomeTimeUnitReports'][0]['clearingAmount'] / 100
        return result
    except Exception as e:
        print(e)
        
# 根据type标签获取当日或者上月收入情况，type = today or month
def main(type):
    # 获取当前日期和时间
    now = datetime.now()
    # 计算上个月的日期
    first_day_of_this_month = now.replace(day=1)  
    last_month = now.replace(month=now.month-1)
    last_month_str = last_month.strftime('%Y-%m')
    last_month_start_day = (now - relativedelta(months=1)).replace(day=1).strftime("%Y-%m-%d")
    last_month_end_day = (first_day_of_this_month - relativedelta(days=1)).strftime("%Y-%m-%d")
    
    # 初始化token
    token = sqb_login('username', 'password')
    today = datetime.now().strftime("%Y-%m-%d")
    if type == 'today':
        today_amounts = get_bill_details(today, token)
        try:
            # pushplus token
            response = requests.get(f"https://www.pushplus.plus/send?token=<your_token>&title={today}到账金额&content={today_amounts}元&template=html")
            if response.status_code == 200:
                print(f"消息推送成功: {today_amounts} 元")
                print(response.text)
            else:
                print('消息推送失败')
        except Exception as e:
                print(e)
    elif type == 'month':
        month_amounts = get_month_bill(start_date=last_month_start_day, end_date=last_month_end_day, token=token)
        try:
            # pushplus token
            response = requests.get(f"https://www.pushplus.plus/send?token=<your_token>&title={last_month_str}月度到账金额&content={month_amounts}元&template=html")
            if response.status_code == 200:
                print(f"消息推送成功: {month_amounts} 元")
                print(response.text)
            else:
                print('消息推送失败')
        except Exception as e:
                print(e)
            
        
if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="ShouQianBa Bill")
    parser.add_argument("type", help="Amounts Type")

    args = parser.parse_args()

    main(args.type)