[Python] 纯文本查看 复制代码
# -*- coding:utf-8 -*-
# @ FileName :dyzbrecorder.py
# @ Author :dltest@52pojie
import datetime
import tkinter as tk
import tkinter.font as tkFont
from tkinter import *
import tkinter.ttk as ttk
import random
import requests
import re
import os
import urllib.request
import urllib.parse
import sys
from bs4 import BeautifulSoup
import time
import json
# import vlc 播放
import configparser
import subprocess
import logging
import threading
import string
import jsonpath
thd = set()
LOG_LINE_NUM =0
class App:
def __init__(self, root):
self.editdate = '2022/10/14'
self.initUI(root)
self.initData()
self._log('直播间地址为分享链接https://v.douyin.com/xxxxxxx/,xxxxxxx这部分的7个字符')
def initData(self):
self.tag = 'odd'
self.count = 0
self.dir = ''
self.flag = True
# self.GLineEdit_863.insert(0,'M6e8gAp')
self.base_url = 'https://v.douyin.com/' #https://v.douyin.com/M6e8gAp/
self.baseinfo = 'https://www.iesdouyin.com/web/api/v2/user/info/?'
self.headers = {'User-Agent': 'Mozilla/5.0 (Linux; U; Android 8.1.0; en-US; Nexus 6P Build/OPM7.181205.001) '
'AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 '
'Chrome/57.0.2987.108 UCBrowser/12.11.1.1197 Mobile Safari/537.36'}
def initUI(self,root):
#setting title
root.title(f"直播录制之抖音版 {self.editdate} dltest@52pojie")
#setting window size
width=684
height=516
screenwidth = root.winfo_screenwidth()
screenheight = root.winfo_screenheight()
alignstr = '%dx%d+%d+%d' % (width, height, (screenwidth - width) / 2, (screenheight - height) / 2)
root.geometry(alignstr)
root.resizable(width=False, height=False)
self.GLabel_257=tk.Label(root)
ft = tkFont.Font(family='Times',size=10)
self.GLabel_257["font"] = ft
self.GLabel_257["fg"] = "#333333"
self.GLabel_257["justify"] = "center"
self.GLabel_257["text"] = "直播间地址"
self.GLabel_257.place(x=10,y=10,width=70,height=25)
self.GLineEdit_863=tk.Entry(root)
self.GLineEdit_863["borderwidth"] = "1px"
ft = tkFont.Font(family='Times',size=10)
self.GLineEdit_863["font"] = ft
self.GLineEdit_863["fg"] = "#333333"
self.GLineEdit_863["justify"] = "center"
self.GLineEdit_863["text"] = "address"
self.GLineEdit_863.place(x=90,y=10,width=144,height=30)
self.GButton_168=tk.Button(root)
self.GButton_168["bg"] = "#efefef"
ft = tkFont.Font(family='Times',size=10)
self.GButton_168["font"] = ft
self.GButton_168["fg"] = "#000000"
self.GButton_168["justify"] = "center"
self.GButton_168["text"] = "开始"
self.GButton_168.place(x=250,y=10,width=95,height=30)
self.GButton_168["command"] = self.GButton_168_command
self.GButton_896=tk.Button(root)
self.GButton_896["bg"] = "#efefef"
ft = tkFont.Font(family='Times',size=10)
self.GButton_896["font"] = ft
self.GButton_896["fg"] = "#000000"
self.GButton_896["justify"] = "center"
self.GButton_896["text"] = "停止"
self.GButton_896.place(x=410,y=10,width=94,height=30)
self.GButton_896["command"] = self.GButton_896_command
self.GLineEdit_371=tk.Text(root)
self.GLineEdit_371["borderwidth"] = "1px"
ft = tkFont.Font(family='Times',size=10)
self.GLineEdit_371["font"] = ft
self.GLineEdit_371["fg"] = "#333333"
self.GLineEdit_371.place(x=10,y=50,width=666,height=454)
self.GButton_448=tk.Button(root)
self.GButton_448["bg"] = "#efefef"
ft = tkFont.Font(family='Times',size=10)
self.GButton_448["font"] = ft
self.GButton_448["fg"] = "#000000"
self.GButton_448["justify"] = "center"
self.GButton_448["text"] = "打开文件夹"
self.GButton_448.place(x=560,y=10,width=94,height=30)
self.GButton_448["command"] = self.GButton_448_command
def GButton_168_command(self): #开始
inurl = self.base_url+self.GLineEdit_863.get()
self.getData(inurl)
def getData(self,inurl):
text_data=self.get_live_data(inurl)
if text_data.startswith('error'):
self._log(text_data)
return
live_data=json.loads(text_data)
zhubo=live_data["data"]["room"]["owner"]['nickname']
sen_text = re.compile(u'[\u4E00-\u9FA5|\s\w]').findall(zhubo)
zhubo = "".join(sen_text)
self._log(f'解析成功，主播名：{zhubo}')
hls_pull_url = live_data["data"]["room"]["stream_url"]['hls_pull_url_map']['FULL_HD1']
self._log(f'直播地址{hls_pull_url}')
now = time.strftime("%Y-%m-%d-%H-%M-%S",time.localtime(time.time()))
if not os.path.exists(zhubo):
os.makedirs('./'+zhubo)
self.dir =os.getcwd() + '\\' + zhubo
print(self.dir)
filename=zhubo + '_' + now + ".ts"
ffmpeg_path = "ffmpeg"
file = os.getcwd() + '\\' +zhubo +'\\'+ filename
print(file)
self._log(f'保存路径{file}')
start = datetime.datetime.now()
self.download(ffmpeg_path,hls_pull_url,file)
self.flag = True
obj1 = threading.Thread(target=self.displayinfo, args=({start,zhubo}))
obj1.setDaemon(True)
obj1.start()
def displayinfo(self,start,zhubo):
time.sleep(10)
while self.flag:
time.sleep(30)
self._log('循环值守录制抖音直播')
self._log("x"*60)
end = datetime.datetime.now()
self._log(f"{zhubo} 正在录制中")
self._log('总共录制时间: ' +str(end - start))
self._log("x"*60)
return
def download(self,ffmpeg_path,hls_pull_url,file):
try:
_output = subprocess.Popen([
ffmpeg_path, "-y",
"-v","verbose",
"-rw_timeout","10000000", # 10s
"-loglevel","error",
"-hide_banner",
# "-user_agent",headers["User-Agent"],
"-analyzeduration","2147483647",
"-probesize","2147483647",
"-i",hls_pull_url,
'-bufsize','5000k',
"-map","0",
"-sn","-dn",
"-f","mpegts",
# "-bsf:v","h264_mp4toannexb",
# "-c","copy",
"-c:v","copy",
'-max_muxing_queue_size','64',
"{path}".format(path=file),
], stderr = subprocess.STDOUT,stdin=subprocess.PIPE)
except subprocess.CalledProcessError as exc:
print(str(exc.output) +" 发生错误的行数: "+str(exc.__traceback__.tb_lineno) )
self.flag = False
global thd
thd.add(_output)
def GButton_896_command(self):
self.flag = False
global thd
for i in thd:
i.stdin.write('q'.encode("GBK"))
i.communicate()
thd.clear()
def GButton_448_command(self): #打开目录
path = self.dir
if path:
self.open_fp(path)
def open_fp(self, fp):
import platform
systemType: str = platform.platform() # 获取系统类型
if 'mac' in systemType: # 判断以下当前系统类型
fp: str = fp.replace("\\", "/") # mac系统下,遇到`\\`让路径打不开,不清楚为什么哈,觉得没必要的话自己可以删掉啦,18行那条也是
subprocess.call(["open", fp])
else:
fp: str = fp.replace("/", "\\") # win系统下,有时`/`让路径打不开
try:
os.startfile(fp)
except:
self._log(f'{"-" * 20}文件还未下载{"-" * 20}')
def get_live_data(self,url):
#短连接解析长链接
response = requests.get(url,headers=self.headers)
dy_url = response.url
print(url)
#取直播间ID
reflow_id_list = re.findall(".*reflow/(.*)\?u_code=.*", dy_url)
if len(reflow_id_list)==0:
sp = dy_url.split('?')
if 'share/user' in dy_url and len(sp) == 2:
param = sp[1]
self.userinfourl = self.baseinfo + param
userinfo = requests.get(self.userinfourl,headers=self.headers).json()
self.nickname = userinfo['user_info']['nickname']
sen_text = re.compile(u'[\u4E00-\u9FA5|\s\w]').findall(self.nickname)
self.nickname = "".join(sen_text)
if len(self.nickname) < 1:
self.nickname = '名字全为符号'
return f'error:取直播间ID错误，{self.nickname}还未开播'
else:
return 'error:取直播间ID错误，链接有问题'
else:
reflow_id = reflow_id_list[0]
self._log('获取直播间ID成功')
self._log('开始请求直播间数据')
#请求直播间数据
live_api = "https://webcast.amemv.com/webcast/room/reflow/info/?verifyFp=verify_l7u0qa4v_lx77IkKY_kSHg_411A_9JqE_9h6LMXjaPvb" \
+ random.choice(string.ascii_uppercase) + "&type_id=0&live_id=1&room_id=" + reflow_id \
+ "&sec_user_id=&app_id=1128&msToken=HZq7LCnGU_J6-_ZIP8z7ugoxa3TO4kiefFKCiLe139sbMkgLan0uEVijUk8B7aBNseUuwGfvGToHXv7" \
"VgGGI4ELJPWBedkZZdZtq9se_dK2gQ_dP-Rw" + random.choice(string.ascii_lowercase) + "&X-Bogus=DFSzswVOye0ANHQbSKxn6VXAIQ5" \
+ random.choice(string.ascii_lowercase)
live_data=requests.get(live_api,headers=self.headers).text
self._log('请求直播间数据完毕')
return live_data
def _log(self, logmsg):
# logmsg = logmsg[:25] if len(logmsg) > 25 else logmsg
global LOG_LINE_NUM
current_time = self.get_current_time()
logmsg_in = str(current_time) + " " +str(logmsg) + "\n" # 换行
self.GLineEdit_371.tag_config("even", background='#e0e0e0')
self.GLineEdit_371.tag_config("odd", background='#ffffff')
self.tag = 'odd' if self.tag == 'even' else 'even'
if LOG_LINE_NUM <= 27:
self.GLineEdit_371.insert('end', logmsg_in, self.tag)
LOG_LINE_NUM = LOG_LINE_NUM + 1
else:
self.GLineEdit_371.delete(1.0, 2.0)
self.GLineEdit_371.insert('end', logmsg_in, self.tag)
def get_current_time(self):
current_time = time.strftime('%H:%M:%S', time.localtime(time.time()))
return current_time
def on_closing() :
global thd
if len(thd)>0:
for i in thd:
i.stdin.write('q'.encode("GBK"))
i.communicate()
print ("Window closed")
root.destroy()
if __name__ == "__main__":
root = tk.Tk()
app = App(root)
root.protocol("WM_DELETE_WINDOW", on_closing)
root.mainloop()
[复制链接]
发表于 2022-10-15 14:09
|
发表于 2022-10-14 15:13
|楼主
收藏
淘帖
有用
分享到朋友圈
