今天分享几段工作和生活中常用的代码。它们是最基本的功能和操作，大多数出现频率很高。其中很多可以直接使用，也可以简单修改后放入自己的项目中。日期生成很多时候我们需要批量生成日期。有很多方法。这里有两段代码获取过去N天的日期importdatetimedefget_nday_list(n):before_n_days=[]foriinrange(1,n+1)[::-1]:before_n_days.append(str(datetime.date.today()-datetime.timedelta(days=i)))returnbefore_n_daysa=get_nday_list(30)print(a)输出：['2021-12-23','2021-12-24','2021-12-25','2021-12-26','2021-12-27','2021-12-28','2021-12-29','2021-12-30','2021-12-31','2022-01-01','2022-01-02','2022-01-03','2022-01-04','2022-01-05','2022-01-06','2022-01-07','2022-01-08','2022-01-09','2022-01-10','2022-01-11','2022-01-12','2022-01-13','2022-01-14','2022-01-15','2022-01-16','2022-01-17','2022-01-18','2022-01-19','2022-01-20','2022-01-21']生成一个时间段内的日期importdatetimedefcreate_assist_date(datestart=None,dateend=None):#创建日期辅助表ifdatestart无：datestart='2016-01-01'如果dateend为无：dateend=datetime.datetime.now().strftime('%Y-%m-%d')#转换为日期格式datestart=datetime.datetime.strptime(datestart,'%Y-%m-%d')dateend=datetime.datetime.strptime(dateend,'%Y-%m-%d')date_list=[]date_list.append(datestart.strftime('%Y-%m-%d'))whiledatestartPie:background_color_js=("newecharts.graphic.LinearGradient(0,0,0,1,""[{offset:0,color:'#C86589'},{offSet:1,color:'#06a7ff'}],false)")c=(pie(init_opts=opts.initopts(bg_color=jscode(background_color_js))).dius=["30%","75%"],center=["45%","50%"],rosetype="radius",label_opts=opts.LabelOpts(formatter="{b}:{c}"),)).set_global_opts(title_opts=opts.TitleOpts(title=""),))returncrequests库调用据统计，requests库是Python家族中被引用次数最多的第三方库，可见其强大地位！发送GET请求importrequestheaders={'user-agent':'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/96.0.4664.110Safari/537.36','cookie':'some_cookie'}response=requests.request("GET",url,headers=headers)发送POST请求importrequestspayload={}files=[]headers={'user-agent':'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/96.0.4664.110Safari/537.36','cookie':'some_cookie'}response=requests.request("POST",url,headers=headers,data=payload,files=files)根据一定的条件循环请求，比如根据生成的日期defget_data(mydate):date_list=create_assist_date(mydate)url=“https://test.test”files=[]headers={'user-agent':'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/96.0.4664.110Safari/537.36','cookie':''}fordindate_list:payload={'p':'10','day':d,'nodeid':'1','t':'itemsbydate','c':'node'}foriinrange(1,100):payload['p']=str(i)print("获取页面%s中%s的数据"%(d,str(i)))响应=requests.request("POST",url,headers=headers,data=payload,files=files)items=response.json()['data']['items']ifitems:保存(md）否则：breakpython操作各种数据库操作redisredisimportredisdefredisdefredis_conn_pool（）：pool=redis.connectionpool（host='localhost'，port=6379，decode_respons=truetrue）写入Redisfromredis_connimportredis_conn_poolrd=redis_conn_pool()rd.set('test_data','mytest')操作MongoDB连接MongoDBfrompymongoimportMongoClientconn=MongoClient("mongodb://%s:%s@ipaddress:49974/mydb"%('username','password'))db=conn.mydbmongo_collection=db.mydata批量插入数据res=requests.get(url,params=query).json()commentList=res['data']['commentList']mongo_collection.insert_many(commentList)操作MySQL连接MySQLimportMySQLdb#开启数据库连接db=MySQLdb.connect("localhost","testuser","test123","TESTDB",charset='utf8')#使用cursor()方法获取操作游标cursor=db.cursor()执行SQL语句#使用execute方法执行SQL语句cursor.execute("SELECTVERSION()")#使用fetchone()方法获取一条数据data=cursor.fetchone()print"Databaseversion:%s"%data#关闭数据库连接db.close()Output:Databaseversion:5.0.45本地文件排序和文件排序涉及的要求比较多，我分享的h这里是将多个本地CSV文件整合到一个文件中importpandasaspdimportosdf_list=[]foriinos.listdir():if"csv"ini:day=i.split('.')[0].split('_')[-1]df=pd.read_csv(i)df['day']=daydf_list.append(df)df=pd.concat(df_list,axis=0)df.to_csv("total.txt",index=0)多线程代码和多线程的实现方式有很多种，我们选择最熟悉的方式importthreadingimporttimeexitFlag=0classmyThread(threading.Thread):def__init__(self,threadID,name,delay):threading.Thread.__init__(self)self.threadID=threadIDdeaydelayself.name=nameserun(self):print("启动线程:"+self.name)print_time(self.name,self.delay,5)print("退出线程:"+self.name)defprint_time(threadName,delay,counter):whilecounter:ifexitFlag:threadName.exit()time.sleep(delay)print("%s:%s"%(threadName,time.ctime(time.time())))counter-=1#创建新的Threadthread1=myThread(1,"Thread-1",1)thread2=myThread(2,"Thread-2",2)#开启一个新线程thread1.start()thread2.start()thread1.join()thread2.join()print("exitmainthread")异步编程代码异步爬取网站importasyncioimportaiohttpimportaiofilesasyncdefget_html(session,url):try:asyncwithsession.get(url=url,timeout=8)asresp:ifnotresp.status//100==2:print(resp.status)print("crawl",url,"else"):Resp.encoding='UTF-8'Text=AWAITResp.text()ReturnTextExceptExceptionasE:Print("Error",E)AWAITGET_HTML(session,URL)使用异步请求。还需要使用异步，即处处异步asyncdefdownload(title_list,content_list):asyncwithaiofiles.open('{}.txt'.format(title_list[0]),'a',utf-8')作为f:awaitf.write('{}'.format(str(content_list)))
