菜单

Administrator
发布于 2025-02-09 / 83 阅读
0
1

BOSS直聘自动投递接入AI分析

新的BOSS直聘脚本 新增AI分析岗位匹配度

"""
@Author: WY
@Date: 2025-02-08
@Description: NOW ENJOY IT
"""
import json
import re
from random import random
import loguru
from DrissionPage import Chromium, ChromiumOptions

from conf.conf import url_zp

# loguru配置
loguru.logger.add("log/boss.log", rotation="1 MB", encoding="utf-8", backtrace=True, diagnose=True)


class ZhiPuAi:
    def __init__(self):
        self.co1 = ChromiumOptions().set_paths(local_port=9111, user_data_path=r'D:\code_py\boss_auto\data')
        self.url = url_zp
        # 随机1-3秒
        self.random_int = random() * 3 + 1
        self.xpath_chat = "xpath://textarea[@class='scroll-display-none']"
        self.xpath_send = "xpath://*[@id='search-input-box']/div[2]/div[3]"
        self.xpath_data = 'xpath://div[@class="answer-content flex1"]'
        self.judge_data_end_xpath = 'xpath://div[@class="panel"]'

    def get_tab(self):
        tab_new = Chromium(addr_or_opts=self.co1).get_tab()
        loguru.logger.info('打开ZhipuAi')
        tab_new.get(self.url)
        tab_new.wait(self.random_int)
        return tab_new

    def get_search(self, tab_zp, job_data):
        tab_zp.ele(self.xpath_chat).input(job_data)
        tab_zp.ele(self.xpath_send).click()
        # tab_zp.wait(120)
        for i in range(20):
            text_data1 = tab_zp.ele(self.xpath_data).text
            tab_zp.wait(8)
            text_data2 = tab_zp.ele(self.xpath_data).text
            if text_data1 == text_data2:
                return text_data1
            else:
                continue


class Boss:
    def __init__(self):
        self.co1 = ChromiumOptions().set_paths(local_port=9111, user_data_path=r'D:\code_py\boss_auto\data')
        self.url = 'https://www.zhipin.com/'
        self.sjkf = 'https://www.zhipin.com/web/geek/job?query=%E9%94%80%E5%94%AE&city=101010100&position=100508'
        # 随机1-3秒
        self.random_int = random() * 3 + 1
        self.xpath_tag_list = "xpath://ul[@class='job-keyword-list']/li"
        self.job_detail_xpath = "xpath://div[@class='job-sec-text']"
        self.job_title_xpath = "xpath://div[@class='name']/h1"
        self.job_bonus_xpath = "xpath://div[@class='job-tags']/span"
        self.job_salary_xpath = "xpath://div[@class='name']/span[@class='salary']"
        self.job_city_xpath = "xpath://div[@class='info-primary']/p/a"
        self.job_year_xpath = "xpath://div[@class='info-primary']/p/span[1]"
        self.job_degree_xpath = "xpath://div[@class='info-primary']/p/span[2]"
        self.start_chat_xpath = "xpath://div[@class='btn-container']/div/a"
        self.chat_close_xpath = "xpath://div[@class='dialog-title']/a/i"
        self.chat_input_xpath = "xpath://div[@class='edit-area']/textarea"
        self.chat_send_xpath = "xpath://div[@class='edit-area']/div"
        self.next_page_xpath = "xpath://a/i[@class='ui-icon-arrow-right']"

    def get_tab(self):
        tab_new = Chromium(addr_or_opts=self.co1).get_tab()
        loguru.logger.info('打开boss直聘')
        tab_new.get(self.url)
        tab_new.wait(self.random_int)
        return tab_new

    def get_latest_tab(self):
        tab_new = Chromium(addr_or_opts=self.co1).latest_tab
        loguru.logger.info('切换最新窗口')
        tab_new.wait(self.random_int)
        return tab_new

    def judge_similar(self, tags_in, job_list_in, job_title_in, job_bonus_in, job_salary_in, job_city_in, job_year_in,
                      job_degree_in):
        json_data = {
            "tags": tags_in,
            "job_list": job_list_in,
            "job_title": job_title_in,
            "job_bonus": job_bonus_in,
            "job_salary": job_salary_in,
            "job_city": job_city_in,
            "job_year": job_year_in,
            "job_degree": job_degree_in
        }
        print(json_data)
        while True:
            zhipu = ZhiPuAi()
            tab_zp = zhipu.get_tab()
            data = zhipu.get_search(tab_zp, json_data)
            loguru.logger.info(data)
            # 使用正则提取百分比
            rate_list = re.findall(r'(\d+\.?\d*)%', data)
            if rate_list and len(rate_list) == 1:
                rate = rate_list[0]
                loguru.logger.info('相似度:{}'.format(rate))
                return int(rate)
            elif rate_list and len(rate_list) > 1:
                # todo 有待商榷,先取最大值
                rate = max(rate_list)
                loguru.logger.info('相似度:{}'.format(rate))
                return int(rate)
            else:
                # 让zhipuai重新搜索
                loguru.logger.warning('未找到相似度,重新搜索')
                continue

    # return rate

    def get_search(self, tab_search, keyword):
        loguru.logger.info('搜索关键词:{}'.format(keyword))
        tab_search.ele('@class=ipt-search').input(keyword)  # 输入搜索内容
        # exit(0)
        tab_search.wait(self.random_int)
        tab_search.ele('@ka=search_box_index').click()  # 点击搜索按钮
        tab_search.wait(self.random_int)
        loguru.logger.info('获取搜索结果')

        tab_search.get(tab_search.url + '&experience=104&industry=100020&position=100508,100507,100511,100506,100104,100512,100514,100122')
        for i in range(1, 10):
            job_list = tab1.eles('@class=job-card-wrapper')
            if len(job_list) == 0:
                loguru.logger.info('未找到相关职位,网站更新了?')
                exit(-1)
            for job in job_list:
                job.click()
                tab_latest = self.get_latest_tab()
                tags_in = self.get_tags(tab_latest)
                job_list_in = self.get_job_detail(tab_latest)
                job_title_in = self.get_job_title(tab_latest)
                job_bonus_in = self.get_job_bonus(tab_latest)
                job_salary_in = self.get_job_salary(tab_latest)
                job_city_in = self.get_job_city(tab_latest)
                job_year_in = self.get_job_year(tab_latest)
                job_degree_in = self.get_job_degree(tab_latest)
                # 快照!
                job_url = tab_latest.url
                rate_similar = self.judge_similar(tags_in, job_list_in, job_title_in, job_bonus_in, job_salary_in,
                                                  job_city_in,
                                                  job_year_in, job_degree_in)
                if rate_similar >= 80:
                    loguru.logger.info('相似度大于80%,打招呼')
                    tab_latest.get(job_url)
                    self.get_start_chat(tab_latest)
                    self.chat_input(tab_latest,rate_similar)
                    self.chat_send(tab_latest)
                    self.chat_close(tab_latest)
                tab_latest.close()
            loguru.logger.info('第{}页'.format(i))
            tab_search.ele(self.next_page_xpath).click()
        loguru.logger.info('搜索完成10页')

    def get_tags(self, tab_latest):
        tags = tab_latest.eles(self.xpath_tag_list)
        tag_list = [tag.text for tag in tags]
        if tag_list:
            loguru.logger.info('职位标签:{}'.format(tag_list))
            return tag_list
        else:
            loguru.logger.warning('该职位没有标签,HR真懒啊')
            return None

    def get_job_detail(self, tab_latest):
        job_detail = tab_latest.ele(self.job_detail_xpath).text
        if job_detail:
            loguru.logger.info('职位详情:{}'.format(job_detail))
            return job_detail
        else:
            loguru.logger.warning('该职位没有详情???那还投个毛')
            return None

    def get_job_title(self, tab_latest):
        job_title = tab_latest.ele(self.job_title_xpath).text
        if job_title:
            loguru.logger.info('职位名称:{}'.format(job_title))
            return job_title
        else:
            loguru.logger.warning('该职位没有名称???那应该不可能吧')
            return None

    def get_job_bonus(self, tab_latest):
        job_bonuss = tab_latest.eles(self.job_bonus_xpath)
        job_bonus = [job_bonus1.text for job_bonus1 in job_bonuss]
        if job_bonus:
            loguru.logger.info('职位福利:{}'.format(job_bonus))
            return job_bonus
        else:
            loguru.logger.warning('该职位没有福利???不太行啊')
            return None

    def get_job_salary(self, tab_latest):
        job_salary = tab_latest.ele(self.job_salary_xpath).text
        if job_salary:
            loguru.logger.info('职位薪资:{}'.format(job_salary))
            return job_salary
        else:
            loguru.logger.warning('该职位没有薪资???')
            return None

    def get_job_city(self, tab_latest):
        job_city = tab_latest.ele(self.job_city_xpath).text
        if job_city:
            loguru.logger.info('职位城市:{}'.format(job_city))
            return job_city
        else:
            loguru.logger.warning('该职位没有城市???')
            return None

    def get_job_year(self, tab_latest):
        job_year = tab_latest.ele(self.job_year_xpath).text
        if job_year:
            loguru.logger.info('职位年限:{}'.format(job_year))
            return job_year
        else:
            loguru.logger.warning('该职位没有年限???')
            return None

    def get_job_degree(self, tab_latest):
        job_degree = tab_latest.ele(self.job_degree_xpath).text
        if job_degree:
            loguru.logger.info('职位学历:{}'.format(job_degree))
            return job_degree
        else:
            loguru.logger.warning('该职位没有学历要求???')
            return None

    def get_start_chat(self, tab_latest):
        start_chat = tab_latest.ele(self.start_chat_xpath).wait(self.random_int)
        if start_chat:
            loguru.logger.info('开始聊天')
            start_chat.click()
        else:
            loguru.logger.warning('该职位没有开始聊天按钮???')

    def chat_close(self, tab_latest):
        chat_close = tab_latest.ele(self.chat_close_xpath).wait(self.random_int)
        if chat_close:
            loguru.logger.info('关闭聊天')
            chat_close.click()
        else:
            loguru.logger.warning('该职位没有关闭聊天按钮???')

    def chat_input(self, tab_latest,rate):
        chat_input = tab_latest.ele(self.chat_input_xpath).wait(self.random_int)
        if chat_input:
            loguru.logger.info('输入聊天内容')
            chat_input.input(f'你好,我对岗位很感兴趣,我想问可以发一份简历吗,谢谢(如果不满足可以跟我说一下不满足的方面吗,万分感谢)')
        else:
            loguru.logger.warning('该职位没有聊天输入框???')

    def chat_send(self, tab_latest):
        chat_send = tab_latest.ele(self.chat_send_xpath)
        if chat_send:
            loguru.logger.info('发送聊天内容')
            chat_send.click()
        else:
            loguru.logger.warning('该职位没有发送聊天按钮???')


if __name__ == '__main__':
    boss = Boss()
    tab1 = boss.get_tab()
    # exit(0)
    boss.get_search(tab1, '数据开发')
    # zhipu = ZhiPuAi()
    # tab_zp = zhipu.get_tab()
    # data=zhipu.get_search(tab_zp,"{'tags': ['Nginx', 'C++', 'Pandas', 'MySQL', 'Numpy', 'Python'], 'job_list': '* IT相关专业\n* 熟练掌握python编程技术,会使用numpy,pandas,matplotlib第三方库\n* 熟练掌握C++编程优先\n* 熟练掌握Qt开发优先\n* 有股票、期货、等金融基本知识,开发A股、期货专用策略具备量化交易经\n* 热爱软件开发\n职责:\n* 参与软件界面需求对接\n* 设计并实现软件界面功能\n* 编写软件需求文档\n* 参与软件测试', 'job_title': 'Python C++ 开发', 'job_bonus': ['股票期权', '五险一金', '股票期权', '五险一金'], 'job_salary': '7-12K', 'job_city': '北京', 'job_year': '3-5年', 'job_degree': '学历不限'}")
    # print(data)


评论