76 lines
3.4 KiB
Python
76 lines
3.4 KiB
Python
"""Extract serviceToken using passToken and save to ~/.mi.token"""
|
|
import asyncio, json, base64, hashlib, secrets, string
|
|
from pathlib import Path
|
|
from aiohttp import ClientSession
|
|
|
|
PASS_TOKEN = "V1:DXmurwq2/R1BHTELu6obCYX7Rq/4OZACAywwHdbXYQdJStLxyGwZzcsUMnk6tnJfQvY6HNI1whU+hcSk8uE2Odjpi89ldPfcJTiZ9Tzm4tcaXJakpJa4yk+FuEFO3bqcF941B0MznsQT+HGEETHP/CHAKpKlxavwIZNwuH2TYJH9gFMEhxzKf5vDASX86lyVPUjXOZF1e+qN0+62zJU6HTorL2fiYXYWL+ikS3mKCHQCaB/+NjqLEGohcmXsCKEEyE6ImJbk6nnQB/EobhcK6GhE8zHSL8MAXhppsfhCt+flp0ymq5ntBGnqHshdySMQaQxklVv8JNvVCr5FdZs4kQ=="
|
|
USER_ID = "1136458602"
|
|
|
|
async def main():
|
|
async with ClientSession() as s:
|
|
headers = {"User-Agent": "Mozilla/5.0"}
|
|
device_id = "".join(secrets.choice(string.ascii_uppercase + string.digits) for _ in range(16))
|
|
cookies = {
|
|
"sdkVersion": "3.9",
|
|
"deviceId": device_id,
|
|
"userId": USER_ID,
|
|
"passToken": PASS_TOKEN,
|
|
}
|
|
|
|
url = "https://account.xiaomi.com/pass/serviceLogin?sid=micoapi&_json=true"
|
|
async with s.get(url, cookies=cookies, headers=headers) as r:
|
|
raw = await r.text()
|
|
data = json.loads(raw[11:])
|
|
|
|
print(f"serviceLogin code: {data.get('code')}")
|
|
ssecurity = data["ssecurity"]
|
|
location = data["location"]
|
|
nonce = str(data["nonce"])
|
|
new_pass = data.get("passToken", PASS_TOKEN)
|
|
|
|
nsec = f"nonce={nonce}&{ssecurity}"
|
|
client_sign = base64.b64encode(hashlib.sha1(nsec.encode()).digest()).decode()
|
|
sts_url = f"{location}&clientSign={client_sign}"
|
|
|
|
async with s.get(sts_url, headers=headers) as r2:
|
|
st_cookie = r2.cookies.get("serviceToken")
|
|
if not st_cookie:
|
|
print(f"❌ 错误：未能获取 serviceToken。请检查 PASS_TOKEN 是否过期。")
|
|
print(f"响应状态码: {r2.status}")
|
|
return
|
|
service_token = st_cookie.value
|
|
|
|
token = {
|
|
"deviceId": device_id,
|
|
"userId": USER_ID,
|
|
"passToken": new_pass,
|
|
"micoapi": [ssecurity, service_token],
|
|
}
|
|
|
|
token_path = Path(__file__).resolve().parent / ".mi.token"
|
|
token_path.write_text(json.dumps(token, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"Token saved to {token_path}")
|
|
print(f" userId: {USER_ID}")
|
|
print(f" passToken: OK")
|
|
print(f" ssecurity: OK")
|
|
print(f" serviceToken: OK ({service_token[:30]}...)")
|
|
|
|
# --- 新增推送至 NAS 逻辑 ---
|
|
nas_url = "http://5kan.cc:42000/sync"
|
|
headers = {"X-Auth-Key": "my_private_key_123"} # 注意：请确保与服务器端的 SECRET_KEY 一致
|
|
|
|
print(f"\n正在同步至 NAS ({nas_url})...")
|
|
try:
|
|
async with s.post(nas_url, json=token, headers=headers, timeout=15) as resp:
|
|
if resp.status == 200:
|
|
result = await resp.json()
|
|
print(f"✅ NAS 同步成功: {result.get('msg')}")
|
|
else:
|
|
text = await resp.text()
|
|
print(f"❌ NAS 同步失败 (状态码 {resp.status}): {text}")
|
|
except Exception as e:
|
|
print(f"❌ 无法连接到 NAS: {e}")
|
|
print("请检查：1. 宝塔是否放行 42000 端口；2. 路由器是否做了端口转发；3. 接收端脚本是否正在运行。")
|
|
|
|
asyncio.run(main())
|