python+selenium上传本地文件

Posted 2021-02-27 modentime

迅雷号自媒体视频文件自动上传，贴标签发布

难点

本地文件上传，通过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:UsersAdministratorDesktopvideo‘

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(‘ ‘, ‘‘),))

#             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(‘ ‘, ‘‘),))

                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