#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import sys
import time
import traceback
import json
import random
import datetime
import threading
import collections
import sys
if 2 == sys.version_info.major:
reload(sys)
sys.setdefaultencoding('utf-8')
import os
import sys
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# at py3 all str type is unicode, str encode type is byte
# at py2 str type is unicode or encode, all trade as byte
# pyinstaller -F -w -i favicon.ico curFile.py
# pyinstaller --distpath . -F -w -i favicon.ico test.py
# pip install openpyxl==2.6.4
# pip install python-docx==0.8.10
# pip install PyMySQL==0.9.3
# pip install redis==3.5.3
# pip install PyQt4-4.11.4-cp27-cp27m-win32.whl
# pip install numpy==1.16.6
# pip install pandas==0.24.2
global run_path
global cur_path
global cur_file
cur_path = os.path.dirname(os.path.abspath(__file__).decode('GB18030', 'ignore'))
cur_file = os.path.basename(os.path.abspath(__file__).decode('GB18030', 'ignore'))
(cur_name, cur_extn) = os.path.splitext(cur_file)
import logging
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s [%(levelname)-8s][%(filename)s:%(lineno)d]: %(message)s')
logger = logging.getLogger(__name__)
# console = logging.StreamHandler()
# console.setLevel(logging.INFO)
# logger.addHandler(console)
os.makedirs(os.path.join(cur_path, 'log')) if not os.path.exists(os.path.join(cur_path, 'log')) else None
handler = logging.FileHandler(os.path.join(cur_path, 'log', '{}.log'.format(cur_name)))
handler.setLevel(logging.INFO)
handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)-8s][%(filename)s:%(lineno)d]: %(message)s'))
logger.addHandler(handler)
debug = True
def func_test(**argv):
try:
logger.info('func_test init')
logger.info('argv:{}'.format(argv))
logger.info('argv:{}'.format(json.dumps(argv, ensure_ascii=False, sort_keys=True, default=str)))
logger.info('func_test exit')
return True
except:
logger.error(traceback.format_exc())
logger.error('func_test fail')
finally:
logger.info('func_test clear')
pass
return False
class Producer(threading.Thread):
def __init__(self, t_name, deque):
threading.Thread.__init__(self, name=t_name)
self.deque = deque
def run(self):
func_producer = lambda name, deque: logger.info('Thread:{} deque:{}'.format(name, len(deque)))
func_producer(self.getName(), self.deque)
return
class Consumer(threading.Thread):
def __init__(self, t_name, deque):
threading.Thread.__init__(self, name=t_name)
self.deque = deque
def run(self):
func_consumer = lambda name, deque: logger.info('Thread:{} deque:{}'.format(name, len(deque)))
func_consumer(self.getName(), self.deque)
return
def main_loop(**argv):
try:
logger.info('main_loop init')
logger.info('argv:{}'.format(argv))
logger.info('argv:{}'.format(json.dumps(argv, ensure_ascii=False, sort_keys=True, default=str)))
main__deque = collections.deque(maxlen=10)
main__deque.append(argv)
producer = Producer('Producer', main__deque)
consumer = Consumer('Consumer', main__deque)
producer.start()
consumer.start()
logger.info('threading active count:{}'.format(threading.active_count()))
producer.join()
consumer.join()
func_test(debug=True)
main__deque.pop()
logger.info('main_loop exit')
return True
except:
logger.error(traceback.format_exc())
logger.error('main_loop fail')
finally:
logger.info('main_loop clear')
pass
return False
def main(*argv):
global run_path
global cur_path
global cur_file
try:
logger.info('{} init'.format(sys._getframe().f_code.co_name))
logger.info('argv:{}'.format(argv))
curr_file = (lambda: sys._getframe().f_code.co_filename) # same as __file__
curr_line = (lambda: sys._getframe().f_lineno)
curr_func = (lambda: sys._getframe(0).f_code.co_name)
last_func = (lambda: sys._getframe(1).f_code.co_name)
logger.info('file:{} line:{} func:{} last_func:{}'.format(curr_file(), curr_line(), curr_func(), last_func()))
# GB18030兼容GBK，GBK兼容GB2312，GB2312兼容ASCII
run_path = os.getcwd().decode('GB18030', 'ignore') # GB2312 => GBK => GB18030
cur_path = os.path.dirname(os.path.abspath(__file__).decode('GB18030', 'ignore'))
cur_file = os.path.basename(os.path.abspath(__file__).decode('GB18030', 'ignore'))
(cur_name, cur_extn) = os.path.splitext(cur_file)
logger.info('run_path:{}'.format(run_path))
logger.info('cur_path:{}'.format(cur_path))
logger.info('cur_file:{}'.format(cur_file))
logger.info('cur_name:{} cur_extn:{}'.format(cur_name, cur_extn))
logger.info('main curr time:{}'.format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f ')))
time_start = time.time()
while True:
main_loop(debug=debug)
time.sleep(random.randint(0, 3))
if debug: break
time_end = time.time()
logger.info('main take time:{}'.format(time_end - time_start))
logger.info('{} exit'.format(sys._getframe().f_code.co_name))
return True
except (NameError, ValueError, TypeError, SystemError):
logger.error(traceback.format_exc())
logger.error('{} error'.format(sys._getframe().f_code.co_name))
except:
logger.error(traceback.format_exc())
logger.error('{} fail'.format(sys._getframe().f_code.co_name))
finally:
logger.info('{} clear'.format(sys._getframe().f_code.co_name))
pass
return False
if __name__ == '__main__':
retCode = main(sys.argv)
exit(not retCode)