You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
371 lines
14 KiB
371 lines
14 KiB
#!/usr/bin/env python
|
|
# -*- coding:utf-8 -*-
|
|
# project: 12月
|
|
# author: NinEveN
|
|
# date: 2021/12/22
|
|
import datetime
|
|
import logging
|
|
import time
|
|
from functools import wraps, WRAPPER_ASSIGNMENTS
|
|
from importlib import import_module
|
|
|
|
from django.core.cache import cache
|
|
from django.http import HttpResponse
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def run_function_by_locker(timeout=60 * 5):
|
|
def decorator(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
start_time = time.time()
|
|
locker = kwargs.get('locker', {})
|
|
if locker:
|
|
kwargs.pop('locker')
|
|
t_locker = {'timeout': timeout, 'locker_key': func.__name__}
|
|
t_locker.update(locker)
|
|
new_locker_key = t_locker.pop('locker_key')
|
|
new_timeout = t_locker.pop('timeout')
|
|
if locker and new_timeout and new_locker_key:
|
|
with cache.lock(new_locker_key, timeout=new_timeout, **t_locker):
|
|
logger.info(f"{new_locker_key} exec {func} start. now time:{time.time()}")
|
|
res = func(*args, **kwargs)
|
|
else:
|
|
res = func(*args, **kwargs)
|
|
logger.info(f"{new_locker_key} exec {func} finished. used time:{time.time() - start_time} result:{res}")
|
|
return res
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def call_function_try_attempts(try_attempts=3, sleep_time=2, failed_callback=None):
|
|
def decorator(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
res = False, {}
|
|
start_time = time.time()
|
|
for i in range(try_attempts):
|
|
res = func(*args, **kwargs)
|
|
status, result = res
|
|
if status:
|
|
return res
|
|
else:
|
|
logger.warning(f'exec {func} failed. {try_attempts} times in total. now {sleep_time} later try '
|
|
f'again...{i}')
|
|
time.sleep(sleep_time)
|
|
if not res[0]:
|
|
logger.error(f'exec {func} failed after the maximum number of attempts. Failed:{res[1]}')
|
|
if failed_callback:
|
|
logger.error(f'exec {func} failed and exec failed callback {failed_callback.__name__}')
|
|
failed_callback(*args, **kwargs, result=res)
|
|
logger.info(f"exec {func} finished. time:{time.time() - start_time} result:{res}")
|
|
return res
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def magic_wrapper(func, *args, **kwargs):
|
|
@wraps(func)
|
|
def wrapper():
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def magic_notify(notify_rules, timeout=30 * 24 * 60 * 60):
|
|
"""
|
|
:param notify_rules:
|
|
:param timeout:
|
|
:return:
|
|
"""
|
|
now_time = datetime.datetime.now().date()
|
|
for notify_rule in notify_rules:
|
|
notify_cache = notify_rule['cache']
|
|
if notify_rule['func']():
|
|
notify_data = notify_cache.get_storage_cache()
|
|
if notify_data is None:
|
|
notify_cache.set_storage_cache([now_time + datetime.timedelta(days=i) for i in notify_rule['notify']],
|
|
timeout)
|
|
magic_notify(notify_rules)
|
|
elif isinstance(notify_data, list):
|
|
if len(notify_data) == 0:
|
|
return
|
|
else:
|
|
notify_data.append(now_time)
|
|
notify_data.sort()
|
|
is_today = False
|
|
if notify_data[0] == notify_data[1]:
|
|
is_today = True
|
|
notify_data = list(set(notify_data))
|
|
notify_data.sort()
|
|
n_index = notify_data.index(now_time)
|
|
if n_index == 0 and not is_today:
|
|
return
|
|
notify_data = notify_data[n_index + 1:]
|
|
|
|
for func in notify_rule['notify_func']:
|
|
try:
|
|
func()
|
|
except Exception as e:
|
|
logger.error(f'func {func.__name__} exec failed Exception:{e}')
|
|
notify_cache.set_storage_cache(notify_data, timeout)
|
|
|
|
else:
|
|
notify_cache.del_storage_cache()
|
|
|
|
|
|
def import_from_string(dotted_path):
|
|
"""
|
|
Import a dotted module path and return the attribute/class designated by the
|
|
last name in the path. Raise ImportError if the import failed.
|
|
"""
|
|
try:
|
|
module_path, class_name = dotted_path.rsplit('.', 1)
|
|
except ValueError as err:
|
|
raise ImportError(f"{dotted_path} doesn't look like a module path") from err
|
|
|
|
module = import_module(module_path)
|
|
|
|
try:
|
|
return getattr(module, class_name)
|
|
except AttributeError as err:
|
|
raise ImportError(f'Module "{module_path}" does not define a "{class_name}" attribute/class') from err
|
|
|
|
|
|
def magic_call_in_times(call_time=24 * 3600, call_limit=6, key=None):
|
|
def decorator(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
cache_key = f'magic_call_in_times_{func.__name__}'
|
|
if key:
|
|
cache_key = f'{cache_key}_{key(*args, **kwargs)}'
|
|
cache_data = cache.get(cache_key)
|
|
if cache_data:
|
|
if cache_data > call_limit:
|
|
err_msg = f'{func} not yet started. cache_key:{cache_key} call over limit {call_limit} in {call_time}'
|
|
logger.warning(err_msg)
|
|
return False, err_msg
|
|
else:
|
|
cache.incr(cache_key, 1)
|
|
else:
|
|
cache.set(cache_key, 1, call_time)
|
|
start_time = time.time()
|
|
try:
|
|
res = func(*args, **kwargs)
|
|
logger.info(
|
|
f"exec {func} finished. time:{time.time() - start_time} cache_key:{cache_key} result:{res}")
|
|
status = True
|
|
except Exception as e:
|
|
res = str(e)
|
|
logger.info(f"exec {func} failed. time:{time.time() - start_time} cache_key:{cache_key} Exception:{e}")
|
|
status = False
|
|
|
|
return status, res
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
class MagicCacheData(object):
|
|
@staticmethod
|
|
def make_cache(timeout=60 * 10, invalid_time=0, key_func=None, timeout_func=None):
|
|
"""
|
|
:param timeout_func:
|
|
:param timeout: 数据缓存的时候，单位秒
|
|
:param invalid_time: 数据缓存提前失效时间，单位秒。该cache有效时间为 cache_time-invalid_time
|
|
:param key_func: cache唯一标识，默认为所装饰函数名称
|
|
:return:
|
|
"""
|
|
|
|
def decorator(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
cache_key = f'magic_cache_data_{func.__name__}'
|
|
if key_func:
|
|
cache_key = f'{cache_key}_{key_func(*args, **kwargs)}'
|
|
|
|
cache_time = timeout
|
|
if timeout_func:
|
|
cache_time = timeout_func(*args, **kwargs)
|
|
|
|
n_time = time.time()
|
|
res = cache.get(cache_key)
|
|
if res:
|
|
while res.get('status') != 'ok':
|
|
time.sleep(0.5)
|
|
logger.warning(
|
|
f'exec {func} wait. data status is not ok. cache_time:{cache_time} cache_key:{cache_key} cache data exist result:{res}')
|
|
res = cache.get(cache_key)
|
|
|
|
if res and n_time - res.get('c_time', n_time) < cache_time - invalid_time:
|
|
logger.info(
|
|
f"exec {func} finished. cache_time:{cache_time} cache_key:{cache_key} cache data exist result:{res}")
|
|
return res['data']
|
|
else:
|
|
res = {'c_time': n_time, 'data': '', 'status': 'ready'}
|
|
cache.set(cache_key, res, cache_time)
|
|
try:
|
|
res['data'] = func(*args, **kwargs)
|
|
res['status'] = 'ok'
|
|
cache.set(cache_key, res, cache_time)
|
|
logger.info(
|
|
f"exec {func} finished. time:{time.time() - n_time} cache_time:{cache_time} cache_key:{cache_key} result:{res}")
|
|
except Exception as e:
|
|
logger.error(
|
|
f"exec {func} failed. time:{time.time() - n_time} cache_time:{cache_time} cache_key:{cache_key} Exception:{e}")
|
|
|
|
return res['data']
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
@staticmethod
|
|
def invalid_cache(key):
|
|
cache_key = f'magic_cache_data_{key}'
|
|
res = cache.delete(cache_key)
|
|
logger.warning(f"invalid_cache cache_key:{cache_key} result:{res}")
|
|
|
|
|
|
class MagicCacheResponse(object):
|
|
def __init__(self, timeout=60 * 10, invalid_time=0, key_func=None, callback_func=None):
|
|
self.timeout = timeout
|
|
self.key_func = key_func
|
|
self.invalid_time = invalid_time
|
|
self.callback_func = callback_func
|
|
|
|
@staticmethod
|
|
def invalid_cache(key):
|
|
cache_key = f'magic_cache_response_{key}'
|
|
res = cache.delete(cache_key)
|
|
logger.warning(f"invalid_response_cache cache_key:{cache_key} result:{res}")
|
|
|
|
def __call__(self, func):
|
|
this = self
|
|
|
|
@wraps(func, assigned=WRAPPER_ASSIGNMENTS)
|
|
def inner(self, request, *args, **kwargs):
|
|
return this.process_cache_response(
|
|
view_instance=self,
|
|
view_method=func,
|
|
request=request,
|
|
args=args,
|
|
kwargs=kwargs,
|
|
)
|
|
|
|
return inner
|
|
|
|
def process_cache_response(self,
|
|
view_instance,
|
|
view_method,
|
|
request,
|
|
args,
|
|
kwargs):
|
|
func_key = self.calculate_key(
|
|
view_instance=view_instance,
|
|
view_method=view_method,
|
|
request=request,
|
|
args=args,
|
|
kwargs=kwargs
|
|
)
|
|
func_name = f'{view_instance.__class__.__name__}_{view_method.__name__}'
|
|
cache_key = f'magic_cache_response_{func_name}'
|
|
if func_key:
|
|
cache_key = f'{cache_key}_{func_key}'
|
|
|
|
timeout = self.calculate_timeout(view_instance=view_instance)
|
|
n_time = time.time()
|
|
res = cache.get(cache_key)
|
|
if res and n_time - res.get('c_time', n_time) < timeout - self.invalid_time:
|
|
logger.info(f"exec {func_name} finished. cache_key:{cache_key} cache data exist result:{res}")
|
|
content, status, headers = res['data']
|
|
response = HttpResponse(content=content, status=status)
|
|
for k, v in headers.values():
|
|
response[k] = v
|
|
else:
|
|
response = view_method(view_instance, request, *args, **kwargs)
|
|
response = view_instance.finalize_response(request, response, *args, **kwargs)
|
|
response.render()
|
|
|
|
if not response.status_code >= 400:
|
|
# django 3.0 has not .items() method, django 3.2 has not ._headers
|
|
if hasattr(response, '_headers'):
|
|
headers = response._headers.copy()
|
|
else:
|
|
headers = {k: (k, v) for k, v in response.items()}
|
|
data = (
|
|
response.rendered_content,
|
|
response.status_code,
|
|
headers
|
|
)
|
|
res = {'c_time': n_time, 'data': data}
|
|
cache.set(cache_key, res, timeout)
|
|
|
|
self.callback_check(view_instance=view_instance,
|
|
view_method=view_method,
|
|
request=request,
|
|
args=args,
|
|
kwargs=kwargs,
|
|
cache_key=cache_key)
|
|
|
|
logger.info(
|
|
f"exec {func_name} finished. time:{time.time() - n_time} cache_key:{cache_key} result:{res}")
|
|
|
|
if not hasattr(response, '_closable_objects'):
|
|
response._closable_objects = []
|
|
|
|
return response
|
|
|
|
def calculate_key(self,
|
|
view_instance,
|
|
view_method,
|
|
request,
|
|
args,
|
|
kwargs):
|
|
if isinstance(self.key_func, str):
|
|
key_func = getattr(view_instance, self.key_func)
|
|
else:
|
|
key_func = self.key_func
|
|
if key_func:
|
|
return key_func(
|
|
view_instance=view_instance,
|
|
view_method=view_method,
|
|
request=request,
|
|
args=args,
|
|
kwargs=kwargs,
|
|
)
|
|
|
|
def calculate_timeout(self, view_instance, **_):
|
|
if isinstance(self.timeout, str):
|
|
self.timeout = getattr(view_instance, self.timeout)
|
|
return self.timeout
|
|
|
|
def callback_check(self,
|
|
view_instance,
|
|
view_method,
|
|
request,
|
|
args,
|
|
kwargs,
|
|
cache_key):
|
|
if isinstance(self.callback_func, str):
|
|
callback_func = getattr(view_instance, self.callback_func)
|
|
else:
|
|
callback_func = self.callback_func
|
|
if callback_func:
|
|
return callback_func(
|
|
view_instance=view_instance,
|
|
view_method=view_method,
|
|
request=request,
|
|
args=args,
|
|
kwargs=kwargs,
|
|
cache_key=cache_key
|
|
)
|
|
|
|
|
|
cache_response = MagicCacheResponse
|
|
|