dwd_user_studio_add_content.py

# -*- coding: utf-8 -*-

'''
用户添加工作室记录,数据量不大，全量更新
'''
import logging
import os
import sys

from common.mysql_uitl import save_result, fetch_all, save_etl_log
from common.time_util import now_str, YMDHMS_FORMAT


logging.basicConfig(format="%(asctime)s %(name)s:%(levelname)s:%(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.INFO)
file_name = sys.argv[0]
task_file = os.path.split(__file__)[-1].split(".")[0]


def dwd_user_studio_add_content(data_dt):
    # 公募基金
    public_fund_dict = query_studio_add_public_fund()
    row1 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', public_fund_dict, file_name)

    # 私募基金
    private_equity_dict = query_studio_add_private_equity()
    row2 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', private_equity_dict, file_name)

    # 白名单基金
    tamp_fund_dict = query_studio_add_tamp_fund()
    row3 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', tamp_fund_dict, file_name)

    # 私有基金
    private_fund_dict = query_studio_add_private_fund()
    row4 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', private_fund_dict, file_name)
    #
    # 无净值产品
    ifa_nonav_dcit = query_studio_add_ifa_nonav()
    row5 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', ifa_nonav_dcit, file_name)

    # 直播
    live_dict = query_studio_add_live()
    row6 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', live_dict, file_name)

    # 课程
    course_dict = query_studio_add_course()
    row7 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', course_dict, file_name)

    # 短视频
    short_video_dict = query_studio_add_short_video()
    row8 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', short_video_dict, file_name)

    # 探普文章
    tamp_article_dict = query_studio_add_tamp_article()
    row9 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', tamp_article_dict, file_name)

    # 私有文章
    self_article_dict = query_studio_add_self_article()
    row10 = save_result('tamp_data_dwd', 'dwd_user_studio_add_content', self_article_dict, file_name)

    row = row1 + row2 + row3 + row4 + row5 + row6 + row7 + row8 + row9 + row10
    now_time = now_str(YMDHMS_FORMAT)
    save_etl_log('tamp_data_dwd', 'dwd_user_studio_add_content', data_dt, row, 'done', task_file, now_time)


def query_studio_add_public_fund():
    '''添加公募基金'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'public_fund' as res_type
                ,p.target_id as res_id
                ,t.fund_name as res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join  tamp_fund.tx_fund_info t
        on 		p.target_id = t.id
        where   p.`type` = 'product'
        and 	p.sub_type = '0'
    '''
    public_fund_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return public_fund_dict


def query_studio_add_private_equity():
    '''添加私募基金'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'private_equity' as res_type
                ,p.target_id as res_id
                ,t.fund_name as res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join  tamp_fund.fund_info t
        on 		p.target_id = t.id
        where   p.`type` = 'product'
        and 	p.sub_type = '1'
    '''
    private_equity_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return private_equity_dict


def query_studio_add_tamp_fund():
    '''添加白名单基金'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'tamp_fund' res_type
                ,p.target_id as res_id
                ,t.fund_name as res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join  tamp_product.fund_info t
        on 		p.target_id = t.id
        where   p.`type` = 'product'
        and 	p.sub_type = '2'
    '''
    tamp_fund_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return tamp_fund_dict


def query_studio_add_private_fund():
    '''添加私有基金'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'private_fund' as res_type
                ,p.target_id as res_id
                ,t.fund_name as res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join  tamp_fund.ifa_imported_fund_info t
        on 		p.target_id = t.id
        where   p.`type` = 'product'
        and 	p.sub_type = '3'
    '''
    public_fund_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return public_fund_dict


def query_studio_add_ifa_nonav():
    '''添加无净值产品'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'ifa_nonav' as res_type
                ,p.target_id as res_id
                ,t.fund_name as res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join tamp_fund.ifa_imported_nonav_fund t
        on 		p.target_id = t.id
        where   p.`type` = 'product'
        and 	p.sub_type = '4'
    '''
    ifa_nonav_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return ifa_nonav_dict


def query_studio_add_live():
    '''添加直播'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'live' as res_type
                ,p.target_id as res_id
                ,t.zt_name as res_name
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join tamp_zhibo.zhibo_theme t
        on 		p.target_id = t.id
        where   p.`type` = 'zhibo'
    '''
    live_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return live_dict


def query_studio_add_course():
    '''添加课程'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'course' as res_type
                ,p.target_id as res_id
                ,t.title as res_name 
                ,p.course_packageId as course_id
                ,q.main_title as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join tamp_course.course t
        on 		p.target_id = t.id
        inner   join  tamp_course.course_package q 
        on      p.course_packageId = q.id 
        where   p.`type` = 'course'
    '''
    course_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return course_dict


def query_studio_add_short_video():
    '''添加短视频'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'short_video' as res_type
                ,p.target_id as res_id
                ,t.res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join tamp_analysis.short_video_view t
        on 		p.target_id = t.res_id
        where   p.`type` = 'shortVideo'
    '''
    short_video_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return short_video_dict


def query_studio_add_tamp_article():
    '''添加探普文章'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,case when t.ar_title is not null then 'tamp_article'
                      when q.af_title is not null then 'self_article'
                      else ''
                end as res_type
                ,p.target_id as res_id
                ,coalesce(t.ar_title,q.af_title) as res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        left    join tamp_user.article t
        on 		p.target_id = t.id
        left    join tamp_user.friend_article q
        on 		p.target_id = q.id 
        where   p.`type` = 'article' 
        and     p.sub_type = 'tamp'
    '''
    tamp_article_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return tamp_article_dict


def query_studio_add_self_article():
    '''添加私有文章'''
    function_name = sys._getframe().f_code.co_name
    logging.info(f'{function_name} start')
    sql = '''
         select p.id 
                ,p.user_id
                ,'self_article' as res_type
                ,p.target_id as res_id
                ,t.af_title as res_name 
                ,p.course_packageId as course_id
                ,'' as course_name
                ,p.create_time
                ,p.update_time
                ,p.delete_tag 
        from    tamp_user.user_poster_attachement p
        inner   join tamp_user.friend_article t
        on 		p.target_id = t.id
        where   p.`type` = 'article' 
        and     p.sub_type = 'self'
    '''
    self_article_dict = fetch_all(sql, None)
    logging.info(f'{function_name} success')
    return self_article_dict


if __name__ == '__main__':
    dwd_user_studio_add_content()