py-tiktok-robot/tiktok/data_handler.py

"""

"""
import json
import re

import jsonpath

item_require_fields = [
    'id', 'author', 'authorStats', 'createTime', 'desc', 'imagePost', 'stats',' bitrateInfo', 'cover', 'CategoryType'
]


def get_video_by_response(response):
    item_struct = get_item_json2(response.text())
    item_struct['cover'] = item_struct.get('video').get('cover')
    item_struct = {k: v for k, v in item_struct.items() if k in item_require_fields}
    return item_struct


def get_video_json(html_content):
    # 逐行读取文件
    inside_items = False  # 标志是否进入 items 部分
    items_buffer = ""  # 临时保存 JSON 部分

    for line in html_content.splitlines():
        # 通过简单的规则修复非标准 JSON 格式（可以根据实际情况定制）
        line = line.strip()  # 去掉多余的空白字符

        if '"itemStruct"' in line:
            items_buffer = '{'
            inside_items = True  # 发现目标字段
            continue

        # 处理 items 数组的部分
        if inside_items:
            items_buffer += line  # 累积读取多行

            # 如果找到了 JSON 数组的结束
            if '}' in items_buffer:
                try:
                    # 尝试解析 JSON
                    data = json.loads(items_buffer)
                    # 获取第一个 item
                    return data
                except json.JSONDecodeError:
                    continue  # 如果解析失败，继续读取下一行
                    # '<script id="__UNIVERSAL_DATA_FOR_REHYDRATION__" type="application/json">'
    return None  # 如果没有找到匹配项


def get_item_json2(html_content):
    # 逐行读取文件
    for line in html_content.splitlines():
        # 通过简单的规则修复非标准 JSON 格式（可以根据实际情况定制）
        line = line.strip()  # 去掉多余的空白字符

        if '"itemStruct"' in line:
            script_pattern = re.compile(r'<script\s+id="__UNIVERSAL_DATA_FOR_REHYDRATION__"\s+type="application/json"[^>]*>(.*?)</script>', re.DOTALL)

            # 查找所有匹配的 <script> 标签内容
            json_str = re.findall(script_pattern, line)
            data = json.loads(json_str[0])
            jsonpath_expr = '$..itemStruct'
            data = jsonpath.jsonpath(data, jsonpath_expr)
            if data:
                return data[0]
    return None  # 如果没有找到匹配项

def get_detail_by_response(response):
    json = response.json()
    item_struct = json.get('itemInfo').get('itemStruct')
    item_struct = {k: v for k, v in item_struct.items() if k in item_require_fields}
    return item_struct

# with open('../aweme_video.txt', 'r') as file:
#     data = get_video_json(file.read())
#     print(data)