Skip to content

Commit

Permalink
✨ 支持共享套餐各号码用量查询接口
Browse files Browse the repository at this point in the history
  • Loading branch information
@Cp0204
Cp0204 committed May 12, 2024
1 parent b0d57c8 commit e5b7dd1
Show file tree
Hide file tree
Showing 3 changed files with 95 additions and 57 deletions.
3 changes: 3 additions & 0 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,9 @@ docker run -d \

返回流量包明细

- `http://127.0.0.1:10000/qryShareUsage`

返回共享套餐各号码用量

- `http://127.0.0.1:10000/summary`

Expand Down
27 changes: 18 additions & 9 deletions app/api_server.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,10 +64,10 @@ def login():
save_login_info(login_info)
return jsonify(data), 200
else:
return jsonify({"message": "登录失败"}), 400
return jsonify(data), 400


def query_data(query_func):
def query_data(query_func, **kwargs):
"""
查询数据，如果本地没有登录信息或密码不匹配，则尝试登录后再查询
"""
Expand All @@ -81,22 +81,21 @@ def query_data(query_func):
login_info = load_login_info()
if phonenum in login_info and login_info[phonenum]["password"] == password:
telecom.set_login_info(login_info[phonenum])
data = query_func()
data = query_func(**kwargs)
if data.get("responseData"):
return jsonify(data), 200
# 重新登录
login_data, status_code = login()
login_data = json.loads(login_data.data)
if status_code == 200:
telecom.set_login_info(
json.loads(login_data.data)["responseData"]["data"]["loginSuccessResult"]
)
data = query_func()
telecom.set_login_info(login_data["responseData"]["data"]["loginSuccessResult"])
data = query_func(**kwargs)
if data:
return jsonify(data), 200
else:
return jsonify({"message": "查询失败"}), 400
return jsonify(data), 400
else:
return jsonify({"message": "手机号或密码错误"}), 400
return jsonify(login_data), 400


@app.route("/qryImportantData", methods=["POST", "GET"])
Expand All @@ -111,6 +110,16 @@ def user_flux_package():
return query_data(telecom.user_flux_package)


@app.route("/qryShareUsage", methods=["POST", "GET"])
def qry_share_usage():
"""查询共享用量接口"""
if request.method == "POST":
data = request.get_json() or {}
else:
data = request.args
return query_data(telecom.qry_share_usage, billing_cycle=data.get("billing_cycle"))


@app.route("/summary", methods=["POST", "GET"])
def summary():
"""查询重要数据简化接口"""
Expand Down
122 changes: 74 additions & 48 deletions telecom_class.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,19 +10,30 @@

class Telecom:
def __init__(self):
self.login_info = None
self.login_info = {}
self.phonenum = None
self.password = None
self.token = None
self.client_type = "#9.6.1#channel50#iPhone 14 Pro#"
self.headers = {
"Accept": "application/json",
"Content-Type": "application/json; charset=UTF-8",
"Connection": "Keep-Alive",
"Accept-Encoding": "gzip",
"user-agent": "iPhone 14 Pro/9.6.1",
}

def set_login_info(self, login_info):
self.login_info = login_info
self.phonenum = login_info.get("phoneNbr", None)
self.password = login_info.get("password", None)
self.token = login_info.get("token", None)

def trans_phone(self, phone_num):
def trans_phone(self, phonenum):
result = ""
for char in phone_num:
result += chr(ord(char) + 2 & 65535)
caesar_size = 2 if phonenum.startswith("1") else -2
for char in phonenum:
result += chr(ord(char) + caesar_size & 65535)
return result

def encrypt(self, str):
Expand Down Expand Up @@ -50,90 +61,72 @@ def do_login(self, phonenum, password):
phonenum = phonenum or self.phonenum
password = password or self.password
ts = datetime.now().strftime("%Y%m%d%H%M00")
enc = f"iPhone 14 13.2.3{phonenum}{phonenum}{ts}{password}0$$$0."
enc_str = f"iPhone 14 13.2.3{phonenum}{phonenum}{ts}{password}0$$$0."
body = {
"content": {
"fieldData": {
"accountType": "",
"authentication": password,
"deviceUid": f"3{phonenum}",
"isChinatelecom": "0",
"loginAuthCipherAsymmertric": self.encrypt(enc),
"loginAuthCipherAsymmertric": self.encrypt(enc_str),
"loginType": "4",
"phoneNum": self.trans_phone(phonenum),
"systemVersion": "13.2.3",
},
"attach": "iPhone",
},
"headerInfos": {
"clientType": "#9.6.1#channel50#iPhone 14 Pro#",
"code": "userLoginNormal",
"clientType": self.client_type,
"timestamp": ts,
"shopId": "20002",
"source": "110003",
"sourcePassword": "Sid98s",
"timestamp": ts,
"userLoginName": phonenum,
},
}
headers = {
"Accept": "application/json",
"Content-Type": "application/json; charset=UTF-8",
"Connection": "Keep-Alive",
"Accept-Encoding": "gzip",
}
response = requests.post(
"https://appgologin.189.cn:9031/login/client/userLoginNormal",
headers=headers,
headers=self.headers,
json=body,
)
# print(response.text)
data = response.json()
return data
print(response.text)
return response.json()

def qry_important_data(self, token=""):
token = token or self.login_info["token"]
provinceCode = self.login_info["provinceCode"] or "600101"
cityCode = self.login_info["cityCode"] or "8441900"
def qry_important_data(self, **kwargs):
ts = datetime.now().strftime("%Y%m%d%H%M00")
body = {
"content": {
"fieldData": {
"provinceCode": provinceCode,
"cityCode": cityCode,
"provinceCode": self.login_info["provinceCode"] or "600101",
"cityCode": self.login_info["cityCode"] or "8441900",
"shopId": "20002",
"isChinatelecom": "0",
"account": self.trans_phone(self.phonenum),
},
"attach": "test",
},
"headerInfos": {
"clientType": "#9.6.1#channel50#iPhone X Plus#",
"timestamp": ts,
"code": "userFluxPackage",
"clientType": self.client_type,
"timestamp": ts,
"shopId": "20002",
"source": "110003",
"sourcePassword": "Sid98s",
"token": token,
"userLoginName": self.phonenum,
"token": kwargs.get("token") or self.token,
},
}
headers = {
"Accept": "application/json",
"Content-Type": "application/json; charset=UTF-8",
"Connection": "Keep-Alive",
"Accept-Encoding": "gzip",
}
response = requests.post(
"https://appfuwu.189.cn:9021/query/qryImportantData",
headers=headers,
headers=self.headers,
json=body,
)
# print(response.text)
data = response.json()
return data
return response.json()

def user_flux_package(self, token=""):
token = token or self.login_info["token"]
def user_flux_package(self, **kwargs):
ts = datetime.now().strftime("%Y%m%d%H%M00")
body = {
"content": {
Expand All @@ -145,29 +138,62 @@ def user_flux_package(self, token=""):
"attach": "test",
},
"headerInfos": {
"clientType": "#9.6.1#channel50#iPhone X Plus#",
"timestamp": ts,
"code": "userFluxPackage",
"clientType": self.client_type,
"timestamp": ts,
"shopId": "20002",
"source": "110003",
"sourcePassword": "Sid98s",
"token": token,
"userLoginName": self.phonenum,
"token": kwargs.get("token") or self.token,
},
}
headers = {
"Accept": "application/json",
"Content-Type": "application/json; charset=UTF-8",
"Connection": "Keep-Alive",
"Accept-Encoding": "gzip",
}
response = requests.post(
"https://appfuwu.189.cn:9021/query/userFluxPackage",
headers=headers,
headers=self.headers,
json=body,
)
# print(response.text)
return response.json()

def qry_share_usage(self, **kwargs):
billing_cycle = kwargs.get("billing_cycle") or datetime.now().strftime("%Y%m")
ts = datetime.now().strftime("%Y%m%d%H%M00")
body = {
"content": {
"attach": "test",
"fieldData": {
"billingCycle": billing_cycle,
"account": self.trans_phone(self.phonenum),
},
},
"headerInfos": {
"code": "qryShareUsage",
"clientType": self.client_type,
"timestamp": ts,
"shopId": "20002",
"source": "110003",
"sourcePassword": "Sid98s",
"userLoginName": self.phonenum,
"token": kwargs.get("token") or self.token,
},
}
response = requests.post(
"https://appfuwu.189.cn:9021/query/qryShareUsage",
headers=self.headers,
json=body,
)
data = response.json()
# 返回的号码字段加密，需做解密转换
if data.get("responseData").get("data").get("sharePhoneBeans"):
for item in data["responseData"]["data"]["sharePhoneBeans"]:
item["sharePhoneNum"] = self.trans_phone(item["sharePhoneNum"])
for share_type in data["responseData"]["data"]["shareTypeBeans"]:
for share_info in share_type["shareUsageInfos"]:
for share_amount in share_info["shareUsageAmounts"]:
share_amount["phoneNum"] = self.trans_phone(
share_amount["phoneNum"]
)
return data

def to_summary(self, data, phonenum=""):
Expand Down

0 comments on commit e5b7dd1

Please sign in to comment.