scan_geek_list.py
8.68 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
#!/usr/bin/env python3
"""扫描 BOSS直聘"牛人"候选人列表(推荐/精选/最新)，用 uiautomator dump 读屏幕上已经
渲染出来的候选人卡片文字，落地到本地 SQLite。不调用任何 App 私有网络接口——只是
模拟人手动划列表的操作(adb input swipe) + 读取当前屏幕控件文本(adb uiautomator dump)。
用法:
python3 scan_geek_list.py # 扫描所有已发现的岗位 tab x 推荐/精选/最新
python3 scan_geek_list.py --job 产品助理 --tab 精选
python3 scan_geek_list.py --max-swipes 30 # 每个 tab 最多滑动多少次(防止卡死)
"""
import argparse
import hashlib
import re
import sqlite3
import subprocess
import time
import xml.etree.ElementTree as ET
from pathlib import Path
ADB_BIN = "adb" # 假设已在 PATH 里(见 pull_dbs.sh 里加过 platform-tools)
DUMP_REMOTE = "/sdcard/window_dump.xml"
DUMP_LOCAL = Path(__file__).parent / "data" / "_window_dump.xml"
DB_PATH = Path(__file__).parent / "data" / "geek_list.db"
BOTTOM_NAV_TEXTS = {"牛人", "搜索", "消息", "我的"}
TOPBAR_TEXTS = {"推荐", "精选", "最新", "筛选"}
META_RE = re.compile(r"(应届生|经验|在校|\d+年).*[|｜].*(K|面议|元)")
EDU_RE = re.compile(r".+ · .+")
def adb(*args, timeout=15):
return subprocess.run([ADB_BIN] + list(args), capture_output=True, text=True, timeout=timeout)
def dump_screen_texts():
adb("shell", "uiautomator", "dump", DUMP_REMOTE)
DUMP_LOCAL.parent.mkdir(parents=True, exist_ok=True)
adb("pull", DUMP_REMOTE, str(DUMP_LOCAL))
root = ET.parse(str(DUMP_LOCAL)).getroot()
out = []
for node in root.iter("node"):
t = node.attrib.get("text", "").strip()
if t:
out.append((t, node.attrib.get("bounds", "")))
return out
def find_bounds(texts_with_bounds, label):
for t, b in texts_with_bounds:
if t == label:
m = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", b)
if m:
x1, y1, x2, y2 = map(int, m.groups())
return (x1 + x2) // 2, (y1 + y2) // 2
return None
def tap(x, y):
adb("shell", "input", "tap", str(x), str(y))
time.sleep(1)
def swipe_up():
# 从屏幕中下部往上划一屏，留出顶部 tab 栏和底部导航栏不划到
adb("shell", "input", "swipe", "1000", "1700", "1000", "700", "400")
time.sleep(1.3)
def parse_cards(texts):
"""把一屏候选人卡片的文本节点，按已知卡片结构切成一张张候选人记录。"""
texts = [t for t in texts if t not in BOTTOM_NAV_TEXTS and t not in TOPBAR_TEXTS and t not in ("北京",)]
cards = []
i = 0
n = len(texts)
while i < n:
# 找 meta 行(如 "26年应届生 | 本科 | 6-9K")作为卡片锚点
if META_RE.search(texts[i]):
name = texts[i - 1] if i - 1 >= 0 else ""
badge = ""
# name 前一格可能是徽章("在线"/"刚刚活跃")而不是姓名，姓名再往前一格
if name in ("在线", "刚刚活跃", "今日活跃") and i - 2 >= 0:
badge = name
name = texts[i - 2]
meta = texts[i]
j = i + 1
focus_line = ""
education = ""
tags = []
description = ""
if j < n and (texts[j].startswith("求职期望") or texts[j].startswith("最近关注")):
focus_line = texts[j]
j += 1
if j < n and EDU_RE.match(texts[j]) and len(texts[j]) < 40:
education = texts[j]
j += 1
# 标签: 连续的短文本(<12 字符)，直到遇到长段描述或下一张卡片的 meta 行
while j < n and len(texts[j]) < 12 and not META_RE.search(texts[j]):
tags.append(texts[j])
j += 1
if j < n and not META_RE.search(texts[j]):
description = texts[j]
j += 1
cards.append(
{
"name": name,
"badge": badge,
"meta": meta,
"focus_line": focus_line,
"education": education,
"tags": "、".join(tags),
"description": description,
}
)
i = j
else:
i += 1
return cards
def ensure_db():
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
con = sqlite3.connect(str(DB_PATH))
con.execute(
"""CREATE TABLE IF NOT EXISTS candidates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_tab TEXT,
sub_tab TEXT,
name TEXT,
badge TEXT,
meta TEXT,
focus_line TEXT,
education TEXT,
tags TEXT,
description TEXT,
dedup_key TEXT UNIQUE,
captured_at INTEGER
)"""
)
con.commit()
return con
def dedup_key(job_tab, sub_tab, card):
# 只用姓名+meta做键：同一张卡片在相邻两屏重叠截取时，education/tags 是否被
# 完整读到可能不一致，用它们做键会把同一个人拆成两条。
raw = f"{job_tab}|{sub_tab}|{card['name']}|{card['meta']}"
return hashlib.sha1(raw.encode("utf-8")).hexdigest()
def scan_current_tab(con, job_tab, sub_tab, max_swipes=25):
seen_keys = set()
new_total = 0
stale_rounds = 0
for _ in range(max_swipes):
texts = [t for t, _ in dump_screen_texts()]
cards = parse_cards(texts)
added_this_round = 0
for c in cards:
key = dedup_key(job_tab, sub_tab, c)
if key in seen_keys:
continue
seen_keys.add(key)
row = con.execute(
"SELECT id, length(description) FROM candidates WHERE dedup_key = ?", (key,)
).fetchone()
if row is None:
con.execute(
"INSERT INTO candidates (job_tab, sub_tab, name, badge, meta, focus_line, education, tags, description, dedup_key, captured_at) "
"VALUES (?,?,?,?,?,?,?,?,?,?,strftime('%s','now'))",
(
job_tab,
sub_tab,
c["name"],
c["badge"],
c["meta"],
c["focus_line"],
c["education"],
c["tags"],
c["description"],
key,
),
)
added_this_round += 1
new_total += 1
elif len(c["description"]) > (row[1] or 0):
# 同一个人前后两屏都拍到了，取信息更完整(描述更长)的那次覆盖
con.execute(
"UPDATE candidates SET badge=?, focus_line=?, education=?, tags=?, description=? WHERE id=?",
(c["badge"], c["focus_line"], c["education"], c["tags"], c["description"], row[0]),
)
con.commit()
if added_this_round == 0:
stale_rounds += 1
if stale_rounds >= 2:
break
else:
stale_rounds = 0
swipe_up()
print(f" [{job_tab} / {sub_tab}] 新增 {new_total} 条")
return new_total
def switch_job_and_tab(job_label, sub_tab_label):
texts = dump_screen_texts()
pos = find_bounds(texts, job_label)
if pos:
tap(*pos)
texts = dump_screen_texts()
pos = find_bounds(texts, sub_tab_label)
if pos:
tap(*pos)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--job", action="append", help="职位 tab 名(可重复)，不填则自动探测顶部两个职位 tab")
ap.add_argument("--tab", action="append", default=None, help="子 tab: 推荐/精选/最新，不填则三个都扫")
ap.add_argument("--max-swipes", type=int, default=25)
args = ap.parse_args()
sub_tabs = args.tab or ["推荐", "精选", "最新"]
con = ensure_db()
jobs = args.job
if not jobs:
# 从当前屏幕顶部读职位 tab 名(比如 "产品助理" / "Prompt Engin...")
texts = [t for t, _ in dump_screen_texts()]
jobs = [t for t in texts[:2] if t not in BOTTOM_NAV_TEXTS and t not in TOPBAR_TEXTS]
print("自动探测到职位 tab:", jobs)
for job in jobs:
for sub_tab in sub_tabs:
print(f"=== 切换到 {job} / {sub_tab} ===")
switch_job_and_tab(job, sub_tab)
scan_current_tab(con, job, sub_tab, max_swipes=args.max_swipes)
total = con.execute("SELECT count(*) FROM candidates").fetchone()[0]
print(f"完成，库里累计 {total} 条候选人记录 -> {DB_PATH}")
con.close()
if __name__ == "__main__":
main()