ddrwode/boss_dp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

haha

Browse Source
This commit is contained in:
27942
2026-03-06 01:58:40 +08:00
parent e5becbae86
commit bb951351d7
1 changed files with 175 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

245
1.py
View File

@@ -66,75 +66,192 @@ def _connect_bit_browser():
return ChromiumPage(addr_or_opts=co)
def main(filters):
from DrissionPage.errors import NoRectError
# 自动获取「所有岗位」时使用的选择器（按顺序尝试，取第一个能匹配到 1~100 个元素的结果）
# 若页面结构变化导致拿不到岗位列表，可在此追加或调整选择器
JOB_LIST_SELECTORS = [
"x://*[contains(@class,'job-item') or contains(@class,'position-item')]",
"x://li[contains(@class,'job')]",
"x://div[contains(@class,'job-list')]/div",
"x://ul[contains(@class,'job')]/li",
"x://*[contains(@class,'recommend-job')]//*[contains(@class,'item')]",
"x://*[contains(@class,'job-list')]/*",
"x://a[contains(@href,'job')]",
]
def _get_container(page):
"""推荐牛人内容在 iframe recommendFrame 内，统一在此取容器。"""
try:
return page.get_frame("recommendFrame")
except Exception:
return page
def _get_all_position_elements(container):
"""在推荐页 iframe 内获取左侧「所有岗位」可点击元素列表；找不到则返回空列表。"""
for sel in JOB_LIST_SELECTORS:
try:
eles = container.eles(sel, timeout=2)
if eles and 1 <= len(eles) <= 100:
return eles
except Exception:
continue
return []
def _apply_filter_and_confirm(container, filters):
"""在容器内点「筛选」、选条件、点「确定」。"""
container.ele("x://*[contains(text(),'筛选')]").click()
time.sleep(2)
for item in filters:
container.ele(f"x://*[contains(text(),'{item}')]").click()
time.sleep(random.random() * 1.5)
container.ele("x://*[contains(text(),'确定')]").click()
def _geek_key(item):
"""牛人去重键：优先 encryptGeekId / geekId，否则用姓名。"""
card = item.get("geekCard") or {}
return card.get("encryptGeekId") or card.get("geekId") or card.get("geekName") or ""
def _greet_one_geek(page, container, item):
"""对单个牛人：找姓名 → 滚动到视图 → 点打招呼。返回是否成功。"""
geekName = (item.get("geekCard") or {}).get("geekName", "")
if not geekName:
return False
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:
print(f" 跳过未找到：{geekName}")
return False
name_ele.run_js("this.scrollIntoView({block: 'center', behavior: 'auto'})")
time.sleep(0.6)
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:
print(f" 未找到「打招呼」按钮：{geekName}")
return False
greet_btn.click(by_js=True)
time.sleep(random.uniform(0.5, 1.2))
return True
def _greet_geek_list_skip_greeted(page, container, geek_list, greeted_keys):
"""
对当前包里的牛人列表逐个打招呼，只打尚未在 greeted_keys 里的人，并写入 greeted_keys。
返回本次新打招呼的人数。
"""
n = 0
for item in geek_list or []:
k = _geek_key(item)
if not k or k in greeted_keys:
continue
if _greet_one_geek(page, container, item):
greeted_keys.add(k)
n += 1
return n
def main(filters, position_names=None, greet_target=None):
"""
推荐牛人流程：多岗位循环，所有岗位合计达到目标人数即停；不够则从第一个岗位再跑一轮直到够了。
- filters: 筛选条件列表，如 ["初中及以下", "离职-随时到岗"]；为空则不做筛选。
- position_names: 若传入列表则按名称点击这些岗位；若为 None 则自动获取页面所有岗位并全部循环一遍。
- greet_target: 所有岗位合计目标打招呼人数，如 50。每岗位只抓当前包（不滚动）；一轮跑完若未达目标则从第一个岗位再跑一轮，直到总人数够了或一轮无新增为止。
"""
if USE_LOCAL_CHROME:
page = _connect_local_chrome()
else:
page = _connect_bit_browser()
# 先启动监听再执行动作，否则拿不到数据包（见 DP 文档）
page.listen.start('wapi/zpjob/rec/geek/list')
page.get("https://www.zhipin.com/web/chat/recommend")
time.sleep(2)
container = _get_container(page)
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
if position_names:
positions = [("name", name) for name in position_names]
else:
page.get("https://www.zhipin.com/web/chat/recommend")
res = page.listen.wait(timeout=30)
job_eles = _get_all_position_elements(container)
if job_eles:
n = len(job_eles)
print(f"自动识别到 {n} 个岗位，将依次处理")
positions = [("index", i) for i in range(n)]
else:
positions = [("current", None)]
if not res:
raise RuntimeError("未捕获到目标请求 wapi/zpjob/rec/geek/list")
greeted_keys = set()
total_greeted = 0
round_num = 0
for i in res.response.body["zpData"]["geekList"]:
print(i)
while True:
round_num += 1
if greet_target is not None:
print(f"--- 第 {round_num} 轮，当前已打招呼 {total_greeted}/{greet_target} ---")
round_added = 0
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:
for pos_type, pos_value in positions:
if greet_target is not None and total_greeted >= greet_target:
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)
container = _get_container(page)
label = "当前"
if pos_type == "name":
job_ele = container.ele(f'x://*[contains(text(),"{pos_value}")]', timeout=5)
if not job_ele:
print(f"未找到岗位：{pos_value}，跳过")
continue
job_ele.click(by_js=True)
label = pos_value
elif pos_type == "index":
job_eles = _get_all_position_elements(container)
if pos_value >= len(job_eles):
continue
job_eles[pos_value].click(by_js=True)
label = f"第{pos_value + 1}个岗位"
time.sleep(2)
container = _get_container(page)
if filters:
_apply_filter_and_confirm(container, filters)
packets = page.listen.wait(count=2, timeout=30)
res = packets[-1] if packets else None
else:
res = page.listen.wait(timeout=30)
if not res:
print(f"岗位「{label}」未捕获到 geek/list，跳过")
continue
geek_list = res.response.body.get("zpData", {}).get("geekList") or []
if not geek_list:
continue
n = _greet_geek_list_skip_greeted(page, container, geek_list, greeted_keys)
total_greeted += n
round_added += n
if n:
print(f"岗位「{label}」本包 {len(geek_list)} 人，新打招呼 {n} 人，累计 {total_greeted}" + (f"/{greet_target}" if greet_target else ""))
if greet_target is None:
break
if total_greeted >= greet_target:
print(f"已达目标，共打招呼 {total_greeted} 人")
break
if round_added == 0:
print(f"本轮无新增，已打招呼 {total_greeted} 人，结束")
break
time.sleep(1)
def main1():
@@ -169,20 +286,8 @@ def main1():
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=["初中及以下", '离职-随时到岗'])
# greet_target=50：所有岗位合计打招呼 50 人；一轮不够则从第一个岗位再跑，直到够了或一轮无新增
main(filters=["初中及以下", "离职-随时到岗"], greet_target=50)
# main1()
# main2()