189 lines
6.7 KiB
Python
189 lines
6.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
本地脚本：由 DrissionPage (DP) 控制浏览器（本地谷歌 Chrome 或比特浏览器）。
|
||
- 使用本地谷歌：USE_LOCAL_CHROME = True，会启动/连接本机 Chrome。
|
||
- 使用比特浏览器：USE_LOCAL_CHROME = False，需先启动比特浏览器客户端（API 端口 54345）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import random
|
||
import sys
|
||
import time
|
||
|
||
from pathlib import Path
|
||
|
||
# 保证从项目根目录运行时可导入 worker 包
|
||
_ROOT = Path(__file__).resolve().parent
|
||
if str(_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(_ROOT))
|
||
|
||
# ---------- 选择控制对象 ----------
|
||
USE_LOCAL_CHROME = True # True=本地谷歌 Chrome，False=比特浏览器
|
||
|
||
# 本地谷歌 Chrome 配置（仅当 USE_LOCAL_CHROME=True 时生效）
|
||
CHROME_DEBUG_PORT = 9222 # 调试端口；若为 None 则由脚本自动启动 Chrome
|
||
CHROME_PATH = None # 例如 r"C:\Program Files\Google\Chrome\Application\chrome.exe"，None 用系统默认
|
||
|
||
# 比特浏览器配置（仅当 USE_LOCAL_CHROME=False 时生效）
|
||
BIT_API_BASE = "http://127.0.0.1:54345"
|
||
BROWSER_NAME = "测试2"
|
||
BROWSER_ID = None
|
||
|
||
|
||
def _connect_local_chrome():
|
||
"""连接或启动本地谷歌 Chrome，返回 ChromiumPage。"""
|
||
from DrissionPage import ChromiumPage, ChromiumOptions
|
||
|
||
co = ChromiumOptions()
|
||
if CHROME_PATH:
|
||
co.set_browser_path(CHROME_PATH)
|
||
if CHROME_DEBUG_PORT is not None:
|
||
# 连接已开启调试端口的 Chrome（需先手动启动：chrome --remote-debugging-port=9222）
|
||
co.set_local_port(CHROME_DEBUG_PORT)
|
||
print(f"正在连接本机 Chrome（调试端口 {CHROME_DEBUG_PORT}）...")
|
||
page = ChromiumPage(addr_or_opts=co)
|
||
else:
|
||
# 由 DrissionPage 自动启动 Chrome
|
||
print("正在启动本地谷歌 Chrome...")
|
||
page = ChromiumPage(addr_or_opts=co)
|
||
print("已连接本地 Chrome。")
|
||
return page
|
||
|
||
|
||
def _connect_bit_browser():
|
||
"""通过比特浏览器 API 打开并连接，返回 ChromiumPage。"""
|
||
from worker.bit_browser import BitBrowserAPI
|
||
from DrissionPage import ChromiumPage, ChromiumOptions
|
||
|
||
print("正在连接比特浏览器 API...")
|
||
bit_api = BitBrowserAPI(BIT_API_BASE)
|
||
print("正在打开比特浏览器...")
|
||
cdp_addr, port, browser_id = bit_api.open_browser(
|
||
browser_id=BROWSER_ID, name=BROWSER_NAME, remark=None
|
||
)
|
||
print(f"已打开浏览器 ID={browser_id}, CDP 端口={port}")
|
||
co = ChromiumOptions().set_local_port(port=port)
|
||
return ChromiumPage(addr_or_opts=co)
|
||
|
||
|
||
def main(filters):
|
||
from DrissionPage.errors import NoRectError
|
||
|
||
if USE_LOCAL_CHROME:
|
||
page = _connect_local_chrome()
|
||
else:
|
||
page = _connect_bit_browser()
|
||
|
||
# 先启动监听再执行动作，否则拿不到数据包（见 DP 文档）
|
||
page.listen.start('wapi/zpjob/rec/geek/list')
|
||
|
||
if filters:
|
||
main2(page, filters)
|
||
# 有筛选时：page.get 会触发第 1 个 geek/list，点「确定」触发第 2 个；取最后一个才是筛选后的结果
|
||
packets = page.listen.wait(count=2, timeout=30)
|
||
res = packets[-1] if packets else None
|
||
else:
|
||
page.get("https://www.zhipin.com/web/chat/recommend")
|
||
res = page.listen.wait(timeout=30)
|
||
|
||
if not res:
|
||
raise RuntimeError("未捕获到目标请求 wapi/zpjob/rec/geek/list")
|
||
|
||
for i in res.response.body["zpData"]["geekList"]:
|
||
print(i)
|
||
|
||
geekName = i["geekCard"]["geekName"] # 姓名
|
||
geekDegree = i["geekCard"]["geekDegree"] # 学历
|
||
expectPositionName = i["geekCard"]["expectPositionName"] # 期待职位
|
||
expectLocationName = i["geekCard"]["expectLocationName"] # 地区
|
||
applyStatusDesc = i["geekCard"]["applyStatusDesc"] # 当前状态离职0随时到岗之类的
|
||
ageDesc = i["geekCard"]["ageDesc"] # 年龄
|
||
lowSalary = i["geekCard"]["lowSalary"] # 最低要求工资
|
||
highSalary = i["geekCard"]["highSalary"] # 最高要求工资
|
||
|
||
"""三个动作：1. 找到姓名 2. 滚动到那里 3. 点击打招呼"""
|
||
try:
|
||
container = page.get_frame("recommendFrame")
|
||
except Exception:
|
||
container = page
|
||
|
||
# 1. 找到这个姓名
|
||
name_ele = container.ele(f'x://span[contains(text(),"{geekName}")]', timeout=5)
|
||
if not name_ele:
|
||
name_ele = container.ele(f'x://span[text()="{geekName}"]', timeout=2)
|
||
if not name_ele:
|
||
raise Exception(f"未找到姓名：{geekName}")
|
||
|
||
# 2. 滚动到那里
|
||
name_ele.run_js("this.scrollIntoView({block: 'center', behavior: 'auto'})")
|
||
time.sleep(0.8)
|
||
|
||
# 3. 点击打招呼（先在该人所在卡片内找）
|
||
parent = name_ele.parent()
|
||
greet_btn = None
|
||
for _ in range(8):
|
||
if not parent:
|
||
break
|
||
greet_btn = parent.ele('x://span[text()="打招呼"]', timeout=0.5) or parent.ele(
|
||
'x://*[contains(text(),"打招呼")]', timeout=0.5)
|
||
if greet_btn:
|
||
break
|
||
parent = parent.parent()
|
||
if not greet_btn:
|
||
greet_btn = container.ele('x://span[text()="打招呼"]', timeout=2) or container.ele(
|
||
'x://*[contains(text(),"打招呼")]', timeout=1)
|
||
if not greet_btn:
|
||
raise Exception("未找到「打招呼」按钮")
|
||
greet_btn.click(by_js=True)
|
||
|
||
|
||
def main1():
|
||
if USE_LOCAL_CHROME:
|
||
page = _connect_local_chrome()
|
||
else:
|
||
page = _connect_bit_browser()
|
||
|
||
page.listen.start('wapi/zpblock/recommend/filters')
|
||
|
||
# 示例：打开一个页面（可选）
|
||
page.get("https://www.zhipin.com/web/chat/recommend")
|
||
res = page.listen.wait()
|
||
|
||
filters = {}
|
||
for i in res.response.body["zpData"]["vipFilter"]["filters"]:
|
||
print(i)
|
||
|
||
if i["name"] == "年龄":
|
||
print(i["start"])
|
||
print(i["end"])
|
||
|
||
filters[i["name"]] = range(int(i["start"]), int(i["end"]) + 1)
|
||
else:
|
||
datas = []
|
||
for i1 in i["options"]:
|
||
print(i1["name"])
|
||
datas.append(i1["name"])
|
||
|
||
filters[i["name"]] = datas
|
||
|
||
print(filters)
|
||
|
||
|
||
def main2(page, filters):
|
||
"""在同一 page 上打开推荐页、点筛选、选条件、点确定；点击确定后会触发 wapi/zpjob/rec/geek/list，由 main() 的 listen 捕获。"""
|
||
page.get("https://www.zhipin.com/web/chat/recommend")
|
||
|
||
page.ele("x://*[contains(text(),'筛选')]").click()
|
||
time.sleep(3)
|
||
for i in filters:
|
||
page.ele(f"x://*[contains(text(),'{i}')]").click()
|
||
time.sleep(random.random() * 2)
|
||
|
||
page.ele("x://*[contains(text(),'确定')]").click()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main(filters=["初中及以下", '离职-随时到岗'])
|
||
# main1()
|
||
# main2()
|