python----/智慧校园/本部/消费数据/24-26消费信息/demo.py

from requests import get
import time
import os
import pandas as pd
import tqdm
from datetime import datetime, timedelta

teamId = 19
ip = "http://band.hxzhxy.cn"
param = {
    "username": "xtl",
    "password": "xia123456",
    "client_id": "client",
    "grant_type": "password",
    "client_secret": "123456",
}

data = get(url=f"{ip}/oauth/token", params=param).json()
token = data["access_token"]
header = {
    "content-type": "application/json",
    "authorization": f"Bearer {token}",
}

url = f"{ip}/pay/trads/findPayTtadsNews"
output_dir = r"e:\project\python\工具\智慧校园\本部\消费数据\24-26消费信息"

FIELD_MAP = {
    "gradeName": "年级名称",
    "termName": "交易地点",
    "teamId": "团队ID",
    "stuId": "学生ID",
    "areaName": "设备区域名称",
    "bagName": "交易账户",
    "userNumb": "学号/工号",
    "dealCount": "交易流水号/笔次",
    "className": "班级名称",
    "remark": "备注/充值类型",
    "cardCode": "物理卡号",
    "dealType": "消费方式",
    "cardValue": "余额",
    "userXm": "姓名",
    "cardNo": "卡号",
    "createDate": "入库时间",
    "dealValue": "交易金额",
    "recordType": "记录类型",
    "id": "记录ID",
    "recordId": "终端原始记录ID",
    "userId": "用户唯一ID",
    "dealTime": "交易时间"
}

OUTPUT_COLUMNS = list(FIELD_MAP.values())


def day_range(start_date, end_date):
    current = start_date
    while current <= end_date:
        yield current
        current += timedelta(days=1)


def month_range(start_date, end_date):
    current = datetime(start_date.year, start_date.month, 1)
    end_month = datetime(end_date.year, end_date.month, 1)
    while current <= end_month:
        yield current
        if current.month == 12:
            current = datetime(current.year + 1, 1, 1)
        else:
            current = datetime(current.year, current.month + 1, 1)


def to_timestamp_ms(dt):
    return int(dt.timestamp() * 1000)


def fetch_day_records(day_dt):
    day_start = datetime(day_dt.year, day_dt.month, day_dt.day, 0, 0, 0)
    day_end = datetime(day_dt.year, day_dt.month, day_dt.day, 23, 59, 59)

    begin_time = to_timestamp_ms(day_start)
    end_time = to_timestamp_ms(day_end)

    page_number = 0
    page_size = 100
    day_records = []

    while True:
        params = {
            "teamId": teamId,
            "pageNumber": page_number,
            "pageSize": page_size,
            "screen": 1,
            "beginTime": begin_time,
            "endTime": end_time,
            "name": "",
            "recordType": 1,
            "termName": "",
            "type": 1,
            "classId": "",
            "gradeId": ""
        }

        resp = get(url=url, params=params, headers=header, timeout=60)
        resp.raise_for_status()
        result = resp.json()

        if not result.get("success"):
            raise Exception(f"接口返回失败: {result}")

        data = result.get("data") or {}
        content = data.get("content") or []
        total_pages = data.get("totalPages", 0)
        total_elements = data.get("totalElements", 0)

        day_records.extend(content)

        print(
            f"{day_dt.strftime('%Y-%m-%d')} 第 {page_number + 1}/{max(total_pages, 1)} 页，"
            f"当前页 {len(content)} 条，当天累计 {len(day_records)} 条，总计 {total_elements} 条"
        )

        page_number += 1
        if page_number >= total_pages or total_pages == 0:
            break

        time.sleep(0.1)

    return day_records


def parse_date(date_str):
    return datetime.strptime(date_str, "%Y-%m-%d")


def convert_records_to_df(records):
    rows = []
    for record in records:
        row = {}
        for en_key, cn_key in FIELD_MAP.items():
            row[cn_key] = record.get(en_key)
        rows.append(row)
    if not rows:
        return pd.DataFrame(columns=OUTPUT_COLUMNS)
    return pd.DataFrame(rows, columns=OUTPUT_COLUMNS)


def get_month_start_end(month_dt, start_date, end_date):
    month_start = datetime(month_dt.year, month_dt.month, 1)
    if month_dt.month == 12:
        next_month = datetime(month_dt.year + 1, 1, 1)
    else:
        next_month = datetime(month_dt.year, month_dt.month + 1, 1)
    month_end = next_month - timedelta(days=1)

    if month_start < start_date:
        month_start = start_date
    if month_end > end_date:
        month_end = end_date

    return month_start, month_end


def save_month_excel(month_dt, records):
    df = convert_records_to_df(records)
    file_name = f"{month_dt.strftime('%Y%m')}_学生消费记录.xlsx"
    file_path = os.path.join(output_dir, file_name)
    df.to_excel(file_path, index=False)
    return file_path, len(df)


def main():
    print("开始拉取数据...")
    start_date_str = os.getenv("START_DATE", "2024-01-01")
    end_date_str = os.getenv("END_DATE", "2026-05-19")

    start_date = parse_date(start_date_str)
    end_date = parse_date(end_date_str)

    all_month_list = list(month_range(start_date, end_date))
    total_records = 0
    success_months = 0

    with tqdm.tqdm(all_month_list, desc="按月拉取消费记录") as month_pbar:
        for month_dt in month_pbar:
            try:
                month_start, month_end = get_month_start_end(month_dt, start_date, end_date)
                month_records = []
                month_days = list(day_range(month_start, month_end))

                for day_dt in month_days:
                    day_records = fetch_day_records(day_dt)
                    month_records.extend(day_records)
                    time.sleep(0.2)

                file_path, count = save_month_excel(month_dt, month_records)
                total_records += count
                success_months += 1
                month_pbar.set_postfix({
                    "month": month_dt.strftime("%Y-%m"),
                    "count": count,
                    "total": total_records
                })
                print(f"已输出: {file_path}")
            except Exception as e:
                print(f"{month_dt.strftime('%Y-%m')} 拉取失败: {e}")

    print("\n完成")
    print(f"成功月份: {success_months}")
    print(f"总记录数: {total_records}")
    print(f"输出目录: {output_dir}")


if __name__ == "__main__":
    main()