# This is a sample Python script. # Press ⌃R to execute it or replace it with your code. # Press Double ⇧ to search everywhere for classes, files, tool windows, actions, and settings. import hashlib import json import os import subprocess import time import traceback import urllib import datetime import requests import pymysql from contextlib import contextmanager from sqlalchemy import create_engine, Numeric, Float from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, DateTime, Text import ssl ssl._create_default_https_context = ssl._create_unverified_context HOST = 'rm-bp1nx318263k95yo3318.mysql.rds.aliyuncs.com' PORT = '3306' DATABASE = 'uservideo_bi' USERNAME = 'majin' PASSWORD = 'E5d2c960fdf3f5f0be5a27eea2f906ef' DB_URI = "mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset=utf8".format(username=USERNAME, password=PASSWORD, host=HOST, port=PORT, db=DATABASE) # HOST = 'rm-bp1k5853td1r25g3n690.mysql.rds.aliyuncs.com' # PORT = '3306' # DATABASE = 'mpad' # USERNAME = 'majin' # PASSWORD = 'e5d2c960fdf3f5f0be5a27eea2f906ef' # DB_URI = "mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset=utf8".format(username=USERNAME, # password=PASSWORD, # host=HOST, port=PORT, # db=DATABASE) Base = declarative_base() class WECHAT_AD_PUBLISHER_ADUNIT_GENERAL(Base): __tablename__ = 'wechat_ad_publisher_adunit_general' id = Column(Integer, primary_key=True) ad_unit_id = Column(String(1000)) ad_unit_name = Column(String(1000)) ad_slot = Column(String(1000)) click_count = Column(Integer, default=0) click_rate = Column(Float, default=0.0) date = Column(String(1000)) ecpm = Column(String(1000)) exposure_count = Column(Integer, default=0) exposure_rate = Column(Float, default=0.0) income = Column(Integer, default=0) req_succ_count = Column(Integer, default=0) app_type = Column(Integer, default=0) slot_str = Column(String(1000)) date_str = Column(String(1000)) def __init__(self): print(f"AUNIT_GENERAL.init app_type = {self.app_type}, ad_unit_id = {self.ad_unit_id}") def __repr__(self): return '' % self.ad_unit_id class WECHAT_AD_PUBLISHER_ADPOS_GENERAL(Base): __tablename__ = 'wechat_ad_publisher_adpos_general' id = Column(Integer, primary_key=True) ad_slot = Column(String(1000)) click_count = Column(Integer, default=0) click_rate = Column(Float, default=0.0) date = Column(String(1000)) ecpm = Column(String(1000)) exposure_count = Column(Integer, default=0) exposure_rate = Column(Float, default=0.0) income = Column(Integer, default=0) req_succ_count = Column(Integer, default=0) app_type = Column(Integer, default=0) slot_str = Column(String(1000)) date_str = Column(String(1000)) def __init__(self): print(f"ADPOS_GENERAL.init app_type = {self.app_type}, ad_slot = {self.ad_slot}") def __repr__(self): return '' % self.ad_slot @contextmanager def session_maker(session=None, session_factory=None): try: if session_factory is None: engine = create_engine(DB_URI) session_factory = sessionmaker(bind=engine) if session is None: session = session_factory() yield session except: session.rollback() raise else: session.commit() # logger.debug('session.commit(){}'.format(session)) finally: session.close() # logger.debug('session.close(){}'.format(session)) def add_ad_data(data, app_type): # Use a breakpoint in the code line below to debug your script. print(f'Hi, add_ad_data.app_type = {app_type}, data = {data}') # Press ⌘F8 to toggle the breakpoint. stat_item = data['stat_item'] try: with session_maker() as session: wechat_ad_publisher_adunit_general = WECHAT_AD_PUBLISHER_ADUNIT_GENERAL() wechat_ad_publisher_adunit_general.ad_unit_id = data['ad_unit_id'] wechat_ad_publisher_adunit_general.ad_unit_name = data['ad_unit_name'] wechat_ad_publisher_adunit_general.ad_slot = stat_item['ad_slot'] wechat_ad_publisher_adunit_general.click_count = stat_item['click_count'] wechat_ad_publisher_adunit_general.click_rate = stat_item['click_rate'] wechat_ad_publisher_adunit_general.date = stat_item['date'] wechat_ad_publisher_adunit_general.ecpm = stat_item['ecpm'] wechat_ad_publisher_adunit_general.exposure_count = stat_item['exposure_count'] wechat_ad_publisher_adunit_general.exposure_rate = stat_item['exposure_rate'] wechat_ad_publisher_adunit_general.income = stat_item['income'] wechat_ad_publisher_adunit_general.req_succ_count = stat_item['req_succ_count'] wechat_ad_publisher_adunit_general.slot_str = stat_item['slot_str'] wechat_ad_publisher_adunit_general.date_str = stat_item['date'].replace('-','') wechat_ad_publisher_adunit_general.app_type = app_type session.add(wechat_ad_publisher_adunit_general) print(f'add_ad_data is OK!; app_type = {app_type}') except Exception as e: traceback.print_exc() print(f"add_ad_data error: app_type = {app_type}; traceback.format_exc = {traceback.format_exc()}") def add_ad_adpos_data(stat_item, app_type): # Use a breakpoint in the code line below to debug your script. print(f'Hi, add_ad_adpos_data.app_type = {app_type}, stat_time = {stat_item}') # Press ⌘F8 to toggle the breakpoint. try: with session_maker() as session: wechat_ad_publisher_adpos_general = WECHAT_AD_PUBLISHER_ADPOS_GENERAL() wechat_ad_publisher_adpos_general.ad_slot = stat_item['ad_slot'] wechat_ad_publisher_adpos_general.click_count = stat_item['click_count'] wechat_ad_publisher_adpos_general.click_rate = stat_item['click_rate'] wechat_ad_publisher_adpos_general.date = stat_item['date'] wechat_ad_publisher_adpos_general.ecpm = stat_item['ecpm'] wechat_ad_publisher_adpos_general.exposure_count = stat_item['exposure_count'] wechat_ad_publisher_adpos_general.exposure_rate = stat_item['exposure_rate'] wechat_ad_publisher_adpos_general.income = stat_item['income'] wechat_ad_publisher_adpos_general.req_succ_count = stat_item['req_succ_count'] wechat_ad_publisher_adpos_general.slot_str = stat_item['slot_str'] wechat_ad_publisher_adpos_general.date_str = stat_item['date'].replace('-','') wechat_ad_publisher_adpos_general.app_type = app_type session.add(wechat_ad_publisher_adpos_general) print(f'add_ad_adpos_data is OK; app_type = {app_type}') except Exception as e: traceback.print_exc() print(f"add_ad_adpos_data error: app_type = {app_type}; traceback.format_exc = {traceback.format_exc()}") def post_inform(url, content_text): url = url data = json.dumps(content_text) data = bytes(data, 'utf8') print(f"post_inform data = {data}") headers = {"Content-Type": 'application/json'} req = urllib.request.Request(url=url, headers=headers, data=data) try: resp = urllib.request.urlopen(req, timeout=10).read() print(f"post_inform resp = {resp.decode('utf-8')}") return resp.decode('utf-8') except Exception as e: print(e) def get_inform(url): url = url headers = {"Content-Type": 'application/json'} print(f"get_inform url = {url}") req = urllib.request.Request(url=url, headers=headers) try: resp = urllib.request.urlopen(req, timeout=10).read() print(f"get_inform resp = {resp.decode('utf-8')}") return resp.decode('utf-8') except Exception as e: print(e) def get_mp_info(app_type): datestr = datetime.datetime.strftime(datetime.datetime.now() - datetime.timedelta(days=+1), '%Y-%m-%d') print(f"get_mp_info: app_type = {app_type} date = {datestr}") time_str = time.strftime("%Y:%m:%d %H") print(f"get_mp_info: app_type= {app_type} time = {time_str}") md5 = hashlib.md5('{}'.format(time_str).encode(encoding='UTF-8')).hexdigest() print(f"get_mp_info: app_type = {app_type} md5 = {md5}") getliveaccesstoken_url = "https://longvideoapi.piaoquantv.com/longvideoapi/weixin/getWxAccessToken/{}".format(app_type) print(f"get_mp_info getliveaccesstoken_url = {getliveaccesstoken_url}") ret = get_inform(getliveaccesstoken_url) data = json.loads(ret).get('data',{}) print(f"get_mp_info app_type = {app_type} getWxAccessToken date = {data}") with session_maker() as session: task = session.query(WECHAT_AD_PUBLISHER_ADUNIT_GENERAL).filter_by(date=datestr,app_type=app_type).first() if task is None: getweanalysisappiddailyvisittrend_url = 'https://api.weixin.qq.com/publisher/stat?action=publisher_adunit_general&access_token={}&page=1&page_size=100&start_date={}&end_date={}'.format( data, datestr, datestr) print(f"get_mp_info app_type = {app_type} publisher/stat adunit = {getweanalysisappiddailyvisittrend_url}") ret = get_inform(getweanalysisappiddailyvisittrend_url) print(f"get_mp_info app_type = {app_type} publisher/stat adunit result = {ret}") list = json.loads(ret).get('list',[]) for item in list: add_ad_data(item, app_type) task = session.query(WECHAT_AD_PUBLISHER_ADPOS_GENERAL).filter_by(date=datestr, app_type=app_type).first() if task is None: getweanalysisappiddailyvisittrend_url = 'https://api.weixin.qq.com/publisher/stat?action=publisher_adpos_general&access_token={}&page=1&page_size=100&start_date={}&end_date={}'.format( data, datestr, datestr) print(f"get_mp_info app_type = {app_type} publisher/stat adops = {getweanalysisappiddailyvisittrend_url}") ret = get_inform(getweanalysisappiddailyvisittrend_url) print(f"get_mp_info app_type = {app_type} publisher/stat adops result = {ret}") list = json.loads(ret).get('list',[]) for item in list: add_ad_adpos_data(item, app_type) summary = json.loads(ret)['summary'] summary['ad_slot'] = 'SLOT_ID_WEAPP_ALL' summary['date'] = datestr summary['slot_str'] = 'summary' add_ad_adpos_data(summary, app_type) # Press the green button in the gutter to run the script. if __name__ == '__main__': app_type_list = [0,2,3,4,5,6,17,18,19,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36] # app_type_list = [2,23,24,25] for app_type in app_type_list: print(f"start app_type = {app_type}") try: get_mp_info(app_type) except Exception as e: print(f"app_type {app_type} get data error: {traceback.format_exc()}") print(f"end app_type = {app_type}") print("") # See PyCharm help at https://www.jetbrains.com/help/pycharm/