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": "06609f63-1a0c-46c2-9f61-9acebe289e79",
    #     "brand": "Redmi",
    #     "os": "android",
    #     "content-type": "application/json; charset=utf-8",
    #     "authori-zation": "",
    #     "systemversion": "32",
    #     "lang": "zh",
    #     "verse-ua": "d7b3b338008806f1b20427173b983e29",
    #     "version": "2.5.3",
    #     "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()