分类 Python 下的文章 - Lan小站-嗯，不错！

2021-09-01

fastapi sqlalchemy 多线程使用由于业务需求，抓取数据后直接由python端入库，然后为了效率考虑多线程肯定是基本操作啦，之前为了提升抓取速度已经用上了异步，将速度由原来的20多s提到1s然后速度太快导致被滑块验证拦截，于是加上了代理IP。今天用我之前fastapi操作MySQL那套去操作数据库的时候，发现多线程并不好用，各种报错经过百度，发现了一篇sqlalchemy 多线程创建session:https://blog.csdn.net/Gragon_Shao/article/details/112786197于是我把我之前用的那套模板改了一下from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, scoped_session from Config import config if config.DBType == 'sqlite': # 使用SQLite数据库 SQLALCHEMY_DATABASE_URL = f"sqlite:///{config.DataBase}" engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine) else: # 使用MySQL数据库 SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{config.UserName}:{config.Password}@{config.Host}:{config.Port}/{config.DataBase}" engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True, pool_size=5, pool_timeout=30, pool_recycle=1) SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine) session = scoped_session(SessionLocal) Base = declarative_base() def get_db(): db = session try: yield db finally: db.remove() 在就是直接去调用session就可以了，使用完之后再removedef test(task, db: scoped_session = session): db.query(ta).filter(ta.id == task.id).delete() db.add(ta(**task.to_dict())) db.commit() db.remove()

2021年09月01日
734 阅读
0 评论
0 点赞

2021-08-17

Python异步请求对大数量请求也太友好了，Python异步的复习刚进入公司，由于对抓取这块比较有经验，然后刚好业务也是有一部分抓取的。于是我的任务就先是这些数据采集。采用异步请求之后的效果：采用同步请求之前的效果：其实这个只是20来条数据，Python也才发送了40多次网络请求，然后差别已经是十多秒的差距了。对于企业级来说，肯定是影响用户体验了，所以我当时考虑的是多线程或协程。然后我就先从协程开始的。由于异步请求这块也就刚开始学习爬虫的时候学了学，后面也就很少去用了，顶多就无脑多线程。一开始以及写好了同步请求的代码，也就两个网络请求，一个get，一个post，post需要传参和请求头（有一丢丢反扒）。大概就是这样的，涉及关键部分的网址都打码了。然后我就打开了万能的搜索引擎：把前面几个大概都看了下，最后还是选择了第一篇（事实证明我眼光不错，也就只踩了一个坑），为啥？百万并发，多牛逼。https://www.cnblogs.com/shenh/p/9090586.html首先它是将同步和异步的效果进行了一个对比：下面通过举例来对比同步代码和异步代码编写方面的差异，其次看下两者性能上的差距，我们使用sleep(1)模拟耗时1秒的io操>作。import time def hello(): time.sleep(1) def run(): for i in range(5): hello() print('Hello World:%s' % time.time()) # 任何伟大的代码都是从Hello World 开始的！ if __name__ == '__main__': run()输出：（间隔约是1s）Hello World:1527595175.4728756 Hello World:1527595176.473001 Hello World:1527595177.473494 Hello World:1527595178.4739306 Hello World:1527595179.474482异步代码import time import asyncio # 定义异步函数 async def hello(): asyncio.sleep(1) print('Hello World:%s' % time.time()) def run(): for i in range(5): loop.run_until_complete(hello()) loop = asyncio.get_event_loop() if __name__ =='__main__': run()输出：Hello World:1527595104.8338501 Hello World:1527595104.8338501 Hello World:1527595104.8338501 Hello World:1527595104.8338501 Hello World:1527595104.8338501async def 用来定义异步函数，其内部有异步操作。每个线程有一个事件循环，主线程调用asyncio.get_event_loop()时会创建事件循环，你需要把异步的任务丢给这个循环的run_until_complete()方法，事件循环会安排协同程序的执行。aiohttp，用于并发请求如果需要并发http请求怎么办呢，通常是用requests，但requests是同步的库，如果想异步的话需要引入aiohttp。这里引入一个类，from aiohttp import ClientSession，首先要建立一个session对象，然后用session对象去打开网页。session可以进行多项操作，比如post, get, put, head等。基本用法：async with ClientSession() as session: async with session.get(url) as response:aiohttp异步实现的例子：import asyncio from aiohttp import ClientSession tasks = [] url = "https://www.baidu.com/{}" async def hello(url): async with ClientSession() as session: async with session.get(url) as response: response = await response.read() print(response) if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(hello(url))首先async def 关键字定义了这是个异步函数，await 关键字加在需要等待的操作前面，response.read()等待request响应，是个耗IO操作。然后使用ClientSession类发起http请求。多链接异步访问如果我们需要请求多个URL该怎么办呢，同步的做法访问多个URL只需要加个for循环就可以了。但异步的实现方式并没那么容易，在之前的基础上需要将hello()包装在asyncio的Future对象中，然后将Future对象列表作为任务传递给事件循环。import time import asyncio from aiohttp import ClientSession tasks = [] url = "https://www.baidu.com/{}" async def hello(url): async with ClientSession() as session: async with session.get(url) as response: response = await response.read() # print(response) print('Hello World:%s' % time.time()) def run(): for i in range(5): task = asyncio.ensure_future(hello(url.format(i))) tasks.append(task) if __name__ == '__main__': loop = asyncio.get_event_loop() run() loop.run_until_complete(asyncio.wait(tasks))输出：Hello World:1527754874.8915546 Hello World:1527754874.899039 Hello World:1527754874.90004 Hello World:1527754874.9095392 Hello World:1527754874.9190395收集http响应好了，上面介绍了访问不同链接的异步实现方式，但是我们只是发出了请求，如果要把响应一一收集到一个列表中，最后保存到本地或者打印出来要怎么实现呢，可通过asyncio.gather(*tasks)将响应全部收集起来，具体通过下面实例来演示。import time import asyncio from aiohttp import ClientSession tasks = [] url = "https://www.baidu.com/{}" async def hello(url): async with ClientSession() as session: async with session.get(url) as response: # print(response) print('Hello World:%s' % time.time()) return await response.read() def run(): for i in range(5): task = asyncio.ensure_future(hello(url.format(i))) tasks.append(task) result = loop.run_until_complete(asyncio.gather(*tasks)) print(result) if __name__ == '__main__': loop = asyncio.get_event_loop() run()输出Hello World:1527765369.0785167 Hello World:1527765369.0845182 Hello World:1527765369.0910277 Hello World:1527765369.0920424 Hello World:1527765369.097017 [b'<!DOCTYPE html>\r\n\r\n<html>\r\n<head>\r\n......异常解决假如你的并发达到2000个，程序会报错：ValueError: too many file descriptors in select()。报错的原因字面上看是 Python 调取的 select 对打开的文件有最大数量的限制，这个其实是操作系统的限制，linux打开文件的最大数默认是1024，windows默认是509，超过了这个值，程序就开始报错。这里我们有三种方法解决这个问题：1.限制并发数量。（一次不要塞那么多任务，或者限制最大并发数量）2.使用回调的方式。3.修改操作系统打开文件数的最大限制，在系统里有个配置文件可以修改默认值，具体步骤不再说明了。不修改系统默认配置的话，个人推荐限制并发数的方法，设置并发数为500，处理速度更快。#coding:utf-8 import time,asyncio,aiohttp url = 'https://www.baidu.com/' async def hello(url,semaphore): async with semaphore: async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.read() async def run(): semaphore = asyncio.Semaphore(500) # 限制并发量为500 to_get = [hello(url.format(),semaphore) for _ in range(1000)] #总共1000任务 await asyncio.wait(to_get) if __name__ == '__main__': # now=lambda :time.time() loop = asyncio.get_event_loop() loop.run_until_complete(run()) loop.close()看完其实已经差不多了，然后发现似乎没有传递请求头，于是看了另外一篇。https://zhuanlan.zhihu.com/p/45159102自定义请求头headers = {'content-type': 'image/gif'} session.post(url, data=data, headers=headers)发现其实和requests差不多异步请求的分块chunk并发控制又在这篇博客发现可以分块https://www.hhtjim.com/aiohttp-asyncio-asynchronous-network-basic-operation-request.html自行chunk操作自己按照所有任务的list列表进行chunk切割，然后分块进行请求，每块中固定chunk数量的任务。基本可以实现想要的并发限制操作async def _bulk_task(num,current_page = 1): """批量创建异步任务 """ task = [] for i in range(num):# 每次10个连接并发进行请求 task.append(asyncio.create_task(get(current_page))) current_page += 1 return await asyncio.gather(*task) # 主要进行chunk操作的函数 def run_task(total,chunk,offset_start_page = 1): """运行分块处理的批量任务 Arguments: total int 总请求数 chunk int 每次并发请求数 offset_start_page int 初始分块开始的页数（偏移页数），正常默认为1 Yields: 返回收集的异步任务运行结果 """ length = math.ceil(total/chunk) for i in range(length): start_page = i * chunk + offset_start_page # 当前分块开始的页数 haldle_num = chunk# 当前需要并发处理的数量 #处理结尾的块 if i == length - 1: # print(':::',chunk,start_page + chunk - offset_start_page) haldle_num = min(chunk,total + offset_start_page - start_page) # print('当前分块下标：{},当前分块需要处理的总数：{},当前分块开始页数:{}'.format(i,haldle_num,start_page)) rel = asyncio.run(_bulk_task(haldle_num,start_page)) yield rel rel = run_task(123,10)# 123总任务每10条并发请求 for i in rel: print(i)综上内容，完成了我对异步的复习，然后将时间效率大大提高。然后就是我说的踩坑的地方，在启动任务的时候，第一篇博客是这样写的loop.run_until_complete(asyncio.gather(*tasks))然而，这是老版本的，好像在某次更新之后就不能这样写了不然会报错RuntimeError: This event loop is already running in python在网上找了一会没有答案，突然想起，第一次学的时候好像也有说过启动问题，于是我改成这样asyncio.gather(*tasks)成功自救

2021年08月17日
640 阅读
0 评论
0 点赞

2021-07-30

万维书刊网所有期刊邮箱地址爬取由于之前要写论文，然后还要投稿，但是有些投稿还需要钱，所以我就爬取了某网站下的免版面费的所有期刊的邮箱地址。然后就小写了一下代码，用以批量爬取，并保存到本地的表格，到时候可以直接批量发送邮件。因为考虑到分类比较多，然后速度比较慢，所以直接上了多线程# -*- coding: utf-8 -*- """ ------------------------------------------------- @ Author ：Lan @ Blog ：www.lanol.cn @ Date ： 2021/7/30 @ Description：I'm in charge of my Code ------------------------------------------------- """ import random import time import requests import parsel import threading def start_down(target, value): html = parsel.Selector(requests.get(f'http://*.com/{target}').text) tou_di_url = html.xpath("//li[@class='bu']/a/@href").extract() with open(f'{value.replace("/", "-")}.csv', 'a+', encoding='gbk') as f: for content_url in tou_di_url: try: content_html = parsel.Selector(requests.get(f'http://*.com/{content_url}').text) title = content_html.xpath( "//div[@class='jjianjie']/div[@class='jjianjietitle']/h1[@class='jname']/text()").extract_first() if 'Email投稿' in title: contact = dict(zip((i.replace(' ', '').replace('\r', '').replace('\n', '') for i in content_html.xpath("//div[@class='sclistclass']//p[2]/text()").extract()), (i.replace(' ', '').replace('\r', '').replace('\n', '') for i in content_html.xpath("//div[@class='sclistclass']//p[3]/text()").extract()))) print(title, contact) f.write(f'{title},{contact}\n') time.sleep(random.randint(1, 4)) f.flush() except: time.sleep(random.randint(1, 4)) if __name__ == '__main__': url = 'http://*.com/NoLayoutFee.aspx?pg=1&hxid=8&typeid=27' type_html = parsel.Selector(requests.get(url).text) types = type_html.xpath("//div[@class='typenamelist']/p/a/text()").extract() urls = type_html.xpath("//div[@class='typenamelist']/p/a/@href").extract() for index, value, in enumerate(types): print(f'正在采集分类{value}') threading.Thread(target=start_down, args=(urls[index], value,)).start(){cloud title="Week9 期刊邮箱.zip" type="lz" url="https://vast.lanzoui.com/iq3b1s0yxsd" password=""/}

2021年07月30日
287 阅读
0 评论
0 点赞

2021-07-24

面向对象—Python基础授课备课第四章面向对象概念类是用来描述具有相同的属性和方法的对象的集合。它定义了该集合中每个对象所共有的属性和方法。对象是类的实例。瞎比比这一章可就没什么好讲的了，在Python里面万物皆对象。Python也和C#一样，定义一个类，然后类就有属性和方法，也可以写构造函数。class Student: def __init__(self, name, age): self.name = name self.age = age def gan_fan(self): print(f'{self.name}开始干饭了')类的继承和重写class Person: def __init__(self, height): self.height = height class Student(Person): def __init__(self, name, age, height): super().__init__(height) self.name = name self.age = age def gan_fan(self): print(f'{self.name}开始干饭了，{self.height}') if __name__ == '__main__': stu = Student('Lan', 18, '1.8') stu.gan_fan()模块化一个项目如果都写在一个文件里面，那会导致很难管理，并且很容易产生重复代码所以呢，我做项目的时候，基本上就是根据功能分文件。比如看我下面这个FastApi项目。在这个plugins的目录里面，放着许多的不同用途的代码，比如，算了不比如了，我的中式英语满级导入模块直接importimport requestsimport之后我还要给它起个名字，方便我调用import requests as req从某个文件导入某一个函数from datetime import datetime从某个文件夹导入某个文件import Tool.answerGet从某个文件夹导入某个文件的某个函数from Tool.Lan超级CC工具 import kill自己写模块直接新建一个py文件 demo1.pydef find_max(a, b): return a if a > b else b在另外一个文件当中 demo2.pyfrom demo1 import find_max find_max(2, 3)

2021年07月24日
247 阅读
0 评论
0 点赞

2021-07-23

函数—Python基础授课备课第三章函数自定义函数C#的方法≈Python的函数定义函数以找最大数的实现为例，分别由C#和Python来写首先是C#public int find_max(int num1, int num2) { if (num1 > num2) { return num1; } else { return num2; } }然后是Pythondef find_max(a, b): """这是一个获取最大值的函数""" if a > b: return a else: return b因此相较于C#，Python定义一个函数只需要以下几点：def关键词方法名圆括号里面放参数，都不需要定义类型冒号，然后换行缩进方法体可有可无的return此图来自菜鸟教程函数传值与返回在C#里面我们知道，参数分实参和形参，反正就是要一个带数据类型的关键词在Python中，参数定义有下面这几种方式(arg1,arg2,...)：直接放参数名接受的(arg1.arg2=value2,...) 给参数设置默认值的(arg1,arg2,*args) 直接懒得定义的，除了前面两个定义的参数外，其他的都存在args这个元祖里面(arg1,arg2,**kwargs) 直接懒得定义的之我不仅懒得定义但我还就要给默认值的，除了已定义的，其他的都存在kwargs字典里面Python里面的返回值就比较随意了，想返回我就来个return，不想返回我就不写，诶我就是玩儿。函数的调用直接函数名然后括号里面放值就可以了呗，轻轻松松。 find_max(1, 2)这样就完成了调用，我们还可以把这个函数当成一个值赋给变量。 max_value = find_max(1, 2)甚至还可以直接输出这个函数，如有返回值就输出返回值，没有就输出？自己试。 print(find_max(1, 2))Python传的是值还是地址在C#里面，如果传的是地址，在方法里面改变变量，外面的也会变，那在Python里面呢？在Python里面都不是，而是“传对象引用”如果接受到的是可变对象，那就可以修改原始值：列表，字典如果接受到的是不可变对象，那就不可以修改原始对象：字符串，元组，数字Python字符串的那些骚操作这里就没什么好说的了，直接这样查看字符串有哪些方法了。Python列表的那些骚操作同上，也没什么好说的了集合的方法常用函数print()输出type()数据类型len()长度

2021年07月23日
221 阅读
0 评论
0 点赞

2021-07-12

软考成绩出来了，你过了吗？软考成绩查询接口分析及Python实现没什么用的废话很久没有写爬虫相关的文章了，因为现在大多数时间都在写Web相关的代码。今天朋友问我软考成绩咋样，我才知道成绩出来了。查了一下，成绩很理想，不出乎我的意料之外。因为：雨过天晴，糟糕的近期，总算迎来了晴天这不是重点，今天来研究一下软考成绩查询网站。官网查成绩地址： https://query.ruankao.org.cn/score/main页面及网络请求分析验证码分析可以看见，页面中有两个查询方式，根据准考证号查询以及根据身份证号查询，以及查询都需要验证码。看到验证码，首先看下他的逻辑有没有问题，就是看能不能跳过二维码。所以我直接点击提交。它发送了一个网络请求VerifyCaptcha，然后返回的preview是验证码错误。到这里还不能确定他的验证码是否可以跳过继续。我们输入正确的验证码后点击查询。可以看见这次的返回值为ok，并且进行了查询，返回了查询结果为空。我们看一下这个result的网络请求。请求方式：Post发送了表单数据：stage: 2021年上半年xm: 11zjhm: 11jym: 8221select_type: 1上述参数直接猜都猜得出吧。年份，姓名，证件号码，验证码，查询方式。然后我们用Python写一个请求试一下。import requests url = 'https://query.ruankao.org.cn//score/result' data = { 'stage': '2021年上半年', 'xm': 'lan', 'zjhm': '666', 'jym': '7777', 'select_type': '1', } print(requests.post(url=url, data=data).json())看来验证码不能跳过，所以有以下几种方法。保存验证码，用户输入后再提交请求。将验证码上传至第三方验证码识别平台，返回验证码后提交请求。这里为了方便采用第一种。过验证码看一下他验证码是怎么请求的。GET请求，获取验证码图片。所以新建一个长链接Session请求先请求一次验证码获取接口，将图片保存在本地，import requests import time session = requests.session() url = 'https://query.ruankao.org.cn//score/result' with open('a.png', 'wb') as f: f.write(session.get(f'https://query.ruankao.org.cn//score/captcha?{time.time() * 1000}').content)然后等用户打开验证码图片识别之后输入验证码再发送查询成绩的网络请求。data = { 'stage': '2021年上半年', 'xm': '姓名', 'zjhm': '证件号码', 'jym': input('请输入验证码'), 'select_type': '1', } print(session.post(url=url, data=data).json())查询成功，然后你就可以再结合结合放到Web啥的

2021年07月12日
530 阅读
0 评论
0 点赞

2021-06-30

2021年06月30日
295 阅读
1 评论
1 点赞