spider_work/leka_spider/leke_login.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/3/24 17:53
import requests
# from loguru import logger
from mysql_pool import MySQLConnectionPool
from settings import *

# logger.remove()
# logger.add("./logs/login_{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
#            format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
#            level="DEBUG", retention="7 day")


def get_token(log):
    log.debug("Start fetching token...........")
    headers = {
        "User-Agent": "Dart/3.5 (dart:io)",
        "Accept-Encoding": "gzip",
        "Content-Type": "application/json",
        "deviceid": "763f77b1-cc16-4369-ac39-a03206ecfb48",
        "brand": "Redmi",
        "os": "android",
        "content-type": "application/json; charset=utf-8",
        "authori-zation": "",
        "systemversion": "32",
        "lang": "zh",
        "verse-ua": "12fe307ef8850bdd64f6dcf1986944ae",
        "version": "2.0.0",
        "isphysicaldevice": "true",
        "cid": ""
    }
    url = "https://api.luckycards.com.cn/api/front/login/pwd"
    data = {
        "phone": "19521500850",
        "pwd": "Pass2025",
        "spreaderInviteCode": ""
    }
    response = requests.post(url, headers=headers, json=data)
    # print(response.text)
    resp_json = response.json()
    if resp_json["code"] == 200:
        token = resp_json["data"]["token"]
        log.debug("Fetch token success!")
        return token
    else:
        log.error(f"Fetch token failed! Msg:{resp_json['message']}")
        return None

@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def login_main():
    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=logger)
    if not sql_pool.check_pool_health():
        logger.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    token = get_token(logger)
    if token is None:
        logger.error("获取token失败")
        raise RuntimeError("获取token失败")
    else:
        sql_pool.update_one_or_dict(table="leka_token", data={"token": token}, condition={"id": 1})
        logger.debug("------------------ 更新token成功 ------------------")


if __name__ == '__main__':
    login_main()