X音采集
開源倉庫
https://gitee.com/erma0/douyin
介紹
Python取資料 + Vue寫界面 + Aria2下載
根據X音各種鏈接或各種id，通過網頁介面采集視頻作品，并下載作品到本地，
支持用戶主頁鏈接或sec_uid/話題挑戰和音樂原聲鏈接或ID，
支持下載喜歡串列（需喜歡串列可見），
① 2000多本Python電子書（主流和經典的書籍應該都有了）
② Python標準庫資料（最全中文版）
③ 專案原始碼（四五十個有趣且經典的練手專案及原始碼）
④ Python基礎入門、爬蟲、web開發、大資料分析方面的視頻（適合小白學習）
⑤ Python學習路線圖（告別不入流的學習）
當然在學習Python的道路上肯定會困難，沒有好的學習資料，怎么去學習呢？ 學習Python中有不明白推薦加入交流Q群號：928946953 群里有志同道合的小伙伴，互幫互助， 群里有不錯的視頻學習教程和PDF！ 還有大牛解答！
使用
0x00 安裝依賴
在程式目錄打開命令列，輸入
復制代碼 隱藏代碼
pip install -r requirements.txt0x01 使用UI界面
雙擊打開啟動.bat，或者在程式目錄打開命令列，輸入
復制代碼 隱藏代碼
python ui.py0x02 直接修改douyin.py中相關引數使用
完全不懂Python的朋友用命令列或操作界面，
0x03 從命令列使用exec.py
- 直接運行可查看命令串列，或使用-h引數查看幫助 復制代碼 隱藏代碼
python exec.py python exec.py -h python exec.py download -h python exec.py download_batch -h - 使用函式名呼叫程式 復制代碼 隱藏代碼
--type 指定下載型別，默認值：--type=user --limit 指定采集數量，默認值：--limit=0（不限制）例如采集某用戶全部作品： 復制代碼 隱藏代碼
python exec.py download https://v.douyin.com/xxxx/ python exec.py download 用戶的secuid例如采集某用戶喜歡的前10個作品： 復制代碼 隱藏代碼
python exec.py download MS4wLjABAAAAl7TJWjJJrnu11IlllB6Mi5V9VbAsQo1N987guPjctc8 --type=like --limit=10 python exec.py download 用戶的secuid例如采集某音樂原聲前10個作品： 復制代碼 隱藏代碼
python exec.py download https://v.douyin.com/xxxx/ --type=music --limit=10 python exec.py download 音樂ID --type=music --limit=10
TODO
- [x] 采集用戶作品
- [x] 呼叫Aria2下載
- [x] 話題/原聲作品采集
- [x] 喜歡作品采集
- [x] 匯入檔案批量采集
- [x] 命令列呼叫
- [x] 用webview寫界面
- [x]
打包exe不打包了，直接裝個Python環境更簡單
知識點
X音相關
- 網頁介面恢復了，一次請求即可取回資料
- UID幾乎沒用了，拼不成主頁鏈接了，所有介面都是sec_uid
- signature可固定了，不用再扣JS了
- 作品中直接包含無水印視頻地址了，不需要移動端UA也可跳轉
- 話題/音樂作品數目
- 2021.04.02 喜歡串列也有資料了
Aria2相關
- aria2p庫使用體驗還不錯
- 大部分Aria2下載都是通過rpc介面實作的，這個也一樣
- 需要自己下載Aria2c.exe來開啟服務，所以要用代碼實作自動啟動服務
- 若檔案已存在則跳過下載的方法：
- --auto-file-renaming=false 可行，但控制臺使用會報錯，雖然報錯不影響
- -c 可行，且控制臺不報錯
- 添加下載任務時通過指定options = {'out': filename}指定檔案名，即-d
- Aria2會根據指定路徑及指定檔案名自動創建下載目錄
- Aria2指定路徑及檔案名中不能傳入非法字串（*|等），所以寫了Download.title2path靜態方法
- 監聽事件要手動停止，不停止會阻塞行程，導致程式無法關閉
- 未發現實時獲取任務進度及下載速度的函式，自己寫了回圈監聽回呼方法
Python相關
- 通過os.popen或subprocess.Popen實作子行程打開程式，無界面，不阻塞
- 繼承父類后重寫init時，通過super().init()呼叫父類構造方法
- 繼承父類后重寫方法時，不能重寫私有方法，不能讀取私有成員
- 引數指定型別提示挺好用，方便呼叫引數的函式時自動補全
- 可通過if 'PROGRAMFILES(X86)' in os.environ簡單判斷系統是否為64位
- Pylance的自動匯入依賴功能很好用，就是感覺時靈時不靈，重新開關后又可以用
- vscode默認啟動路徑是當前專案路徑，在launch.json中加一句"cwd": "${fileDirname}",即可，不過自動補全pylance就無法識別相對目錄了
- 用pipreqs一鍵生成當前專案依賴：cmd切換到專案路徑，pipreqs ./ --encoding=utf-8 --force
命令列模塊fire相關
- 最簡單的方法就是直接一個fire.Fire()，暴露全部函式
- 如果用類或物件暴露，類引數需要單獨指定
- 組合命令需要用不同的類，暴露的類中引入需用組合命令的類，但是在這個批量下載的場景下感覺比較繁瑣，所以直接加了個引數，分兩個函式來呼叫
UI模塊pywebview相關
- 可以把一個類的實體暴露給頁面js_api，通過pywebview.api.func().then(() => {})呼叫Python函式
- 也可以把Flask等服務實體暴露給頁面js_api（無需url引數），在內部實作index.html
- Python中通過window.evaluate_js('JS代碼')呼叫JS方法
- 在UI中，類的初始化無法傳參，所以需要重新定義init
- 在UI中，需要公開的類實體方法不能以下劃線_開頭
- 創建UI時設定的視窗寬高，好像和網頁中大小不一樣，值需要比網頁中大一些
X音采集部分原始碼
復制代碼 隱藏代碼 # -*- encoding: utf-8 -*- ''' @File : douyin.py @Time : 2021年03月12日 18:16:57 星期五 @Author : erma0 @Version : 1.0 @Link : https://erma0.cn @Desc : X音用戶作品采集 ''' import json import os import time from urllib.parse import parse_qs, urlparse import requests from download import Download class Douyin(object): """ X音用戶類 采集作品串列 """ def __init__(self, param: str, limit: int = 0): """ 初始化用戶資訊 引數自動判斷：ID/URL """ self.limit = limit self.http = requests.Session() self.url = '' self.type = 'unknow' self.download_path = '暫未定義目錄' # ↑ 預定義屬性，避免呼叫時未定義 ↑ self.param = param.strip() self.sign = 'TG2uvBAbGAHzG19a.rniF0xtrq' # sign可以固定 self.__get_type() # 判斷當前任務型別：鏈接/ID self.aria2 = Download() # 初始化Aria2下載服務，先不指定目錄了，在設定檔案名的時候再加入目錄 self.has_more = True self.finish = False # 字典格式方便入庫用id做key/取值/修改對應資料，但是表格都接收陣列 self.videosL = [] #串列格式 # self.videos = {} #字典格式 self.gids = {} # gid和作品序號映射 def __get_type(self): """ 判斷當前任務型別 鏈接/ID """ if '://' in self.param: # 鏈接 self.__url2redirect() else: # ID self.id = self.param def __url2redirect(self): """ 取302跳轉地址 短連接轉長鏈接 """ headers = { # 以前作品需要決議去水印，要用到移動端UA，現在不用了 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 Mobile/15E148 Safari/604.1 Edg/89.0.4389.82' } try: r = self.http.head(self.param, headers=headers, allow_redirects=False) self.url = r.headers['Location'] except: self.url = self.param def __url2id(self): try: self.id = urlparse(self.url).path.split('/')[3] except: self.id = '' def __url2uid(self): try: query = urlparse(self.url).query self.id = parse_qs(query)['sec_uid'][0] except: self.id = '' def get_sign(self): """ 網頁sign演算法，現在不需要了，直接固定 """ self.sign = 'TG2uvBAbGAHzG19a.rniF0xtrq' return self.sign def get_user_info(self): """ 取用戶資訊 查詢結果在 self.user_info """ if self.url: self.__url2uid() url = 'https://www.iesdouyin.com/web/api/v2/user/info/?sec_uid=' + self.id try: res = self.http.get(url).json() info = res.get('user_info', dict()) except: info = dict() self.user_info = info # 下載路徑 username = '{}_{}_{}'.format(self.user_info.get('short_id', '0'), self.user_info.get('nickname', '無昵稱'), self.type) self.download_path = Download.title2path(username) # 需提前處理非法字串 def get_challenge_info(self): """ 取話題挑戰資訊 查詢結果在 self.challenge_info """ if self.url: self.__url2id() url = 'https://www.iesdouyin.com/web/api/v2/challenge/info/?ch_id=' + self.id try: res = self.http.get(url).json() info = res.get('ch_info', dict()) except: info = dict() self.challenge_info = info # 話題挑戰下載路徑 username = '{}_{}_{}'.format(self.challenge_info.get('cid', '0'), self.challenge_info.get('cha_name', '無標題'), self.type) self.download_path = Download.title2path(username) # 需提前處理非法字串 def get_music_info(self): """ 取音樂原聲資訊 查詢結果在 self.music_info """ if self.url: self.__url2id() url = 'https://www.iesdouyin.com/web/api/v2/music/info/?music_id=' + self.id try: res = self.http.get(url).json() info = res.get('music_info', dict()) except: info = dict() self.music_info = info # 音樂原聲下載路徑 username = '{}_{}_{}'.format(self.music_info.get('mid', '0'), self.music_info.get('title', '無標題'), self.type) self.download_path = Download.title2path(username) # 需提前處理非法字串 def crawling_users_post(self): """ 采集用戶作品 """ self.type = 'post' self.__crawling_user() def crawling_users_like(self): """ 采集用戶喜歡 """ self.type = 'like' self.__crawling_user() def crawling_challenge(self): """ 采集話題挑戰 """ self.type = 'challenge' self.get_challenge_info() # 取當前資訊，用做下載目錄 # https://www.iesdouyin.com/web/api/v2/challenge/aweme/?ch_id=1570693184929793&count=9&cursor=9&aid=1128&screen_limit=3&download_click_limit=0&_signature=AXN-GQAAYUTpqVxkCT6GHQFzfg url = 'https://www.iesdouyin.com/web/api/v2/challenge/aweme/' cursor = '0' while self.has_more: params = { "ch_id": self.id, "count": "21", # 可調大 初始值：9 "cursor": cursor, "aid": "1128", "screen_limit": "3", "download_click_limit": "0", "_signature": self.sign } try: res = self.http.get(url, params=params).json() cursor = res['cursor'] self.has_more = res['has_more'] self.__append_videos(res) except: print('話題挑戰采集出錯') print('話題挑戰采集完成') def crawling_music(self): """ 采集音樂原聲 """ self.type = 'music' self.get_music_info() # 取當前資訊，用做下載目錄 # https://www.iesdouyin.com/web/api/v2/music/list/aweme/?music_id=6928362875564067592&count=9&cursor=18&aid=1128&screen_limit=3&download_click_limit=0&_signature=5ULmIQAAhRYNmMRcpDm2COVC5j url = 'https://www.iesdouyin.com/web/api/v2/music/list/aweme/' cursor = '0' while self.has_more: params = { "music_id": self.id, "count": "21", # 可調大 初始值：9 "cursor": cursor, "aid": "1128", "screen_limit": "3", "download_click_limit": "0", "_signature": self.sign } try: res = self.http.get(url, params=params).json() cursor = res['cursor'] self.has_more = res['has_more'] self.__append_videos(res) except: print('音樂原聲采集出錯') print('音樂原聲采集完成') def __crawling_user(self): """ 采集用戶作品/喜歡 """ self.get_user_info() # 取當前用戶資訊，昵稱用做下載目錄 max_cursor = 0 # https://www.iesdouyin.com/web/api/v2/aweme/like/?sec_uid=MS4wLjABAAAAaJO9L9M0scJ_njvXncvoFQj3ilCKW1qQkNGyDc2_5CQ&count=21&max_cursor=0&aid=1128&_signature=2QoRnQAAuXcx0DPg2DVICdkKEY&dytk= # https://www.iesdouyin.com/web/api/v2/aweme/post/?sec_uid=MS4wLjABAAAAaJO9L9M0scJ_njvXncvoFQj3ilCKW1qQkNGyDc2_5CQ&count=21&max_cursor=0&aid=1128&_signature=DrXeeAAAbwPmb.wFM3e63w613m&dytk= url = 'https://www.iesdouyin.com/web/api/v2/aweme/{}/'.format(self.type) while self.has_more: params = { "sec_uid": self.id, "count": "21", "max_cursor": max_cursor, "aid": "1128", "_signature": self.sign, "dytk": "" } try: res = self.http.get(url, params=params).json() max_cursor = res['max_cursor'] self.has_more = res['has_more'] self.__append_videos(res) except: print('作品采集出錯') print('作品采集完成') def __append_videos(self, res): """ 資料入庫 """ if res.get('aweme_list'): for item in res['aweme_list']: info = item['statistics'] info.pop('forward_count') info.pop('play_count') info['desc'] = Download.title2path(item['desc']) # 需提前處理非法字串 info['uri'] = item['video']['play_addr']['uri'] info['play_addr'] = item['video']['play_addr']['url_list'][0] info['dynamic_cover'] = item['video']['dynamic_cover']['url_list'][0] info['status'] = 0 # 下載進度狀態；等待下載：0，下載中：0.xx；下載完成：1 # 串列格式 self.videosL.append(info) # 字典格式 # self.videos[info['aweme_id']] = info # 此處可以直接添加下載任務，不過考慮到下載占用網速,影響采集程序，所以采集完再下載 if self.limit: more = len(self.videos) - self.limit if more >= 0: # 如果給出了限制采集數目，超出的洗掉后直接回傳 self.has_more = False # 串列格式 self.videosL = self.videosL[:self.limit] # 字典格式 # for i in range(more): # self.videos.popitem() # return else: # 還有作品的情況下沒回傳資料則進入這里 print('未采集完成，但回傳作品串列為空') def download_all(self): """ 作品抓取完成后，統一添加下載任務 可選擇在外部注冊回呼函式，監聽下載任務狀態 """ for id, video in enumerate(self.videosL): # for id, video in self.videos.items(): gid = self.aria2.download(url=video['play_addr'], filename='{}/{}_{}.mp4'.format(self.download_path, video['aweme_id'], video['desc']) # ,options={'gid': id} # 指定gid ) self.gids[gid] = id # 因為傳入gid必須16位，所以就不指定gid了，另存一個字典映射 print('下載任務投遞完成')
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/296029.html
標籤：Python