1）python,现在用过的，后面推出每日练习语法与深度挖掘应用

Posted 2020-11-11 zhitest

python

1, Mac 系统自带的python路径

/System/Library/Frameworks/Python.framework/Version里面存放多个版本

可通过：启动python

import sys

print sys.path 查看路径

/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python27.zip

 

 

2, mac 里面可以启动idle新建.py 文件，然后打开cmd用Mac自带的 执行 python 路径/.py文件

windows里面可以用notepad++ 编写，然后 ctrl + f5配置一下，作为run

 

3,mac 不支持汉字支持

https://www.jianshu.com/p/8333f08ec87d

 

4，自动发送邮件 http://www.runoob.com/python/python-email.html

#! /usr/bin/env python

#! /usr/bin/env python

# coding=utf-8

import sys

import os

import urllib

import urllib2

import unittest

import json

import time

 

reload(sys)

sys.setdefaultencoding("utf-8")

 

from email.mime.text import MIMEText

from email.mime.multipart import MIMEMultipart

import smtplib

 

def sendmail():

msg = MIMEMultipart()

#att = MIMEText(open(r‘C:\Users\alice02.yang\Desktop\brandid.txt‘, ‘rb‘).read(), ‘base64‘, ‘gb2312‘)

#att["Content-Type"] = ‘application/octet-stream‘

#att["Content-Disposition"] = ‘attachment; filename="brandid.txt"‘

#msg.attach(att)

 

strTo = (‘[email protected]‘,‘[email protected]‘)

msg[‘to‘] = ‘;‘.join(strTo)

msg[‘from‘] = ‘[email protected]‘

msg[‘subject‘] = ‘brandid‘

 

try:

server = smtplib.SMTP()

server.connect(‘smtp.163.com‘)

server.login(‘[email protected]‘,‘yzh1990a1a1‘)

server.sendmail(msg[‘from‘], strTo,msg.as_string())

server.quit()

print ‘finish‘

except Exception, e:

print str(e)

 

if __name__ == ‘__main__‘:

sendmail()

 

 

linux --> cmd 使用crontab -e

然后 编写 下午六点20发送 20 18 * * * cd /Users/vip/pythonTestScript; python testAutoSendMail.py 退出

 

 

对被反垃圾邮件网关隔离邮件放行的操作步骤：

http://wiki.corp.vipshop.com/pages/viewpage.action?pageId=35029549&src=search

 

 

开始→程序→附件→系统工具→计划任务→打开添加任务计划。进入任务计划向导。

 

利用“任务计划”，可以将任何脚本、程序或文档安排在某个最方便的时间运行。“任务计划”在每次启动 Windows XP 的时候启动并在后台运行。

 

5,连接数据库操作

linux 要安装MySQLdb插件

 

import MySQLdb

 

def conncur():

conn=MySQLdb.connect(host=‘10.198.184.184‘,user=‘root‘,passwd=‘admin‘,port=3306,charset=‘GBK‘)

cur=conn.cursor()

return conn,cur

 

def conSql():

try:

conn,cur=conncur()

cur.execute(‘drop database if exists vip_test_pythonMysql‘)

cur.execute(‘create database if not exists vip_test_pythonMysql‘)

conn.select_db(‘vip_test_pythonMysql‘)

#sqlfile=open(r‘D:\vipd.sql‘)

#sql=sqlfile.readlines()

 

#sqlfile.close()

 

#for line in sql:

#cur.execute(line)

#conn.commit()

cur.close()

conn.close()

except MySQLdb.Error,e:

print "Mysql Error %d: %s" % (e.args[0], e.args[1])

 

if __name__ == ‘__main__‘:

conSql()

 

6，多线程

# coding:gbk

import time

import urllib

import urllib2

import threading

from Queue import Queue

from time import sleep

 

 

 

 

#THREAD_NUM = 10

#ONE_WORKER_NUM = 500

#LOOP_SLEEP = 0.01

 

 

THREAD_NUM = 100

ONE_WORKER_NUM = 10

LOOP_SLEEP = 0.5

 

 

PERF_TEST_URL=‘‘

 

ERROR_NUM = 0

 

def get(url,para):

url_value = urllib.urlencode(para)

fullurl=url+‘?‘+url_value

print fullurl

data=urllib2.urlopen(fullurl,timeout=20)

json=data.read()

return json

 

 

 

 

def doWork(index):

t = threading.currentThread()

#print "["+t.name+" "+str(index)+"] "+PERF_TEST_URL

 

try:

url=‘http://10.199.144.188/vips-mobile-operation-center-api/cms/content/list/get_rule_resp‘

para= build_para()

json = get(url,para)

print json

except urllib2.URLError, e:

print "["+t.name+" "+str(index)+"] "

print e

global ERROR_NUM

ERROR_NUM += 1

 

 

def working():

t = threading.currentThread()

print "["+t.name+"] Sub Thread Begin"

 

i = 0

while i < ONE_WORKER_NUM:

i += 1

doWork(i)

sleep(LOOP_SLEEP)

 

print "["+t.name+"] Sub Thread End"

 

def build_para():

para={}

para[‘mid‘]=‘123456789‘

para[‘function‘]=‘testfunction‘

para[‘appname‘]=‘testapp‘

para[‘warehouse‘]=‘VIP_SH‘

para[‘tag‘]=‘C10‘

para[‘client‘]=‘ipad‘

para[‘areaid‘]=‘103101‘

para[‘mchannel‘]=‘10‘

return para

 

 

def main():

t1 = time.time()

Threads = []

 

for i in range(THREAD_NUM):

t = threading.Thread(target=working, name="T"+str(i))

t.setDaemon(True)

Threads.append(t)

for t in Threads:

t.start()

for t in Threads:

t.join()

 

print "main thread end"

t2 = time.time()

print "========================================"

print "URL:", PERF_TEST_URL

print "task num:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM*ONE_WORKER_NUM

print "total time cost:", t2-t1

print "per req time cost( sec):", (t2-t1) / (THREAD_NUM*ONE_WORKER_NUM)

print "req per sec:", 1 / ((t2-t1) / (THREAD_NUM*ONE_WORKER_NUM))

print "error num:", ERROR_NUM

 

 

if __name__ == "__main__":

main()

 

7，单元测试操作

https://blog.csdn.net/huilan_same/article/details/52944782

 

写好TestCase，然后由TestLoader加载TestCase到TestSuite，然后由TextTestRunner来运行TestSuite，运行的结果保存在TextTestResult中，我们通过命令行或者unittest.main()执行时，main会调用TextTestRunner中的run来执行，或者我们可以直接通过TextTestRunner来执行用例。

import unittest

class TestCdi(unittest.TestCase):

def setUp(self):

conSql()

 

#------------------------across hit one-----------------------------

#------------------------hit Challenger-----------------------------

def testbroute_2(self):

 

if __name__ == ‘__main__‘:

unittest.main()

 

每个测试方法均以  开头，否则是不被unittest识别的

 

8，post/get请求操作

 

#! /usr/bin/env python

# coding=utf-8

import urllib2

import urllib

import httplib

import json

url=‘http://10.199.144.226/vips-infra-cdi-rule-engine-service/cdi_rule_engine_service/get_cdi_resp/‘

para={}

para={}

var={}

f1=open(r"C:\Users\alice02.yang\Desktop\my.txt",‘w‘)

para["function"]="Recommend"

para["appName"]="ADS"

#var["mid"]="123456789"

var["warehouse"]="VIP_NH"

#var["tag"]="B"

var["mid"]="11111111111"

#var["client"]="iphone"

#var["areaid"]="103107"

#var["mchannel"]="1"

var["user_id"]="12345"

#var["version"]="2.11"

para["variables"]=var

url_v = urllib.urlencode(para)

url_value=url_v.replace("%27",‘%22‘)

fullurl=url+‘?‘+url_value

headers = {"Content-Type":"application/x-www-form-urlencoded","Connection":"Keep-Alive","Referer":"http://10.199.144.226/vips-infra-cdi-rule-engine-service/cdi_rule_engine_service/get_cdi_resp/"}

print fullurl

conn = httplib.HTTPConnection("10.199.144.226");

conn.request("POST","/vips-infra-cdi-rule-engine-service/cdi_rule_engine_service/get_cdi_resp/",url_value,headers);

response = conn.getresponse()

 

print response.read()

f1.close()

conn.close()

 

 

import hashlib

import json

import unittest

import urllib2

 

def getmd5code(src):

#print src

m1 = hashlib.md5()

m1.update(src)

return m1.hexdigest()

 

 

 

 

9，连接redis集群操作

 

#! /usr/bin/env python

# coding=utf-8

 

import sys

import os

import json

import getopt

import urllib2

from urllib import urlopen

from rediscluster import StrictRedisCluster

 

 

reload(sys)

sys.setdefaultencoding( "utf-8" )

 

# URL = "http://rank-admin.vip.vip.com/rank/category/findCategoryListInfoByTreeId"

# DATAURL = "http://rank-admin.vip.vip.com/rank/redis/findBigDataFormRedisCluster"

# ticket = "ST-328962-SqdDyIf25geXGIlfdxhj-cas.oa.vipshop.com"

# DELETE_DATA_URL = "".join([URL, "?ticket=", ticket, "&treeId=160"])

# INFO_DATA_URL = "".join([DATAURL, "?ticket=", ticket, "&key="])

 

URL = "http://rank-admin.vip.vip.com/rank/category/findCategoryListInfoByTreeId"

DATAURL = "http://rank-admin.vip.vip.com/rank/redis/findBigDataFormRedisCluster"

opener = urllib2.build_opener()

opener.addheaders.append((‘Cookie‘,‘_const_cas_assertion_=ST-454039-HGKO3KBGkZiVpAgLCb2x-cas.oa.vipshop.com‘))

DELETE_DATA_URL = "".join([URL, "?treeId=117"])

INFO_DATA_URL = "".join([DATAURL, "?key="])

 

f1=open(r‘C:\Users\alice02.yang\Desktop\ss\117mappingId.txt‘,‘w‘)

 

def redis_cluster(key,value):

print key

print value

redis_nodes = [{‘host‘:‘10.198.184.184‘,‘port‘:6379},

{‘host‘:‘10.198.184.184‘,‘port‘:6380},

{‘host‘:‘10.198.184.185‘,‘port‘:6379},

{‘host‘:‘10.198.184.185‘,‘port‘:6380},

{‘host‘:‘10.198.184.186‘,‘port‘:6379},

{‘host‘:‘10.198.184.186‘,‘port‘:6380}

]

try:

redisconn = StrictRedisCluster(startup_nodes=redis_nodes)

#rc = StrictRedisCluster(startup_nodes=startup_nodes, decode_responses=True)

except Exception,e:

print "Connect Error!"

sys.exit(1)

try:

redisconn.hmset(key,value)

except:

pass

print "sssss"

print redisconn.hgetall(key)

 

 

def rank():

catIdList=[]

#print DELETE_DATA_URL

#content = urlopen(DELETE_DATA_URL).read()

content = opener.open(DELETE_DATA_URL).read()

print content

content = json.loads(content)

datas = content["data"];

#print datas

for data in datas:

for i in data[‘nodes‘]:

try:

for j in i[‘nodes‘][0][‘mappingCatIds‘]:

catIdList.append(str(j))

print i[‘nodes‘][0][‘mappingCatIds‘]

except:

pass

return catIdList

 

 

 

def getBigDataInfo(cat):

adict = {}

for i in cat:

warehouse=["VIP_SH", "VIP_CD", "VIP_HZ", "VIP_BJ", "VIP_NH"]

#warehouse=["VIP_ZZ", "VIP_ZZKG", "VIP_GZNS", "VIP_JC", "HT_GZZY", "HT_GZFLXY", "VIP_NBJCBS", "HT_NBYC"]

for wareh in warehouse:

key = str(i) + ":" + str(wareh) + ":3105"

print key

#print INFO_DATA_URL + key

bigDataInfo = urlopen(INFO_DATA_URL + key).read()

bigDataInfo = json.loads(bigDataInfo)

bigdatas = bigDataInfo["data"];

try:

print bigdatas["value"]

adict[key]=bigdatas["value"]

except:

pass

return adict

#for k in bigdatas["value"]:

#print k

#redis_cluster(key,k,bigdatas["value"][k])

 

 

def main():

cat = rank()

#cat=[7494,1018]

print cat

print >>f1,"\n".join(cat)

#adict = getBigDataInfo(cat)

#for key,value in adict.items():

#print key

#print value

#redis_cluster(key,value)

 

if __name__ == "__main__":

main()

 

10，excel表格操作

 

import xlrd

import json

data = xlrd.open_workbook(r‘C:\Users\alice02.yang\Desktop\tt.xlsx‘)

table = data.sheet_by_name(u‘Sheet1‘)

a=table.col_values(6)

b=table.col_values(7)

dics={}

result=[]

for i in range(len(a)):

dics={}

print a[i]

if len(a[i])==0:

break

dics[‘name‘]=a[i]

dics[‘value‘]=b[i]

print dics

result.append(dics)

print result

dict_res=json.dumps(result,ensure_ascii=False)

print dict_res.decode(‘utf-8‘).encode(‘gbk‘)

 

11，基本操作

http://www.runoob.com/python/python-json.html

 

异常

:=(,"w" fhwrite"这是一个测试文件，用于测试异常!!" :print"Error: 没有找到文件或读取文件失败"else fhclose

 

import os

import time

import datetime

overdueName=list()

oneDayAgo = (datetime.datetime.now() - datetime.timedelta(days = 1))

oneDayAgoTimeStamp = int(time.mktime(oneDayAgo.timetuple()))

def del_files(path):

for root , dirs, files in os.walk(path):

 

for name in files:

if name.endswith(".xls"):

timeStr = name.split(".xls")[0].split("cdi_rule_")[1];

timeStrToInt = int(time.mktime(time.strptime(timeStr,‘%Y-%m-%d‘)))

if timeStrToInt <= oneDayAgoTimeStamp:

overdueName.append(name)

if len(overdueName)>=10:

for temp in overdueName:

os.remove(os.path.join(path, temp))

print ("Delete File: " + os.path.join(path, temp))

 

if __name__ == "__main__":

#path = ‘/tmp‘

path = ‘E:‘

del_files(path)