MENU

躺着投币之抖音辅助篇

July 24, 2021 • Read: 989 • 其他杂类阅读设置

0x01 背景介绍

平时刷抖音有不少有趣的视频,有的随手点个喜欢有的点喜欢收藏留着今后再看,但久而久之这些碎片化内容没被整理出来,看完也就忘了导致收藏夹越来越多,想把这些视频都下载下来做个整理,把有用小视频存档素材或者转化成摘录,但抖音本身没有批量下载视频的功能,因此需要脚本实现如下两个我的需求:

  • 批量下载个人中心点赞收藏的无水印视频
  • 边下载边取消喜欢收藏信息

0x02 需求实现

本着代码能白嫖就白嫖的理念,写之前在Github发现TikTokDownload可以批量下载视频,于是在这个脚本基础上进行加工,主要去掉了一些冗余代码和增加了取消喜欢的模块;批量下载接口只要没有开隐私保护不需要认证,取消喜欢接口需要携带认证的headers。在抓取请求时发现抖音做了SSL Pinning限制正常bp同网段抓包不行,放进虚拟机抓包或rvictl即可正常抓包:

调整测试脚本如下:


#!/usr/bin/env python
# -*- encoding: utf-8 -*-

# @Modify   : https://github.com/Johnserf-Seed/TikTokDownload
# @Summary  : Batch download tiktok video and  cancel like
# @Usage    : Configure login_header and personal_home_page and run

import os
import re
import json
import time
import requests
import traceback


class TikTok(object):
    def __init__(self):
        self.per_page_download_count = 34
        self.number = 0
        self.flag = True
        self.sec_uid = None
        self.mode = "like"  # post|like
        self.headers = {
            'user-agent': 'Mozilla/5.0 (Linux; Android 8.0; Pixel 2 Build/OPD3.170816.012) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Mobile Safari/537.36 Edg/87.0.664.66'
        }
        self.login_header = {
            'sdk-version': '2',
            'passport-sdk-version': '18',
            'x-Tt-Token': 'xxx-1.0.1',
            'User-Agent': 'okhttp/3.10.0.1',
            'X-SS-REQ-TICKET': 'xxx',
            'Accept-Encoding': 'gzip, deflate',
            'Cookie': 'xxx',
            'X-Ladon': 'xxxx/xxxx',
            'X-Khronos': 'xxxx',
            'X-Gorgon': 'xxxx'
        }
        self.personal_home_page = "https://v.douyin.com/xxxx/"
        self.init_url()

    def init_url(self, max_cursor=0):
        try:
            r = requests.get(url=self.personal_home_page)
            self.sec_uid = re.findall('/user/(.*?)\?', str(r.url))[0] if \
                re.findall(r'/user/(.*?)\?', str(r.url))[0] else r.url[28:83]
            api_url = 'https://www.iesdouyin.com/web/api/v2/aweme/%s/?sec_uid=' \
                      '%s&count=%s&max_cursor=%s&aid=1128&_signature=PDHVOQAAX' \
                      'MfFyj02QEpGaDwx1S&dytk=' % (
                          self.mode, self.sec_uid, str(self.per_page_download_count), max_cursor)
            response = requests.get(
                url=api_url, headers=self.headers, timeout=30)
            html = json.loads(response.content.decode())
            max_cursor = html['max_cursor']
            result = html['aweme_list']
            self.parseVideo(result, max_cursor)
        except BaseException:
            traceback.print_exc()

    def parseVideo(self, video_result, max_cursor):
        author_list, video_list, aweme_id, nickname = [], [], [], []
        for _ in range(self.per_page_download_count):
            try:
                author_list.append(str(video_result[_]['desc']))
                video_list.append(
                    str(video_result[_]['video']['play_addr']['url_list'][0]))
                aweme_id.append(str(video_result[_]['aweme_id']))
                nickname.append(str(video_result[_]['author']['nickname']))
            except BaseException:
                pass
        self.downloadVideo(
            author_list,
            video_list,
            nickname,
            aweme_id,
            max_cursor)

    def downloadVideo(
            self,
            author_list,
            video_list,
            nickname,
            aweme_id,
            max_cursor):
        for _ in range(self.per_page_download_count):
            try:
                video = requests.get(
                    video_list[_], headers=self.headers, timeout=30)
                start = time.time()
                init_size, chunk_size, content_size = 0, 1024, int(
                    video.headers["content-length"])
                try:
                    os.makedirs(
                        os.path.join(
                            "./download/",
                            self.mode,
                            nickname[_]))
                except BaseException:
                    pass
                if video.status_code == 200:
                    self.number += 1
                    print('*' * 80)
                    print(
                        '\r' + '[  No  ]: %s | %s | %s' %
                        (self.number, max_cursor, aweme_id[_]))
                    print('[  Video  ]: ' +
                          ''.join(author_list[_][:30]) +
                          '...' +
                          '({size:.2f} MB)'.format(size=content_size /
                                                        chunk_size /
                                                        1024))
                    video_url = os.path.join(
                        "./download/",
                        self.mode,
                        nickname[_],
                        re.sub(
                            r'[\\/:*?"<>|\r\n]+',
                            "_",
                            author_list[_]) +
                        ".mp4")
                    if not os.path.exists(video_url):
                        with open(video_url, 'wb') as file:
                            for data in video.iter_content(
                                    chunk_size=chunk_size):
                                file.write(data)
                                init_size += len(data)
                                print('\r' + '[Progress]: %s%.2f%%' % (
                                    '>' * int(init_size * 50 / content_size), float(init_size / content_size * 100)),
                                      end=' ')
                            end = time.time()
                            print('\n[Done]: Use: %.2fs' % (end - start))
                    else:
                        print('\n' + '[Pass]: File exists')
                    self.cancel_favorite(aweme_id[_])
            except Exception as e:
                print("Error({}),download next\r".format(e))
                break
        self.next_data(max_cursor)

    def next_data(self, max_cursor):
        api_naxt_post_url = 'https://www.iesdouyin.com/web/api/v2/aweme/%s/?sec_uid=%s' \
                            '&count=%s&max_cursor=%s&aid=1128&_signature=RuMN1wAAJu7w0' \
                            '.6HdIeO2EbjDc&dytk=' % (self.mode, self.sec_uid,
                                                     str(self.per_page_download_count), max_cursor)
        index, result = 0, []
        while self.flag:
            if max_cursor == 0:
                self.flag = False
                return
            index += 1
            time.sleep(0.3)
            try:
                response = requests.get(
                    url=api_naxt_post_url, headers=self.headers)
                html = json.loads(response.content.decode())
                max_cursor = html['max_cursor']
                result = html['aweme_list']
            except BaseException:
                traceback.print_exc()
            self.parseVideo(result, max_cursor)

    def cancel_favorite(self, aweme_id):
        try:
            self.login_header['x-common-params-v2'] = 'channel=App%20Store&version_code=13.3.0&app_name=aweme&vid=' \
                                                      'C8CAFFA5-41C9-47A5-92B9-88A9265FEB4A&app_version=13.3.0&mcc' \
                                                      '_mnc=46011&device_id=69577004385&aid=1128&screen_width=1125' \
                                                      '&openudid=33c4e33cbc64128d9e7f3c3c4c8e04ed94b6c827&os_api=1' \
                                                      '8&os_version=13.5.1&device_platform=iphone&build_number=133' \
                                                      '014&device_type=iPhone11,2&iid=2744012310333677&idfa=784369' \
                                                      '4E-DFBB-40CF-9628-D1846C97CB7D&js_sdk_version=1.85.0.16&cdi' \
                                                      'd=4BC5D665-979C-4834-933F-C79A0C6948DD'
            data = {
                'aweme_id': aweme_id,
                'channel_id': 3,
                'type': 0
            }
            cancel_url = "https://api26-normal-lq.amemv.com/aweme/v1/commit/item/digg/?tma_jssdk_version=1.85.0.16&ac" \
                         "=WIFI&is_vcd=1&"
            req = requests.post(
                cancel_url,
                data=data,
                headers=self.login_header)
            res = json.loads(req.content)
            if res['is_digg']:
                print(
                    "[Cancel]: {} removed the favorite list success".format(aweme_id))
            else:
                print(
                    "[Cancel]: {} removed the favorite list fail".format(aweme_id))
        except BaseException:
            traceback.print_exc()


if __name__ == "__main__":
    RTK = TikTok()
    input('[  ALL Done  ]: Batch download has been completed. Enter any key to exit')
    sys.exit()

image-20210729192849094


0x03 最后随笔

抖音批量下载、批量取消点赞这样的小功能发现网上有不少和我有着同样的需求,甚至在淘宝发现有人卖这样的服务,看来又是一枚可以躺着赚钱,三年起步最高七年的姿势了,猜测对于像批量取消点赞这样需要认证的服务,那些卖家拿到买家的认证信息后可能还会去刷赞刷评论等,花了钱可能还要帮别人"助纣为虐"..

---The END---
  • 文章标题:《躺着投币之抖音辅助篇》
  • 文章作者:Coco413
  • 文章链接:https://www.coco413.com/archives/106/
  • 版权声明:本文为原创文章,仅代表个人观点,内容采用《署名-非商业性使用-相同方式共享 4.0 国际》进行许可,转载请注明出处。
  • Archives QR Code
    QR Code for this page
    Tipping QR Code