python實(shí)現(xiàn)多線程抓取知乎用戶

更新時(shí)間：2016年12月12日 08:51:23 投稿：hebedich

本文給大家分享的是如何使用Python實(shí)現(xiàn)多線程抓取知乎用戶的思路以及代碼，非常的詳細(xì)，作者使用本程序成功抓取到了百萬用戶，有需要的小伙伴可以參考下

需要用到的包：

beautifulsoup4
html5lib
image
requests
redis
PyMySQL

pip安裝所有依賴包：

pip install \
Image \
requests \
beautifulsoup4 \
html5lib \
redis \
PyMySQL

運(yùn)行環(huán)境需要支持中文

測(cè)試運(yùn)行環(huán)境python3.5，不保證其他運(yùn)行環(huán)境能完美運(yùn)行

需要安裝mysql和redis

配置 config.ini 文件，設(shè)置好mysql和redis，并且填寫你的知乎帳號(hào)

向數(shù)據(jù)庫導(dǎo)入 init.sql

Run

開始抓取數(shù)據(jù): python get_user.py
查看抓取數(shù)量: python check_redis.py

效果

總體思路

1.首先是模擬登陸知乎，利用保存登陸的cookie信息
2.抓取知乎頁面的html代碼，留待下一步繼續(xù)進(jìn)行分析提取信息
3.分析提取頁面中用戶的個(gè)性化url，放入redis（這里特別說明一下redis的思路用法，將提取到的用戶的個(gè)性化url放入redis的一個(gè)名為already_get_user的hash table，表示已抓取的用戶，對(duì)于已抓取過的用戶判斷是否存在于already_get_user以去除重復(fù)抓取，同時(shí)將個(gè)性化url放入user_queue的隊(duì)列中，需要抓取新用戶時(shí)pop隊(duì)列獲取新的用戶）
4.獲取用戶的關(guān)注列表和粉絲列表，繼續(xù)插入到redis
5.從redis的user_queue隊(duì)列中獲取新用戶繼續(xù)重復(fù)步驟3

模擬登陸知乎

首先是登陸，登陸功能作為一個(gè)包封裝了在login里面，方便整合調(diào)用

header部分，這里Connection最好設(shè)為close，不然可能會(huì)碰到max retireve exceed的錯(cuò)誤
原因在于普通的連接是keep-alive的但是卻又沒有關(guān)閉

# http請(qǐng)求的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",
  'Connection': 'close'
}

# 驗(yàn)證是否登陸
def check_login(self):
  check_url = 'https://www.zhihu.com/settings/profile'
  try:
    login_check = self.__session.get(check_url, headers=self.headers, timeout=35)
  except Exception as err:
    print(traceback.print_exc())
    print(err)
    print("驗(yàn)證登陸失敗，請(qǐng)檢查網(wǎng)絡(luò)")
    sys.exit()
  print("驗(yàn)證登陸的http status code為：" + str(login_check.status_code))
  if int(login_check.status_code) == 200:
    return True
  else:
    return False

進(jìn)入首頁查看http狀態(tài)碼來驗(yàn)證是否登陸，200為已經(jīng)登陸，一般304就是被重定向所以就是沒有登陸

# 獲取驗(yàn)證碼
def get_captcha(self):
  t = str(time.time() * 1000)
  captcha_url = 'http://www.zhihu.com/captcha.gif?r=' + t + "&type=login"
  r = self.__session.get(captcha_url, headers=self.headers, timeout=35)
  with open('captcha.jpg', 'wb') as f:
    f.write(r.content)
    f.close()
    # 用pillow 的 Image 顯示驗(yàn)證碼
    # 如果沒有安裝 pillow 到源代碼所在的目錄去找到驗(yàn)證碼然后手動(dòng)輸入
  '''try:
    im = Image.open('captcha.jpg')
    im.show()
    im.close()
  except:'''
  print(u'請(qǐng)到 %s 目錄找到captcha.jpg 手動(dòng)輸入' % os.path.abspath('captcha.jpg'))
  captcha = input("請(qǐng)輸入驗(yàn)證碼\n>")
  return captcha

獲取驗(yàn)證碼的方法。當(dāng)?shù)卿洿螖?shù)太多有可能會(huì)要求輸入驗(yàn)證碼，這里實(shí)現(xiàn)這個(gè)功能

# 獲取xsrf
def get_xsrf(self):
  index_url = 'http://www.zhihu.com'
  # 獲取登錄時(shí)需要用到的_xsrf
  try:
    index_page = self.__session.get(index_url, headers=self.headers, timeout=35)
  except:
    print('獲取知乎頁面失敗，請(qǐng)檢查網(wǎng)絡(luò)連接')
    sys.exit()
  html = index_page.text
  # 這里的_xsrf 返回的是一個(gè)list
  BS = BeautifulSoup(html, 'html.parser')
  xsrf_input = BS.find(attrs={'name': '_xsrf'})
  pattern = r'value=\"(.*?)\"'
  print(xsrf_input)
  self.__xsrf = re.findall(pattern, str(xsrf_input))
  return self.__xsrf[0]

獲取xsrf，為什么要獲取xsrf呢，因?yàn)閤srf是一種防止跨站攻擊的手段，具體介紹可以看這里csrf
在獲取到xsrf之后把xsrf存入cookie當(dāng)中，并且在調(diào)用api的時(shí)候帶上xsrf作為頭部，不然的話知乎會(huì)返回403

# 進(jìn)行模擬登陸
def do_login(self):
  try:
    # 模擬登陸
    if self.check_login():
      print('您已經(jīng)登錄')
      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('請(qǐng)輸入你的用戶名\n> ')
        self.password = input("請(qǐng)輸入你的密碼\n> ")
  except Exception as err:
    print(traceback.print_exc())
    print(err)
    sys.exit()
  if re.match(r"^1\d{10}$", self.username):
    print("手機(jī)登陸\n")
    post_url = 'http://www.zhihu.com/login/phone_num'
    postdata = {
      '_xsrf': self.get_xsrf(),
      'password': self.password,
      'remember_me': 'true',
      'phone_num': self.username,
    }
  else:
    print("郵箱登陸\n")
    post_url = 'http://www.zhihu.com/login/email'
    postdata = {
      '_xsrf': self.get_xsrf(),
      'password': self.password,
      'remember_me': 'true',
      'email': self.username,
    }
  try:
    login_page = self.__session.post(post_url, postdata, headers=self.headers, timeout=35)
    login_text = json.loads(login_page.text.encode('latin-1').decode('unicode-escape'))
    print(postdata)
    print(login_text)
    # 需要輸入驗(yàn)證碼 r = 0為登陸成功代碼
    if login_text['r'] == 1:
      sys.exit()
  except:
    postdata['captcha'] = self.get_captcha()
    login_page = self.__session.post(post_url, postdata, headers=self.headers, timeout=35)
    print(json.loads(login_page.text.encode('latin-1').decode('unicode-escape')))
  # 保存登陸cookie
  self.__session.cookies.save()

這個(gè)就是核心的登陸功能啦，非常關(guān)鍵的就是用到了requests庫，非常方便的保存到session
我們這里全局都是用單例模式，統(tǒng)一使用同一個(gè)requests.session對(duì)象進(jìn)行訪問功能，保持登錄狀態(tài)的一致性

最后主要調(diào)用登陸的代碼為

# 創(chuàng)建login對(duì)象
lo = login.login.Login(self.session)
# 模擬登陸
if lo.check_login():
  print('您已經(jīng)登錄')
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('請(qǐng)輸入你的用戶名\n> ')
    password = input("請(qǐng)輸入你的密碼\n> ")
  lo.do_login(username, password)

知乎模擬登陸到此就完成啦

知乎用戶抓取

def __init__(self, threadID=1, name=''):
  # 多線程
  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='cookie')
  self.session.keep_alive = False
  try:
    self.session.cookies.load(ignore_discard=True)
  except:
    print('Cookie 未能加載')
  finally:
    pass
  # 創(chuàng)建login對(duì)象
  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("請(qǐng)安裝redis或檢查redis連接配置")
    sys.exit()
  # 初始化數(shù)據(jù)庫連接
  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("請(qǐng)檢查數(shù)據(jù)庫配置")
    sys.exit()
  # 初始化系統(tǒng)設(shè)置
  self.max_queue_len = int(self.config.get("sys", "max_queue_len"))

這個(gè)是get_user.py的構(gòu)造函數(shù)，主要功能就是初始化mysql連接、redis連接、驗(yàn)證登陸、生成全局的session對(duì)象、導(dǎo)入系統(tǒng)配置、開啟多線程。

# 獲取首頁html
def get_index_page(self):
  index_url = 'https://www.zhihu.com/'
  try:
    index_html = self.session.get(index_url, headers=self.headers, timeout=35)
  except Exception as err:
    # 出現(xiàn)異常重試
    print("獲取頁面失敗，正在重試......")
    print(err)
    traceback.print_exc()
    return None
  finally:
    pass
  return index_html.text
# 獲取單個(gè)用戶詳情頁面
def get_user_page(self, name_url):
  user_page_url = 'https://www.zhihu.com' + str(name_url) + '/about'
  try:
    index_html = self.session.get(user_page_url, headers=self.headers, timeout=35)
  except Exception as err:
    # 出現(xiàn)異常重試
    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 = 'https://www.zhihu.com' + str(name_url) + '/followers'
  try:
    index_html = self.session.get(user_page_url, headers=self.headers, timeout=35)
  except Exception as err:
    # 出現(xiàn)異常重試
    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 = 'https://www.zhihu.com' + str(name_url) + '/followers'
  try:
    index_html = self.session.get(user_page_url, headers=self.headers, timeout=35)
  except Exception as err:
    # 出現(xiàn)異常重試
    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標(biāo)簽
  for a in user_a:
    if a:
      self.add_wait_user(a.get('href'))
    else:
      continue

這一部分的代碼就是用于抓取各個(gè)頁面的html代碼

# 加入帶抓取用戶隊(duì)列，先用redis判斷是否已被抓取過
def add_wait_user(self, name_url):
  # 判斷是否已抓取
  self.threadLock.acquire()
  if not self.redis_con.hexists('already_get_user', name_url):
    self.counter += 1
    print(name_url + " 加入隊(duì)列")
    self.redis_con.hset('already_get_user', name_url, 1)
    self.redis_con.lpush('user_queue', name_url)
    print("添加用戶 " + name_url + "到隊(duì)列")
  self.threadLock.release()
# 獲取頁面出錯(cuò)移出redis
def del_already_user(self, name_url):
  self.threadLock.acquire()
  if not self.redis_con.hexists('already_get_user', name_url):
    self.counter -= 1
    self.redis_con.hdel('already_get_user', name_url)
  self.threadLock.release()

用戶加入redis的操作，在數(shù)據(jù)庫插入出錯(cuò)時(shí)我們調(diào)用del_already_user刪除插入出錯(cuò)的用戶

# 分析粉絲頁面獲取用戶的所有粉絲用戶
# @param follower_page get_follower_page()中獲取到的頁面，這里獲取用戶hash_id請(qǐng)求粉絲接口獲取粉絲信息
def get_all_follower(self, name_url):
  follower_page = self.get_follower_page(name_url)
  # 判斷是否獲取到頁面
  if not follower_page:
    return
  BS = BeautifulSoup(follower_page, 'html.parser')
  # 獲取關(guān)注者數(shù)量
  follower_num = int(BS.find('span', text='關(guān)注者').find_parent().find('strong').get_text())
  # 獲取用戶的hash_id
  hash_id = \
    json.loads(BS.select("#zh-profile-follows-list")[0].select(".zh-general-list")[0].get('data-init'))[
      'params'][
      'hash_id']
  # 獲取關(guān)注者列表
  self.get_xsrf(follower_page) # 獲取xsrf
  post_url = 'https://www.zhihu.com/node/ProfileFollowersListV2'
  # 開始獲取所有的關(guān)注者 math.ceil(follower_num/20)*20
  for i in range(0, math.ceil(follower_num / 20) * 20, 20):
    post_data = {
      'method': 'next',
      'params': 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(
        'latin-1').decode(
        'unicode-escape')
      pattern = re.compile(r"class=\"zm-item-link-avatar\"[^\"]*\"([^\"]*)", re.DOTALL)
      j = pattern.findall(j)
      for user in j:
        user = user.replace('\\', '')
        self.add_wait_user(user) # 保存到redis
    except Exception as err:
      print("獲取正在關(guān)注失敗")
      print(err)
      traceback.print_exc()
      pass
# 獲取正在關(guān)注列表
def get_all_following(self, name_url):
  following_page = self.get_following_page(name_url)
  # 判斷是否獲取到頁面
  if not following_page:
    return
  BS = BeautifulSoup(following_page, 'html.parser')
  # 獲取關(guān)注者數(shù)量
  following_num = int(BS.find('span', text='關(guān)注了').find_parent().find('strong').get_text())
  # 獲取用戶的hash_id
  hash_id = \
    json.loads(BS.select("#zh-profile-follows-list")[0].select(".zh-general-list")[0].get('data-init'))[
      'params'][
      'hash_id']
  # 獲取關(guān)注者列表
  self.get_xsrf(following_page) # 獲取xsrf
  post_url = 'https://www.zhihu.com/node/ProfileFolloweesListV2'
  # 開始獲取所有的關(guān)注者 math.ceil(follower_num/20)*20
  for i in range(0, math.ceil(following_num / 20) * 20, 20):
    post_data = {
      'method': 'next',
      'params': 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(
        'latin-1').decode(
        'unicode-escape')
      pattern = re.compile(r"class=\"zm-item-link-avatar\"[^\"]*\"([^\"]*)", re.DOTALL)
      j = pattern.findall(j)
      for user in j:
        user = user.replace('\\', '')
        self.add_wait_user(user) # 保存到redis
    except Exception as err:
      print("獲取正在關(guān)注失敗")
      print(err)
      traceback.print_exc()
      pass

調(diào)用知乎的API，獲取所有的關(guān)注用戶列表和粉絲用戶列表，遞歸獲取用戶
這里需要注意的是頭部要記得帶上xsrf不然會(huì)拋出403

# 分析about頁面，獲取用戶詳細(xì)資料
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, 'html.parser')
  # 獲取頁面的具體數(shù)據(jù)
  try:
    nickname = BS.find("a", class_="name").get_text() if BS.find("a", class_="name") else ''
    user_type = name_url[1:name_url.index('/', 1)]
    self_domain = name_url[name_url.index('/', 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('span', text='關(guān)注者').find_parent().find('strong').get_text())
    following_num = int(BS.find('span', text='關(guān)注了').find_parent().find('strong').get_text())
    agree_num = int(re.findall(r'<strong>(.*)</strong>.*贊同', about_page)[0])
    appreciate_num = int(re.findall(r'<strong>(.*)</strong>.*感謝', about_page)[0])
    star_num = int(re.findall(r'<strong>(.*)</strong>.*收藏', about_page)[0])
    share_num = int(re.findall(r'<strong>(.*)</strong>.*分享', 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('title') if BS.find("span",
                                       class_="business item") else ''
    company = BS.find("span", class_="employment item").get('title') if BS.find("span",
                                         class_="employment item") else ''
    school = BS.find("span", class_="education item").get('title') if BS.find("span",
                                        class_="education item") else ''
    major = BS.find("span", class_="education-extra item").get('title') if BS.find("span",
                                           class_="education-extra item") else ''
    job = BS.find("span", class_="position item").get_text() if BS.find("span",
                                      class_="position item") else ''
    location = BS.find("span", class_="location item").get('title') if BS.find("span",
                                         class_="location item") else ''
    description = BS.find("div", class_="bio ellipsis").get('title') if BS.find("div",
                                          class_="bio ellipsis") else ''
    ask_num = int(BS.find_all("a", class_='item')[1].find("span").get_text()) if \
      BS.find_all("a", class_='item')[
        1] else int(0)
    answer_num = int(BS.find_all("a", class_='item')[2].find("span").get_text()) if \
      BS.find_all("a", class_='item')[
        2] else int(0)
    article_num = int(BS.find_all("a", class_='item')[3].find("span").get_text()) if \
      BS.find_all("a", class_='item')[3] else int(0)
    collect_num = int(BS.find_all("a", class_='item')[4].find("span").get_text()) if \
      BS.find_all("a", class_='item')[4] else int(0)
    public_edit_num = int(BS.find_all("a", class_='item')[5].find("span").get_text()) if \
      BS.find_all("a", class_='item')[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 = '''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)'''
    try:
      print("獲取到數(shù)據(jù)：")
      print(replace_data)
      self.db_cursor.execute(replace_sql, replace_data)
      self.db.commit()
    except Exception as err:
      print("插入數(shù)據(jù)庫出錯(cuò)")
      print("獲取到數(shù)據(jù)：")
      print(replace_data)
      print("插入語句：" + self.db_cursor._last_executed)
      self.db.rollback()
      print(err)
      traceback.print_exc()
  except Exception as err:
    print("獲取數(shù)據(jù)出錯(cuò)，跳過用戶")
    self.redis_con.hdel("already_get_user", name_url)
    self.del_already_user(name_url)
    print(err)
    traceback.print_exc()
    pass

最后，到用戶的about頁面，分析頁面元素，利用正則或者beatifulsoup分析抓取頁面的數(shù)據(jù)
這里我們SQL語句用REPLACE INTO而不用INSERT INTO，這樣可以很好的防止數(shù)據(jù)重復(fù)問題

# 開始抓取用戶，程序總?cè)肟?
def entrance(self):
  while 1:
    if int(self.redis_con.llen("user_queue")) < 1:
      self.get_index_page_user()
    else:
      # 出隊(duì)列獲取用戶name_url redis取出的是byte，要decode成utf-8
      name_url = str(self.redis_con.rpop("user_queue").decode('utf-8'))
      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__ == '__main__':
  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()

這里就是多線程的開啟，需要開啟多少個(gè)線程就把4換成多少就可以了

Docker

嫌麻煩的可以參考一下我用docker簡單的搭建一個(gè)基礎(chǔ)環(huán)境：

mysql和redis都是官方鏡像

docker run --name mysql -itd mysql:latest
docker run --name redis -itd mysql:latest

再利用docker-compose運(yùn)行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

最后附上源代碼： GITHUB https://github.com/kong36088/ZhihuSpider

本站下載地址： http://xiazai.jb51.net/201612/yuanma/ZhihuSpider(jb51.net).zip

您可能感興趣的文章: