python+selenium上传本地文件
2021-02-09 16:19
标签：meta this UNC 打包 path 加载完成 com account not 迅雷号自媒体视频文件自动上传，贴标签发布 难点 本地文件上传，通过send_keys(‘文件路径’)的方式实现上传的目的 文件名通过正则匹配的方式进行处理，主要匹配出中文标题名称 处理过程中文件名称中包括中文字符，特殊字符，数字等 视频文件上传是否完成的判断，视频上传的进度条通过js加载，在上传的过程中让程序进入到循环中进行等待进度条的加载完成，根据加载完成后标签的内容来判定是否上传完成 发布是否成功的判断，发布成功前后页面几乎没有变化，通过查看元素查找上传成功和失败有哪些细微的变化，从细微的变化中找到可以依据判断是否上传成功的条件 import sys from threading import Thread from selenium import webdriver from selenium.webdriver.common.keys import Keys #引入Keys类包 import time import re, os from selenium.webdriver.support.wait import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.common.by import By from multiprocessing import Queue from ua import * path_queue = Queue() path = r‘C:\Users\Administrator\Desktop\video‘ run_log = { ‘login_status‘: [], ‘publish_status‘: [], } class XunleiPublish(object): def __init__(self): self.url = ‘http://mp.m.xunlei.com/‘ def login(self, id): self.opt = webdriver.ChromeOptions() self.opt.set_headless() # 设置随机请求头 self.opt.add_argument(‘user-agent=‘ + getheaders()) self.driver = webdriver.Chrome(chrome_options=self.opt,executable_path=‘chromedriver.exe‘) self.driver.get(self.url) time.sleep(1) try: WebDriverWait(self.driver, 10, 0.5).until(EC.presence_of_element_located((By.ID, ‘loginIframe‘))) # 找到iframe标签 self.driver.switch_to.frame(‘loginIframe‘) self.driver.find_element_by_id(‘al_u‘).send_keys(id[:10]) self.driver.find_element_by_id(‘al_p‘).send_keys(id[11:]) self.driver.find_element_by_id(‘al_submit‘).click() # 登陆按钮 self.driver.find_element_by_class_name(‘header_info_name‘) run_log[‘login_status‘].append(‘%s：登陆成功！‘ % id[:10]) self.driver.get(‘http://mp.m.xunlei.com/publish‘) time.sleep(0.1) self.upload(id) except Exception as e: run_log[‘login_status‘].append(‘%s：登陆失败！‘ % id[:10]) self.driver.quit() sys.exit() def upload(self, id): i = 0 while True: if not path_queue.empty(): filepath = path_queue.get() while True: try: self.driver.find_element_by_xpath(‘//input[@type="file"]‘) break except: pass # 找到input标签把所需要上传的文件发送过去 self.driver.find_element_by_xpath(‘//input[@type="file"]‘).send_keys(filepath) time.sleep(2) self.publish(filepath) # 分类标签如果没有分类则发布成功 cate_attr = self.driver.find_element_by_class_name(‘catetxt‘) if cate_attr.text == ‘精确选择分类更有利于推荐‘: i += 1 # print(‘第%d次发布成功！‘ % i) # run_log[‘publish_status‘].append(‘%s第%d次发布成功！‘ % (id[:10], i)) run_log[‘publish_status‘].append(‘{}：第{}次发布成功！‘.format(id[:10], i)) os.remove(filepath) else: os.remove(filepath) self.driver.refresh() if i >= 500: self.driver.quit() sys.exit() else: break self.driver.quit() sys.exit() def publish(self, filepath): title_list = re.compile(‘([^\x00-\xff]+)‘, re.S).findall(filepath) while True: try: upload_progress = self.driver.find_element_by_id(‘showTips‘).find_element_by_tag_name(‘font‘).text if upload_progress == ‘已为您上传成功！‘: break elif upload_progress == ‘上传失败！‘: os.remove(filepath) break except: pass time.sleep(2) try: if title_list: self.driver.find_element_by_xpath(‘//input[@id="title"]‘).send_keys(‘,‘.join(title_list)) self.driver.find_element_by_id(‘topicInput‘).send_keys(‘搞笑‘) self.driver.find_element_by_class_name(‘span_squ‘).click() self.driver.find_elements_by_class_name(‘option_item‘)[0].click() # 发送文字到input框按下回车键 self.driver.find_element_by_id(‘tag-selectized‘).send_keys(‘搞笑‘, Keys.ENTER) self.driver.find_element_by_id(‘prot_check‘).click() # 点击同意平台服务协议按钮 self.driver.find_element_by_id(‘publishBtn‘).click() # 点击发布按钮 time.sleep(2) except: self.driver.refresh() def file(self): for filename in os.listdir(path): filepath = path + ‘\\‘ + filename path_queue.put(filepath) # if __name__ == ‘__main__‘: # XunleiPublish().file() # thread_list = [] # with open(‘account_info.txt‘, ‘r‘, encoding=‘utf-8‘) as f: # s = f.readlines() # for id in s[21:]: # t = Thread(target=XunleiPublish().login, args=(id.replace(‘\n‘, ‘‘),)) # thread_list.append(t) # t.start() # for i in thread_list: # i.join() # 返回登录日志 def login_record(): while 1: if run_log[‘login_status‘]: return run_log[‘login_status‘] time.sleep(0.1) # 返回发布日志 def publish_record(): while 1: if run_log[‘publish_status‘]: return run_log[‘publish_status‘] time.sleep(0.1) if __name__ == ‘__main__‘: Pass pyqt界面 # -*- coding: utf-8 -*- # Form implementation generated from reading ui file ‘untitled.ui‘ # # Created by: PyQt5 UI code generator 5.11.3 # # WARNING! All changes made in this file will be lost! import ctypes import win32con from PyQt5 import QtCore, QtWidgets from PyQt5.QtCore import QThread, pyqtSignal from PyQt5.QtWidgets import QApplication, QMainWindow from thr_publish import * import sys, time SESSION_DATA = False SHOW_S_P = False class Worker(QThread): valueChanged = pyqtSignal(int) handle = -1 def run(self): global SESSION_DATA, EXIT_COND try: self.handle = ctypes.windll.kernel32.OpenThread( win32con.PROCESS_ALL_ACCESS, False, int(QThread.currentThreadId()) ) except Exception as e: print(‘get thread handle failed‘, e) # 循环发送信号 while True: if SESSION_DATA: self.valueChanged.emit(1024) SESSION_DATA = False time.sleep(0.1) def exi_thread(self): os._exit(122) class Ui_MainWindow(QMainWindow): thread_list = [] def __init__(self): super(Ui_MainWindow, self).__init__() self.publish_record_count = 0 for func in [self.output_login_status, self.publish_record]: thr = Thread(target=func) thr.setDaemon(True) thr.start() # 子线程 self._thread = Worker(self) self._thread.finished.connect(self._thread.deleteLater) # self._thread.valueChanged.connect(ex.create_c) self._thread.start() def setupUi(self, MainWindow): MainWindow.setObjectName("MainWindow") MainWindow.resize(1222, 665) self.centralwidget = QtWidgets.QWidget(MainWindow) self.centralwidget.setObjectName("centralwidget") self.gridLayoutWidget = QtWidgets.QWidget(self.centralwidget) self.gridLayoutWidget.setGeometry(QtCore.QRect(10, 0, 911, 591)) self.gridLayoutWidget.setObjectName("gridLayoutWidget") self.gridLayout = QtWidgets.QGridLayout(self.gridLayoutWidget) self.gridLayout.setContentsMargins(0, 0, 0, 0) self.gridLayout.setObjectName("gridLayout") self.tableWidget_2 = QtWidgets.QTableWidget(self.gridLayoutWidget) self.tableWidget_2.setObjectName("tableWidget_2") self.tableWidget_2.setColumnCount(4) self.tableWidget_2.setRowCount(0) item = QtWidgets.QTableWidgetItem() self.tableWidget_2.setHorizontalHeaderItem(0, item) item = QtWidgets.QTableWidgetItem() self.tableWidget_2.setHorizontalHeaderItem(1, item) item = QtWidgets.QTableWidgetItem() self.tableWidget_2.setHorizontalHeaderItem(2, item) item = QtWidgets.QTableWidgetItem() self.tableWidget_2.setHorizontalHeaderItem(3, item) self.gridLayout.addWidget(self.tableWidget_2, 0, 1, 1, 1) self.listWidget = QtWidgets.QListWidget(self.gridLayoutWidget) self.listWidget.setObjectName("listWidget") self.gridLayout.addWidget(self.listWidget, 1, 1, 1, 1) # 运行日志 self.textBrowser_1 = QtWidgets.QTextBrowser(self.listWidget) self.textBrowser_1.setGeometry(QtCore.QRect(2, 2, 521, 260)) self.textBrowser_1.setObjectName("textBrowser") # self.textBrowser_1.append(‘运行日志‘) self.verticalLayoutWidget = QtWidgets.QWidget(self.centralwidget) self.verticalLayoutWidget.setGeometry(QtCore.QRect(920, 0, 301, 131)) self.verticalLayoutWidget.setObjectName("verticalLayoutWidget") self.verticalLayout = QtWidgets.QVBoxLayout(self.verticalLayoutWidget) self.verticalLayout.setContentsMargins(0, 0, 0, 0) self.verticalLayout.setObjectName("verticalLayout") self.pushButton = QtWidgets.QPushButton(self.verticalLayoutWidget) self.pushButton.setObjectName("pushButton") self.verticalLayout.addWidget(self.pushButton) self.horizontalLayoutWidget = QtWidgets.QWidget(self.centralwidget) self.horizontalLayoutWidget.setGeometry(QtCore.QRect(920, 180, 301, 71)) self.horizontalLayoutWidget.setObjectName("horizontalLayoutWidget") self.horizontalLayout = QtWidgets.QHBoxLayout(self.horizontalLayoutWidget) self.horizontalLayout.setContentsMargins(0, 0, 0, 0) self.horizontalLayout.setObjectName("horizontalLayout") self.label = QtWidgets.QLabel(self.horizontalLayoutWidget) self.label.setObjectName("label") self.horizontalLayout.addWidget(self.label) self.lineEdit = QtWidgets.QLineEdit(self.horizontalLayoutWidget) self.lineEdit.setObjectName("lineEdit") self.horizontalLayout.addWidget(self.lineEdit) self.horizontalLayoutWidget_2 = QtWidgets.QWidget(self.centralwidget) self.horizontalLayoutWidget_2.setGeometry(QtCore.QRect(920, 280, 301, 80)) self.horizontalLayoutWidget_2.setObjectName("horizontalLayoutWidget_2") self.horizontalLayout_2 = QtWidgets.QHBoxLayout(self.horizontalLayoutWidget_2) self.horizontalLayout_2.setContentsMargins(0, 0, 0, 0) self.horizontalLayout_2.setObjectName("horizontalLayout_2") self.label_2 = QtWidgets.QLabel(self.horizontalLayoutWidget_2) self.label_2.setObjectName("label_2") self.horizontalLayout_2.addWidget(self.label_2) self.lineEdit_2 = QtWidgets.QLineEdit(self.horizontalLayoutWidget_2) self.lineEdit_2.setObjectName("lineEdit_2") self.horizontalLayout_2.addWidget(self.lineEdit_2) MainWindow.setCentralWidget(self.centralwidget) self.menubar = QtWidgets.QMenuBar(MainWindow) self.menubar.setGeometry(QtCore.QRect(0, 0, 1222, 26)) self.menubar.setObjectName("menubar") MainWindow.setMenuBar(self.menubar) self.statusbar = QtWidgets.QStatusBar(MainWindow) self.statusbar.setObjectName("statusbar") MainWindow.setStatusBar(self.statusbar) self.retranslateUi(MainWindow) QtCore.QMetaObject.connectSlotsByName(MainWindow) def retranslateUi(self, MainWindow): _translate = QtCore.QCoreApplication.translate MainWindow.setWindowTitle(_translate("MainWindow", "迅雷号")) item = self.tableWidget_2.horizontalHeaderItem(0) item.setText(_translate("MainWindow", "账号")) item = self.tableWidget_2.horizontalHeaderItem(1) item.setText(_translate("MainWindow", "密码")) item = self.tableWidget_2.horizontalHeaderItem(2) item.setText(_translate("MainWindow", "登陆状态")) item = self.tableWidget_2.horizontalHeaderItem(3) item.setText(_translate("MainWindow", "上传数量")) self.pushButton.setText(_translate("MainWindow", "登陆账号")) self.pushButton.clicked.connect(self.login) self.label.setText(_translate("MainWindow", "文件路径：")) self.label_2.setText(_translate("MainWindow", "添加话题：")) # 点击登陆 def login(self): XunleiPublish().file() self.textBrowser_1.append(‘开始登陆，上传发布...‘) # thread_list = [] with open(‘account_info.txt‘, ‘r‘, encoding=‘utf-8‘) as f: s = f.readlines() for id in s[:5]: t = Thread(target=XunleiPublish().login, args=(id.replace(‘\n‘, ‘‘),)) self.thread_list.append(t) t.setDaemon(True) t.start() # for i in self.thread_list: # i.join() # 日志更新 def output_login_status(self): # 登陆成功输出 while True: # 登陆日志 login_record_list = login_record() if login_record_list: for i in login_record_list: self.textBrowser_1.append(i) self.textBrowser_1.moveCursor(self.textBrowser_1.textCursor().End) login_record_list.remove(i) time.sleep(0.1) # 发布日志 def publish_record(self): while True: publish_record_list = publish_record() if publish_record_list: for record in publish_record_list: self.textBrowser_1.append(record) self.textBrowser_1.moveCursor(self.textBrowser_1.textCursor().End) publish_record_list.remove(record) time.sleep(0.1) if __name__ == ‘__main__‘: app = QApplication(sys.argv) MainWindow = QMainWindow() ui = Ui_MainWindow() ui.setupUi(MainWindow) MainWindow.show() sys.exit(app.exec_()) # 打包封装为.exe可执行文件 # pyinstaller -F -w untitled.py -p thr_publish.py -p ua.py --hidden-import untitled --hidden-import thr_publish --hidden-import ua 原文链接：https://blog.csdn.net/qq_41866851/java/article/details/95781660 python+selenium上传本地文件 标签：meta this UNC 打包 path 加载完成 com account not 原文地址：https://www.cnblogs.com/modentime/p/12749647.html
上一篇：springBoot整合RabbitMQ(新手整合请勿喷)
下一篇：去除.net core 发布的时候生成的多余语言文件夹cs,de...，多余的Microsoft.CodeAnalysis语言资源文件
文章标题：python+selenium上传本地文件
文章链接：http://soscw.com/index.php/essay/53172.html