第如何在Pycharm中制作自己的爬蟲代碼模板目錄寫作背景爬蟲代碼在Pycharm中設(shè)置代碼模板

寫作背景

最近本菜雞有幾個(gè)網(wǎng)站想要爬，每個(gè)爬蟲的代碼不一樣，但有某種聯(lián)系，可以抽出一部分通用的代碼制成模板，減少代碼工作量，于是就有了這篇文章。

如果覺得我這篇文章寫的好的話，能不能給我點(diǎn)個(gè)贊，評(píng)論、收藏一條龍(☆▽☆)。如果要點(diǎn)個(gè)關(guān)注的話也不是不可以。

如果有什么問題，還請(qǐng)各位大佬提出，不勝感激。

爬蟲代碼

我的爬蟲代碼都是使用的自己寫的多線程。

因?yàn)槲业拇a能力很差，所以如果代碼有哪里讓各位大佬倍感不適，請(qǐng)及時(shí)在評(píng)論區(qū)指出，謝謝各位大佬。

我的代碼如下：

#!/usr/bin/python3

#-*-coding=utf-8-*-

#@Author:lhys

#@FileName:proxy_tool.py

importrequests

importthreading

timeout=300

lock=threading.Lock()

#請(qǐng)求頭用自己的

headers={

'':''

classMyProxy:

def__init__(self,proxy_api='',proxy_server='',max_use=5000,try_count=5):

ifnot(proxy_apiorproxy_server):

raiseTypeError('Proxy_apiandproxy_servercannotbeemptyatthesametime.')

xies=Noneifnotproxy_serverelse{

'http':proxy_server,

'https':proxy_server

#代理API

xy_api=proxy_api

#代理IP最大使用次數(shù)

self.max_use=max_use

#測(cè)試代理IP次數(shù)，超過次數(shù)即認(rèn)為代理IP不可用

self.try_count=try_count

#是否爬蟲請(qǐng)求出錯(cuò)，如果出錯(cuò)，直接更換IP

self.flag=0

#代理IP剩余生存時(shí)間

xy_ttl=0

#各種鎖

self.lock=threading.Lock()

self.ttl_lock=threading.Lock()

self.flag_lock=threading.Lock()

defset_flag(self):

self.flag_lock.acquire()

self.flag=1

self.flag_lock.release()

defget_flag(self):

self.flag_lock.acquire()

flag=self.flag

self.flag_lock.release()

returnflag

defdecrease_ttl(self):

self.ttl_lock.acquire()

xy_ttl-=1

self.ttl_lock.release()

defget_ttl(self):

self.ttl_lock.acquire()

ttl=xy_ttl

self.ttl_lock.release()

returnttl

defset_ttl(self):

self.ttl_lock.acquire()

xy_ttl=self.max_use

self.ttl_lock.release()

defget_proxy(self):

self.lock.acquire()

proxy=xies

self.lock.release()

returnproxy

defset_proxy(self):

ifxy_ttl0andself.flag==0:

return

old=xies

ifself.flag==1:

fortry_countinrange(self.try_count):

try:

requests.get('',headers=headers,proxies=old,timeout=timeout)

print(f'Testproxy{old}successfully.')

return

exceptrequests.exceptions.ProxyErrororrequests.exceptions.ConnectionErrororrequests.exceptions.ConnectTimeout:

print(f'Testproxy{old}failed.')

break

exceptExceptionase:

print(e)

ifnotxy_api:

raiseValueError('代理IP不可用，且代理IPAPI未設(shè)置。')

whileTrue:

res=requests.get(xy_api)

#這一部分按照自己的代理IP文檔來，僅供參考

try:

ifres.json()["ERRORCODE"]=="0":

ip,port=res.json()["RESULT"][0]['ip'],res.json()["RESULT"][0]['port']

self.lock.acquire()

xies={

'http':'http://%s:%s'%(ip,port),

'https':'http://%s:%s'%(ip,port)

print(f'Setproxy:{ip}:{port}.')

self.flag=0

self.lock.release()

self.set_ttl()

return

else:

print(f'Setproxyfailed.')

exceptExceptionase:

print(e)

Proxy=MyProxy()

defrequest_by_proxy(url,use_proxy=True):

whileTrue:

try:

#使用代理

ifuse_proxy:

proxy_ttl=Proxy.get_ttl()

print(proxy_ttl)

#如果超過最大使用次數(shù)或者請(qǐng)求出現(xiàn)錯(cuò)誤，重新設(shè)置IP

ifproxy_ttl=0orProxy.get_flag():

Proxy.set_proxy()

print(Proxy.get_ttl())

proxy=Proxy.get_proxy()

lock.acquire()

res=requests.get(url,headers=headers,proxies=proxy,timeout=timeout)

lock.release()

Proxy.decrease_ttl()

returnres

else:

res=requests.get(url,headers=headers,timeout=timeout)

returnres

exceptrequests.exceptions.ProxyErroraspe:

ifuse_proxy:

lock.release()

print(f'Proxy{Pxies}isnotavailable,reason:{pe}.')

Proxy.set_flag()

exceptrequests.ex