1. import os
    2. import re
    3. from queue import Queue
    4. from threading import Thread
    6. import requests
    9. class Demo(Thread):
    10.     def __init__(self, queue):
    11.         Thread.__init__(self)
    12.         self.queue = queue
    13.         self.headers = {
    14.             "user-agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.114 Mobile Safari/537.36"
    15.         }
    17.     def run(self):
    18.         while True:
    19.             video_info = self.queue.get()
    20.             try:
    21.                 self.parse(video_info)
    22.             finally:
    23.                 self.queue.task_done()
    25.     def parse(self, video_info):
    26.         file_name = re.sub(r"[/\\:*?\"<>|]", "_", video_info['desc'])
    27.         if not file_name:
    28.             file_name = video_info["aweme_id"]
    30.         file_name = video_info["sort_name"] + " " + file_name
    31.         if os.path.exists(file_name + ".mp4"):
    32.             print("文件存在,{}.mp4".format(file_name))
    33.             return
    34.         video_url = video_info['video']['play_addr']['url_list'][0]
    35.         with open(file_name + ".mp4", 'wb') as f:
    36.             f.write(requests.get(url=video_url, headers=self.headers).content)
    37.         print('下载成功: {}.mp4'.format(file_name))
    40. class DouYin:
    41.     def __init__(self):
    42.         self.share_url = input('粘贴分享链接：')
    43.         # self.share_url = '在抖音，记录美好生活！ https://v.douyin.com/ekkTsYw/'
    44.         self.headers = {
    45.             "user-agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.114 Mobile Safari/537.36"
    46.         }
    47.         self.sort_num = 1
    48.         self.video_list = []
    49.         self.queue = Queue()
    51.     # 解析粘贴内容里的url
    52.     def parse_share_url(self):
    53.         return re.findall('[a-z]+://[\S]+', self.share_url, re.I | re.M)[0]
    55.     def get_sec_uid(self):
    56.         url = self.parse_share_url()
    57.         location = requests.get(url, headers=self.headers, allow_redirects=False).headers['location']
    58.         return re.findall('sec_uid=(.*?)&', location)[0]
    60.     # 创建用户名目录
    61.     def init_save_dir(self, sec_uid):
    62.         userinfo = requests.get(url='https://www.iesdouyin.com/web/api/v2/user/info/?sec_uid={}'.format(sec_uid), headers=self.headers).json()
    63.         save_path = userinfo['user_info']['nickname']
    64.         if not os.path.exists(save_path):
    65.             os.mkdir(save_path)
    66.         os.chdir(save_path)  # 变更工作目录
    68.     def get_video_list(self, sec_uid, max_cursor=0):
    69.         url = "https://www.iesdouyin.com/web/api/v2/aweme/post/?sec_uid={}&count=21&max_cursor={}".format(sec_uid, max_cursor)
    70.         # yield
    71.         resp = requests.get(url, headers=self.headers)
    72.         data = resp.json()
    73.         if data["aweme_list"]:
    74.             for video_info in data["aweme_list"]:
    75.                 self.video_list.append(video_info)
    76.             print("下一页", url)
    77.             self.get_video_list(sec_uid, data["max_cursor"])
    79.     def run(self):
    80.         sec_uid = self.get_sec_uid()
    81.         self.init_save_dir(sec_uid)
    82.         self.get_video_list(sec_uid)
    84.         for x in range(10):
    85.             worker = Demo(self.queue)
    86.             worker.daemon = True
    87.             worker.start()
    89.         self.video_list.reverse()
    90.         for video_info in self.video_list:
    91.             video_info["sort_name"] = "[{}]".format(self.sort_num)
    92.             self.sort_num += 1
    93.             self.queue.put(video_info)
    95.         self.queue.join()
    98. if __name__ == '__main__':
    99.     DouYin().run()