2021年12月

因为网站比较敏感, 所以具体网站就不说了, 直接说逻辑部分

为了降低数据提取的错误率, 所以使用了python的slimit库对js代码进行提取处理

对网站源码的JavaScript进行分析后, 发现在其中一个script标签内的js代码是利用抽取混淆的, 并用flashvars开头的变量存储

首先用python将该js代码进行提取

response = requests.get(url=url, proxies=proxies)
script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()

1 2	response = requests.get(url=url, proxies=proxies) script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()

将该部分代码格式化后可以看到键名为mediaDefinitions的值是数组, 数组内存储着videoUrl

在这里可以看到, 链接是被抽取的进行拼接后即可还原真实地址

下面会使用到python的slimit库的ast进行还原

# 将js代码转成结构树
tree = Parser().parse(script)

1 2	# 将js代码转成结构树 tree = Parser().parse(script)

通过smlit的Parser类的parse方法, 对js代码转换为ast结构树

获取到结构树后, 需要自定义类, 并继承ASTVisitor, 自定义访问者遍历节点对节点进行抽取

下面代码我先对mediaDefinitions和qualityItems对象进行抽取还原

class VarStatement_Visitor(ASTVisitor):

    # 自定义访问者，重写VarStatement节点访问逻辑
    def visit_VarStatement(self, node):
        Identifier, Object = node.children()[0].children()
        # 获取flashvars的节点
        if 'flashvars' in Identifier.value:
            for i in Object.properties:
                left, right = i.children()
                # mediaDefinitions数组
                if left.value == '"mediaDefinitions"':
                    # 还原字典
                    for item in right.items:
                        media_data = dict()
                        for medias in item.properties:
                            media_left, media_right = medias.children()
                            if isinstance(media_right, ast.Array):
                                data_list = [i.value for i in media_right.items]
                                media_data[media_left.value[1:-1]] = data_list
                            else:
                                if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':
                                    media_data[media_left.value[1:-1]] = media_right.value
                                else:
                                    media_data[media_left.value[1:-1]] = media_right.value[1:-1]
                        flashvars.append(media_data)
        # 获取qualityItems的节点
        if 'qualityItems' in Identifier.value:
            for i in Object.items:
                media_data = dict()
                for medias in i.properties:
                    media_left, media_right, = medias.children()
                    if isinstance(media_right, ast.Number):
                        media_data[media_left.value[1:-1]] = media_right.value
                    else:
                        media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')
                flashvars.append(media_data)

class VarStatement_Visitor(ASTVisitor):

# 自定义访问者，重写VarStatement节点访问逻辑

def visit_VarStatement(self, node):

Identifier, Object = node.children()[0].children()

# 获取flashvars的节点

if 'flashvars' in Identifier.value:

for i in Object.properties:

left, right = i.children()

# mediaDefinitions数组

if left.value == '"mediaDefinitions"':

# 还原字典

for item in right.items:

media_data = dict()

for medias in item.properties:

media_left, media_right = medias.children()

if isinstance(media_right, ast.Array):

data_list = [i.value for i in media_right.items]

media_data[media_left.value[1:-1]] = data_list

else:

if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1]

flashvars.append(media_data)

# 获取qualityItems的节点

if 'qualityItems' in Identifier.value:

for i in Object.items:

media_data = dict()

for medias in i.properties:

media_left, media_right, = medias.children()

if isinstance(media_right, ast.Number):

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')

flashvars.append(media_data)

还原mediaDefinitions和qualityItems对象后, 继续创建访问者对抽取链接进行还原

进入的节点仍然是VarStatement, 因为视频的链接由多个变量并且不定数量进行拼接, 所以下面使用递归方式并使用了get_Identifier方法对抽取变量进行获取还原

class Media_Visitor(ASTVisitor):

    def __init__(self, i, *args, **kwargs):
        # 视频所在的序号
        self.i = i
        # 映射关系
        self.identifier = {}
        # 映射顺序
        self.identifiers_list = []
        super(*args, **kwargs)

    # 递归获取映射顺序
    def get_Identifier(self, node, identifiers_list):
        left, right = node.children()
        identifiers_list.append(self.identifier[right.value])
        if isinstance(left, ast.BinOp):
            self.get_Identifier(left, identifiers_list)
        else:
            identifiers_list.append(self.identifier[left.value])

    def visit_VarStatement(self, node):
        Identifier, BinOp = node.children()[0].children()
        # 获取该函数地址的映射顺序
        if 'media_' + str(self.i) in Identifier.value:
            # 计算该视频的真实地址
            self.get_Identifier(BinOp, self.identifiers_list)
            # 填充视频地址
            flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])
        # 根据映射关系进行恢复
        elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):
            if isinstance(BinOp, ast.String):
                self.identifier[Identifier.value] = BinOp.value[1:-1]
            else:
                self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])

class Media_Visitor(ASTVisitor):

def __init__(self, i, *args, **kwargs):

# 视频所在的序号

self.i = i

# 映射关系

self.identifier = {}

# 映射顺序

self.identifiers_list = []

super(*args, **kwargs)

# 递归获取映射顺序

def get_Identifier(self, node, identifiers_list):

left, right = node.children()

identifiers_list.append(self.identifier[right.value])

if isinstance(left, ast.BinOp):

self.get_Identifier(left, identifiers_list)

else:

identifiers_list.append(self.identifier[left.value])

def visit_VarStatement(self, node):

Identifier, BinOp = node.children()[0].children()

# 获取该函数地址的映射顺序

if 'media_' + str(self.i) in Identifier.value:

# 计算该视频的真实地址

self.get_Identifier(BinOp, self.identifiers_list)

# 填充视频地址

flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])

# 根据映射关系进行恢复

elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):

if isinstance(BinOp, ast.String):

self.identifier[Identifier.value] = BinOp.value[1:-1]

else:

self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])

下面是完整代码:

import re
import time
import traceback
import requests_html
# 将js代码转为AST结构树
from slimit.parser import Parser
# 自定义访问者
from slimit.visitors.nodevisitor import ASTVisitor
# 判断类型
from slimit import ast
import os

requests = requests_html.HTMLSession()
flashvars = list()


class VarStatement_Visitor(ASTVisitor):

    # 自定义访问者，重写VarStatement节点访问逻辑
    def visit_VarStatement(self, node):
        Identifier, Object = node.children()[0].children()
        # 获取flashvars的节点
        if 'flashvars' in Identifier.value:
            for i in Object.properties:
                left, right = i.children()
                # mediaDefinitions数组
                if left.value == '"mediaDefinitions"':
                    # 还原字典
                    for item in right.items:
                        media_data = dict()
                        for medias in item.properties:
                            media_left, media_right = medias.children()
                            if isinstance(media_right, ast.Array):
                                data_list = [i.value for i in media_right.items]
                                media_data[media_left.value[1:-1]] = data_list
                            else:
                                if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':
                                    media_data[media_left.value[1:-1]] = media_right.value
                                else:
                                    media_data[media_left.value[1:-1]] = media_right.value[1:-1]
                        flashvars.append(media_data)
        # 获取qualityItems的节点
        if 'qualityItems' in Identifier.value:
            for i in Object.items:
                media_data = dict()
                for medias in i.properties:
                    media_left, media_right, = medias.children()
                    if isinstance(media_right, ast.Number):
                        media_data[media_left.value[1:-1]] = media_right.value
                    else:
                        media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')
                flashvars.append(media_data)


class Media_Visitor(ASTVisitor):

    def __init__(self, i, *args, **kwargs):
        # 视频所在的序号
        self.i = i
        # 映射关系
        self.identifier = {}
        # 映射顺序
        self.identifiers_list = []
        super(*args, **kwargs)

    # 递归获取映射顺序
    def get_Identifier(self, node, identifiers_list):
        left, right = node.children()
        identifiers_list.append(self.identifier[right.value])
        if isinstance(left, ast.BinOp):
            self.get_Identifier(left, identifiers_list)
        else:
            identifiers_list.append(self.identifier[left.value])

    def visit_VarStatement(self, node):
        Identifier, BinOp = node.children()[0].children()
        # 获取该函数地址的映射顺序
        if 'media_' + str(self.i) in Identifier.value:
            # 计算该视频的真实地址
            self.get_Identifier(BinOp, self.identifiers_list)
            # 填充视频地址
            flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])
        # 根据映射关系进行恢复
        elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):
            if isinstance(BinOp, ast.String):
                self.identifier[Identifier.value] = BinOp.value[1:-1]
            else:
                self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])
                

def main(url):
    response = requests.get(url=url)
    script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()
    # 将js代码转成结构树
    tree = Parser().parse(script)
    VarStatement_Visitor().visit(tree)
    for i, _ in enumerate(flashvars):
        Media_Visitor(i).visit(tree)
    print(flashvars)

import re

import time

import traceback

import requests_html

# 将js代码转为AST结构树

from slimit.parser import Parser

# 自定义访问者

from slimit.visitors.nodevisitor import ASTVisitor

# 判断类型

from slimit import ast

import os

requests = requests_html.HTMLSession()

flashvars = list()

class VarStatement_Visitor(ASTVisitor):

# 自定义访问者，重写VarStatement节点访问逻辑

def visit_VarStatement(self, node):

Identifier, Object = node.children()[0].children()

# 获取flashvars的节点

if 'flashvars' in Identifier.value:

for i in Object.properties:

left, right = i.children()

# mediaDefinitions数组

if left.value == '"mediaDefinitions"':

# 还原字典

for item in right.items:

media_data = dict()

for medias in item.properties:

media_left, media_right = medias.children()

if isinstance(media_right, ast.Array):

data_list = [i.value for i in media_right.items]

media_data[media_left.value[1:-1]] = data_list

else:

if media_left.value == '"defaultQuality"' or media_left.value == '"remote"':

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1]

flashvars.append(media_data)

# 获取qualityItems的节点

if 'qualityItems' in Identifier.value:

for i in Object.items:

media_data = dict()

for medias in i.properties:

media_left, media_right, = medias.children()

if isinstance(media_right, ast.Number):

media_data[media_left.value[1:-1]] = media_right.value

else:

media_data[media_left.value[1:-1]] = media_right.value[1:-1].replace('\\/', '/')

flashvars.append(media_data)

class Media_Visitor(ASTVisitor):

def __init__(self, i, *args, **kwargs):

# 视频所在的序号

self.i = i

# 映射关系

self.identifier = {}

# 映射顺序

self.identifiers_list = []

super(*args, **kwargs)

# 递归获取映射顺序

def get_Identifier(self, node, identifiers_list):

left, right = node.children()

identifiers_list.append(self.identifier[right.value])

if isinstance(left, ast.BinOp):

self.get_Identifier(left, identifiers_list)

else:

identifiers_list.append(self.identifier[left.value])

def visit_VarStatement(self, node):

Identifier, BinOp = node.children()[0].children()

# 获取该函数地址的映射顺序

if 'media_' + str(self.i) in Identifier.value:

# 计算该视频的真实地址

self.get_Identifier(BinOp, self.identifiers_list)

# 填充视频地址

flashvars[self.i]['videoUrl'] = ''.join(self.identifiers_list[::-1])

# 根据映射关系进行恢复

elif isinstance(BinOp, ast.String) or (len(BinOp.children()) == 2 and isinstance(BinOp.children()[0], ast.String) and isinstance(BinOp.children()[1], ast.String)):

if isinstance(BinOp, ast.String):

self.identifier[Identifier.value] = BinOp.value[1:-1]

else:

self.identifier[Identifier.value] = ''.join([i.value[1:-1] for i in BinOp.children()])

def main(url):

response = requests.get(url=url)

script = filter(lambda x: 'flashvars' in x, response.html.xpath('//script//text()')).__next__()

# 将js代码转成结构树

tree = Parser().parse(script)

VarStatement_Visitor().visit(tree)

for i, _ in enumerate(flashvars):

Media_Visitor(i).visit(tree)

print(flashvars)

原本的Django本地运行的时候是正常的, 当部署线上的时候就一大堆问题出现了.

因为我对Linux并不熟悉, 所以就选择了宝塔的python项目管理器来部署

在部署上线之前, 要添加几个文件.

常规的要在项目根目录加上uwsgi.ini文件, 文件内容如下

[uwsgi]
master = true
processes = 1
threads = 2
chdir = /www/wwwroot/api
wsgi-file = /www/wwwroot/api/zeno_api/wsgi.py  # wsgi文件的绝对路径
http = 0.0.0.0:520  # 0.0.0.0:端口号
logto = /www/wwwroot/api/logs/error.log  # 日志保存路径
chmod-socket = 660
vacuum = true
max-requests = 1000

[uwsgi]

master = true

processes = 1

threads = 2

chdir = /www/wwwroot/api

wsgi-file = /www/wwwroot/api/zeno_api/wsgi.py # wsgi文件的绝对路径

http = 0.0.0.0:520 # 0.0.0.0:端口号

logto = /www/wwwroot/api/logs/error.log # 日志保存路径

chmod-socket = 660

vacuum = true

max-requests = 1000

以上内容需要注意的是, wsgi-file填写的是wsgi.py文件的绝对路径加上文件夹名字. 网上很多都是只填路径的, 部署后会直接报错. 而wsgi.py是Django在建项目的时候就自带了的, 不需修改

部署成功之后, 如果有域名可以映射到域名上, 点击上面的映射填上域名即可, 映射成功之后可以使用域名访问

如果有使用Django自带的admin后台, 会发现项目部署后admin后台的样式会没了. 找了好几天百度都没有解决, 后来还是看官方文档解决的…

首先是要在settings.py下加上STATIC_ROOT, 然后在这个路径下手动创建一个static文件夹

STATIC_ROOT = '/www/wwwroot/api/zeno_api/static/'

1	STATIC_ROOT = '/www/wwwroot/api/zeno_api/static/'

然后再去主app下的urls.py修改以下代码

from django.conf import settings
from django.conf.urls.static import static

urlpatterns = [
    # ... the rest of your URLconf goes here ...
] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)