154 lines
4.5 KiB
Python
154 lines
4.5 KiB
Python
|
||
"""user_resolver.py
|
||
|
||
将“数字小红书号（例如 11432840954）”解析成 profile_id（/user/profile/{profile_id} 里的那串）。
|
||
- 使用 Playwright 打开搜索结果页（更接近真实浏览器，风险更低）
|
||
- 解析 HTML：根据 '小红书号：<keyword>' 定位，再提取附近的 profile href
|
||
- 强制缓存：data/profile_map.json，避免重复搜索造成风控压力
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import random
|
||
import re
|
||
from html import unescape
|
||
from typing import Dict, Optional, Tuple
|
||
|
||
from playwright.async_api import Browser
|
||
|
||
from app import config
|
||
|
||
PROFILE_MAP_FILE = os.path.join(config.DATA_DIR, "profile_map.json")
|
||
|
||
|
||
def _load_map() -> Dict[str, str]:
|
||
if not os.path.exists(PROFILE_MAP_FILE):
|
||
return {}
|
||
try:
|
||
with open(PROFILE_MAP_FILE, "r", encoding="utf-8") as f:
|
||
return json.load(f) or {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _save_map(m: Dict[str, str]) -> None:
|
||
tmp = PROFILE_MAP_FILE + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(m, f, ensure_ascii=False, indent=2)
|
||
os.replace(tmp, PROFILE_MAP_FILE)
|
||
|
||
|
||
def extract_profile_id_from_search_html(html: str, keyword: str) -> Optional[str]:
|
||
"""从 search_result HTML 中解析出对应 keyword 的 profile_id。"""
|
||
if not html:
|
||
return None
|
||
|
||
html = unescape(html)
|
||
marker = f"小红书号：{keyword}"
|
||
idx = html.find(marker)
|
||
if idx == -1:
|
||
return None
|
||
|
||
start = max(0, idx - 5000)
|
||
end = min(len(html), idx + 5000)
|
||
chunk = html[start:end]
|
||
|
||
m = re.search(r'href="\/user\/profile\/([0-9a-fA-F]+)', chunk)
|
||
if not m:
|
||
return None
|
||
return m.group(1)
|
||
|
||
|
||
def parse_line_to_keyword_or_profile(line: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
|
||
"""解析一行 txt，返回 (keyword数字, profile_id, label备注) 三者之一。
|
||
|
||
允许格式：
|
||
- 11432840954
|
||
- 11432840954 张三
|
||
- https://www.xiaohongshu.com/user/profile/<profile_id> ...
|
||
- /user/profile/<profile_id> ...
|
||
- profile_id(十六进制串)
|
||
- 支持行尾 # 备注
|
||
"""
|
||
raw = line.strip()
|
||
if not raw:
|
||
return None, None, None
|
||
|
||
# 整行注释
|
||
if raw.startswith("#") or raw.startswith("//"):
|
||
return None, None, None
|
||
|
||
# 拆备注：支持 # 以及连续空格
|
||
label = None
|
||
if "#" in raw:
|
||
raw, label = raw.split("#", 1)
|
||
raw = raw.strip()
|
||
label = label.strip() or None
|
||
|
||
parts = raw.split()
|
||
core = parts[0].strip() if parts else ""
|
||
if len(parts) > 1 and not label:
|
||
label = " ".join(parts[1:]).strip() or None
|
||
|
||
# URL/相对路径中直接提取 profile_id
|
||
m = re.search(r"/user/profile/([0-9a-fA-F]+)", core)
|
||
if m:
|
||
return None, m.group(1), label
|
||
|
||
# 可能直接就是 profile_id（十六进制串，长度>=8）
|
||
if re.fullmatch(r"[0-9a-fA-F]{8,}", core):
|
||
return None, core, label
|
||
|
||
# 纯数字：小红书号
|
||
if re.fullmatch(r"\d{5,20}", core):
|
||
return core, None, label
|
||
|
||
return None, None, label
|
||
|
||
|
||
async def resolve_keyword_to_profile_id(browser: Browser, keyword: str, *, slow_mode: bool = True) -> Tuple[Optional[str], str]:
|
||
"""将数字 keyword 解析成 profile_id，并写入缓存。返回 (profile_id, message)。"""
|
||
m = _load_map()
|
||
if keyword in m:
|
||
return m[keyword], "cache"
|
||
|
||
# 使用新的 context（不强制登录）
|
||
context = await browser.new_context()
|
||
page = await context.new_page()
|
||
|
||
try:
|
||
url = f"{config.XIAOHONGSHU_BASE_URL}/search_result?keyword={keyword}&source=unknown"
|
||
await page.goto(url, wait_until="domcontentloaded", timeout=20000)
|
||
|
||
# 轻微等待/滚动（更像真人浏览）
|
||
if slow_mode:
|
||
await page.wait_for_timeout(int(random.uniform(800, 1600)))
|
||
# 小幅滚动一下
|
||
try:
|
||
await page.mouse.wheel(0, random.randint(200, 600))
|
||
except Exception:
|
||
pass
|
||
await page.wait_for_timeout(int(random.uniform(400, 1000)))
|
||
|
||
html = await page.content()
|
||
pid = extract_profile_id_from_search_html(html, keyword)
|
||
if not pid:
|
||
return None, "not_found"
|
||
|
||
m[keyword] = pid
|
||
_save_map(m)
|
||
return pid, "ok"
|
||
except Exception as e:
|
||
return None, f"error:{e}"
|
||
finally:
|
||
try:
|
||
await page.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
await context.close()
|
||
except Exception:
|
||
pass
|