标签：str dict key new 自用 ns mydb
# -*- coding: utf-8 -*-
"""
@Author : Klaus
@Time : 2022/9/6 10:38
"""
import requests, os, time, sqlalchemy, pymysql, pyperclip, json
import pandas as pd
from sqlalchemy import create_engine
def get_data_summary(number):
"""
获取数据概览数据,
:return:
"""
url = 'testurl'
headers = {
'authority': 'hz-mydata.alibaba.com',
'accept': '*/*',
'accept-language': 'zh-CN,zh;q=0.9',
'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
'cookie': 'cookie_str',
'origin': 'origin',
'referer': 'referer',
'sec-ch-ua': '"Chromium";v="104", " Not A;Brand";v="99", "Google Chrome";v="104"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36',
}
params = {
"selected": str(number),
"region": "us",
}
response = requests.get(url, headers=headers, params=params)
return response
def mytools():
"""
分析数据概览
在chrome中按F12，在开发者工具中，选择网络，搜索 '',复制链接即可
:return:
"""
selected = 3
# dc_list = []
zhuzhanghao_list = ['forpark', 'gaochengtools']
for number in range(selected, -1, -1):
if number < 0:
break
else:
response = get_data_summary(number)
temp_dict = response.json()
new_dict_one = temp_dict['data']['returnValue'][0]
new_dict = dict()
table = 'DATA_SUMMARY'
new_dict['id'] = 'null'
for key in new_dict_one.keys():
k_dict = new_dict_one[key]
if isinstance(k_dict, dict):
for k_key in k_dict.keys():
# print(key + k_key)
new_dict[key + k_key] = k_dict[k_key]
else:
new_dict[key] = new_dict_one[key]
# print(key)
# print(new_dict)
new_dict['zhuzhanghao'] = zhuzhanghao_list[0]
sq = get_insert_sql(table, new_dict)
# print(sq)
# zeng(mydb, sq)
# dc_list.append(new_dict)
time.sleep(5)
cha_sql1 = f"select statDatevalue,shopUvvalue,shopUvcycleCrc,shopPvvalue,shopPvcycleCrc,searchImplsvalue,searchImplscycleCrc,natureExposureCntvalue,natureClickCntcycleCrc,p4pExposureCntvalue,p4pExposureCntcycleCrc,searchClicksvalue,searchClickscycleCrc,fbUvvalue,fbUvcycleCrc,fbPvvalue,fbPvcycleCrc,tmUvvalue,tmUvcycleCrc,natureClickCntvalue,natureClickCntcycleCrc,p4pClickCntvalue,p4pClickCntcycleCrc,ordAmtvalue,ordAmtcycleCrc,ordCntvalue,ordCntcycleCrc,replyRatevalue,replyRatecycleCrc,fst5minReplyRate30dvalue,fst5minReplyRate30dcycleCrc,avgReplyTimevalue,avgReplyTimecycleCrc,highQualityFbUvRatevalue from data_summary where zhuzhanghao='{zhuzhanghao_list[0]}' order by statDatevalue desc limit 0,14"
cha_sql2 = f"select statDatevalue,shopUvvalue,shopUvcycleCrc,shopPvvalue,shopPvcycleCrc,searchImplsvalue,searchImplscycleCrc,natureExposureCntvalue,natureClickCntcycleCrc,p4pExposureCntvalue,p4pExposureCntcycleCrc,searchClicksvalue,searchClickscycleCrc,fbUvvalue,fbUvcycleCrc,fbPvvalue,fbPvcycleCrc,tmUvvalue,tmUvcycleCrc,natureClickCntvalue,natureClickCntcycleCrc,p4pClickCntvalue,p4pClickCntcycleCrc,ordAmtvalue,ordAmtcycleCrc,ordCntvalue,ordCntcycleCrc,replyRatevalue,replyRatecycleCrc,fst5minReplyRate30dvalue,fst5minReplyRate30dcycleCrc,avgReplyTimevalue,avgReplyTimecycleCrc,highQualityFbUvRatevalue from data_summary where zhuzhanghao='{zhuzhanghao_list[0]}' order by statDatevalue desc limit 7,14"
print(cha_sql1)
df1 = read_from_mydb(cha_sql1, mydb)
df2 = read_from_mydb(cha_sql2, mydb)
print(df1)
xlsxpath1 = r'E:\data_summary1.xlsx'
xlsxpath2 = r'E:\data_summary2.xlsx'
df1.to_excel(xlsxpath1, index=0)
df2.to_excel(xlsxpath2, index=0)
def connect(mydb):
"""
连接数据库
:param mydb: 数据库名称 show databases
:return:
"""
# default,连接串格式为 "数据库类型+数据库驱动://数据库用户名:数据库密码@IP地址:端口/数据库"
databaseType = 'mysql'
databaseDriver = 'pymysql'
hostName = '127.0.0.1'
port = '3306'
username = 'root'
password = '12345678'
database = mydb
engine_url = '{}+{}://{}:{}@{}:{}/{}?charset=utf8'.format( \
databaseType, databaseDriver, username, password, hostName, port, database)
conn = create_engine(engine_url)
# mysql-python
# engine = create_engine('mysql+mysqldb://username:password@127.0.0.1:9527/AiTestOps')
# MySQL-connector-python
# engine = create_engine('mysql+mysqlconnector://username:password@127.0.0.1:9527/AiTestOps')
return conn
def read_from_mydb(sql, mydb):
conn = connect(mydb)
df = pd.read_sql(sql, conn)
return df
def get_insert_sql(table, dic):
"""
生成insert的sql语句
:param table，插入记录的表名
:param dic,插入的数据，字典
"""
sql = 'insert into %s set ' % table
sql += dict_2_str(dic)
return sql
def dict_2_str(anydict):
"""
将字典变成，key='value',key='value' 的形式
"""
tmplist = []
for k, v in anydict.items():
v = str(v).replace("'", ''')
if v == 'null':
ns = "%s=null " % (str(k))
elif v.lower() == 'false' or v is False:
ns = "%s=0" % (str(k))
# ns = " " + str(k) + "=0"
elif v.lower() == 'true' or v is True:
ns = "%s=1" % (str(k))
# ns = " " + str(k) + "=1"
else:
ns = "%s='%s'" % (str(k), str(v))
# ns = " " + str(k) + "='" + str(v) + "' " # 不使用此方式拼接字符串，是为了避免注入攻击
tmplist.append(ns)
return ','.join(tmplist)
def dict_2_str_and(anydict):
"""
将字典变成，key='value' and key='value'的形式
"""
tmplist = []
for k, v in anydict.items():
if isinstance(v, str):
ns = "%s='%s'" % (str(k), str(v))
elif isinstance(v, int):
ns = "%s=%s" % (str(k), v)
# ns = " " + str(k) + "='" + str(v) + "' "
tmplist.append(ns)
return ' and '.join(tmplist)
标签：str,dict,key,new,自用,ns,mydb 来源： https://www.cnblogs.com/lauff/p/16661140.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。