test

标签：存储   mysq   efault   author   self   and   uid   har   数据量   

CREATE TABLE `account_uid_ip` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`app_id` char(20) NOT NULL DEFAULT ‘zjh‘,
`account_id` bigint(20) DEFAULT NULL,
`client_ip` char(20) DEFAULT NULL,
`record_time` datetime DEFAULT NULL,
`updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `app_id` (`app_id`,`account_id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

# -*- coding:utf-8 -*-
import time
from datetime import date, timedelta
import pandas as pd
from pandas import *
from config.log_path_config import *
from config.database_config import DATAMIND_DATABASE_CONFIG
from utils.log_util import LogManage
from analysis import BaseAnalysis


__version__ = ‘1.0.0.0‘
"""
@brief  :   简介
@details:   账号注册ip分析脚本
@author :   tliu
@date   :   2019-06-04
"""

log = LogManage(__name__)


class RegisterIpAnalysis(BaseAnalysis):
    """
    {"app": "dfh",
    "channel": "kxdwc",
    "date_time": "2019-06-04 14:53:09",
    "distinct_id": "41336",
    "event_id": "36",
    "event_type": "IP",
    "game_id": "",
    "ip": "58.49.115.114",
    "onlyid": "201906041453098176840002394773",
    "package": "com.zhajinhua.kxdwc01",
    "project": "dfh",
    "room_id": "",
    "system": "android",
    "table_id": "",
    "timestamp": "1559631189",
    "type": "game_event",
    "version": "5.0.0"}
    """
    def __init__(self, current_date):
        super(RegisterIpAnalysis, self).__init__(current_date, DATAMIND_DATABASE_CONFIG)
        log.info("RegisterIpAnalysis __init__ :%s" % current_date)
        self.register_log_file = self.merge_log(REGISTER_IP_LOG, current_date, "register_ip")
        # self.register_log_file = SAVE_LOG_PATH + "23/logs"

    def analysis(self):
        log.info("start handle %s" % self.register_log_file)
        df = pd.read_json(self.register_log_file, lines=True)
        df = df[[‘app‘, ‘distinct_id‘, ‘ip‘, ‘date_time‘]]
        df.columns = [‘app_id‘, ‘account_id‘, ‘client_ip‘, "record_time"]
        self.save_to_account_uid_ip(df)
        self.close()

    def save_to_account_uid_ip(self, dst_df):
        """
        将dataframe结果集存储到表中
        :param dst_df: dataframe对象
        :return:
        """
        if dst_df.shape[0] <= 0:
            return
        log.info("开始存储注册ip表,数据量：{0}".format(dst_df.shape))
        sql = list()
        sql.append("replace into account_uid_ip(app_id, account_id, client_ip, "
                   "record_time) values ")
        for row_index, row in dst_df.iterrows():
            sql.append("(‘%s‘,%s,‘%s‘,‘%s‘)," %
                       (row[‘app_id‘], row[‘account_id‘], row[‘client_ip‘], row[‘record_time‘]))
        sql = "".join(sql)[:-1]
        self.mysql_con.excute(sql)


def insert_register_ip_data(count=0):
    log.info("insert_login_ip_data -->执行注册日志分析")
    t1 = time.time()
    try:
        now_time = datetime.now()
        log.info(now_time)
        # 小于凌晨2点的情况下还要处理前一天的数据防止0点的数据没有处理进去
        if now_time.hour <= 2:
            today = date.today() - timedelta(count + 1)
            RegisterIpAnalysis(today).analysis()
        # 处理当天的注册数据
        today = date.today() - timedelta(count)
        RegisterIpAnalysis(today).analysis()
        t2 = time.time() - t1
        log.info("insert_login_ip_data -->结束注册日志分析,spend：%s" % t2)
    except Exception as e:
        log.info(e)

REGISTER_IP_LOG = BASE_LOG_PATH + "36/"                                         # 注册ip日志

if __name__ == ‘__main__‘:
    insert_register_ip_data()

test

标签：存储   mysq   efault   author   self   and   uid   har   数据量   

原文地址：https://www.cnblogs.com/niuniuc/p/11027270.html