match_spider/zbytb/crawler/login.py

import json
import re
import threading
from collections import namedtuple
from pathlib import Path

import requests
from requests.utils import dict_from_cookiejar

_lock = threading.Lock()
ROOT_PATH = Path(__file__).parent.parent

JSON_LOGIN_COOKIE = (ROOT_PATH / 'config/login_cookie.json').resolve()

User = namedtuple('User', ['username', 'password'])


def _open_file():
    try:
        fp = open(JSON_LOGIN_COOKIE, encoding='utf-8')
    except FileNotFoundError:
        fp = open(JSON_LOGIN_COOKIE, 'w+', encoding='utf-8')
    return fp


def load_login_cookies(user_name: str):
    fp = _open_file()
    try:
        cookies: dict = json.load(fp).get(user_name)
        return cookies
    except json.decoder.JSONDecodeError:
        pass
    fp.close()


def update_login_cookies(user_name: str, update_val: dict):
    """
    更新登录 cookie 内容

    Args:
        user_name: 账号
        update_val: 需要更新的cookie内容

    """
    with _lock:
        fp = open(JSON_LOGIN_COOKIE, encoding='utf-8')
        user_maps: dict = json.load(fp)
        login_cookies: dict = user_maps.get(user_name)
        if login_cookies is not None and len(update_val) > 0:
            login_cookies.update(update_val)
            user_login_info = {user_name: login_cookies}
            user_maps.update(user_login_info)
            wp = open(JSON_LOGIN_COOKIE, 'w+', encoding='utf-8')
            wp.write(json.dumps(user_maps, indent=4))
            wp.close()
        fp.close()


def save_login_cookies(user_name: str, login_cookie: dict):
    with _lock:
        fp = _open_file()
        '''内容存在就加载到内存,不存在就设置为空字典'''
        try:
            user_maps: dict = json.load(fp)
        except json.decoder.JSONDecodeError:
            user_maps = {}
        # print(user_maps)

        if user_name not in user_maps:
            user_maps.setdefault(user_name, login_cookie)
        else:
            cookies = {user_name: login_cookie}
            user_maps.update(cookies)

        wp = open(JSON_LOGIN_COOKIE, 'w+', encoding='utf-8')
        wp.write(json.dumps(user_maps, indent=4))
        fp.close()
        wp.close()


def login_status_check(response):
    userid = re.findall('var destoon_userid = (\d+);', response.text)
    userid = "".join(userid).strip()
    username = re.findall('var destoon_username = \'(.*?)\'', response.text)
    username = "".join(username).strip()
    # print(">>> _user ", (userid, username))
    if userid == '0' and username == '':
        # 登录失效
        return True
    else:
        # 登录有效
        return False


def login(user_name: str, password: str):
    headers = {
        "Connection": "keep-alive",
        "Accept": "application/json, text/javascript, */*; q=0.01",
        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        "X-Requested-With": "XMLHttpRequest",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36",
        "Origin": "https://www.zbytb.com",
        "Referer": "https://www.zbytb.com/member/login.php",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
    }
    data = {
        "username": str(user_name),
        "password": str(password),
        "cookietime": "on"
    }
    login_url = "https://www.zbytb.com/member/ajaxlogin.php"
    response = requests.post(login_url, headers=headers, data=data)
    login_status = response.json()
    # print(user_name + ' ' + login_status.get('msg'))
    if login_status.get('msg') == '登录成功':
        login_cookie = dict_from_cookiejar(response.cookies)
        save_login_cookies(user_name, login_cookie)
        return login_cookie