#!/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()