获取看准网A股评论数据，部分数据需要登录才能获取，需要给浏览器加上代理。
# -*- coding: utf-8 -*-
"""================================================================================================================
@date : 2022/11/19 17:06
@function : 看准网
股票代码 公司全称 搜索到的公司 点评人数1 点评人数2 总得分 薪酬福利总得分 职业发展总得分 公司认同总得分 工作条件总得分 工作生活平衡总得分 用户类型：匿名用户等 任职/在职（工作过）那块 评论内容 评论时间 评论总得分 评论总得分 评论薪酬福利得分 评论薪酬福利得分 评论职业发展得分 评论职业发展得分 评论公司认同得分 评论公司认同得分 评论工作条件得分 评论工作条件得分 评论工作生活平衡得分 评论工作生活平衡得分
================================================================================================================"""
import re
import csv
import time
import random
import pandas as pd
from bs4 import BeautifulSoup
from loguru import logger
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from concurrent.futures import ThreadPoolExecutor
from utils.request import Request
from get_company_name import KanZhunSeleniumNoLogin
class KanZhun:
def __init__(self):
self.selenium_no_login = KanZhunSeleniumNoLogin()
self.headers = {'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'accept-encoding': 'gzip, deflate, br',
'accept-language': 'zh-CN,zh;q=0.9',
'cache-control': 'max-age=0',
# 'cookie': 'wd_guid=f803fd1a-c3d3-4e11-b92f-d95d4dc9ca37; W_CITY_S_V=50; __t=nG8GtfU74cmValk; ac=18512143810; historyState=state; __lh_v1=4_131286; AB_T=abva; __g=-; Hm_lvt_1f6f005d03f3c4d854faec87a0bee48e=1668955744,1668957607,1669035462,1669188029; __c=1669191789; __l=r=https%3A%2F%2Fwww.kanzhun.com%2Ffirm%2Freview%2F0nJ839q-Fg~~%2F&l=%2Fqrcode%2FcreateQrCode%3FaccountId%3D9%26page%3Dmodules%2Fcompany%2Fpages%2Fcompany%2Fcompany%26param%3Dc%3D5664736%26s%3D1%26n%3D10; JSESSIONID=""; R_SCH_CY_V=899085|25055|1800226|31558|2091844; __a=16527091.1668955743.1669188029.1669191789.2652.6.177.2652; t=nG8GtfU74cmValk; Hm_lpvt_1f6f005d03f3c4d854faec87a0bee48e=1669293547',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36'
}
self.rq = Request()
option = Options()
option.add_experimental_option('debuggerAddress', '127.0.0.1:9202')
self.driver = webdriver.Chrome(options=option)
self.wait = WebDriverWait(self.driver, 60)
self.score_pattern = re.compile('综合评分(.*?)分：薪酬福利怎么样：(.*?)分、工作条件怎么样：(.*?)分、工作生活平衡怎么样：(.*?)分、职业发展怎么样：(.*?)分、公司认同度怎么样：(.*?)分。')
csv_f = open('result1.csv', 'a+', newline='', encoding='utf-8')
self.csv_writer = csv.writer(csv_f)
# header = ['股票代码','公司全称','搜索到的公司','点评人数1','点评人数2','总得分','薪酬福利总得分','职业发展总得分','公司认同总得分',
# '工作条件总得分','工作生活平衡总得分','用户类型：匿名用户等','任职/在职','评论内容','评论时间','评论总得分','评论总得分',
# '评论薪酬福利得分','评论薪酬福利得分','评论职业发展得分','评论职业发展得分','评论公司认同得分','评论公司认同得分',
# '评论工作条件得分','评论工作条件得分','评论工作生活平衡得分','评论工作生活平衡得分']
# self.csv_writer.writerow(header)
def add_to_csv(self, datas):
self.csv_writer.writerows(datas)
def get_company(self):
df = pd.read_excel('companys.xlsx')
companys = df.values.tolist()
print(len(companys), companys[:3])
return companys
def parse_star(self, word):
return 1 if word in ['很差', '很不满意', '很不认同'] else 2 if word in ['差', '不满意', '不认同'] else 3 if word in ['一般'] else 4 if word in ['很棒', '好', '满意', '认同'] \
else 5 if word in ['完美！', '非常好', '满意', '非常满意', '非常认同'] else ''
def parse_exception(self, obj):
rs = obj.result()
if rs:
print('error', rs)
def treat_one_comment(self, i, l, company_code, comment_code):
"""单一评论详情页，不需要登录"""
logger.info(f'{i}/{l} {company_code} {comment_code}')
url = f'https://www.kanzhun.com/firm/review/{company_code}/{comment_code}.html'
print('comment_url', url)
z = 0
while z < 5:
response = self.rq.requests_get(url, headers=self.headers)
if type(response) == int:
z += 1
print(f'try {z} times:{url}')
else:
# print(type(response))
break
else:
raise
soup = BeautifulSoup(response.text, 'lxml')
username = soup.select('.comment-detail .user span')[0].text.strip()
status = soup.select('.user-info')[0].text.replace(username, '').strip()
content = soup.select('.content-wrap .text-content')[0].text.replace('\n', ' ').replace('\r', ' ').replace('\u200b', ' ').replace('\xa0', '').strip()
time_str = soup.select('.publish span')[0].text.strip()
total_score_text = soup.select('.comment-box .top .kz-rate-text')[0].text.strip()
total_score = self.parse_star(total_score_text)
score1_text = soup.select('.comment-box .center .kz-rate-text')[0].text.strip()
score1 = self.parse_star(score1_text)
score2_text = soup.select('.comment-box .center .kz-rate-text')[1].text.strip()
score2 = self.parse_star(score2_text)
score3_text = soup.select('.comment-box .center .kz-rate-text')[2].text.strip()
score3 = self.parse_star(score3_text)
score4_text = soup.select('.comment-box .center .kz-rate-text')[3].text.strip()
score4 = self.parse_star(score4_text)
score5_text = soup.select('.comment-box .center .kz-rate-text')[4].text.strip()
score5 = self.parse_star(score5_text)
data = [username, status, content, time_str, total_score_text, total_score, score1_text, score1, score2_text, score2,
score3_text, score3, score4_text, score4, score5_text, score5]
print(data)
self.data.append(data)
# return data
def comment_page(self, company_code, id, name):
"""评论页，需要登录"""
comment_url = f'https://www.kanzhun.com/firm/review/{company_code}/'
print('comment_url', comment_url)
self.driver.get(comment_url)
time.sleep(3)
if '暂无公司点评' in self.driver.page_source or '如果你在这里工作过，欢迎分享工作体验' in self.driver.page_source:
print('暂无公司点评')
self.add_to_csv([[id, name, name]])
return
if '哎呀，我们没办法找到这个页面了' in self.driver.page_source:
print('哎呀，我们没办法找到这个页面了')
raise
# self.add_to_csv([[id, name, name]])
# return
total_comment_num1 = self.driver.find_element(By.CSS_SELECTOR, '.page-tab .number').text.strip()
try:
total_comment_num2 = self.driver.find_element(By.CSS_SELECTOR, '#list .title-content .number').text.strip()
except:
total_comment_num2 = 0
print('暂无公司点评')
self.add_to_csv([[id, name, name]])
return
print('total_comment_num1', total_comment_num1, 'total_comment_num2', total_comment_num2)
# 单项评分
description = self.driver.find_element(By.CSS_SELECTOR, '[name="description"]').get_attribute('content')
try:
scores = re.search(self.score_pattern, description).groups()
except AttributeError:
scores = ('', '', '', '', '', '')
print(scores)
i = 0
flag = 0 # [连续] 5次不增长则刷新
while i < 80:
print('i', i)
js = 'window.scrollTo(0, 1000000)'
self.driver.execute_script(js)
time.sleep(2)
js = "return document.documentElement.scrollHeight"
height = self.driver.execute_script(js)
js = f'window.scrollTo(0, {height * random.choice([0.5,0.6,0.7,0.8])})'
self.driver.execute_script(js)
time.sleep(2)
js = f'window.scrollTo(0, {height * random.choice([0.5,0.6,0.7,0.8])})'
self.driver.execute_script(js)
time.sleep(2)
comments = self.driver.find_elements(By.CSS_SELECTOR, '.bala-list_VWnqh .kz-bala-card_d73dq')
print('comments', len(comments))
if len(comments) == 10:
flag += 1
if flag == 5:
self.driver.refresh()
print('flag=5, refresh')
flag = 0
continue
if '没有更多了' in self.driver.page_source:
print('没有更多了')
break
else:
i += 1
continue
else:
raise Exception
comment_codes = [re.search("review/.*?/(.*?).html", i.find_element(By.CSS_SELECTOR, '.header a').get_attribute('href')).group(1) for i in comments]
self.data = []
exec = ThreadPoolExecutor(max_workers=5)
for index, comment_code in enumerate(comment_codes):
# self.treat_one_comment(index, len(comments), company_code, comment_code)
exec.submit(self.treat_one_comment, index, len(comments), company_code, comment_code).add_done_callback(self.parse_exception)
exec.shutdown(wait=True)
self.data = [[id, name, name, total_comment_num1, total_comment_num2] + list(scores) + i for i in self.data]
print('data', len(self.data))
self.add_to_csv(self.data)
def search_one(self, company):
"""搜索公司名称，取第一个，这里无需登录"""
id, name = company[0], company[1]
id = '%06d' % id
print('id', id)
# name = '中国南玻集团股份有限公司'
try:
first_company_code = self.selenium_no_login.search_one(name)
except:
first_company_code = self.selenium_no_login.search_one(name)
if not first_company_code:
print('暂无公司')
self.add_to_csv([[id, name, '']])
return
try:
self.comment_page(first_company_code, id, name)
except:
# 重新点击登录
print('重新点击登录')
try:
self.driver.find_element(By.CSS_SELECTOR, '.btn-login').click()
time.sleep(3)
except:
pass
try:
self.comment_page(first_company_code, id, name)
except:
time.sleep(6)
self.comment_page(first_company_code, id, name)
def main(self):
companys = self.get_company()
# companys = [('002695','江西煌上煌集团食品股份有限公司')]
for index, company in enumerate(companys):
if index < 0:
continue
logger.info(f'{index}/{len(companys)} {company}')
self.search_one(company)
if __name__ == '__main__':
c = KanZhun()
c.main()
# c.search_one()
# c.comment_page('1Xd62dw~', '000012', '中国南玻集团股份有限公司')
get_company_name.py
# -*- coding: utf-8 -*-
"""================================================================================================================
@date : 2022/11/20 9:26
@function : 看准网 搜索公司名称 这里处理不需要登录的部分
================================================================================================================"""
import re
import time
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException
from utils.proxy import selenium_proxy_xiaoxiang
class KanZhunSeleniumNoLogin:
def __init__(self):
self.driver = selenium_proxy_xiaoxiang()
self.wait = WebDriverWait(self.driver, 30)
self.score_pattern = re.compile('综合评分(.*?)分：薪酬福利怎么样：(.*?)分、工作条件怎么样：(.*?)分、工作生活平衡怎么样：(.*?)分、职业发展怎么样：(.*?)分、公司认同度怎么样：(.*?)分。')
def search_one(self, name):
"""搜索公司名称，取第一个，这里无需登录"""
url = f'https://www.kanzhun.com/search?pageNum=1&query={name}&type=1'
x = 0
while x < 5:
try:
self.driver.get(url)
if '没有找到相关结果' in self.driver.page_source:
print('没有找到相关结果')
return
self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, '.search-title')))
break
except:
x += 1
print(f'try {x} times')
else:
raise
first_company_name = self.driver.find_element(By.CSS_SELECTOR, '.company-list .company-info .middle a[href*="firm/info"]').get_attribute('title').replace('（', '(').replace('）', ')')
first_company_code = self.driver.find_element(By.CSS_SELECTOR, '.company-list .company-info .middle a[href*="firm/info"]').get_attribute('href').split('/info/')[-1].split('.')[0]
print('first_company_name', first_company_name, 'first_company_code', first_company_code)
if first_company_name == name:
return first_company_code
else:
print(f'no result:{first_company_name}')
return None
if __name__ == '__main__':
c = KanZhunSeleniumNoLogin()
# c.comment_page()
欢迎关注：爬虫王者
