#!/usr/bin/env python # -*- coding: utf-8 -*- import time from collections import namedtuple from gm_rpcd.all import context from gm_types.user import ( STAT_LOG_USER_ACTION, STAT_LOG_ACTION_CONTENT_TYPE, ) class SocialStatLogForUserAction(object): """ 整理用户行为埋点数据 """ _ctx_t_enum_list = ["diary", "question", "user_post", "article"] CONTENT_TYPE = namedtuple("CONTENT_TYPE", _ctx_t_enum_list)( **dict(zip(_ctx_t_enum_list, _ctx_t_enum_list)) ) @classmethod def _format_ctx_type_data(cls, ctx_type): ctx_mapping_dic = { cls.CONTENT_TYPE.diary: STAT_LOG_ACTION_CONTENT_TYPE.DIARY, cls.CONTENT_TYPE.question: STAT_LOG_ACTION_CONTENT_TYPE.QUESTION, cls.CONTENT_TYPE.user_post: STAT_LOG_ACTION_CONTENT_TYPE.USER_POST, cls.CONTENT_TYPE.article: STAT_LOG_ACTION_CONTENT_TYPE.ARTICLE, } return ctx_mapping_dic.get(ctx_type, "") @classmethod def convert_user_action_stat_log(cls, _data): _data.update({ "timestamp": time.time(), }) context.request_logger.app(**_data) @classmethod def stat_log_for_like(cls, **kwargs): _ctx_type = cls._format_ctx_type_data(kwargs.get("content_type", "")) _data = { "user_id": kwargs.get("user_id", 0), "content_id": kwargs.get("content_id", 0), "action": STAT_LOG_USER_ACTION.LIKE, "content_type": _ctx_type, } cls.convert_user_action_stat_log(_data) @classmethod def stat_log_for_favor(cls, **kwargs): _ctx_type = cls._format_ctx_type_data(kwargs.get("content_type", "")) _data = { "user_id": kwargs.get("user_id", 0), "content_id": kwargs.get("content_id", 0), "action": STAT_LOG_USER_ACTION.COLLECT, "content_type": _ctx_type, } cls.convert_user_action_stat_log(_data) @classmethod def stat_log_for_reply(cls, **kwargs): _ctx_type = cls._format_ctx_type_data(kwargs.get("content_type", "")) _data = { "user_id": kwargs.get("user_id", 0), "content_id": kwargs.get("content_id", 0), "action": STAT_LOG_USER_ACTION.COMMENT, "content_type": _ctx_type, } cls.convert_user_action_stat_log(_data)