t Merge branch 'master' of git.wanmeizhensuo.com:alpha/physical into test

Makefile

@@ -5,4 +5,4 @@ crontab:
 =======
 >>>>>>> fe8ae25876e5baba8581e3a79f5440c3f7f21dca
 celery:
-	celery -A physical worker -c 1 -Q vest -l debug
+	celery -A physical worker -c 1 -Q tapir-alpha -l debug --max-tasks-per-child == 500

libs/timelib.py

@@ -29,8 +29,8 @@ def get_rand_time(hourlow=0, hourup=13, minutelow=0, minuteup=60):
     # todo redis会自动给加8个小时，所以这边先写死减少8小时
     now_time = NOW + timedelta(hours=hours, minutes=minutes) - timedelta(hours=8)
     time = eta_2_push_time(now_time.strftime("%Y-%m-%d %H:%M:%S"))
-    print(datetime.fromtimestamp(time))
-    return datetime.fromtimestamp(time)
+    print(datetime.fromtimestamp(time, pytz.timezone('Asia/Shanghai')))
+    return datetime.fromtimestamp(time, pytz.timezone('Asia/Shanghai'))
 
 
 if __name__ == '__main__':

physical/celery.py

@@ -13,6 +13,14 @@ app = Celery('physical')
 # - namespace='CELERY' means all celery-related configuration keys
 #   should have a `CELERY_` prefix.
 app.config_from_object('django.conf:settings', namespace='CELERY')
+app.conf.ONCE = {
+  'backend': 'celery_once.backends.Redis',
+  'settings': {
+    'url': settings.CELERY_BROKER_URL,
+    'default_timeout': 60 * 60
+  }
+}
+
 
 # Load task modules from all registered Django app configs.
 app.autodiscover_tasks()

physical/celery_task_router.py

@@ -9,8 +9,6 @@ class CeleryTaskRouter(object):
     queue_task_map = {
         "tapir-alpha": [
             'injection.data_sync.tasks.write_to_es',
-        ],
-        "vest": [
             'vest.request.auto_request.click',
             'vest.request.auto_request.reply',
             'vest.request.auto_request.follow',

requirements.txt

@@ -14,6 +14,7 @@ gevent==1.3.7
 pypinyin==0.34.1
 numpy==1.16.2
 lz4==2.1.6
+celery_once==3.0.1
 
 git+ssh://git@git.wanmeizhensuo.com/backend/gm-rpcd.git@master
 git+ssh://git@git.wanmeizhensuo.com/backend/helios.git@master

vest/request/auto_request.py

@@ -8,6 +8,7 @@ import json
 import redis
 import smtplib
 from celery import shared_task
+from celery_once import QueueOnce
 from libs.cache import redis_client
 from email.mime.text import MIMEText
 from email.utils import formataddr
@@ -114,7 +115,7 @@ def logins(user_id):
         return None
 
 
-@shared_task
+@shared_task(retry_kwargs={'max_retries': 0}, base=QueueOnce, once={'graceful': True, 'unlock_before_run': True})
 def click(cookies_get, id):
     # 点赞
     try:
@@ -148,7 +149,7 @@ def click(cookies_get, id):
         logging.error("catch exception,logins:%s" % traceback.format_exc())
 
 
-@shared_task
+@shared_task(retry_kwargs={'max_retries': 0}, base=QueueOnce, once={'graceful': True, 'unlock_before_run': True})
 def reply(cookies_get, id, content):
     try:
         post_dict = {
@@ -272,7 +273,7 @@ def get_comments():
         return None
 
 
-@shared_task
+@shared_task(retry_kwargs={'max_retries': 0}, base=QueueOnce, once={'graceful': True, 'unlock_before_run': True})
 def follow(cookies_get, id):
     try:
         post_dict = {
@@ -438,7 +439,7 @@ def set_reply_to_redis():
         logging.error("catch exception,logins:%s" % traceback.format_exc())
 
 
-@shared_task
+@shared_task(retry_kwargs={'max_retries': 0}, base=QueueOnce, once={'graceful': True, 'unlock_before_run': True})
 def reply2(cookies_get, id, content, replied_id):
     try:
         post_dict = {
@@ -459,7 +460,7 @@ def reply2(cookies_get, id, content, replied_id):
         logging.error("catch exception,logins:%s" % traceback.format_exc())
 
 
-@shared_task
+@shared_task(retry_kwargs={'max_retries': 0}, base=QueueOnce, once={'graceful': True, 'unlock_before_run': True})
 def pictorial_reply(cookies_get, id, content):
     try:
         post_dict = {