Python 登錄移動查詢話費

通過python爬蟲方式登錄10086后臺，查詢話費。# -*- coding: utf-8 -*-

# @Time : 2019-02-22 09:52

# @Author : cxa

# @File : beijing_crawler.py

# @Software: PyCharm

import requests

import time

import base64

from Cryptodome.Cipher import PKCS1_v1_5

from Cryptodome.PublicKey import RSA

from io import BytesIO

import random

from PIL import Image

import re

class BeiJingCrawler():

headers = {

"Accept": "application/json, text/javascript, */*; q=0.01",

"Accept-Encoding": "gzip, deflate, br",

"Accept-Language": "zh-CN,zh;q=0.9",

"Connection": "keep-alive",

"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",

"Host": "login.10086.cn",

"Origin": "https://login.10086.cn",

"Referer": "https://login.10086.cn/html/bj/iloginnew.html?1551250969321",

"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.109 Safari/537.36",

"X-Requested-With": "XMLHttpRequest",

}

def __init__(self, phone):

self.session = requests.Session()

self.phone_number = phone

def get_timestamp(self):

return int(time.time() * 1000)

def check_uid_available(self):

url = "https://login.10086.cn/checkUidAvailable.action"

req = self.session.get(url, headers=self.headers)

return req

def chk_number_action(self):

'''

檢查是否為北京移動，是返回true,不是返回flase

:return:

'''

url = "https://login.10086.cn/chkNumberBjAction.action"

data = {

"userName": self.phone_number

}

req = self.session.post(url, headers=self.headers, data=data)

return req

def loadToken(self):

url = "https://login.10086.cn/loadToken.action"

data = {

"userName": self.phone_number

}

req = self.session.post(url, headers=self.headers, data=data)

return req

def sendRandomCodeAction(self, result):

self.headers["Xa-before"] = result

url = "https://login.10086.cn/sendRandomCodeAction.action"

data = {

"userName": self.phone_number,

"type": "POST",

"channelID": "00100"

}

req = self.session.post(url, headers=self.headers, data=data)

return req

def sendflag(self, ):

t = self.get_timestamp()

url = "https://login.10086.cn/sendflag.htm"

data = {

"timestamp": t

}

req = self.session.get(url, headers=self.headers, params=data)

return req

def send_sms_code(self):

r = self.check_uid_available()

code = r.json().get("code")

if code == "1":

r1 = self.chk_number_action()

flag = r1.text

if flag:

r2 = self.loadToken()

token = r2.json().get("result")

print("token", token)

r4 = self.sendflag()

print("r4", r4.text)

r3 = self.sendRandomCodeAction(token)

if r3.text == "0":

print("驗證碼發送成功!")

elif r3.text == "1":

print("驗證碼發送頻繁,請一分鐘后再重試！")

elif r3.text == "2":

print("驗證碼發送頻繁,已經超過最大次數")

def get_password(self, k):

rsa_publickey = '''-----BEGIN PUBLIC KEY-----

MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAsgDq4OqxuEisnk2F0EJF

mw4xKa5IrcqEYHvqxPs2CHEg2kolhfWA2SjNuGAHxyDDE5MLtOvzuXjBx/5YJtc9

zj2xR/0moesS+Vi/xtG1tkVaTCba+TV+Y5C61iyr3FGqr+KOD4/XECu0Xky1W9Zm

maFADmZi7+6gO9wjgVpU9aLcBcw/loHOeJrCqjp7pA98hRJRY+MML8MK15mnC4eb

ooOva+mJlstW6t/1lghR8WNV8cocxgcHHuXBxgns2MlACQbSdJ8c6Z3RQeRZBzyj

fey6JCCfbEKouVrWIUuPphBL3OANfgp0B+QG31bapvePTfXU48TYK0M5kE+8Lgbb

WQIDAQAB

-----END PUBLIC KEY-----'''

key = RSA.import_key(rsa_publickey)

passwd = PKCS1_v1_5.new(key)

text = base64.b64encode(passwd.encrypt(bytes(k, encoding='utf-8')))

return str(text, encoding='utf-8')

def login(self, code=None):

num = self.phone_number

if not code:

self.sms_code = input("請輸入短信驗證碼:\n")

self.sms_ser = input("請輸入服務密碼")

else:

self.sms_code = code

login_url = "https://login.10086.cn/touchBjLogin.action"

data = {

"account": num,

"accountType": "01",

"channelID": "00100",

"password": self.get_password(self.sms_code),

"protocol": "https:",

"pwdType": "02",

"rememberMe": "0",

"timestamp": self.get_timestamp(),

}

req = self.session.post(login_url, headers=self.headers, data=data)

return req

def verify_captcha(self, code):

'''驗證驗證碼是否正確'''

flag = self.checkzh(code)

if flag == 1:

'''1表示中文三個字的驗證碼'''

code = self.str_to_hex(code)

req = self.session.get(f"https://login.10086.cn/verifyCaptcha?inputCode={code}")

resultCode = req.get("resultCode")

return resultCode

def str_to_hex(self, s):

return ''.join([hex(ord(c)).replace('0x', '') for c in s])

def checkzh(self, s):

"""判斷驗證碼是否為中文"""

pat = re.compile("^[\u4E00-\u9FA5]{3}$")

res = pat.findall(s)

return len(res)

def get_verify_code_img(self):

url = "https://login.10086.cn/captchazh.htm?type=12"

buff = self.session.get(url).content

image = Image.open(BytesIO(buff))

image.show()

def get_authimg(self):

url = f"https://shop.10086.cn/i/authImg?t={random.random()}"

buff = self.session.get(url).content

image = Image.open(BytesIO(buff))

image.show()

def bill(self, res):

r_json = res.json()

print("登錄狀態", r_json)

code = r_json.get("result")

desc = r_json.get("desc")

while code != "0":

code = input("請輸入正確的短信驗證碼:")

print("開始發送圖片驗證碼")

self.get_verify_code_img()

img_code = input("獲取圖片驗證碼,請輸入圖片驗證碼")

rescode = self.verify_captcha(img_code)

while rescode != "0":

img_code = input("圖片驗證碼錯誤,請輸入圖片驗證碼:")

rescode = self.verify_captcha(img_code)

print("圖片驗證碼輸入正確")

res = self.login(code)

code = res.get("result")

print("短信驗證碼輸入正確")

r_json = res.json()

print(r_json)

assert_acceptURL = r_json.get("assertAcceptURL")

artifact = r_json.get("artifact")

init_url = "https://service.bj.10086.cn/poffice/package/ywcx.action?PACKAGECODE=CXNEW"

data = {

"artifact": artifact,

"backUrl": init_url

}

print("artifactUrl", assert_acceptURL)

headers = self.headers

headers.update({'Host': 'service.bj.10086.cn'})

req = self.session.get(assert_acceptURL, headers=self.headers, params=data, verify=False,

allow_redirects=False)

req.encoding = req.apparent_encoding

redirect_url = req.headers['Location']

headers.update({'Referer': ''})

redirect_res1 = self.session.get(url=redirect_url, headers=headers)

redirect_res1.encoding = 'utf-8'

bill_detail_url = "https://service.bj.10086.cn/poffice/package/xdcx/xdcxShow.action?PACKAGECODE=XD"

headers.update({'Referer': 'http://service.bj.10086.cn/poffice/jsp/service/fee/fee.jsp'})

bill_req = self.session.get(bill_detail_url, headers=headers, verify=False)

print(bill_req.text)

yzmcheck_url = "https://service.bj.10086.cn/poffice/package/xdcx/userYzmCheck.action"

data = {

"yzCheckCode": self.sms_ser,

"PACKAGECODE": "XD",

}

yzmcheck_req = self.session.post(yzmcheck_url, headers=headers, data=data)

yzmcheck_json = yzmcheck_req.json()

message = yzmcheck_json.get("message")

if message.upper() == "Y":

for item in ("201901", "201812"):

data = {

"PACKAGECODE": "XD",

"xdFlag": "GSM",

"month": item,

}

headers.update(

{'Referer': 'https://service.bj.10086.cn/poffice/package/xdcx/xdcxShow.action?PACKAGECODE=XD'})

xdsxshow_url = "https://service.bj.10086.cn/poffice/package/xdcx/xdcxShow.action"

xdsxshow_req = self.session.post(xdsxshow_url, headers=headers, data=data)

print(f"{item}通話詳單", xdsxshow_req.text)

else:

print("登錄失敗", message)

def base64encode(self, passwd):

text = base64.b64encode(bytes(passwd, encoding='utf-8')).decode("utf-8")

return text

if __name__ == '__main__':

num = "xxxxxxx"

bj = BeiJingCrawler(num)

bj.send_sms_code() # 發送短信驗證碼

res = bj.login()

bj.bill(res)

文章出處：北京移動

總結

以上是生活随笔為你收集整理的python话费充值_Python 登录移动查询话费的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。