Download_video --功能說明
@功能1：檢測能否連接到數(shù)據(jù)庫、獲取數(shù)據(jù)庫集合中總共的數(shù)據(jù)（條）
@功能2：連接數(shù)據(jù)庫查詢數(shù)據(jù)
@功能3：批量下載視頻、異步非阻塞多線程
@功能4：工作進度保存機制
-
實際效果
開啟 “從上一次下載進度的位置啟動.png
關(guān)閉 “從上一次下載進度的位置啟動.png
運行(多線程下載4線程).png
Requests_video --功能說明
@功能1：批量爬取數(shù)據(jù)(發(fā)送請求、獲取標題、獲取高清鏈接、獲取標清鏈接、)
@功能2：可視化進度條
@功能3：連接數(shù)據(jù)庫存儲數(shù)據(jù)
-
實際效果
運行狀態(tài).png
聲明：該文章所有代碼僅供學(xué)習(xí)參考使用，若使用不法途徑本人一概不負。
# -*- encoding: utf-8 -*-
"""
@File : Download_video.py
@Contact : t.ianxi@foxmail.com
@License : // Copyright (C) 2018 Milo Yip<dell>
@Modify Time @Author @Version @Desciption
------------ ------- -------- -----------
2022/8/30 20:55 MuKe~ 1.0 None
"""
import sys
import time
import pymongo
import os
import threading
# ================================================= 檢測網(wǎng)絡(luò)\統(tǒng)計 ========================================================
def Mongodb_find_net_num(Host,Post,input_Nome):
"""
檢測能否連接到數(shù)據(jù)庫、獲取數(shù)據(jù)庫集合中總共的數(shù)據(jù)（條）
:param Host: ip地址
:param Post: 端口號
:param input_Nome: 數(shù)據(jù)庫集合
:return: 返回當(dāng)前數(shù)據(jù)庫集合中總共的數(shù)據(jù)（條）
"""
try: # 捕獲連接數(shù)據(jù)庫時可能發(fā)生的異常
Client = pymongo.MongoClient(host=Host, port=Post) # 連接數(shù)據(jù)庫
db = Client[Mongodb_name] # 數(shù)據(jù)庫名稱
Data_db = db[input_Nome] # 集合名稱
Mongodb_num = Data_db.find().count() # 查詢數(shù)據(jù)庫當(dāng)前集合中數(shù)據(jù)的總共數(shù)量（條）
return Mongodb_num # 返回統(tǒng)計后的數(shù)量
except: # 發(fā)生異常后，并返回False
return False
# ==================================================== 查詢數(shù)據(jù)庫 ========================================================
def Mongodb_post(Host,Post,Mongodb_name,input_Nome,page_data):
"""
連接數(shù)據(jù)庫獲取數(shù)據(jù)
:param Host: ip地址
:param Post: 端口號
:param Mongodb_name:數(shù)據(jù)庫名稱
:param input_Nome:數(shù)據(jù)庫集合
:param Yema: ‘當(dāng)前頁面’id
:return: 字典、數(shù)據(jù)內(nèi)容：（作品名稱、上架時間、下載鏈接）
"""
Client = pymongo.MongoClient(host=Host, port=Post) # 連接數(shù)據(jù)庫
db = Client[Mongodb_name] # 數(shù)據(jù)庫名稱
print("[INFO] 當(dāng)前數(shù)據(jù)庫名稱：" + input_Nome) # 輸出集合名稱
Data_db = db[input_Nome] # 輸出集合名稱
results = Data_db.find({"寫入序號":str(page_data)}) # 查詢下數(shù)據(jù)庫中的key值
for key in results: # 歷遍獲取到的字典數(shù)據(jù)
# print("video_nome：", key['作品標題'] + ' ' + key['時間數(shù)據(jù)'], key['標清下載'])
Mongodb_DataDB = {
"Data_db_name":key['作品標題'],
"Data_db_page":key['頁面位置'],
"Data_db_date":key['時間數(shù)據(jù)'],
"Data_db_download":key['標清下載']
}
return Mongodb_DataDB # 對字典中的數(shù)據(jù)重命名,并返回
# ==================================================== 下載文件 =======================================================
def Download_URL_video(Mongodb_DataDB,video_path):
"""
下載視頻、異步非阻塞多進程
Mongodb_DataDB = {
:param video_name: 作品名稱
:param video_date: 上架時間
:param video_downloadurl: 下載鏈接
:param video_path: 文件存儲路徑
}
:return: Nome
"""
video_name = Mongodb_DataDB['Data_db_name'] # 獲取字典中key值
video_date = Mongodb_DataDB['Data_db_date']
video_downloadurl = Mongodb_DataDB['Data_db_download']
print('==============================================================================================================')
print("[INFO] "+video_name)
print("[INFO] "+video_date)
print("[++++] "+video_path)
print("[INFO] "+video_downloadurl)
# you-get使用方法參考文章 https://github.com/soimort/you-get/blob/develop/README.md
# you-get -o 路徑 -O 文件名 鏈接
# print(f'you-get -o {video_path} -O {video_name} {video_date} {video_downloadurl}')
if not os.path.exists(video_path + '\\' + video_name + '.mp4'): # 判斷文件是否存在
os.system(f'you-get -o {video_path} -O {video_name} {video_date} {video_downloadurl}') # 通過傳參調(diào)用’you-get‘下載文件
else:
print("[INFO] 已存在此作品",video_name + '.mp4')
# =================================================== 記錄下載進度 ====================================================
def Detect_MP4(video_path,page_data):
"""
寫入下載進度數(shù)據(jù)（計數(shù)）
:param video_path: 文件的存儲路徑
:param page_data: 數(shù)據(jù)庫“當(dāng)前頁面”數(shù)據(jù)
:return: None
"""
try: # 用于捕獲IO偶爾會發(fā)生的意外（如：寫入文件被占用）
with open(video_path+'\\count.txt','w',encoding='utf-8') as count:
count.write(str(page_data + 1)) # 寫入字符串 （計數(shù)）
except OSError as Os: # 捕獲并輸出異常
print(f'[INFO] 無法寫入下載進度，異常原因：{Os}')
# ==================================================== 配置信息 ======================================================
Host = "10.52.xx.xx" # Mongodb數(shù)據(jù)庫地址(ip)
Post = 27017 # Mongodb數(shù)據(jù)庫端口
Mongodb_name = "video_URL" # Mongodb數(shù)據(jù)庫名
# (批量操作)Mongodb數(shù)據(jù)庫集合名，以及起始位置
URL_class_dict = {
"vlg_抖音":2,
"國產(chǎn)網(wǎng)劇":2,
"卡通動漫":2,
"微視頻":3
}
# input_Nome = "yazhouqingse" # (單個操作)Mongodb數(shù)據(jù)庫集合名
OFF_ON_page = True # 開啟/關(guān)閉 “從上一次下載進度的位置啟動”（若讀取失敗或者為0則默認為字典的默認初始化值）
video_path = r'D:\Pyfile\腳本\視頻url收集' # 文件存儲路徑
thread_time = int(0.2) # 線程減速
thread_list = [] # 存儲線程
thread_num = int(os.cpu_count()) # 定義線程數(shù)量(檢測cpu核數(shù))
# ==================================================== 程序入口(main) ====================================================
if __name__=='__main__': # 定義一個main
# 該方法只適合(批量操作)，數(shù)據(jù)庫集合只有一個情況不需要一下三行代碼
for D in URL_class_dict: # 歷遍包含數(shù)據(jù)庫集合名稱字典及起始位置
input_Nome = D # 獲取key（集合名稱）
page_data = URL_class_dict[D] # 獲取起始位置
Mongodb_num = Mongodb_find_net_num(Host, Post, input_Nome) # 實例化“檢測網(wǎng)絡(luò)\統(tǒng)計”方法
if Mongodb_num != False: # 判斷數(shù)據(jù)庫連接是否異常
print("[INFO] 數(shù)據(jù)庫已連接")
for i in range(1,Mongodb_num): # 歷遍數(shù)據(jù)庫當(dāng)前集合中數(shù)據(jù)的總數(shù)(條)
if OFF_ON_page == True: # 判斷“配置信息”如果用戶開啟則讀取上一次的下載進度
try: # 捕獲IO在讀寫時偶爾會發(fā)生的意外
with open(video_path+'\\count.txt','r',encoding='utf-8') as count0: # 讀取“count.txt”配置文件(用于存儲下載進度)
page_data = int(count0.read()) # 讀取數(shù)值
print(f'[++++] 上一次進度停留在：{page_data}')
except: # 捕獲異常并賦值為字典的默認初始化值作為數(shù)據(jù)庫查詢條件
print("[INFO] 存儲進度文件“count.txt”已丟失！")
with open(video_path + '\\count.txt', 'w+', encoding='utf-8'):
pass
page_data = URL_class_dict[D]
print(f"[INFO] 已改為字典的默初始化值{page_data}")
Mongodb_DataDB = Mongodb_post(Host, Post,Mongodb_name,input_Nome,page_data) # 實例化“查詢數(shù)據(jù)庫”方法
t = threading.Thread(target=Download_URL_video, args=(Mongodb_DataDB,video_path)) # 創(chuàng)建多線程、調(diào)用“下載文件”方法
thread_list.append(t) # 向列表中加入線程池
t.start() # 啟動線程
print('[INFO] 線程池',thread_list[0]) # 輸出線程池信息
while len(thread_list) > thread_num: # 統(tǒng)計線程池總數(shù)，若大于獲取到cpu核數(shù)（如：4核）
thread_list = [x for x in thread_list if x.is_alive()] # 歷遍線程池中的線程，判斷線程是否在運行
time.sleep(thread_time) # 設(shè)備性能較差防止循環(huán)過快
# print('剩余線程池________' + str(thread_list))
if page_data <= Mongodb_num: # 判斷如果數(shù)據(jù)庫中總數(shù)小于當(dāng)前下載進度 則繼續(xù)寫入下載進度(用于切換數(shù)據(jù)庫集合時防止下載進度累計增加)
Detect_MP4(video_path,page_data) # 實例化“記錄下載進度”方法，寫入下載進度一邊下次啟動使用
elif page_data >= Mongodb_num:
print(f"[INFO] 已達到數(shù)據(jù)庫閾值!{input_Nome}集合總數(shù)：{Mongodb_num}")
time.sleep(10)
Detect_MP4(video_path,0) # 判斷如果數(shù)據(jù)庫中總數(shù)大于當(dāng)前下載進度 則重置寫入下載進度(用于切換數(shù)據(jù)庫集合時重置下載進度)
break # 跳出循環(huán)切換數(shù)據(jù)庫集合
elif Mongodb_num == False: # 判斷“檢測網(wǎng)絡(luò)\統(tǒng)計”是否異常，并輸出信息
print("[INFO] 數(shù)據(jù)庫連接失??！")
# -*- encoding: utf-8 -*-
"""
@File : Requests_video.py
@Contact : t.ianxi@foxmail.com
@License : // Copyright (C) 2018 Milo Yip<dell>
@Modify Time @Author @Version @Desciption
------------ ------- -------- -----------
2022/7/4 22:23 MuKe~ 1.0 None
"""
"""
@File : Requests_video.py
@Contact : t.ianxi@foxmail.com
@License : // Copyright (C) 2018 Milo Yip<dell>
@Modify Time @Author @Version @Desciption
------------ ------- -------- -----------
2022/7/4 22:23 MuKe~ 1.0 None
"""
import time
import requests
import csv
import os
import datetime
import pymongo
from lxml import etree
from tqdm import tqdm
# 網(wǎng)頁類目，字典：卡通動漫、國產(chǎn)網(wǎng)劇、vlg_抖音、微視頻，列表：開始位置，結(jié)束位置
URL_class_dict = {
"卡通動漫":[2,58],
"國產(chǎn)網(wǎng)劇":[2,123],
"vlg_抖音:[2,135],
"微視頻":[3,78]
}
for i in URL_class_dict.items():
il = list(i)
input_Start = il[1][0] # 起始頁
input_Stop = il[1][1] # 終止頁
input_Nome = il[0] # 數(shù)據(jù)庫集合名(表)、網(wǎng)頁視頻類目參數(shù)
# ======================================================================================================================
def GET_urlnome():
"""
發(fā)送請求、獲取標題、獲取高清鏈接、獲取標清鏈接、
:return:"頁面位置":i+1,
"時間數(shù)據(jù)":time_data,
"作品標題":nome_video,
"高清下載":HD_video,
"""
for i in range(input_Start,input_Stop): # 實現(xiàn)翻頁功能
URL = "https://taxm.com/{Class_url}/index_{i}.html".format(i=i,Class_url=input_Nome)
headers = {
"referer": "https://taxm.com/",
"User-Agent": "Mozilla/4.0 (Windows NT 11.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4430.85 Safari/537.16"
}
# 參考文章https://www.cnblogs.com/Renyi-Fan/p/13266518.html#:~:text=requests.Session,%28%29%EF%BC%9A%E7%BB%B4%E6%8C%81%E4%BC%9A%E8%AF%9D%EF%BC%8C%E5%8F%AF%E4%BB%A5%E8%AE%A9%E6%88%91%E4%BB%AC%E5%9C%A8%E8%B7%A8%E8%AF%B7%E6%B1%82%E6%97%B6%E4%BF%9D%E5%AD%98%E6%9F%90%E4%BA%9B%E5%8F%82%E6%95%B0%EF%BC%8C%E8%BF%99%E6%A0%B7%E5%8F%AF%E4%BB%A5%E5%BE%97%E5%88%B0%E4%B8%80%E4%B8%AAsession%E5%AF%B9%E8%B1%A1%EF%BC%8C%E8%80%8C%E8%BF%99%E4%B8%AAsession%E5%AF%B9%E8%B1%A1%E6%98%AF%E5%8F%AF%E4%BB%A5%E5%8F%91%E9%80%81%E5%90%84%E7%A7%8D%E8%AF%B7%E6%B1%82%E7%9A%84
Session = requests.Session() # Session() 持會話，可以讓我們在跨請求時保存某些參數(shù)，這樣可以得到一個session對象，而這個session對象是可以發(fā)送各種請求的
GET_url = Session.get(URL,headers=headers).text # 發(fā)送get請求，并調(diào)用text函數(shù)將數(shù)據(jù)類型轉(zhuǎn)換為文本信息
GET_html = etree.HTML(GET_url) # 使用lxml庫的etree下的HTML()方法解析text文本為html
bue = tqdm(range(1,36)) # 實例tqdm對象 歷遍url，從1開始36結(jié)束
for i_s in bue:
Ye_url = GET_html.xpath('/html/body/div[11]/div[5]/ul/li[{i}]/a/@href'.format(i=int(i_s))) # 獲取作品鏈接 href屬于屬性
video_url = "https://taxm.com"+Ye_url[0] # 拼接作品url(每頁共有36個作品)
response = Session.get(video_url, headers=headers) # 調(diào)用Session的get()方法發(fā)送請求，參數(shù)(url，請求頭)
response.encoding = response.apparent_encoding # html文本為外語，轉(zhuǎn)換編碼 注意：沒有轉(zhuǎn)換會出現(xiàn)亂碼
video_html = etree.HTML(response.text)# 發(fā)送get請求，并調(diào)用text函數(shù)將數(shù)據(jù)類型轉(zhuǎn)換為文本信息，并使用lxml庫的etree下的HTML()方法解析text文本為html
HD_video = video_html.xpath('/html/body/div[11]/div[5]/div[7]/ul[2]/li[1]/a/@href') # 獲取標清鏈接
nome_video = video_html.xpath('/html/body/div[11]/div[3]/span[4]/text()') # 獲取作品標題
# print("標清下載:",HD_video)
# print("作品標題：",nome_video) # 進度條標題，格式化字符串參數(shù)(頁碼，標題)
bue.set_description('第{State}頁:{print_nome}'.format(State=i+1,print_nome=nome_video))
now = datetime.datetime.now() # 實例化datetime庫的now()方法
time_data = now.strftime('%Y-%m-%d %H:%M:%S') # 獲取當(dāng)前時間
input_Mongdb = {
"頁面位置":i+1,
"時間數(shù)據(jù)":time_data,
"作品標題":nome_video[0],
"高清下載":HD_video[0],
}
# print(input_Mongdb)
# 實例化save_to_mongo()(自定義的方法，連接Mongdb數(shù)據(jù)庫、存儲數(shù)據(jù))
save_to_mongo(result=input_Mongdb)
# 定義一個數(shù)據(jù)庫
MONGO_DB = 'video_URL'
# 指定集合名(表)
MONGO_COLLECTION = input_Nome
# 連接MongoDB數(shù)據(jù)庫參數(shù)一為主機或IP，參數(shù)二為端口號默認27017 (連接數(shù)據(jù)庫)
client = pymongo.MongoClient(host='localhost',port=27017)
# 指定數(shù)據(jù)庫
db = client[MONGO_DB]
# ======================================================================================================================
def save_to_mongo(result):
"""
:param result:結(jié)果
:return: None
"""
try:
# 調(diào)用insert()方法將數(shù)據(jù)插入到Mongdb數(shù)據(jù)庫中，result變量是get_products()方法中傳來的product變量(包含著商品相關(guān)數(shù)據(jù))
if db[MONGO_COLLECTION].insert(result):
# print("儲存到MongDB成功！")
pass
except Exception:
print("儲存到MongDB失??！")
if __name__=='__main__':
GET_urlnome()
