本帖最后由 话痨司机啊 于 2022-5-25 20:32 编辑
【源码下载地址】链接: https://pan.baidu.com/s/1lNZ6Cns3oWPhQuCiTpCMaw?pwd=en25 提取码: en25
【成品exe下载地址】链接: https://pan.baidu.com/s/1Va1Pz-Ue2Jq815BjpuvUig?pwd=vsb2 提取码: vsb2
【演示说明下载地址】链接: https://pan.baidu.com/s/1AtNzUuquaXsLPFdk_NR8bw?pwd=hxpb 提取码: hxpb
【说明】 运行程序之前需要打开PC端微信并登录,配置好配置文件,如果下载成品的,运行一次配置文件就自动生成了,这时候你可以自定义上传文件夹,默认的是桌面的uploads(如果没有,请自行新建文件夹并命名uploads,将需要上传的文件放到uploads)。
核心代码:
[Python] 纯文本查看 复制代码
"""
************************************
Description: 文件批量上传到微信
Author: @话痨
Github: [url=https://github.com/jianala]https://github.com/jianala[/url]
Date: 2022-05-16 10:35:09
FilePath: \电脑操控类\wetchat_uploads.py
LastEditors: @话痨
LastEditTime: 2022-05-16 10:36:58
善始者实繁,克终者盖寡。
************************************
"""
import logging
import os
import subprocess
from functools import wraps
import time
import uiautomation as auto
from rich.console import Console
from configparefile import ConfigFile
logging.basicConfig(level=logging.INFO,
filename='wetchat_uploads.log',
filemode='a',
format='%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
def exception_handler(func):
"""
description: 异常捕获装饰器
return {函数调用返回值}}
"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logging.error(e)
return wrapper
def is_configfile_exists():
"""
description: 判断配置文件是否存在,不存在则创建并写入配置文件
"""
_cfg = ConfigFile('wechat_config.ini')
if not _cfg.is_file_exist():
_cfg.set_config('WeChat_path', 'file_path',
'C:\\Program Files\\Tencent\\WeChat\\WeChat.exe')
_cfg.set_config('uploads_path', 'dir_path',
'C:\\Users\\Administrator\\Desktop\\uploads')
class WechatUploads:
def __init__(self) -> None:
self.wx = auto.WindowControl(Name='微信')
self.cgf = ConfigFile('wechat_config.ini')
self.console = Console()
self.dir_uploads = os.path.normpath(
self.cgf.read_config()['uploads_path']['dir_path'])
print(
"""
************************************\r\n
_______________#########_______________________
______________############_____________________
______________#############____________________
_____________##__###########___________________
____________###__######_#####__________________
____________###_#######___####_________________
___________###__##########_####________________
__________####__###########_####_______________
________#####___###########__#####_____________
_______######___###_########___#####___________
_______#####___###___########___######_________
______######___###__###########___######_______
_____######___####_##############__######______
____#######__#####################_#######_____
____#######__##############################____
___#######__######_#################_#######___
___#######__######_######_#########___######___
___#######____##__######___######_____######___
___#######________######____#####_____#####____
____######________#####_____#####_____####_____
_____#####________####______#####_____###______
______#####______;###________###______#________
________##_______####________####______________
************************************ \r\n
-*-批量上传微信文件-*-
"""
)
@exception_handler
def is_WeChat_start(self):
"""
description: 判断微信是否启动,如果启动则为真,否则调用微信
return {*}
"""
if self.wx.Exists():
self.wx.SwitchToThisWindow()
logging.info('微信已经启动')
return True
else:
subprocess.Popen(self.cgf.read_config().get(
'WeChat_path', 'file_path'))
raise Exception('微信未启动,请启动微信后再次运行该程序!')
@exception_handler
def get_file_list(self, path):
"""
description: 获取文件夹的文件列表
return {*}
"""
file_list = []
for root, dirs, files in os.walk(path):
for file in files:
file_list.append(os.path.join(root, file))
return file_list
@exception_handler
def upload_target(self, upload_target):
"""
description: 判断上传目标是存在
params: upload_target {str} 上传目标
return {*}
"""
hh = self.wx.ListControl(name='会话').GetChildren()
for h in hh:
if h.Name == upload_target:
return upload_target
raise Exception('上传目标不存在')
def wechat_api(self, file_name, target):
"""
description: 上传接口
return {*}
"""
self.wx.ListItemControl(Name=target).Click(simulateMove=False)
self.wx.ButtonControl(Name='发送文件').Click(simulateMove=False)
self.wx.EditControl(Name='文件名(N):').SendKeys(file_name)
self.wx.SplitButtonControl(Name='打开(O)').Click(simulateMove=False)
self.wx.ButtonControl(Name='sendBtn').Click(simulateMove=False)
# 调用微信上传文件
def upload_file(self, upload_target):
"""
description: 上传文件
return {*}
"""
if self.is_WeChat_start():
_upload_target = self.upload_target(upload_target)
_file_list = self.get_file_list(self.dir_uploads)
for _file in _file_list:
self.wechat_api(_file, _upload_target)
now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
self.console.print("[yellow]创建时间:{}\n文件:{} [green]状态:上传成功!".format(now, os.path.basename(_file)))
logging.info('{}上传成功'.format(_file))
if __name__ == "__main__":
is_configfile_exists()
wx = WechatUploads()
target = input('请输入上传目标(直接回车就默认为文件传输助手):').strip()
target = target if not target == '' else '文件传输助手'
wx.upload_file(target)
运行效果反馈,已经测试一周了,稳定运行,虚拟机 每天晚上7点55分开机,然后7点57分自动登录微信,8点整发送文件和发送信息提醒群内人员收听并通过钉钉机器人反馈发送情况,剩余文件等信息,8点02分虚拟机关机(当然为了不发送重复文件还将发送的文件移除到另一个文件夹以后待用)。
, |