灌水功能

.idea/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

click/views/auto_click.py

@@ -7,11 +7,12 @@ import time
 from moment.views.process_time import get_content_time_by_create_time, get_click_follow_time_by_create_time, \
     get_vest_userid
 from moment.views.process_time import save_data_to_kafka
+from libs.error import logging_exception
 
 logger = logging.getLogger(__name__)
 
 
-def auto_click_user(card_info):
+def auto_click_user(card_info, after_day=False):
     """
 
     发帖后2小时：[1-3]个赞
@@ -47,9 +48,22 @@ def auto_click_user(card_info):
                     "all_follow_id" not in card_info and "all_push_time" not in card_info):
 
                 ###根据创建时间回答等级去获取对应的下发时间
-                get_time = get_click_follow_time_by_create_time(create_time, content_level, action_type="follow")
-                if len(get_time) > 0:
 
+                while True:
+
+                    repeat_time = 1
+                    if card_info['type'] == "get_write_answer_userinfo":
+                        repeat_time = 0
+                        card_info['have_click_number'] = 0
+
+                    get_time, time_region = get_click_follow_time_by_create_time(create_time, content_level,
+                                                                                 action_type="click",
+                                                                                 after_day=after_day,
+                                                                                 card_info=card_info,
+                                                                                 repeat_time=repeat_time)
+                    logging.info("get get_time and time_region:%s" % get_time)
+
+                    if len(get_time) > 0 and time_region != 3:
                         ###拿到下发时间后根据下发的时间个数去拿对应个数的马甲账号
                         userids = get_vest_userid(need_comment_num=len(get_time))
                         for i in range(0, len(userids)):
@@ -60,10 +74,12 @@ def auto_click_user(card_info):
                             card_info['all_push_time'] = get_time
                             card_info["need_pust_num"] = len(userids)
                             card_info["have_pust_num"] = 0
-                    save_data_to_kafka(card_info)  # 存储数据
-
+                        save_data_to_kafka(card_info)
+                        break
                     else:
-                    pass
+                        if time_region == 3:
+                            break
+                        repeat_time += 1
 
             else:  # 代表还有push好的时间没有下发完成 需要继续使用这些
 
@@ -74,14 +90,15 @@ def auto_click_user(card_info):
                 card_info['all_push_time'].remove(current_push_time)
 
                 if len(card_info["all_follow_id"]) > 0:
-                    card_info["current_user_id"] = card_info["all_follow_id"].pop()
+                    card_info["current_user_id"] = card_info["all_follow_id"][0]
 
                 if len(card_info["all_follow_id"]) > 0:
-                    card_info['current_push_time'] = card_info['all_push_time'].pop()
+                    card_info['current_push_time'] = card_info['all_push_time'][0]
 
                 card_info["have_pust_num"] = card_info["have_pust_num"] + 1
                 save_data_to_kafka(card_info)  # 存储数据
                 logging.info("get--------click--------------card_info:%s" % card_info)
 
     except:
+        logging_exception()
         logging.error("catch exception,err_msg:%s" % traceback.format_exc())

click/views/click_fun.py

+from libs.rpc import get_rpc_invoker
+import traceback
+from libs.error import logging_exception
+import logging
+
+
+def click(card_info):
+    try:
+        rpc_invoker = get_rpc_invoker()
+        data = rpc_invoker['qa/irrigation/create_answer_vote'](user_id=card_info['current_user_id'],
+                                                               answer_id=card_info['card_id']).unwrap()
+
+    except:
+        logging_exception()
+        logging.error("catch exception,err_msg:%s" % traceback.format_exc())

comment/views/auto_comment.py

@@ -6,11 +6,12 @@ import random
 import time
 from moment.views.process_time import get_content_time_by_create_time, get_vest_userid_and_comment, get_vest_userid
 from moment.views.process_time import save_data_to_kafka
+from libs.error import logging_exception
 
 logger = logging.getLogger(__name__)
 
 
-def auto_comment_user(card_info):
+def auto_comment_user(card_info, after_day=False):
     """
     当天发布之后
 
@@ -48,34 +49,53 @@ def auto_comment_user(card_info):
             create_time = card_info["create_time"]
             content_level = card_info["content_level"]
             card_info["action_type"] = "comment"
-
             tag_names = card_info['tag_names']
+
             # 当满足这些条件的时候 代表已经完成了下发的所有的时间或者还没有下发过时间 需要给予新的时间
             if ("all_follow_id" in card_info and "all_push_time" in card_info and len(
                     card_info['all_follow_id']) == 0 and len(card_info['all_push_time']) == 0) or (
-                    "all_follow_id" not in card_info and "all_push_time" not in card_info):
-
-                ###根据创建时间回答等级去获取对应的下发时间
-                get_time = get_content_time_by_create_time(create_time, content_level, action_type="follow")
-                if len(get_time) > 0:
-
+                    "all_follow_id" not in card_info and "all_push_time" not in card_info and card_info[
+                'type'] == "get_write_answer_userinfo"):
+
+                while True:
+
+                    repeat_time = 1
+                    if card_info['type'] == "get_write_answer_userinfo":
+                        repeat_time = 0
+                        card_info['have_comment_number'] = 0
+
+                    get_time, time_region = get_content_time_by_create_time(create_time, content_level,
+                                                                            action_type="comment",
+                                                                            after_day=after_day,
+                                                                            card_info=card_info,
+                                                                            repeat_time=repeat_time)
+                    logging.info("get get_time and time_region:%s" % get_time)
+                    if len(get_time) > 0 and time_region != 3:
                         ###拿到下发时间后根据下发的时间个数去拿对应个数的马甲账号
-                    all_content = get_vest_userid_and_comment(need_comment_num=len(get_time), tag_names=tag_names)
-                    all_userids = get_vest_userid(need_comment_num=len(get_time))
-
-                    for i in range(0, len(all_userids)):
-                        card_info["all_follow_id"] = all_userids
-                        card_info["current_user_id"] = all_userids[0]
+                        all_content = get_vest_userid_and_comment(need_comment_num=len(get_time),
+                                                                  tag_names=tag_names,
+                                                                  card_id=card_info['card_id'])
+
+                        userids = get_vest_userid(need_comment_num=len(get_time))
+                        for i in range(0, len(userids)):
+                            card_info["all_follow_id"] = userids
+                            card_info["current_user_id"] = userids[0]
                             card_info['type'] = "have_get_push_time"
                             card_info['current_push_time'] = get_time[0]
                             card_info['all_push_time'] = get_time
-                        card_info["need_pust_num"] = len(all_userids)
+                            card_info["need_pust_num"] = len(userids)
                             card_info["have_pust_num"] = 0
-                        card_info['comment_content'] = all_content[i]
-                    save_data_to_kafka(card_info)  # 存储数据
+                            card_info["all_content"] = all_content
+                            if len(all_content) > 0:
+                                card_info['comment_content'] = all_content[0]
+                        save_data_to_kafka(card_info)
+                        break
                     else:
-                    pass
+                        if time_region == 3:
+                            break
+                        repeat_time += 1
 
+                logging.info("get have_get_after__time:%s" % card_info)
             else:  # 代表还有push好的时间没有下发完成 需要继续使用这些
 
                 logging.info("get-------comment---------------card_info:%s" % card_info)
@@ -85,14 +105,21 @@ def auto_comment_user(card_info):
                 card_info['all_push_time'].remove(current_push_time)
                 card_info["have_pust_num"] = card_info["have_pust_num"] + 1
 
+                comment_content = card_info["comment_content"]
+                card_info["all_content"].remove(comment_content)
+
+                if len(card_info["all_content"]) > 0:
+                    card_info["comment_content"] = card_info["all_content"][0]
+                    
                 if len(card_info["all_follow_id"]) > 0:
-                    card_info["current_user_id"] = card_info["all_follow_id"].pop()
+                    card_info["current_user_id"] = card_info["all_follow_id"][0]
 
                 if len(card_info["all_follow_id"]) > 0:
-                    card_info['current_push_time'] = card_info['all_push_time'].pop()
+                    card_info['current_push_time'] = card_info['all_push_time'][0]
 
                 save_data_to_kafka(card_info)  # 存储数据
                 logging.info("get-------comment---------------card_info:%s" % card_info)
 
     except:
+        logging_exception()
         logging.error("catch exception,err_msg:%s" % traceback.format_exc())

comment/views/comment_fun.py

+from libs.rpc import get_rpc_invoker
+import traceback
+from libs.error import logging_exception
+import logging
+
+
+def comment(card_info):
+    try:
+        rpc_invoker = get_rpc_invoker()
+        rpc_invoker['qa/irrigation/create_answer_reply'](user_id=card_info['current_user_id'],
+                                                                 answer_id=card_info['card_id'],
+                                                                 content=card_info['comment_content']).unwrap()
+
+    except:
+        logging_exception()
+        logging.error("catch exception,err_msg:%s" % traceback.format_exc())

follow/views/auto_follow.py

@@ -7,11 +7,12 @@ import time
 from moment.views.process_time import get_content_time_by_create_time, get_click_follow_time_by_create_time, \
     get_vest_userid
 from moment.views.process_time import save_data_to_kafka
+from libs.error import logging_exception
 
 logger = logging.getLogger(__name__)
 
 
-def auto_follow_user(card_info):
+def auto_follow_user(card_info, after_day=False):
     """
     灌水功能下发关注
     发帖后2小时：[1-3]个粉丝
@@ -48,8 +49,21 @@ def auto_follow_user(card_info):
                     "all_follow_id" not in card_info and "all_push_time" not in card_info):
 
                 ###根据创建时间回答等级去获取对应的下发时间
-                get_time = get_click_follow_time_by_create_time(create_time, content_level, action_type="follow")
-                if len(get_time) > 0:
+                while True:
+
+                    repeat_time = 1
+                    if card_info['type'] == "get_write_answer_userinfo":
+                        repeat_time = 0
+                        card_info['have_follow_number'] = 0
+
+                    get_time, time_region = get_click_follow_time_by_create_time(create_time, content_level,
+                                                                                 action_type="follow",
+                                                                                 after_day=after_day,
+                                                                                 card_info=card_info,
+                                                                                 repeat_time=repeat_time)
+                    logging.info("get get_time and time_region:%s" % get_time)
+
+                    if len(get_time) > 0 and time_region != 3:
 
                         ###拿到下发时间后根据下发的时间个数去拿对应个数的马甲账号
                         userids = get_vest_userid(need_comment_num=len(get_time))
@@ -61,10 +75,14 @@ def auto_follow_user(card_info):
                             card_info['all_push_time'] = get_time
                             card_info["need_pust_num"] = len(userids)
                             card_info["have_pust_num"] = 0
-                    save_data_to_kafka(card_info)  # 存储数据
-
+                        save_data_to_kafka(card_info)
+                        break
                     else:
-                    pass
+                        if time_region == 3:
+                            break
+                        repeat_time += 1
+
+                logging.info("get have_get_after__time:%s" % card_info)
 
             else:  # 代表还有push好的时间没有下发完成 需要继续使用这些
                 logging.info("get-----follow----card_info:%s" % card_info)
@@ -85,6 +103,6 @@ def auto_follow_user(card_info):
 
                 logging.info("get-----follow----card_info:%s" % card_info)
 
-
     except:
+        logging_exception()
         logging.error("catch exception,err_msg:%s" % traceback.format_exc())

follow/views/follow_fun.py

+from libs.rpc import get_rpc_invoker
+import traceback
+from libs.error import logging_exception
+import logging
+
+
+def follow(card_info):
+    try:
+        rpc_invoker = get_rpc_invoker()
+        rpc_invoker['api/irrigation/user_add_follow'](follow_user_id=card_info['current_user_id'],
+                                                      followed_user_id=card_info['card_user_id']).unwrap()
+
+
+    except:
+        logging_exception()
+        logging.error("catch exception,err_msg:%s" % traceback.format_exc())

moment/views/consum.py

@@ -9,79 +9,106 @@ from django.conf import settings
 from follow.views.auto_follow import auto_follow_user
 from click.views.auto_click import auto_click_user
 from comment.views.auto_comment import auto_comment_user
+from moment.views.process_time import save_data_to_kafka
 import datetime
 import time
 import logging
+from libs.error import logging_exception
+from click.views.click_fun import click
+from comment.views.comment_fun import comment
+from follow.views.follow_fun import follow
 
 
 def kafka_consum(topic_name=None):
-    try:
     topic_name = settings.KAFKA_TOPIC_NAME if not topic_name else topic_name
-        consumser_obj = KafkaConsumer(topic_name, bootstrap_servers=[settings.KAFKA_BROKER_LIST])
-        logging.info("get consumser_obj:%s" % consumser_obj)
+    consumser_obj = KafkaConsumer(topic_name, bootstrap_servers=[settings.KAFKA_BROKER_LIST], enable_auto_commit=False,
+                                  auto_commit_interval_ms=100, group_id="vest")
+
+    try:
         while True:
-            msg_dict = consumser_obj.poll(timeout_ms=100)
+            msg_dict = consumser_obj.poll(timeout_ms=100, max_records=10)
+
             for msg_key in msg_dict:
                 consume_msg = msg_dict[msg_key]
                 for msg in consume_msg:
                     card_info = json.loads(msg.value)
+                    if card_info['card_type'] == "tractate777":
+                        logging.info("+++++++++++++++++++++++++新的数据进来了+++++++++++++++++++++++")
                         logging.info("get card_info:%s" % card_info)
                         # 代表当天数据
                         current_push_time = card_info['current_push_time']
                         create_time = card_info['create_time']
-                    action_type = card_info['action_type']
                         nowtime = datetime.datetime.now()
-                    # nowtime = now.strftime('%Y-%m-%d %H:%M:%S')
                         push_time_date = datetime.datetime.strptime(current_push_time, '%Y-%m-%d %H:%M:%S')
-
-                    logging.info("get push_time_date:%s" % push_time_date)
-                    logging.info("get push_time_date:%s" % type(push_time_date))
-
-                    logging.info("get nowtime:%s" % nowtime)
-                    logging.info("get nowtime:%s" % type(nowtime))
-
                         if push_time_date <= nowtime:  # push_time已经到时间了 需要去下发
                             logging.info("get push_time_date:%s" % push_time_date)
                             # 判断如果当前的push_time 和当前的创建时间一样  需要给push_time下发真的push时间
                             if current_push_time == create_time:
-                            auto_follow_user(card_info)
-                            auto_comment_user(card_info)
-                            auto_click_user(card_info)
+                                if action_type == "comment":
+                                    auto_comment_user(card_info, after_day=True)
+
+                                elif action_type == "click":
+                                    auto_click_user(card_info, after_day=True)
+
+                                elif action_type == "follow":
+                                    auto_follow_user(card_info, after_day=True)
 
                                 else:
-                            # current_push_time = card_info['current_push_time']
-                            # current_time = datetime.datetime.strptime(current_push_time, '%Y-%m-%d %H:%M:%S')
-                            # now = datetime.datetime.now()
-                            # 需要判断是否真的已经下发完了 如果是时间过了但是没有下发直接下发了
-                            # 当前已下发完 需要判断最新一次的下发时间是否是今天 是的话直接跳过
-                            if card_info['have_pust_num'] == card_info['need_pust_num']:
-                                if nowtime.day - push_time_date.day == 0:  # 是今天的话直接跳过
                                     pass
+
                             else:
-                                    # 再存储起来？
-                                    pass
-                            # 还有一种是下发时间还没有到 之前没有轮询到
+                                # 当前已下发完 需要判断最新一次的下发时间是否是今天 是的话直接保存起来,l=轮询等待明天再下发
+                                if card_info['have_pust_num'] == card_info['need_pust_num']:
+                                    if nowtime.day - push_time_date.day == 0:  # 今日的已经下发完了，需要去取之后的
+                                        action_type = card_info['action_type']
+                                        logging.info("get-------今天已经下发完了----------------")
+                                        if action_type == "comment":
+                                            auto_comment_user(card_info, after_day=True)
+                                        elif action_type == "click":
+                                            auto_click_user(card_info, after_day=True)
+                                        elif action_type == "follow":
+                                            auto_follow_user(card_info, after_day=True)
                                         else:
-                                if card_info['have_pust_num'] <= card_info['need_pust_num'] and \
+                                            pass
+
+                                else:  # 还有一种是下发时间已到
+                                    action_type = card_info['action_type']
+
+                                    if card_info['have_pust_num'] < card_info['need_pust_num'] and \
                                             push_time_date < nowtime:
+
                                         if action_type == "comment":  # 在这里去调评论的接口
-                                        logging.info("get------------------comment-----------------")
+                                            if 'have_comment_number' in card_info and \
+                                                    card_info['have_comment_number'] < 20:
+                                                card_info["have_comment_number"] += 1
+                                                logging.info("get------xiafa------------vestcomment:%s" % card_info)
+
                                             # 调完接口后需要再次去拿新的push_time的时间
                                             auto_comment_user(card_info)
 
                                         elif action_type == "click":  # 在这里去调点赞的接口
-                                        logging.info("get--------------click---------------------")
+                                            if 'have_click_number' in card_info and card_info['have_click_number'] < 20:
+                                                card_info["have_click_number"] += 1
+                                                logging.info("get------xiafa------------vestclick:%s" % card_info)
 
                                             auto_click_user(card_info)
 
                                         elif action_type == "follow":  # 在这里去调关注的接口
-                                        logging.info("get----------follow-------------------------")
+
+                                            if 'have_follow_number' in card_info and \
+                                                    card_info['have_follow_number'] < 20:
+                                                logging.info("get------xiafa------------vestfollow:%s" % card_info)
+                                                card_info["have_follow_number"] += 1
 
                                             auto_follow_user(card_info)
 
                         else:  # push_time时间未到 需要等待
                             logging.info("push_time时间未到========================需要等待")
+                            save_data_to_kafka(card_info)
                             pass
 
+                    consumser_obj.commit_async()
     except:
+        consumser_obj.close()
+        logging_exception()
         logging.error("catch exception,err_msg:%s" % traceback.format_exc())

moment/views/irrigation.py

@@ -12,20 +12,50 @@ import logging
 
 
 @bind('vest/moment/vest_irrigation')
-def vest_irrigation(card_id=0, card_type=None, create_time="", content_level=0, tag_names=[]):
+def vest_irrigation(card_id=0, card_type=None, card_user_id=None, create_time="", content_level=0, tag_names=[]):
     try:
         producer = KafkaProducer(bootstrap_servers=[settings.KAFKA_BROKER_LIST])
         topic = settings.KAFKA_TOPIC_NAME
-        msg_dict = {
+        comment_msg_dict = {
             "card_id": card_id,
             "card_type": card_type,
             "create_time": create_time,
             "content_level": content_level,
             "tag_names": tag_names,
             "type": "get_write_answer_userinfo",
-            "current_push_time": create_time
+            "current_push_time": create_time,
+            'action_type': 'comment'
+
+        }
+        producer.send(topic, json.dumps(comment_msg_dict).encode())
+
+        follow_msg_dict = {
+            "card_id": card_id,
+            "card_type": card_type,
+            "create_time": create_time,
+            "content_level": content_level,
+            "tag_names": tag_names,
+            "type": "get_write_answer_userinfo",
+            "current_push_time": create_time,
+            'action_type': 'follow',
+            'card_user_id': card_user_id
+
         }
-        producer.send(topic, json.dumps(msg_dict).encode())
+        producer.send(topic, json.dumps(follow_msg_dict).encode())
+
+        click_msg_dict = {
+            "card_id": card_id,
+            "card_type": card_type,
+            "create_time": create_time,
+            "content_level": content_level,
+            "tag_names": tag_names,
+            "type": "get_write_answer_userinfo",
+            "current_push_time": create_time,
+            'action_type': 'click'
+
+        }
+        producer.send(topic, json.dumps(click_msg_dict).encode())
+
         producer.close()
 
     except:

trans2es/admin.py

+from django.contrib import admin
+
+# Register your models here.

trans2es/apps.py

+from django.apps import AppConfig
+
+
+class Trans2EsConfig(AppConfig):
+    name = 'trans2es'

trans2es/management/commands/trans2es_data2es_redis.py

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from django.core.management.base import BaseCommand
+import traceback
+import logging
+from moment.views.consum import kafka_consum
+from libs.error import logging_exception
+
+logger = logging.getLogger('async_debug')
+
+
+class Command(BaseCommand):
+    args = ''
+    help = 'dump data to elasticsearch, parallel'
+
+    from optparse import make_option
+
+    option_list = BaseCommand.option_list + (
+        make_option('-t', '--type', dest='type', help='type name to dump data to elasticsearch', metavar='TYPE',
+                    default=''),
+        make_option('-i', '--index-prefix', dest='index_prefix', help='index name to dump data to elasticsearch',
+                    metavar='INDEX_PREFIX'),
+        make_option('-p', '--parallel', dest='parallel', help='parallel process count', metavar='PARALLEL'),
+        make_option('-s', '--pks', dest='pks', help='specify sync pks, comma separated', metavar='PKS', default=''),
+        make_option('--streaming-slicing', dest='streaming_slicing', action='store_true', default=True),
+        make_option('--no-streaming-slicing', dest='streaming_slicing', action='store_false', default=True),
+        make_option('-S', '--sync_type', dest='sync_type', help='sync data to es', metavar='TYPE', default='')
+    )
+
+    def handle(self, *args, **options):
+        try:
+            if len(options["sync_type"]) and options["sync_type"] == "auto_vest":
+                kafka_consum()
+                logging.info("add a log >>>>>  auto_vest")
+
+        except:
+            logging_exception()
+            logging.error("catch exception,err_msg:%s" % traceback.format_exc())

trans2es/models.py

+from django.db import models
+
+# Create your models here.

trans2es/tests.py

+from django.test import TestCase
+
+# Create your tests here.

trans2es/views.py

+from django.shortcuts import render
+
+# Create your views here.

vest/settings.py

@@ -41,7 +41,8 @@ INSTALLED_APPS = (
     "click",
     "moment",
     'follow',
-    'comment'
+    'comment',
+    'trans2es'
 )
 
 MIDDLEWARE_CLASSES = (
@@ -55,8 +56,6 @@ MIDDLEWARE_CLASSES = (
     'django.middleware.security.SecurityMiddleware',
 )
 
-# ROOT_URLCONF = 'vest.urls'
-
 TEMPLATES = [
     {
         'BACKEND': 'django.template.backends.django.DjangoTemplates',

vest/settings_local.py

 # coding=utf-8
-from celery import *
-
-from celery.schedules import crontab
 
 KAFKA_TOPIC_NAME = "auto_vest"
 KAFKA_BROKER_LIST = "kafka-service-0.kafka-service-headless.test:9092"
@@ -16,19 +13,17 @@ REDIS = {
 }
 
 # from base
-DATABASES = {
-    'ENGINE': 'django.db.backends.mysql',  # 设置为mysql数据库
-    'NAME': 'zhengxing_test',
-    'USER': 'work',
-    'PASSWORD': 'Gengmei1',
-    'HOST': 'bj-cdb-6slgqwlc.sql.tencentcdb.com',
-    'PORT': '62120',
-    'OPTIONS': {
+ENGINE = 'django.db.backends.mysql',  # 设置为mysql数据库
+NAME = 'mimas_test'
+USER = 'work'
+PASSWORD = 'Gengmei1'
+HOST = 'bj-cdb-6slgqwlc.sql.tencentcdb.com'
+PORT = 62120
+
+OPTIONS = {
     "init_command": "SET foreign_key_checks = 0;",
     "charset": "utf8mb4",  # 为了支持emoji表情
-    },
 }
-
 GM_KV_HOSTS = [
     {
         "host": "redis.paas-test.env",