最近有个需求，统计录入人员的实际工作时间，想了下，每隔一段时间做一个客户端把用户当前的窗口标题记录下来，定期传给MySQL，之后再用另外一个脚本分析实际的工作时间，初学Python，没写过什么东西，如有错误，望指正。
客户端如下：
import win32gui
import time
import getpass
import pythoncom
import pyHook
import threading
import MySQLdb
import sys
LOG_LIST = [] # init global log list
LAST_LOG = None # time_stamp of last log.
class Mysql():
def __init__(self):
try:
self.conn = MySQLdb.connect(
host=‘192.168.100.250‘,
user=‘hook_agent‘, passwd=‘qwqwqw480066‘,
db=‘qc_upload‘, port=3306,
charset=‘utf8‘)
except MySQLdb.Error, e:
print ‘Connect to MySQL Failed‘, e
sys.exit(5)
else:
self.cursor = self.conn.cursor()
def __del__(self):
self.cursor.close()
self.conn.close()
def insert(self, sql, params):
print ‘Insert to DB.‘
try:
self.cursor.executemany(sql, params)
except Exception, e:
print ‘Failed to write db‘, e
else:
self.conn.commit()
def get_window_name():
"""return current focus window name"""
window_name = win32gui.GetWindowText(win32gui.GetForegroundWindow())
‘‘‘for fault tolerance. when cursor in StartMenu,
window_name may will return None, and NoneType have no decode method,
cause AttributeError‘‘‘
try:
return window_name.decode(‘gbk‘)[:50]
except AttributeError:
return ‘None‘
def upload(data):
"""upload to server,
if failed, retry every 2 minutes, total 10 times."""
print data
retry = 0
while retry <= 10:
try:
sql_helper = Mysql()
sql = ‘insert into qc_worklog(uid, date, time_stmp, win_title) values(%s, %s, %s, %s)‘
sql_helper.insert(sql, data)
break
except:
time.sleep(120)
retry += 1
continue
finally:
del sql_helper
def log(event):
"""every 120 seconds log the win_name,
and when logs‘ count >= 30, upload."""
global LOG_LIST, LAST_LOG
time_now = int(time.time())
if not LAST_LOG or time_now - LAST_LOG >= 120:
log_list = [
getpass.getuser(), time.strftime("%Y-%m-%d"),
int(time.time()), get_window_name()]
LOG_LIST.append(log_list)
if len(LOG_LIST) >= 30:
upload_t = threading.Thread(target=upload, args=(LOG_LIST,))
upload_t.start()
LOG_LIST = [] # re-init the LOG_LIST
LAST_LOG = time_now
return True # MUST return True, or cursor will NOT able to move.
def main():
hm = pyHook.HookManager()
hm.MouseAll = log
hm.HookMouse()
hm.KeyDown = log
hm.HookKeyboard()
pythoncom.PumpMessages()
if __name__ == ‘__main__‘:
main()
服务器端判断工作时间脚本如下：
#!/bin/env python
import json
import sys
import MySQLdb
import time
import urllib
import urllib2
class Mysql():
def __init__(self):
try:
self.conn = MySQLdb.connect(
host=‘127.0.0.1‘,
user=‘root‘, passwd=‘qwqwqw‘,
db=‘qc_upload‘, port=3306,
charset=‘utf8‘)
except MySQLdb.Error, e:
print ‘Connect to MySQL Failed‘, e
log(e)
sys.exit(5)
else:
self.cursor = self.conn.cursor()
def __del__(self):
self.cursor.close()
self.conn.close()
def query(self, sql, params):
try:
self.cursor.execute(sql, params)
except Exception, e:
print human_time(), ‘Failed to read db‘, e
log(e)
else:
return self.cursor.fetchall()
def human_time():
return time.strftime("%Y-%m-%d %H:%M:%S")
def log(e):
content = human_time() + e.__str__() + ‘\n‘
with open(‘logs.log‘, ‘a‘) as f:
f.write(content)
def calculate(username, day):
"""analyse today‘s log, if keyword(‘LinkDoc‘) in win_title,
add the time to work time. and return work_time"""
# select passed username‘s log
sql_helper = Mysql()
sql = ‘select time_stmp, win_title from qc_worklog where uid=%s and date=%s order by time_stmp‘
logs = sql_helper.query(sql, (username[0], day))
# calculate the current user‘s work time
work_time = 0
for log in logs:
if ‘LinkDoc‘ in log[1]:
if ‘last_time_stmp‘ not in dir():
last_time_stmp = log[0]
delta_time = log[0] - last_time_stmp
if delta_time <= 300:
work_time += delta_time
last_time_stmp = log[0]
return {‘username‘: username[0], ‘qc_time‘: work_time, ‘date‘: day}
def analyse(day=time.strftime("%Y-%m-%d"), action=‘upload‘):
"""analyse user‘s worktime of today"""
sql_helper = Mysql()
sql = ‘select distinct uid from qc_worklog where date=%s‘
# get all distinct username of today
usernames = sql_helper.query(sql, (day,))
# call calculate func. and add to the result list.
result = []
for u in usernames:
result.append(calculate(u, day))
if action == ‘upload‘:
upload(result)
elif action == ‘print‘:
print result
def get_token():
"""get token from dig"""
url = ‘http://192.168.10.38:8089/login/collectorlogin ?collecter_name=dig_api_auth&password=wei712372_knil‘
while True:
try:
req = urllib2.Request(url)
response = urllib2.urlopen(req, timeout=10)
res_data = json.loads(response.read())
break
except urllib2.URLError, e:
log(e)
time.sleep(120)
continue
return res_data[‘data‘][‘token‘]
def upload(result):
"""upload to dig"""
post_value = {‘data‘: json.dumps(result), ‘token‘: get_token()}
post = urllib.urlencode(post_value)
url = ‘http://192.168.10.38:8089/api/saveUserWorktime‘
while True:
try:
req = urllib2.Request(url)
response = urllib2.urlopen(req, post, timeout=10)
res = response.read()
break
except urllib2.URLError, e:
log(e)
time.sleep(120)
continue
log(res)
def print_usage():
print """usage: -p for print today‘s workload.
-p YYYY-MM-DD for print specific day‘s workload
run the script without any args, upload the workload of today
run the srcipt with YYYY-MM-DD, upload the workload of the specific day"""
def main():
args = sys.argv
if len(args) > 1:
# print the workload of today
if ‘-p‘ in args and len(args) == 2:
analyse(action=‘print‘)
# print the workload of specific day
elif ‘-p‘ in args and len(args) == 3 and len(args[2]) == 10:
analyse(day=args[2], action=‘print‘)
# upload the workload of specific day.
elif len(args[1]) == 10:
analyse(day=args[1])
else:
print_usage()
# upload the workload of today
elif len(args) == 1:
analyse()
else:
print_usage()
if __name__ == ‘__main__‘:
main()
时间： 2024-08-08 13:52:57