MiCommunityTool_dev.py

#!/usr/bin/python

import os
import importlib

while True:
    for lib in ['requests', 'ntplib']:
        try:
            importlib.import_module(lib)
        except ModuleNotFoundError:
            os.system(f'pip install {lib}')
            break
    else:
        break

import requests, json, hashlib, urllib.parse, time, sys, os, base64, ntplib
from datetime import datetime, timedelta, timezone
from urllib.parse import parse_qs, urlparse, quote

version = "dev"

print(f"\n[V{version}] For issues or feedback:\n- GitHub: github.com/offici5l/MiCommunityTool/issues\n- Telegram: t.me/Offici5l_Group\n")

User = "offici5l/MiCommunityTool"
headers = {"User-Agent": User}

def login():
    base_url = "https://account.xiaomi.com"
    sid = "18n_bbs_global"

    user = input('\nEnter user: ')
    pwd = input('\nEnter pwd: ')
    hash_pwd = hashlib.md5(pwd.encode()).hexdigest().upper()
    cookies = {}

    def parse(res): return json.loads(res.text[11:])

    r = requests.get(f"{base_url}/pass/serviceLogin", params={'sid': sid, '_json': True}, headers=headers, cookies=cookies)
    cookies.update(r.cookies.get_dict())
    data = {k: v[0] for k, v in parse_qs(urlparse(parse(r)['location']).query).items()}
    data.update({'user': user, 'hash': hash_pwd})

    r = requests.post(f"{base_url}/pass/serviceLoginAuth2", data=data, headers=headers, cookies=cookies)
    cookies.update(r.cookies.get_dict())
    res = parse(r)

    if res["code"] == 70016: exit("invalid user or pwd")
    if 'notificationUrl' in res:
        url = res['notificationUrl']
        if any(x in url for x in ['callback','SetEmail','BindAppealOrSafePhone']): exit(url)

        cookies.update({"NativeUserAgent": base64.b64encode(User.encode()).decode()})
        params = parse_qs(urlparse(url).query)
        cookies.update(requests.get(f"{base_url}/identity/list", params=params, headers=headers, cookies=cookies).cookies.get_dict())

        email = parse(requests.get(f"{base_url}/identity/auth/verifyEmail", params={'_json': True}, cookies=cookies, headers=headers))['maskedEmail']
        quota = parse(requests.post(f"{base_url}/identity/pass/sms/userQuota", data={'addressType': 'EM', 'contentType': 160040}, cookies=cookies, headers=headers))['info']
        print(f"Account Authentication\nemail: {email}, Remaining attempts: {quota}")
        input("\nPress Enter to send the verification code")

        code_res = parse(requests.post(f"{base_url}/identity/auth/sendEmailTicket", cookies=cookies, headers=headers))

        if code_res["code"] == 0: print(f"\nVerification code sent to your {email}")
        elif code_res["code"] == 70022: exit("Sent too many codes. Try again tomorrow.")
        else: exit(code_res)

        while True:
            ticket = input("Enter code: ").strip()
            v_res = parse(requests.post(f"{base_url}/identity/auth/verifyEmail", data={'ticket':ticket, 'trust':True}, cookies=cookies, headers=headers))
            if v_res["code"] == 70014: print("Verification code error")
            elif v_res["code"] == 0:
                cookies.update(requests.get(v_res['location'], headers=headers, cookies=cookies).history[1].cookies.get_dict())
                cookies.pop("pass_ua", None)
                break
            else: exit(v_res)

        r = requests.get(f"{base_url}/pass/serviceLogin", params={'_json': "true", 'sid': sid}, cookies=cookies, headers=headers)
        res = parse(r)

    region = json.loads(requests.get(f"https://account.xiaomi.com/pass/user/login/region", headers=headers, cookies=cookies).text[11:])["data"]["region"]    

    nonce, ssecurity = res['nonce'], res['ssecurity']
    res['location'] += f"&clientSign={quote(base64.b64encode(hashlib.sha1(f'nonce={nonce}&{ssecurity}'.encode()).digest()))}"
    serviceToken = requests.get(res['location'], headers=headers, cookies=cookies).cookies.get_dict()

    micdata = {"userId": res['userId'], "serviceToken": serviceToken, "region": region}
    with open("micdata.json", "w") as f: json.dump(micdata, f)
    return micdata

try:
    with open('micdata.json') as f:
        micdata = json.load(f)
    if not all(micdata.get(k) for k in ("userId", "serviceToken", "region")):
        raise ValueError
    print(f"\nAccount ID: {micdata['userId']}")
    input("Press 'Enter' to continue.\nPress 'Ctrl' + 'd' to log out.")
except (FileNotFoundError, json.JSONDecodeError, EOFError, ValueError):
    if os.path.exists('micdata.json'):
        os.remove('micdata.json')
    micdata = login()

serviceToken = micdata["serviceToken"]

print(f"\nAccount Region: {micdata['region']}")

api = "https://sgp-api.buy.mi.com/bbs/api/global/"

U_state = api + "user/bl-switch/state"
U_apply = api + "apply/bl-auth"
U_info = api + "user/data"

print("\n[INFO]:")
info = requests.get(U_info, headers=headers, cookies=serviceToken).json()['data']

print(f"{info['registered_day']} days in Community")
print(f"LV{info['level_info']['level']} {info['level_info']['level_title']}")
print(f"{info['level_info']['max_value'] - info['level_info']['current_value']} more points to the next level")
print(f"Points: {info['level_info']['current_value']}")

def state_request():
    print("\n[STATE]:")
    try:
        state = requests.get(U_state, headers=headers, cookies=serviceToken).json().get("data", {})
        is_ = state.get("is_pass")
        button_ = state.get("button_state")
        deadline_ = state.get("deadline_format", "")
        if is_ == 1:
            exit(f"You have been granted access to unlock until Beijing time {deadline_} (mm/dd/yyyy)\n")
        msg = {
            1: "Apply for unlocking\n",
            2: f"Account Error Please try again after {deadline_} (mm/dd)\n",
            3: "Account must be registered over 30 days\n"
        }
        print(msg.get(button_, ""))
        if button_ in [2, 3]:
            exit()
    except Exception as e:
        exit(f"state: {e}")

state_request()

session = requests.Session()
session.headers.update(headers)
session.cookies.update(micdata["serviceToken"])

def apply_request():
    print("\n[APPLY]:")
    try:
        apply = session.post(U_apply, json={"is_retry": True})
        print(f"Request elapsed time: {apply.elapsed.total_seconds()} seconds")
        print(f"Server response time: {apply.headers['Date']}")
        if apply.json().get("code") != 0:
            exit(apply.json())
        data_ = apply.json().get("data", {}) or {}
        apply_ = data_.get("apply_result", 0)
        deadline_ = data_.get("deadline_format", "")
        messages = {
            1: "Application Successful",
            4: f"\nAccount Error Please try again after {deadline_} (mm/dd)\n",
            3: f"\nApplication quota limit reached, please try again after {deadline_.split()[0]} (mm/dd) {deadline_.split()[1]} (GMT+8)\n",
            5: "\nApplication failed. Please try again later\n",
            6: "\nPlease try again in a minute\n",
            7: "\nPlease try again later\n"
        }
        print(messages.get(apply_, ""))
        if apply_ == 1:
            state_request()
        elif apply_ in [4, 5, 6, 7]:
            exit()
        elif apply_ == 3:
            return 1
    except Exception as e:
        exit(f"apply: {e}")


def calcul():
    def send_head(_range):
        return [session.head(U_apply).elapsed.total_seconds() for _ in range(_range)]

    send_head(1)
    res = session.head(U_apply)
    h, m, s = map(int, res.headers['Date'][17:25].split(':'))
    s_s = h * 3600 + m * 60 + s
    t_s = 16 * 3600
    if s_s > t_s:
        t_s += 86400
    r_s = t_s - s_s  
    if r_s < 6:
        if r_s == 1:
            send_head(2)
        elif r_s == 2:
            send_head(4)
        elif r_s == 3:
            send_head(6)
        elif r_s == 4:
            send_head(8)
        elif r_s == 5:
            send_head(10)
        return apply_request()

    e_t = send_head(10)
    a_e = round(sum(e_t[3:]) / 7, 1)
    t_r = t_s - s_s - sum(e_t)
    ss = t_s - a_e

    hours, remainder = divmod(ss, 3600)
    minutes, seconds = divmod(remainder, 60)
    microseconds = (seconds - int(seconds)) * 1_000_000

    print(f"request will be sent at: {int(hours):02}:{int(minutes):02}:{int(seconds):02}.{int(microseconds):06} GMT + 8")

    time.sleep(t_r)
    send_head(9)
    return apply_request()

while True:
    result = calcul()
    if result != 1:
        break