315 lines
12 KiB
Python
315 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Paytm Business 登录 API
|
||
支持：OTP 登录、Token 保存/读取、获取 QR 数据、查询流水
|
||
"""
|
||
import requests
|
||
import json
|
||
import os
|
||
from datetime import datetime, timedelta
|
||
|
||
TOKEN_FILE = ".paytm_business_token.json"
|
||
|
||
class PaytmBusinessAPI:
|
||
def __init__(self):
|
||
self.base_url = "https://accounts.paytm.com"
|
||
self.dashboard_url = "https://dashboard.paytm.com"
|
||
self.token = "cDRiLW13ZWItcHJvZHVjdGlvbjplTU01VWpXaW9wNnNFVGwzcDBpcGRvZ0hJdENtTXNibA=="
|
||
self.client_id = "p4b-mweb-production"
|
||
self.session = requests.Session()
|
||
self.csrf_token = None
|
||
self.state = None
|
||
self.session_cookie = None
|
||
self.xsrf_token = None
|
||
self.qr_data = []
|
||
|
||
def _get_headers(self):
|
||
"""通用请求头"""
|
||
return {
|
||
"Authorization": f"Basic {self.token}",
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "Mozilla/5.0 (Linux; Android 12; M2004J7AC Build/SP1A.210812.016; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/143.0.7499.192 Mobile Safari/537.36",
|
||
"Accept": "*/*",
|
||
"Origin": "https://accounts.paytm.com",
|
||
"Referer": "https://accounts.paytm.com/oauth-js-sdk/index.html",
|
||
"Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
|
||
"sec-ch-ua": '"Android WebView";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
|
||
"sec-ch-ua-mobile": "?1",
|
||
"sec-ch-ua-platform": '"Android"',
|
||
"Sec-Fetch-Site": "same-origin",
|
||
"Sec-Fetch-Mode": "cors",
|
||
"Sec-Fetch-Dest": "empty",
|
||
}
|
||
|
||
def save_token(self):
|
||
"""保存 token 到文件"""
|
||
if not self.session_cookie or not self.xsrf_token:
|
||
return
|
||
|
||
data = {
|
||
"session_cookie": self.session_cookie,
|
||
"xsrf_token": self.xsrf_token,
|
||
"qr_data": self.qr_data,
|
||
"saved_at": datetime.now().isoformat()
|
||
}
|
||
|
||
with open(TOKEN_FILE, 'w') as f:
|
||
json.dump(data, f, indent=2)
|
||
print(f"✓ Token 已保存到 {TOKEN_FILE}")
|
||
|
||
def load_token(self):
|
||
"""从文件加载 token"""
|
||
if not os.path.exists(TOKEN_FILE):
|
||
print(f"Token 文件不存在: {TOKEN_FILE}")
|
||
return False
|
||
|
||
try:
|
||
with open(TOKEN_FILE, 'r') as f:
|
||
data = json.load(f)
|
||
|
||
self.session_cookie = data.get("session_cookie")
|
||
self.xsrf_token = data.get("xsrf_token")
|
||
self.qr_data = data.get("qr_data", [])
|
||
|
||
print(f"✓ 已加载 Token (保存于 {data.get('saved_at')})")
|
||
|
||
# 验证 token 是否有效
|
||
if self.verify_token():
|
||
print("✓ Token 有效")
|
||
return True
|
||
else:
|
||
print("✗ Token 已失效")
|
||
return False
|
||
except Exception as e:
|
||
print(f"✗ 加载 Token 失败: {e}")
|
||
return False
|
||
|
||
def _get_dashboard_headers(self):
|
||
"""Dashboard 请求头"""
|
||
cookie_header = self.session_cookie
|
||
if not cookie_header.startswith("SESSION="):
|
||
cookie_header = "SESSION=" + cookie_header
|
||
return {
|
||
"Cookie": cookie_header,
|
||
"X-XSRF-TOKEN": self.xsrf_token,
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
"User-Agent": "Mozilla/5.0 (Linux; Android 12) AppleWebKit/537.36",
|
||
"Origin": "https://dashboard.paytm.com",
|
||
"Referer": "https://dashboard.paytm.com/next/transactions",
|
||
"x-ump-version": "bpay-v2.26.2-12188-g89732e0ebb",
|
||
}
|
||
|
||
def verify_token(self):
|
||
"""验证 token 是否有效"""
|
||
try:
|
||
url = f"{self.dashboard_url}/api/v1/context"
|
||
resp = requests.get(url, headers=self._get_dashboard_headers(), timeout=10)
|
||
return resp.status_code == 200
|
||
except:
|
||
return False
|
||
|
||
def init(self):
|
||
"""初始化，获取 csrfToken"""
|
||
url = f"{self.base_url}/um/authorize/init"
|
||
data = {
|
||
"clientId": self.client_id,
|
||
"responseType": "code",
|
||
"scope": "paytm",
|
||
"redirectUri": "https://dashboard.paytm.com/auth"
|
||
}
|
||
|
||
resp = self.session.post(url, headers=self._get_headers(), json=data)
|
||
result = resp.json()
|
||
|
||
if result.get("status") == "SUCCESS":
|
||
self.csrf_token = result["data"]["authState"]
|
||
print(f"✓ 初始化成功，csrfToken: {self.csrf_token}")
|
||
return self.csrf_token
|
||
else:
|
||
raise Exception(f"初始化失败: {result}")
|
||
|
||
def request_otp(self, mobile, password):
|
||
"""请求 OTP"""
|
||
if not self.csrf_token:
|
||
self.init()
|
||
|
||
url = f"{self.base_url}/um/authorize/proceed"
|
||
data = {
|
||
"userName": mobile,
|
||
"password": password,
|
||
"clientId": self.client_id,
|
||
"csrfToken": self.csrf_token
|
||
}
|
||
|
||
resp = self.session.post(url, headers=self._get_headers(), json=data)
|
||
result = resp.json()
|
||
|
||
if result.get("status") == "SUCCESS":
|
||
self.state = result.get("stateCode")
|
||
print(f"✓ OTP 已发送到 {mobile}")
|
||
return result
|
||
else:
|
||
raise Exception(f"请求 OTP 失败: {result.get('message')}")
|
||
|
||
def verify_otp(self, otp):
|
||
"""验证 OTP，并跟进 redirect 获取 SESSION cookie"""
|
||
if not self.csrf_token or not self.state:
|
||
raise Exception("请先请求 OTP")
|
||
|
||
url = f"{self.base_url}/login/validate/otp"
|
||
data = {
|
||
"otp": otp,
|
||
"state": self.state,
|
||
"csrfToken": self.csrf_token
|
||
}
|
||
|
||
resp = self.session.post(url, headers=self._get_headers(), json=data)
|
||
|
||
try:
|
||
result = resp.json()
|
||
except:
|
||
raise Exception(f"响应格式错误: {resp.text[:200]}")
|
||
|
||
redirect_uri = result.get("redirectUri")
|
||
if not redirect_uri:
|
||
raise Exception(f"OTP 验证失败: {result.get('message')} (code: {result.get('responseCode')})")
|
||
|
||
print(f"✓ OTP 验证成功，跟进 redirect...")
|
||
|
||
# 跟进 redirect 到 dashboard，获取 SESSION 和 XSRF-TOKEN
|
||
dashboard_headers = {
|
||
"User-Agent": "Mozilla/5.0 (Linux; Android 12) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Mobile Safari/537.36",
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
}
|
||
auth_resp = self.session.get(redirect_uri, headers=dashboard_headers, allow_redirects=True)
|
||
print(f"✓ Dashboard 响应: {auth_resp.status_code} ({auth_resp.url})")
|
||
|
||
for cookie in self.session.cookies:
|
||
if cookie.name == "SESSION":
|
||
self.session_cookie = cookie.value
|
||
elif cookie.name == "XSRF-TOKEN":
|
||
self.xsrf_token = cookie.value
|
||
|
||
if not self.session_cookie or not self.xsrf_token:
|
||
raise Exception(f"未能获取 SESSION/XSRF-TOKEN，cookies: {dict(self.session.cookies)}")
|
||
|
||
print(f"✓ SESSION: {self.session_cookie[:30]}...")
|
||
print(f"✓ XSRF-TOKEN: {self.xsrf_token[:30]}...")
|
||
return result
|
||
|
||
def fetch_qr_data(self):
|
||
"""获取 QR 码数据"""
|
||
if not self.session_cookie or not self.xsrf_token:
|
||
raise Exception("请先登录")
|
||
|
||
url = f"{self.dashboard_url}/api/v4/qrcode/fetch/?pageNo=1&pageSize=100"
|
||
resp = requests.get(url, headers=self._get_dashboard_headers())
|
||
|
||
if resp.status_code == 401:
|
||
raise Exception("Token 已失效，需要重新登录")
|
||
|
||
result = resp.json()
|
||
|
||
if resp.status_code == 200 and result.get("response") is not None:
|
||
self.qr_data = result.get("response", [])
|
||
print(f"✓ 获取到 {len(self.qr_data)} 个 QR 码")
|
||
if self.qr_data:
|
||
print(f" VPA: {self.qr_data[0].get('vpa')}")
|
||
return self.qr_data
|
||
else:
|
||
raise Exception(f"获取 QR 数据失败: {result}")
|
||
|
||
def get_transactions(self, start_date=None, end_date=None, page_num=1, page_size=50):
|
||
"""获取交易流水"""
|
||
if not self.session_cookie or not self.xsrf_token:
|
||
raise Exception("请先登录")
|
||
|
||
if not end_date:
|
||
end_date = datetime.now()
|
||
if not start_date:
|
||
start_date = end_date - timedelta(days=1)
|
||
|
||
url = f"{self.dashboard_url}/api/v3/order/list"
|
||
data = {
|
||
"bizTypeList": ["ACQUIRING", "CASHBACK", "SPLIT_PAYMENT"],
|
||
"pageSize": page_size,
|
||
"pageNum": page_num,
|
||
"orderCreatedStartTime": start_date.strftime("%Y-%m-%d") + "T00:00:00+05:30",
|
||
"orderCreatedEndTime": end_date.strftime("%Y-%m-%d") + "T23:59:59+05:30",
|
||
"orderStatusList": ["SUCCESS"],
|
||
"isSort": True
|
||
}
|
||
|
||
resp = requests.post(url, headers=self._get_dashboard_headers(), json=data)
|
||
|
||
if resp.status_code == 401:
|
||
raise Exception("Token 已失效，需要重新登录")
|
||
|
||
result = resp.json()
|
||
result_info = result.get("resultInfo", {})
|
||
|
||
if result_info.get("resultStatus") == "F":
|
||
raise Exception(f"Paytm API error: code={result_info.get('resultCode')}, msg={result_info.get('resultMsg')}")
|
||
|
||
orders = result.get("orderList", [])
|
||
print(f"✓ 获取到 {len(orders)} 笔交易")
|
||
return orders
|
||
|
||
def main():
|
||
api = PaytmBusinessAPI()
|
||
|
||
# 示例账号
|
||
mobile = "7528905079"
|
||
password = "Kular@500"
|
||
|
||
try:
|
||
# 1. 尝试加载已保存的 token
|
||
if api.load_token():
|
||
print("\n使用已保存的 Token")
|
||
else:
|
||
print("\n需要重新登录")
|
||
# 2. 初始化
|
||
api.init()
|
||
|
||
# 3. 请求 OTP
|
||
api.request_otp(mobile, password)
|
||
|
||
# 4. 验证 OTP
|
||
otp = input("请输入收到的 OTP: ")
|
||
api.verify_otp(otp)
|
||
|
||
# 5. 获取 QR 数据
|
||
api.fetch_qr_data()
|
||
|
||
# 6. 保存 token
|
||
api.save_token()
|
||
|
||
# 7. 获取流水
|
||
print("\n查询最近1天的交易流水...")
|
||
transactions = api.get_transactions()
|
||
|
||
if transactions:
|
||
print(f"\n最新 5 笔交易:")
|
||
for i, txn in enumerate(transactions[:5], 1):
|
||
amount = float(txn.get("payMoneyAmount", {}).get("value", 0)) / 100
|
||
customer = txn.get("additionalInfo", {}).get("customerName", "Unknown")
|
||
upi = txn.get("additionalInfo", {}).get("virtualPaymentAddr", "")
|
||
order_id = txn.get("bizOrderId", "")
|
||
print(f"{i}. {txn.get('orderCreatedTime')} | ₹{amount:.2f} | {customer} ({upi}) | {order_id}")
|
||
|
||
txn_file = ".paytm_business_transactions.json"
|
||
with open(txn_file, 'w') as f:
|
||
json.dump(transactions, f, indent=2, ensure_ascii=False)
|
||
print(f"\n✓ {len(transactions)} 笔交易已保存到 {txn_file}")
|
||
else:
|
||
print("暂无交易记录")
|
||
|
||
except Exception as e:
|
||
print(f"\n✗ 错误: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|