ai-yunying/douyin/api/req_api.py

import requests
from requests import Request,Response
import DrissionPage
import douyin.cookies as cookies
from conf.config import logger
import jsonpath

params = {
    "url": "https://www.douyin.com/note/12345678910",
    "source": True
}
def get_user_info():
    row = cookies.get_from_table()

    params = {
    "url": "https://www.douyin.com/user/MS4wLjABAAAAQ6pjxWtdMnxF3LjL6oAGNuU-irhAXJNhhCACx2fTo_IHERUBDVrJJiWCfZQmLi0S",
    "source": "true",
    "cookie": row['cookies'],
    "token": None
    }

    # response = requests.post("http://10.0.0.12:5000/user/", json=params)
    # http://192.168.2.11:5000
    response = requests.post("http://192.168.2.11:5000/user/", json=params)
    if 200 == response.status_code:
        logger.info(f"{response.json()}")
    else:
        logger.error(f"{response.status_code}: {response.raw}")


'''
直播推流数据接口
获取直播推流数据

请求接口：/live/

请求参数

{
  "url": "直播链接，支持多直播，字符串，必需参数",
  "source": "是否返回原始数据，布尔值，可选参数，默认值: false",
  "cookie": "抖音 cookie，字符串，可选参数",
  "token": "自定义参数"
}
响应参数

{
  "data": [
    "直播数据-1，JSON 格式",
    "直播数据-2，JSON 格式",
    "..."
  ],
  "message": "success"
}'''
def rq_live_url():
    row = cookies.get_from_table()
    params = {
        # "url": "https://live.douyin.com/137439305439",
        "url": "https://live.douyin.com/660292215268",
        "source": True,
        "cookie": row['cookies'],
        "token": None
    }
    response = requests.post("http://localhost:5000/live/", json=params)
    logger.info(f"{response.status_code}")
    if 200 == response.status_code:
        data = response.json()
        logger.info(f"{data}")
        return data

def get_anchor(data:dict):
    status_code = jsonpath.jsonpath(data, '$...status_code')
    if status_code and status_code[0] != 0:
        logger.info(f"响应错误：{data}")
        return None
    user = jsonpath.jsonpath(data, '$...user')[0]
    logger.info(f"user {user}")
    nickname = user.get("nickname")
    room_view_stats = jsonpath.jsonpath(data, '$...room_view_stats')[0]
    if not room_view_stats:
        
        logger.info(f"{nickname} 没有开启直播")
        return
    logger.info(f"room_view_stats {room_view_stats}")
    display_value = room_view_stats.get("display_value")
    anchor = room_view_stats.get("display_long_anchor")
    ret = f"主播： {nickname}  |  {anchor} （{display_value}）"
    logger.info(ret)
    return ret


def get_live_data(url):
    display_value, anchor = get_anchor(data.get("data")[0])
    name = data.get("user").get("nickname")
    ret = f"主播： {name}  |  {anchor} （{display_value}）"


from faker import Faker

def main():
    # res = cookies.get_from_table()
    # logger.info(f"{res}")
    rq_live_url()
    # get_user_info()
    # fake = Faker()
    # name = fake.user_agent()
    # print(name)

if __name__ == "__main__":
    main()