beautifulsoup4
html5lib
image
requests
redis
PyMySQL
pip安装所有依赖包：
pip install \ Image \ requests \ beautifulsoup4 \ html5lib \ redis \ PyMySQL
运行环境需要支持中文
测试运行环境python3.5，不保证其他运行环境能完美运行
需要安装mysql和redis
配置
config.ini
文件，设置好mysql和redis，并且填写你的知乎帐号
向数据库导入
init.sql
Run
开始抓取数据:
python get_user.py
查看抓取数量:
python check_redis.py
效果
原因在于普通的连接是keep-alive的但是却又没有关闭
# http请求的header
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.143 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Host": "www.zhihu.com",
"Referer": "https://www.zhihu.com/",
"Origin": "https://www.zhihu.com/",
"Upgrade-Insecure-Requests": "1",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Pragma": "no-cache",
"Accept-Encoding": "gzip, deflate, br",
&＃39;Connection&＃39;: &＃39;close&＃39;
}
# 验证是否登陆
def check_login(self):
check_url = &＃39;https://www.zhihu.com/settings/profile&＃39;
try:
login_check = self.__session.get(check_url, headers=self.headers, timeout=35)
except Exception as err:
print(traceback.print_exc())
print(err)
print("验证登陆失败，请检查网络")
sys.exit()
print("验证登陆的http status code为：" + str(login_check.status_code))
if int(login_check.status_code) == 200:
return True
else:
return False
进入首页查看http状态码来验证是否登陆，200为已经登陆，一般304就是被重定向所以就是没有登陆
# 获取验证码
def get_captcha(self):
t = str(time.time() * 1000)
captcha_url = &＃39;http://www.zhihu.com/captcha.gif?r=&＃39; + t + "&type=login"
r = self.__session.get(captcha_url, headers=self.headers, timeout=35)
with open(&＃39;captcha.jpg&＃39;, &＃39;wb&＃39;) as f:
f.write(r.content)
f.close()
# 用pillow 的 Image 显示验证码
# 如果没有安装 pillow 到源代码所在的目录去找到验证码然后手动输入
&＃39;&＃39;&＃39;try:
im = Image.open(&＃39;captcha.jpg&＃39;)
im.show()
im.close()
except:&＃39;&＃39;&＃39;
print(u&＃39;请到 %s 目录找到captcha.jpg 手动输入&＃39; % os.path.abspath(&＃39;captcha.jpg&＃39;))
captcha = input("请输入验证码\n>")
return captcha
获取验证码的方法。当登录次数太多有可能会要求输入验证码，这里实现这个功能
# 获取xsrf
def get_xsrf(self):
index_url = &＃39;http://www.zhihu.com&＃39;
# 获取登录时需要用到的_xsrf
try:
index_page = self.__session.get(index_url, headers=self.headers, timeout=35)
except:
print(&＃39;获取知乎页面失败，请检查网络连接&＃39;)
sys.exit()
html = index_page.text
# 这里的_xsrf 返回的是一个list
BS = BeautifulSoup(html, &＃39;html.parser&＃39;)
xsrf_input = BS.find(attrs={&＃39;name&＃39;: &＃39;_xsrf&＃39;})
pattern = r&＃39;value=\"(.*?)\"&＃39;
print(xsrf_input)
self.__xsrf = re.findall(pattern, str(xsrf_input))
return self.__xsrf[0]
获取xsrf，为什么要获取xsrf呢，因为xsrf是一种防止跨站攻击的手段，具体介绍可以看这里csrf
在获取到xsrf之后把xsrf存入COOKIE当中，并且在调用api的时候带上xsrf作为头部，不然的话知乎会返回403
# 进行模拟登陆
def do_login(self):
try:
# 模拟登陆
if self.check_login():
print(&＃39;您已经登录&＃39;)
return
else:
if self.config.get("zhihu_account", "username") and self.config.get("zhihu_account", "password"):
self.username = self.config.get("zhihu_account", "username")
self.password = self.config.get("zhihu_account", "password")
else:
self.username = input(&＃39;请输入你的用户名\n> &＃39;)
self.password = input("请输入你的密码\n> ")
except Exception as err:
print(traceback.print_exc())
print(err)
sys.exit()
if re.match(r"^1\d{10}$", self.username):
print("手机登陆\n")
post_url = &＃39;http://www.zhihu.com/login/phone_num&＃39;
postdata = {
&＃39;_xsrf&＃39;: self.get_xsrf(),
&＃39;password&＃39;: self.password,
&＃39;remember_me&＃39;: &＃39;true&＃39;,
&＃39;phone_num&＃39;: self.username,
}
else:
print("邮箱登陆\n")
post_url = &＃39;http://www.zhihu.com/login/email&＃39;
postdata = {
&＃39;_xsrf&＃39;: self.get_xsrf(),
&＃39;password&＃39;: self.password,
&＃39;remember_me&＃39;: &＃39;true&＃39;,
&＃39;email&＃39;: self.username,
}
try:
login_page = self.__session.post(post_url, postdata, headers=self.headers, timeout=35)
login_text = json.loads(login_page.text.encode(&＃39;latin-1&＃39;).decode(&＃39;unicode-escape&＃39;))
print(postdata)
print(login_text)
# 需要输入验证码 r = 0为登陆成功代码
if login_text[&＃39;r&＃39;] == 1:
sys.exit()
except:
postdata[&＃39;captcha&＃39;] = self.get_captcha()
login_page = self.__session.post(post_url, postdata, headers=self.headers, timeout=35)
print(json.loads(login_page.text.encode(&＃39;latin-1&＃39;).decode(&＃39;unicode-escape&＃39;)))
# 保存登陆COOKIE
self.__session.COOKIEs.save()
这个就是核心的登陆功能啦，非常关键的就是用到了requests库，非常方便的保存到session
我们这里全局都是用单例模式，统一使用同一个requests.session对象进行访问功能，保持登录状态的一致性
最后主要调用登陆的代码为
# 创建login对象
lo = login.login.Login(self.session)
# 模拟登陆
if lo.check_login():
print(&＃39;您已经登录&＃39;)
else:
if self.config.get("zhihu_account", "username") and self.config.get("zhihu_account", "username"):
username = self.config.get("zhihu_account", "username")
password = self.config.get("zhihu_account", "password")
else:
username = input(&＃39;请输入你的用户名\n> &＃39;)
password = input("请输入你的密码\n> ")
lo.do_login(username, password)
知乎模拟登陆到此就完成啦
知乎用户抓取
def __init__(self, threadID=1, name=&＃39;&＃39;):
# 多线程
print("线程" + str(threadID) + "初始化")
threading.Thread.__init__(self)
self.threadID = threadID
self.name = name
try:
print("线程" + str(threadID) + "初始化成功")
except Exception as err:
print(err)
print("线程" + str(threadID) + "开启失败")
self.threadLock = threading.Lock()
# 获取配置
self.cOnfig= configparser.ConfigParser()
self.config.read("config.ini")
# 初始化session
requests.adapters.DEFAULT_RETRIES = 5
self.session = requests.Session()
self.session.COOKIEs = COOKIElib.LWPCOOKIEJar(filename=&＃39;COOKIE&＃39;)
self.session.keep_alive = False
try:
self.session.COOKIEs.load(ignore_discard=True)
except:
print(&＃39;COOKIE 未能加载&＃39;)
finally:
pass
# 创建login对象
lo = Login(self.session)
lo.do_login()
# 初始化redis连接
try:
redis_host = self.config.get("redis", "host")
redis_port = self.config.get("redis", "port")
self.redis_con = redis.Redis(host=redis_host, port=redis_port, db=0)
# 刷新redis库
# self.redis_con.flushdb()
except:
print("请安装redis或检查redis连接配置")
sys.exit()
# 初始化数据库连接
try:
db_host = self.config.get("db", "host")
db_port = int(self.config.get("db", "port"))
db_user = self.config.get("db", "user")
db_pass = self.config.get("db", "password")
db_db = self.config.get("db", "db")
db_charset = self.config.get("db", "charset")
self.db = pymysql.connect(host=db_host, port=db_port, user=db_user, passwd=db_pass, db=db_db,
charset=db_charset)
self.db_cursor = self.db.cursor()
except:
print("请检查数据库配置")
sys.exit()
# 初始化系统设置
self.max_queue_len = int(self.config.get("sys", "max_queue_len"))
这个是get_user.py的构造函数，主要功能就是初始化mysql连接、redis连接、验证登陆、生成全局的session对象、导入系统配置、开启多线程。
# 获取首页html
def get_index_page(self):
index_url = &＃39;https://www.zhihu.com/&＃39;
try:
index_html = self.session.get(index_url, headers=self.headers, timeout=35)
except Exception as err:
# 出现异常重试
print("获取页面失败，正在重试......")
print(err)
traceback.print_exc()
return None
finally:
pass
return index_html.text
# 获取单个用户详情页面
def get_user_page(self, name_url):
user_page_url = &＃39;https://www.zhihu.com&＃39; + str(name_url) + &＃39;/about&＃39;
try:
index_html = self.session.get(user_page_url, headers=self.headers, timeout=35)
except Exception as err:
# 出现异常重试
print("失败name_url：" + str(name_url) + "获取页面失败，放弃该用户")
print(err)
traceback.print_exc()
return None
finally:
pass
return index_html.text
# 获取粉丝页面
def get_follower_page(self, name_url):
user_page_url = &＃39;https://www.zhihu.com&＃39; + str(name_url) + &＃39;/followers&＃39;
try:
index_html = self.session.get(user_page_url, headers=self.headers, timeout=35)
except Exception as err:
# 出现异常重试
print("失败name_url：" + str(name_url) + "获取页面失败，放弃该用户")
print(err)
traceback.print_exc()
return None
finally:
pass
return index_html.text
def get_following_page(self, name_url):
user_page_url = &＃39;https://www.zhihu.com&＃39; + str(name_url) + &＃39;/followers&＃39;
try:
index_html = self.session.get(user_page_url, headers=self.headers, timeout=35)
except Exception as err:
# 出现异常重试
print("失败name_url：" + str(name_url) + "获取页面失败，放弃该用户")
print(err)
traceback.print_exc()
return None
finally:
pass
return index_html.text
# 获取首页上的用户列表，存入redis
def get_index_page_user(self):
index_html = self.get_index_page()
if not index_html:
return
BS = BeautifulSoup(index_html, "html.parser")
self.get_xsrf(index_html)
user_a = BS.find_all("a", class_="author-link") # 获取用户的a标签
for a in user_a:
if a:
self.add_wait_user(a.get(&＃39;href&＃39;))
else:
continue
这一部分的代码就是用于抓取各个页面的html代码
# 加入带抓取用户队列，先用redis判断是否已被抓取过
def add_wait_user(self, name_url):
# 判断是否已抓取
self.threadLock.acquire()
if not self.redis_con.hexists(&＃39;already_get_user&＃39;, name_url):
self.counter += 1
print(name_url + " 加入队列")
self.redis_con.hset(&＃39;already_get_user&＃39;, name_url, 1)
self.redis_con.lpush(&＃39;user_queue&＃39;, name_url)
print("添加用户 " + name_url + "到队列")
self.threadLock.release()
# 获取页面出错移出redis
def del_already_user(self, name_url):
self.threadLock.acquire()
if not self.redis_con.hexists(&＃39;already_get_user&＃39;, name_url):
self.counter -= 1
self.redis_con.hdel(&＃39;already_get_user&＃39;, name_url)
self.threadLock.release()
用户加入redis的操作，在数据库插入出错时我们调用del_already_user删除插入出错的用户
# 分析粉丝页面获取用户的所有粉丝用户
# @param follower_page get_follower_page()中获取到的页面，这里获取用户hash_id请求粉丝接口获取粉丝信息
def get_all_follower(self, name_url):
follower_page = self.get_follower_page(name_url)
# 判断是否获取到页面
if not follower_page:
return
BS = BeautifulSoup(follower_page, &＃39;html.parser&＃39;)
# 获取关注者数量
follower_num = int(BS.find(&＃39;span&＃39;, text=&＃39;关注者&＃39;).find_parent().find(&＃39;strong&＃39;).get_text())
# 获取用户的hash_id
hash_id = \
json.loads(BS.select("#zh-profile-follows-list")[0].select(".zh-general-list")[0].get(&＃39;data-init&＃39;))[
&＃39;params&＃39;][
&＃39;hash_id&＃39;]
# 获取关注者列表
self.get_xsrf(follower_page) # 获取xsrf
post_url = &＃39;https://www.zhihu.com/node/ProfileFollowersListV2&＃39;
# 开始获取所有的关注者 math.ceil(follower_num/20)*20
for i in range(0, math.ceil(follower_num / 20) * 20, 20):
post_data = {
&＃39;method&＃39;: &＃39;next&＃39;,
&＃39;params&＃39;: json.dumps({"offset": i, "order_by": "created", "hash_id": hash_id})
}
try:
j = self.session.post(post_url, params=post_data, headers=self.headers, timeout=35).text.encode(
&＃39;latin-1&＃39;).decode(
&＃39;unicode-escape&＃39;)
pattern = re.compile(r"class=\"zm-item-link-avatar\"[^\"]*\"([^\"]*)", re.DOTALL)
j = pattern.findall(j)
for user in j:
user = user.replace(&＃39;\\&＃39;, &＃39;&＃39;)
self.add_wait_user(user) # 保存到redis
except Exception as err:
print("获取正在关注失败")
print(err)
traceback.print_exc()
pass
# 获取正在关注列表
def get_all_following(self, name_url):
following_page = self.get_following_page(name_url)
# 判断是否获取到页面
if not following_page:
return
BS = BeautifulSoup(following_page, &＃39;html.parser&＃39;)
# 获取关注者数量
following_num = int(BS.find(&＃39;span&＃39;, text=&＃39;关注了&＃39;).find_parent().find(&＃39;strong&＃39;).get_text())
# 获取用户的hash_id
hash_id = \
json.loads(BS.select("#zh-profile-follows-list")[0].select(".zh-general-list")[0].get(&＃39;data-init&＃39;))[
&＃39;params&＃39;][
&＃39;hash_id&＃39;]
# 获取关注者列表
self.get_xsrf(following_page) # 获取xsrf
post_url = &＃39;https://www.zhihu.com/node/ProfileFolloweesListV2&＃39;
# 开始获取所有的关注者 math.ceil(follower_num/20)*20
for i in range(0, math.ceil(following_num / 20) * 20, 20):
post_data = {
&＃39;method&＃39;: &＃39;next&＃39;,
&＃39;params&＃39;: json.dumps({"offset": i, "order_by": "created", "hash_id": hash_id})
}
try:
j = self.session.post(post_url, params=post_data, headers=self.headers, timeout=35).text.encode(
&＃39;latin-1&＃39;).decode(
&＃39;unicode-escape&＃39;)
pattern = re.compile(r"class=\"zm-item-link-avatar\"[^\"]*\"([^\"]*)", re.DOTALL)
j = pattern.findall(j)
for user in j:
user = user.replace(&＃39;\\&＃39;, &＃39;&＃39;)
self.add_wait_user(user) # 保存到redis
except Exception as err:
print("获取正在关注失败")
print(err)
traceback.print_exc()
pass
调用知乎的API，获取所有的关注用户列表和粉丝用户列表，递归获取用户
这里需要注意的是头部要记得带上xsrf不然会抛出403
# 分析about页面，获取用户详细资料
def get_user_info(self, name_url):
about_page = self.get_user_page(name_url)
# 判断是否获取到页面
if not about_page:
print("获取用户详情页面失败，跳过，name_url：" + name_url)
return
self.get_xsrf(about_page)
BS = BeautifulSoup(about_page, &＃39;html.parser&＃39;)
# 获取页面的具体数据
try:
nickname = BS.find("a", class_="name").get_text() if BS.find("a", class_="name") else &＃39;&＃39;
user_type = name_url[1:name_url.index(&＃39;/&＃39;, 1)]
self_domain = name_url[name_url.index(&＃39;/&＃39;, 1) + 1:]
gender = 2 if BS.find("i", class_="icon icon-profile-female") else (1 if BS.find("i", class_="icon icon-profile-male") else 3)
follower_num = int(BS.find(&＃39;span&＃39;, text=&＃39;关注者&＃39;).find_parent().find(&＃39;strong&＃39;).get_text())
following_num = int(BS.find(&＃39;span&＃39;, text=&＃39;关注了&＃39;).find_parent().find(&＃39;strong&＃39;).get_text())
agree_num = int(re.findall(r&＃39;(.*).*赞同&＃39;, about_page)[0])
appreciate_num = int(re.findall(r&＃39;(.*).*感谢&＃39;, about_page)[0])
star_num = int(re.findall(r&＃39;(.*).*收藏&＃39;, about_page)[0])
share_num = int(re.findall(r&＃39;(.*).*分享&＃39;, about_page)[0])
browse_num = int(BS.find_all("span", class_="zg-gray-normal")[2].find("strong").get_text())
trade = BS.find("span", class_="business item").get(&＃39;title&＃39;) if BS.find("span",
class_="business item") else &＃39;&＃39;
company = BS.find("span", class_="employment item").get(&＃39;title&＃39;) if BS.find("span",
class_="employment item") else &＃39;&＃39;
school = BS.find("span", class_="education item").get(&＃39;title&＃39;) if BS.find("span",
class_="education item") else &＃39;&＃39;
major = BS.find("span", class_="education-extra item").get(&＃39;title&＃39;) if BS.find("span",
class_="education-extra item") else &＃39;&＃39;
job = BS.find("span", class_="position item").get_text() if BS.find("span",
class_="position item") else &＃39;&＃39;
location = BS.find("span", class_="location item").get(&＃39;title&＃39;) if BS.find("span",
class_="location item") else &＃39;&＃39;
description = BS.find("p", class_="bio ellipsis").get(&＃39;title&＃39;) if BS.find("p",
class_="bio ellipsis") else &＃39;&＃39;
ask_num = int(BS.find_all("a", class_=&＃39;item&＃39;)[1].find("span").get_text()) if \
BS.find_all("a", class_=&＃39;item&＃39;)[
1] else int(0)
answer_num = int(BS.find_all("a", class_=&＃39;item&＃39;)[2].find("span").get_text()) if \
BS.find_all("a", class_=&＃39;item&＃39;)[
2] else int(0)
article_num = int(BS.find_all("a", class_=&＃39;item&＃39;)[3].find("span").get_text()) if \
BS.find_all("a", class_=&＃39;item&＃39;)[3] else int(0)
collect_num = int(BS.find_all("a", class_=&＃39;item&＃39;)[4].find("span").get_text()) if \
BS.find_all("a", class_=&＃39;item&＃39;)[4] else int(0)
public_edit_num = int(BS.find_all("a", class_=&＃39;item&＃39;)[5].find("span").get_text()) if \
BS.find_all("a", class_=&＃39;item&＃39;)[5] else int(0)
replace_data = \
(pymysql.escape_string(name_url), nickname, self_domain, user_type,
gender, follower_num, following_num, agree_num, appreciate_num, star_num, share_num, browse_num,
trade, company, school, major, job, location, pymysql.escape_string(description),
ask_num, answer_num, article_num, collect_num, public_edit_num)
replace_sql = &＃39;&＃39;&＃39;REPLACE INTO
user(url,nickname,self_domain,user_type,
gender, follower,following,agree_num,appreciate_num,star_num,share_num,browse_num,
trade,company,school,major,job,location,description,
ask_num,answer_num,article_num,collect_num,public_edit_num)
VALUES(%s,%s,%s,%s,
%s,%s,%s,%s,%s,%s,%s,%s,
%s,%s,%s,%s,%s,%s,%s,
%s,%s,%s,%s,%s)&＃39;&＃39;&＃39;
try:
print("获取到数据：")
print(replace_data)
self.db_cursor.execute(replace_sql, replace_data)
self.db.commit()
except Exception as err:
print("插入数据库出错")
print("获取到数据：")
print(replace_data)
print("插入语句：" + self.db_cursor._last_executed)
self.db.rollback()
print(err)
traceback.print_exc()
except Exception as err:
print("获取数据出错，跳过用户")
self.redis_con.hdel("already_get_user", name_url)
self.del_already_user(name_url)
print(err)
traceback.print_exc()
pass
最后，到用户的about页面，分析页面元素，利用正则或者beatifulsoup分析抓取页面的数据
这里我们SQL语句用REPLACE INTO而不用INSERT INTO，这样可以很好的防止数据重复问题
# 开始抓取用户，程序总入口
def entrance(self):
while 1:
if int(self.redis_con.llen("user_queue")) <1:
self.get_index_page_user()
else:
# 出队列获取用户name_url redis取出的是byte，要decode成utf-8
name_url = str(self.redis_con.rpop("user_queue").decode(&＃39;utf-8&＃39;))
print("正在处理name_url：" + name_url)
self.get_user_info(name_url)
if int(self.redis_con.llen("user_queue")) <= int(self.max_queue_len):
self.get_all_follower(name_url)
self.get_all_following(name_url)
self.session.COOKIEs.save()
def run(self):
print(self.name + " is running")
self.entrance()
最后，入口
if __name__ == &＃39;__main__&＃39;:
login = GetUser(999, "登陆线程")
threads = []
for i in range(0, 4):
m = GetUser(i, "thread" + str(i))
threads.append(m)
for i in range(0, 4):
threads[i].start()
for i in range(0, 4):
threads[i].join()
这里就是多线程的开启，需要开启多少个线程就把4换成多少就可以了
Docker
嫌麻烦的可以参考一下我用docker简单的搭建一个基础环境：
mysql和redis都是官方镜像
docker run --name mysql -itd mysql:latest docker run --name redis -itd mysql:latest
再利用docker-compose运行python镜像，我的python的docker-compose.yml：
python: container_name: python build: . ports: - "84:80" external_links: - memcache:memcache - mysql:mysql - redis:redis volumes: - /docker_containers/python/www:/var/www/html tty: true stdin_open: true extra_hosts: - "python:192.168.102.140" environment: PYTHONIOENCODING: utf-8
以上就是利用python实现多线程抓取知乎用户方法的详细内容，更多请关注 第一PHP社区 其它相关文章！
![[附源码]计算机毕业设计JAVAjsp医药管理信息系统](https://img3.php1.cn/3cdc5/63cc/3b4/50c0ad62c8b7c2f1.png)
![Kubernetes 1.9.0 Alpha.1 发布公告 [Kubernetes 最新动态]](https://img1.php1.cn/3cd4a/25047/ae9/3397348f4961686e.png)
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有