前言：最近在看統計學，是以很久沒寫代碼了。手癢癢，把之前說的“微信銷量上報自動提醒與統計”做出來了，使用效果如下圖：
在同目錄下準備了2個txt文檔，分别放入标準上報模闆和需要上報帳量的機關，格式如下：
2月15日更新：
1、把自動關機的代碼注釋掉了，如果需要可以自己打開；
2、增加了重複上報的判斷，如果是的話，就發送“**，收到，資料已更新！”
3、新增了模闆的前四個字元判斷，代替了“銷量上報”，可以自定義了；
4、新增了“銷量上報群和總結時間.txt”，可以把想要互動的群放在裡面，也便于修改；
5、感謝同僚胡大師的幫忙，支援多線程了
還有幾個問題：
1、我本來想用多線程來自動到時間就統計今日資料+關機的，但是實際測試下來一直不行，等多線程再學一段時間看看能不能優化；
2、對于模闆和上報人員的要求比較高，不能有一點錯誤，否則就會統計不到，我在想如果用正規表達式模糊比對是否會好一點；
具體代碼如下：
#coding=utf8
import itchat
from itchat.content import *
import re
import datetime
import os
import threading
import time
class Test_Thread(threading.Thread):
# 初始化，時間為time_to_run,字元串類型
def __init__(self, time_to_run):
self.end_time = time_to_run
threading.Thread.__init__(self)
#每10秒判斷一次是否已經總結時間
def run(self):
a = True
while(a):
#每次等待一秒後判斷目前時間是否到達，比如22:00
time.sleep(10)
current_time = datetime.datetime.now()
current_time_format = current_time.hour * 3600 + current_time.minute * 60 + current_time.second
end_time_format = int(self.end_time.split(':')[0]) * 3600 + int(self.end_time.split(':')[1]) * 60
if current_time_format >= end_time_format:
itchat.get_chatrooms()
room_report = itchat.search_chatrooms(name=qun_name)[0]['UserName']
alt_store_str = '、'.join(alt_store)
not_alt_store_str = '、'.join(not_alt_store)
itchat.send('今日共收到{}家門店的銷量上報，已收到：{}，未收到：{}！'.format(len(alt_store), alt_store_str, not_alt_store_str), toUserName=room_report)
total_muban = today_count()
itchat.send(total_muban, toUserName=room_report)
a = False
# os.system("shutdown -s -t 0") #自動關機
def time_to_run(end_time):
'''判斷是否到某一個時間
end_time, 類似22:00'''
current_time = datetime.datetime.now()
current_time_format = current_time.hour * 3600 + current_time.minute * 60 + current_time.second
end_time_format = int(end_time.split(':')[0]) * 3600 + int(end_time.split(':')[1]) * 60
if current_time_format >= end_time_format : return 1
else: return 0
# 在注冊時增加isGroupChat=True将判定為群聊回複
@itchat.msg_register(TEXT, isGroupChat = True)
def groupchat_reply(msg):
'''
msg['Content'],資訊内容
msg['FromUserName'],資訊的發送方ID，如果是群資訊的話就是群ID，是一長串數字，沒有msg['NickName']，是以要用chatroom_dict()進行轉化
msg['ActualNickName'],資訊發送者的昵稱，注意如果是群裡的話就是群裡資訊的實際發送者
'''
d = chatroom_dict()
chatroom_id = msg['FromUserName']
chatroom_name = d[chatroom_id]
if chatroom_name == qun_name: #這裡指定了群，且該群必須儲存在通訊錄裡!
text = msg['Content']
if text == '今日銷量統計':
alt_store_str = '、'.join(alt_store)
not_alt_store_str = '、'.join(not_alt_store)
itchat.send('今日共收到{}家門店的銷量上報，已收到：{}，未收到：{}！'.format(len(alt_store), alt_store_str,
not_alt_store_str), toUserName=chatroom_id)
total_muban = today_count()
itchat.send(total_muban, toUserName=chatroom_id)
# os.system("shutdown -s -t 0") #自動關機
elif str(text).startswith(muban_start):
#把獲得文字中根據模闆提取關鍵字
store_num_list = []
for i in range(0, muban_count):
store_num_list.append(re.search(new_muban, text).group(i + 1))
store = store_num_list[0]
if store in store_dict:
store_dict[store] = store_num_list[1:]
if store in alt_store:
itchat.send('{}，收到，資料已更新！'.format(store), toUserName=chatroom_id)
else:
alt_store.append(store)
alt_store_str = '、'.join(alt_store)
not_alt_store.remove(store)
not_alt_store_str = '、'.join(not_alt_store)
itchat.send('{}，收到!\n目前已收到{}的銷量，尚未收到{}，請在{}前上報，謝謝！'.format(store, alt_store_str, not_alt_store_str, summary_time), toUserName=chatroom_id)
def chatroom_dict():
'''獲得群的ID和群名稱，并組合成字典'''
d = {}
chatrooms = itchat.get_chatrooms(update=True)
for chatroom in chatrooms:
d[chatroom['UserName']] = chatroom['NickName']
return d
def run_itchat():
'''運作微信'''
itchat.auto_login(True)
itchat.run()
def get_muban():
'''讀取模闆内容'''
with open("銷量上報模闆.txt", "r") as file:
muban = file.read()
return muban
def get_store():
'''擷取需要上報帳量的機關名稱并組成清單和字典'''
with open("銷量上報機關.txt", "r") as file:
not_alt_store = []
store_dict = {}
for line in file.readlines():
line = line.strip()
not_alt_store.append(line)
store_dict[line] = []
return not_alt_store, store_dict
def get_qun():
'''獲得要互動的群名稱和總結時間'''
with open("銷量上報群和總結時間.txt", "r") as file:
qun_list = []
for line in file.readlines():
line = line.strip()
qun_list.append(line)
return qun_list
def today_count():
'''合并統計今天已上報的銷量'''
today_value_list = []
for key in store_dict:
if store_dict[key] == []:
continue
else:
today_value_list.append(store_dict[key])
new_list = list_value_combine(today_value_list, muban_count)
total_muban = string_replace(muban, muban_count, new_list)
return total_muban
def list_value_combine(old_list, muban_count):
'''按順序合并清單中多個子清單的值，并獲得一個新的清單'''
new_list = []
old_len = len(old_list)
for j in range(muban_count-1):
temp_num = 0
for i in range(old_len):
temp_num += int(old_list[i][j])
new_list.append(temp_num)
return new_list
def string_replace(muban, muban_count, new_list):
'''根據清單中的元素依次替換字元串中的*'''
for i in range(muban_count):
temp = str(muban).find('*')
if i ==0:
muban ='今日已上報帳量合計：' + muban[temp+1:]
else:
muban = muban[:temp] + str(new_list[i-1]) + muban[temp+1:]
return muban
if __name__ == '__main__':
muban = get_muban()
muban_start = muban[:4]
new_muban = muban.replace('*', '(.*?)') #變成正規表達式的樣式
muban_count = muban.count('*')
qun_name = get_qun()[0] #互動群的名稱
summary_time = get_qun()[1] #總結的時間
store_dict = get_store()[1] #門店字典，key是名稱，value是以清單形式展現的銷量資料
not_alt_store = get_store()[0] #尚未上報帳量的門店清單
alt_store = [] #已經上報了銷量的門店清單
cal1 = Test_Thread(summary_time) # 定義cal1對象，為test_thread類派生，定義22點運作
cal1.start()
run_itchat()