python调用某段代码,python运行部分代码

　　大家好，本篇文章主要讲的是大蟒中经常使用的代码片段，感兴趣的同学赶快来看一看吧，对你有帮助的话记得收藏一下

目录

日期生成

获取过去 N 天的日期

　　导入日期时间

　　定义获取日期列表(n):

　　前n天=[]

　　# [:-1]控制日期排序

　　对于范围(1，n ^ 1)[:-1]:中的我

　　n天之前。append(字符串(日期时间。约会。今天()-日期时间。时间增量(天数=I)))

　　n天前返回

　　a=get_nday_list(30)

　　打印（一份)

　　输出：

　　[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, 2022-01-22, 2022-01-23, 2022-01-24]

生成一段时间区间内的日期

　　导入日期时间

　　def create _ assist _ date(datestart=None，datend=None):

　　# 创建日期辅助表

　　如果日期开始不是：

　　datestart=2016-01-01

　　如果日期不是：

　　datend=日期时间。日期时间。现在().strftime(%Y-%m-%d )

　　# 转为日期格式

　　datestart=日期时间。日期时间。strptime(日期开始， %Y-%m-%d )

　　datend=日期时间。日期时间。strptime(datend， %Y-%m-%d )

　　date_list=[]

　　日期_列表。追加(日期开始。strftime( % Y-% m-% d ))

　　while datestartdateend:

　　# 日期叠加一天

　　datestart=datetime.timedelta(天数=1)

　　# 日期转字符串存入列表

　　日期_列表。追加(日期开始。strftime( % Y-% m-% d ))

　　 return date_list

　　d_list = create_assist_date(datestart=2021-12-27, dateend=2021-12-30)

　　print(d_list)

　　输出：

[2021-12-27, 2021-12-28, 2021-12-29, 2021-12-30]

保存数据到CSV

　　保存数据到 CSV 算是比较常见的操作了，下面代码如果运行正确会生成"2022_data_2022-01-25.csv"文件。

import os
　　def save_data(data, date):
　　 """
　　 :param data:
　　 :param date:
　　 :return:
　　 """
　　 if not os.path.exists(r2022_data_%s.csv % date):
　　 with open("2022_data_%s.csv" % date, "a+", encoding=utf-8) as f:
　　 f.write("标题,热度,时间,url\n")
　　 for i in data:
　　 title = i["title"]
　　 extra = i["extra"]
　　 time = i[time]
　　 url = i["url"]
　　 row = {},{},{},{}.format(title,extra,time,url)
　　 f.write(row)
　　 f.write(\n)
　　 else:
　　 with open("2022_data_%s.csv" % date, "a+", encoding=utf-8) as f:
　　 for i in data:
　　 title = i["title"]
　　 extra = i["extra"]
　　 time = i[time]
　　 url = i["url"]
　　 row = {},{},{},{}.format(title,extra,time,url)
　　 f.write(row)
　　 f.write(\n)
　　data = [{"title": "demo", "extra": "hello", "time": "1998-01-01", "url": "https://www.baidu.com/"}]
　　date = "2022-01-25"
　　save_data(data, date)

requests 库调用

　　据统计，requests 库是 Python 家族里被引用的最多的第三方库，足见其江湖地位之高大！

　　发送 GET 请求

import requests
　　headers = {
　　 user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36,
　　 cookie: some_cookie
　　}
　　response = requests.request("GET", url, headers=headers)

　　发送 POST 请求

import requests
　　payload={}
　　files=[]
　　headers = {
　　 user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36,
　　 cookie: some_cookie
　　}
　　response = requests.request("POST", url, headers=headers, data=payload, files=files)

Python 操作各种数据库

操作 Redis

　　连接 Redis

import redis
　　def redis_conn_pool():
　　 pool = redis.ConnectionPool(host=localhost, port=6379, decode_responses=True)
　　 rd = redis.Redis(connection_pool=pool)
　　 return rd

　　写入 Redis

from redis_conn import redis_conn_pool
　　rd = redis_conn_pool()
　　rd.set(test_data, mytest)

操作 MongoDB

　　连接 MongoDB

from pymongo import MongoClient
　　conn = MongoClient("mongodb://%s:%s@ipaddress:49974/mydb" % (username, password))
　　db = conn.mydb
　　mongo_collection = db.mydata

　　批量插入数据

res = requests.get(url, params=query).json()
　　commentList = res[data][commentList]
　　mongo_collection.insert_many(commentList)

操作 MySQL

　　连接 MySQL

　　import MySQLdb
　　# 打开数据库连接
　　db = MySQLdb.connect("localhost", "testuser", "test123", "TESTDB", charset=utf8 )
　　# 使用cursor()方法获取操作游标 
　　cursor = db.cursor()

　　执行 SQL 语句

# 使用 execute 方法执行 SQL 语句
　　cursor.execute("SELECT VERSION()")
　　# 使用 fetchone() 方法获取一条数据
　　data = cursor.fetchone()
　　print "Database version : %s " % data
　　# 关闭数据库连接
　　db.close()

本地文件整理

　　整理文件涉及需求的比较多，这里分享的是将本地多个 CSV 文件整合成一个文件

import pandas as pd
　　import os
　　df_list = []
　　for i in os.listdir():
　　 if "csv" in i:
　　 day = i.split(.)[0].split(_)[-1]
　　 df = pd.read_csv(i)
　　 df[day] = day
　　 df_list.append(df)
　　df = pd.concat(df_list, axis=0)
　　df.to_csv("total.txt", index=0)

多线程代码

　　多线程也有很多实现方式，我们选择自己最为熟悉顺手的方式即可

import threading
　　import time
　　exitFlag = 0
　　class myThread (threading.Thread):
　　 def __init__(self, threadID, name, delay):
　　 threading.Thread.__init__(self)
　　 self.threadID = threadID
　　 self.name = name
　　 self.delay = delay
　　 def run(self):
　　 print ("开始线程：" + self.name)
　　 print_time(self.name, self.delay, 5)
　　 print ("退出线程：" + self.name)
　　def print_time(threadName, delay, counter):
　　 while counter:
　　 if exitFlag:
　　 threadName.exit()
　　 time.sleep(delay)
　　 print ("%s: %s" % (threadName, time.ctime(time.time())))
　　 counter -= 1
　　# 创建新线程
　　thread1 = myThread(1, "Thread-1", 1)
　　thread2 = myThread(2, "Thread-2", 2)
　　# 开启新线程
　　thread1.start()
　　thread2.start()
　　thread1.join()
　　thread2.join()
　　print ("退出主线程")

异步编程代码

　　异步爬取网站代码示例：

import asyncio
　　import aiohttp
　　import aiofiles
　　async def get_html(session, url):
　　 try:
　　 async with session.get(url=url, timeout=8) as resp:
　　 if not resp.status // 100 == 2:
　　 print(resp.status)
　　 print("爬取", url, "出现错误")
　　 else:
　　 resp.encoding = utf-8
　　 text = await resp.text()
　　 return text
　　 except Exception as e:
　　 print("出现错误", e)
　　 await get_html(session, url)

　　使用异步请求之后，对应的文件保存也需要使用异步，即是一处异步，处处异步

async def download(title_list, content_list):
　　 async with aiofiles.open({}.txt.format(title_list[0]), a,
　　 encoding=utf-8) as f:
　　 await f.write({}.format(str(content_list)))

总结

　　到此这篇关于Python中经常使用的代码片段的文章就介绍到这了,更多相关Python代码片段内容请搜索盛行IT软件开发工作室以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT软件开发工作室！

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。