tieba.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@File    :   tieba.py
@Time    :   2023-01-28 15:54:11
@Author  :   yikoyu
@Version :   1.0
@Desc    :   百度贴吧自动签到
"""


import asyncio
import copy
import hashlib
import logging
import os
import random
import time

import httpx

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# API_URL
LIKIE_URL = "http://c.tieba.baidu.com/c/f/forum/like"
TBS_URL = "http://tieba.baidu.com/dc/common/tbs"
SIGN_URL = "http://c.tieba.baidu.com/c/c/forum/sign"

ENV = os.environ

HEADERS = {
    "Host": "tieba.baidu.com",
    "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 \
    (KHTML, like Gecko) Chrome/39.0.2171.71 Safari/537.36",
}
SIGN_DATA = {
    "_client_type": "2",
    "_client_version": "9.7.8.0",
    "_phone_imei": "000000000000000",
    "model": "MI+5",
    "net_type": "1",
}

# VARIABLE NAME
COOKIE = "Cookie"
BDUSS = "BDUSS"
EQUAL = r"="
EMPTY_STR = r""
TBS = "tbs"
PAGE_NO = "page_no"
ONE = "1"
TIMESTAMP = "timestamp"
DATA = "data"
FID = "fid"
SIGN_KEY = "tiebaclient!!!"
UTF8 = "utf-8"
SIGN = "sign"
KW = "kw"

client = httpx.AsyncClient()


def get_tbs(bduss):
    logger.info("获取tbs开始")
    headers = copy.copy(HEADERS)
    headers.update({COOKIE: EMPTY_STR.join([BDUSS, EQUAL, bduss])})
    try:
        tbs = httpx.get(url=TBS_URL, headers=headers, timeout=5).json()[TBS]
    except Exception as e:
        logger.error("获取tbs出错 %s", e)
        logger.info("重新获取tbs开始")
        tbs = httpx.get(url=TBS_URL, headers=headers, timeout=5).json()[TBS]
    logger.info("获取tbs结束")
    return tbs


def get_favorite(bduss):  # noqa: C901
    logger.info("获取关注的贴吧开始")
    # 客户端关注的贴吧
    returnData = {}
    i = 1
    data = {
        "BDUSS": bduss,
        "_client_id": "wappc_1534235498291_488",
        "from": "1008621y",
        "page_no": "1",
        "page_size": "200",
        "timestamp": str(int(time.time())),
        "vcode_tag": "11",
        **SIGN_DATA,
    }
    data = encodeData(data)

    try:
        res = httpx.post(url=LIKIE_URL, data=data, timeout=5).json()
    except Exception as e:
        logger.error("获取关注的贴吧出错 %s", e)
        return []

    returnData = res
    if "forum_list" not in returnData:
        returnData["forum_list"] = []
    if res["forum_list"] == []:
        return {"gconforum": [], "non-gconforum": []}
    if "non-gconforum" not in returnData["forum_list"]:
        returnData["forum_list"]["non-gconforum"] = []
    if "gconforum" not in returnData["forum_list"]:
        returnData["forum_list"]["gconforum"] = []

    while "has_more" in res and res["has_more"] == "1":
        i = i + 1
        data = {
            "BDUSS": bduss,
            "_client_type": "2",
            "_client_id": "wappc_1534235498291_488",
            "_client_version": "9.7.8.0",
            "_phone_imei": "000000000000000",
            "from": "1008621y",
            "page_no": str(i),
            "page_size": "200",
            "model": "MI+5",
            "net_type": "1",
            "timestamp": str(int(time.time())),
            "vcode_tag": "11",
        }
        data = encodeData(data)
        try:
            res = httpx.post(url=LIKIE_URL, data=data, timeout=5).json()
        except Exception as e:
            logger.error("获取关注的贴吧出错 %s", e)
            continue
        if "forum_list" not in res:
            continue
        if "non-gconforum" in res["forum_list"]:
            returnData["forum_list"]["non-gconforum"].append(
                res["forum_list"]["non-gconforum"]
            )
        if "gconforum" in res["forum_list"]:
            returnData["forum_list"]["gconforum"].append(res["forum_list"]["gconforum"])

    t = []
    for i in returnData["forum_list"]["non-gconforum"]:
        if isinstance(i, list):
            for j in i:
                if isinstance(j, list):
                    for k in j:
                        t.append(k)
                else:
                    t.append(j)
        else:
            t.append(i)
    for i in returnData["forum_list"]["gconforum"]:
        if isinstance(i, list):
            for j in i:
                if isinstance(j, list):
                    for k in j:
                        t.append(k)
                else:
                    t.append(j)
        else:
            t.append(i)
    logger.info("获取关注的贴吧结束")
    return t


def encodeData(data):
    s = EMPTY_STR
    keys = data.keys()
    for i in sorted(keys):
        s += i + EQUAL + str(data[i])
    sign = hashlib.md5((s + SIGN_KEY).encode(UTF8)).hexdigest().upper()
    data.update({SIGN: str(sign)})
    return data


async def client_sign(bduss, tbs, fid, kw):
    # 客户端签到
    logger.info("开始签到贴吧：" + kw)
    data = copy.copy(SIGN_DATA)
    data.update(
        {BDUSS: bduss, FID: fid, KW: kw, TBS: tbs, TIMESTAMP: str(int(time.time()))}
    )
    data = encodeData(data)
    res = await client.post(url=SIGN_URL, data=data, timeout=5)
    return res.json()


def main():
    if "BDUSS" not in ENV:
        logger.error("未配置 BDUSS")
        raise ValueError("BDUSS not configure")

    b = ENV["BDUSS"].split("#")

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    for n, i in enumerate(b):
        logger.info("开始签到第" + str(n) + "个用户" + i)
        tbs = get_tbs(i)
        favorites = get_favorite(i)
        for j in favorites:
            time.sleep(random.randint(1, 4))
            loop.run_until_complete(client_sign(i, tbs, j["id"], j["name"]))
        logger.info("完成第" + str(n) + "个用户签到")

    logger.info("所有用户签到结束")
    loop.close()


if __name__ == "__main__":
    main()