spider_work/yueka_spider/yk_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...........")
    url = "https://api.joycard.xyz/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="yueka_token", data={"token": token}, condition={"id": 1})
        logger.debug("------------------ 更新token成功 ------------------")


if __name__ == '__main__':
    login_main()
    # get_token(logger)