wehub-resource-sync
114 lines
4.7 KiB
Python
114 lines
4.7 KiB
Python
"""从渲染后的 DOM 文本里抽取 LinkedIn 单帖分析指标。
|
||
|
||
LinkedIn 把单帖分析（/analytics/post-summary/）SSR/inline 进页面——不是可拦的
|
||
voyager XHR，所以读 `inner_text` 文本、按已知标签锚点解析。纯函数，可独立测试
|
||
（见 test_extract.py）。
|
||
|
||
LinkedIn 会在**日文 / 英文之间随机切换**界面语言（同一 session 内都可能换），
|
||
所以每个指标都存多语言别名，两套都试。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
|
||
def _to_int(s: str) -> int | None:
|
||
"""'34,057' → 34057；'1.2K' → 1200；'3M' → 3000000；抽不出返回 None。"""
|
||
s = s.strip().replace(",", "")
|
||
m = re.fullmatch(r"([\d.]+)\s*([KMB]?)", s)
|
||
if not m:
|
||
return None
|
||
mult = {"": 1, "K": 1_000, "M": 1_000_000, "B": 1_000_000_000}[m.group(2)]
|
||
try:
|
||
return int(float(m.group(1)) * mult)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
# 单帖分析页（/analytics/post-summary/）。LinkedIn 在 日/英 间**随机切换**语言，
|
||
# 所以每个指标存多语言别名。版式两段：
|
||
# 顶部指标——值在标签上一行（"before"）
|
||
# 互动明细——值在标签下一行（"after"）
|
||
# (labels, key, value_position)
|
||
POST_METRICS = [
|
||
(("インプレッション数", "Impressions"), "impressions", "before"),
|
||
(("リーチしたメンバー", "Members reached"), "reach", "before"),
|
||
(("この投稿からのプロフィール閲覧ユーザー", "Profile viewers from this post"), "profile_views_from_post", "before"),
|
||
(("この投稿で獲得したフォロワー", "Followers gained from this post"), "followers_from_post", "before"),
|
||
(("ソーシャルエンゲージメント", "Social engagements"), "social_engagement", "before"),
|
||
(("リアクション", "Reactions"), "reactions", "after"),
|
||
(("コメント", "Comments"), "comments", "after"),
|
||
(("再投稿", "Reposts"), "reposts", "after"),
|
||
(("保存数", "Saves"), "saves", "after"),
|
||
(("LinkedInでの送信数", "Sends on LinkedIn"), "sends", "after"),
|
||
]
|
||
|
||
_IMPRESSION_LABELS = ("インプレッション数", "Impressions")
|
||
|
||
# 作者署名行的语言标记（"…さんが投稿しました • 4日" / "… posted this • 6d"）。
|
||
_BYLINE_MARKERS = ("さんが投稿しました", "posted this")
|
||
# 顶部指标段之前会出现的小标题（"Discovery" / "調査" 等）——正文与指标的分界。
|
||
_BODY_END_MARKERS = ("調査", "Discovery", "ディスカバリー")
|
||
|
||
|
||
def parse_post_summary(text: str) -> dict:
|
||
"""单帖分析 DOM 文本 → {'metrics': {...}}。支持 日/英（LinkedIn 随机切换）。
|
||
|
||
锚在第一个指标（Impressions）后，避开正文里的数字；每个指标按 value_position
|
||
取前一行 / 后一行。抽不到的指标为 None（接口/版式变更时不致整体崩）。
|
||
"""
|
||
lines = [l.strip() for l in text.splitlines() if l.strip()]
|
||
start = 0
|
||
for i, l in enumerate(lines):
|
||
if l in _IMPRESSION_LABELS:
|
||
start = max(0, i - 1)
|
||
break
|
||
scan = lines[start:]
|
||
out = {key: None for _, key, _ in POST_METRICS}
|
||
for i, line in enumerate(scan):
|
||
for labels, key, pos in POST_METRICS:
|
||
if line not in labels or out[key] is not None:
|
||
continue
|
||
j = i - 1 if pos == "before" else i + 1
|
||
if 0 <= j < len(scan):
|
||
out[key] = _to_int(scan[j])
|
||
return {"metrics": out}
|
||
|
||
|
||
def parse_post_meta(text: str) -> dict:
|
||
"""从单帖分析 DOM 抽作者署名 + 相对发布时间 + 正文。
|
||
|
||
版式（reading order）：署名行（"…さんが投稿しました • 4日" / "… posted this • 6d"）
|
||
→ 正文若干行 → 顶部指标小标题（"Discovery"/"調査"）→ 指标段。
|
||
取署名行与小标题之间的行为正文；抽不到时各字段为空，不报错。
|
||
返回 {'author': str, 'age': str, 'text': str}。
|
||
"""
|
||
lines = [l.strip() for l in text.splitlines() if l.strip()]
|
||
byline_idx = None
|
||
for i, l in enumerate(lines):
|
||
if any(m in l for m in _BYLINE_MARKERS):
|
||
byline_idx = i
|
||
break
|
||
if byline_idx is None:
|
||
return {"author": "", "age": "", "text": ""}
|
||
|
||
byline = lines[byline_idx]
|
||
author, age = "", ""
|
||
if "•" in byline:
|
||
head, _, tail = byline.partition("•")
|
||
age = tail.strip()
|
||
author = head.strip()
|
||
else:
|
||
author = byline.strip()
|
||
for marker in _BYLINE_MARKERS:
|
||
if marker in author:
|
||
author = author.split(marker, 1)[0].strip()
|
||
break
|
||
|
||
body: list[str] = []
|
||
for l in lines[byline_idx + 1:]:
|
||
if l in _IMPRESSION_LABELS or l in _BODY_END_MARKERS:
|
||
break
|
||
body.append(l)
|
||
return {"author": author, "age": age, "text": "\n".join(body).strip()}
|