471 lines
18 KiB
Python
471 lines
18 KiB
Python
"""
|
||
extract_service.py
|
||
=================
|
||
将爬虫原始响应（/api/v1/xhs/wen/overview · /api/v1/xhs/wen/sources）
|
||
清洗为标准结构化格式，与 extract_wenyiwen.py 逻辑一致。
|
||
|
||
在业务后端层面做此转换的好处：
|
||
- 任务结果存入 DB 时已是最干净的格式
|
||
- 前端查看文件、AI 读取数据均直接拿到结构化结果
|
||
- 无需前端再依赖原始爬虫字段名
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 辅助：类型安全取值
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def as_dict(value: Any, path: str, warnings: list[str]) -> dict[str, Any]:
|
||
if isinstance(value, dict):
|
||
return value
|
||
warnings.append(f"{path} 缺失或不是对象")
|
||
return {}
|
||
|
||
|
||
def as_list(value: Any, path: str, warnings: list[str]) -> list[Any]:
|
||
if isinstance(value, list):
|
||
return value
|
||
warnings.append(f"{path} 缺失或不是数组")
|
||
return []
|
||
|
||
|
||
def field(item: dict[str, Any], key: str, path: str, warnings: list[str]) -> Any:
|
||
if key not in item:
|
||
warnings.append(f"{path}.{key} 缺失")
|
||
return None
|
||
return item[key]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 辅助：数值解析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def first_integer(value: Any) -> int | None:
|
||
if not isinstance(value, str):
|
||
return None
|
||
match = re.search(r"\d[\d,]*", value)
|
||
if not match:
|
||
return None
|
||
return int(match.group(0).replace(",", ""))
|
||
|
||
|
||
def numeric_sort_value(value: Any) -> float | None:
|
||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||
return float(value)
|
||
parsed = first_integer(value)
|
||
return float(parsed) if parsed is not None else None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 辅助：时间 / 视频时长格式化
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def format_publish_time(
|
||
value: Any, path: str, warnings: list[str]
|
||
) -> str | None:
|
||
if (
|
||
not isinstance(value, (int, float))
|
||
or isinstance(value, bool)
|
||
or value <= 0
|
||
):
|
||
if value != 0:
|
||
warnings.append(f"{path}.time 缺失或不是有效 Unix 时间戳")
|
||
return None
|
||
try:
|
||
china_tz = timezone(timedelta(hours=8))
|
||
return datetime.fromtimestamp(value, china_tz).strftime("%Y-%m-%d %H:%M:%S")
|
||
except (OverflowError, OSError, ValueError):
|
||
warnings.append(f"{path}.time 超出可转换范围")
|
||
return None
|
||
|
||
|
||
def format_video_duration(
|
||
value: Any, path: str, warnings: list[str]
|
||
) -> str | None:
|
||
if (
|
||
not isinstance(value, (int, float))
|
||
or isinstance(value, bool)
|
||
or value < 0
|
||
):
|
||
warnings.append(f"{path}.duration 缺失或不是非负数")
|
||
return None
|
||
total_seconds = int(value / 1000) if value >= 1000 else int(value)
|
||
hours, remainder = divmod(total_seconds, 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
if hours:
|
||
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
|
||
return f"{minutes:02d}:{seconds:02d}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 辅助：正文清洗
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def clean_full_body(desc: Any, title: Any) -> str | None:
|
||
if not isinstance(desc, str):
|
||
return None
|
||
body = desc.replace("\r\n", "\n").replace("\r", "\n")
|
||
stripped_body = body.lstrip()
|
||
if isinstance(title, str) and title and stripped_body.startswith(title):
|
||
remainder = stripped_body[len(title):]
|
||
if not remainder or remainder[0].isspace():
|
||
body = remainder.lstrip()
|
||
topic_pattern = r"#[^#\n]*?\[话题\]#?"
|
||
body = re.sub(rf"^(?:\s*{topic_pattern})+\s*", "", body)
|
||
trailing_topic = re.search(topic_pattern, body)
|
||
if trailing_topic:
|
||
topic_start = trailing_topic.start()
|
||
prior_topic_block = re.search(
|
||
r"(?s)(?:\s+#[^\n]*)+\s*$", body[:topic_start]
|
||
)
|
||
if prior_topic_block:
|
||
topic_start = prior_topic_block.start()
|
||
else:
|
||
while topic_start > 0 and body[topic_start - 1] in " \t":
|
||
topic_start -= 1
|
||
body = body[:topic_start]
|
||
else:
|
||
body = re.sub(r"(?s)#[^\n]*(?:\n[ \t]*#[^\n]*)*\s*$", "", body)
|
||
body = re.sub(r"(?m)^[ \t]*#+[ \t]*$", "", body)
|
||
body = re.sub(r"(?m)(?:^|\n)[ \t]*(?:标签|话题)[：:][ \t]*$", "", body)
|
||
return body.strip()
|
||
|
||
|
||
def body_char_count(body: str | None) -> int | None:
|
||
if body is None:
|
||
return None
|
||
return len(re.sub(r"\s+", "", body))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 召回明细：建立 (product_name, tag_name) → notes 的查找表
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def build_recall_lookup(
|
||
core_data: dict[str, Any], base_path: str, warnings: list[str]
|
||
) -> tuple[dict[tuple[str, str], dict[str, Any]], bool]:
|
||
raw_note_details = core_data.get("note_details")
|
||
if raw_note_details is None:
|
||
return {}, False
|
||
note_details = as_dict(raw_note_details, f"{base_path}.note_details", warnings)
|
||
products = as_list(
|
||
note_details.get("products"),
|
||
f"{base_path}.note_details.products",
|
||
warnings,
|
||
)
|
||
lookup: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for product_index, raw_product in enumerate(products):
|
||
product_path = f"{base_path}.note_details.products[{product_index}]"
|
||
product = as_dict(raw_product, product_path, warnings)
|
||
product_name = field(product, "product_name", product_path, warnings)
|
||
content_tags = as_list(
|
||
product.get("content_tags"),
|
||
f"{product_path}.content_tags",
|
||
warnings,
|
||
)
|
||
for tag_index, raw_tag in enumerate(content_tags):
|
||
tag_path = f"{product_path}.content_tags[{tag_index}]"
|
||
tag = as_dict(raw_tag, tag_path, warnings)
|
||
tag_name = field(tag, "content_tag", tag_path, warnings)
|
||
if not isinstance(product_name, str) or not isinstance(tag_name, str):
|
||
continue
|
||
key = (product_name, tag_name)
|
||
if key in lookup:
|
||
warnings.append(
|
||
f"召回明细出现重复的产品标签组合：{product_name} / {tag_name}"
|
||
)
|
||
continue
|
||
lookup[key] = {
|
||
"count": field(tag, "count", tag_path, warnings),
|
||
"notes": as_list(tag.get("notes"), f"{tag_path}.notes", warnings),
|
||
"path": tag_path,
|
||
}
|
||
return lookup, True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 召回内容提取
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def extract_recall_notes(
|
||
detail: dict[str, Any], warnings: list[str]
|
||
) -> list[dict[str, Any]]:
|
||
notes = detail["notes"]
|
||
detail_path = detail["path"]
|
||
extracted: list[dict[str, Any]] = []
|
||
for note_index, raw_note in enumerate(notes):
|
||
note_path = f"{detail_path}.notes[{note_index}]"
|
||
note = as_dict(raw_note, note_path, warnings)
|
||
title = note.get("title")
|
||
body = clean_full_body(note.get("desc"), title)
|
||
recall_type = note.get("type")
|
||
recall_content = field(note, "wechat_share_desc", note_path, warnings)
|
||
publish_timestamp = field(note, "time", note_path, warnings)
|
||
publish_time = format_publish_time(publish_timestamp, note_path, warnings)
|
||
if recall_type == "comment":
|
||
extracted.append({
|
||
"召回类型": "comment",
|
||
"评论内容": recall_content,
|
||
"所属笔记点赞数量": field(note, "liked_count", note_path, warnings),
|
||
"发布时间": publish_time,
|
||
})
|
||
continue
|
||
recall_metrics: dict[str, Any] = {}
|
||
if recall_type == "image":
|
||
images_list = as_list(
|
||
field(note, "images_list", note_path, warnings),
|
||
f"{note_path}.images_list",
|
||
warnings,
|
||
)
|
||
recall_metrics = {
|
||
"图片数量": len(images_list),
|
||
"点赞数量": field(note, "liked_count", note_path, warnings),
|
||
"收藏数量": field(note, "collected_count", note_path, warnings),
|
||
}
|
||
elif recall_type == "video":
|
||
recall_metrics = {
|
||
"视频时长": format_video_duration(
|
||
field(note, "duration", note_path, warnings), note_path, warnings
|
||
),
|
||
"点赞": field(note, "liked_count", note_path, warnings),
|
||
"收藏": field(note, "collected_count", note_path, warnings),
|
||
}
|
||
extracted.append({
|
||
"序号": note_index + 1,
|
||
"召回类型": recall_type,
|
||
"笔记标题": title,
|
||
"发布时间": publish_time,
|
||
"召回内容": recall_content,
|
||
"正文内容": body,
|
||
"字数": body_char_count(body),
|
||
"评论数量": note.get("comments_count"),
|
||
**recall_metrics,
|
||
**(
|
||
{} if recall_type in {"image", "video"} else {"评论内容": None}
|
||
),
|
||
"笔记ID": note.get("note_id"),
|
||
"笔记链接": note.get("note_url"),
|
||
})
|
||
return extracted
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 核心：定位 data / data.data.complex_detail
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def locate_core_data(
|
||
root: dict[str, Any], warnings: list[str]
|
||
) -> tuple[dict[str, Any], str]:
|
||
outer_data = as_dict(root.get("data"), "data", warnings)
|
||
if isinstance(outer_data.get("complex_detail"), dict):
|
||
return outer_data, "data"
|
||
inner_data = outer_data.get("data")
|
||
if isinstance(inner_data, dict) and isinstance(
|
||
inner_data.get("complex_detail"), dict
|
||
):
|
||
return inner_data, "data.data"
|
||
warnings.append("未在 data 或 data.data 下找到 complex_detail")
|
||
return {}, "data"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 顶层 extract：对外入口
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def extract(payload: dict[str, Any]) -> dict[str, Any]:
|
||
"""
|
||
对爬虫原始响应 JSON 执行完整清洗，追加 result.extracted 字段。
|
||
|
||
输入 payload：crawler_client 返回的原始爬虫 JSON（即存入 result_json 的内容）
|
||
返回值：在原 payload 上追加 { "extracted": { ...清洗后结构... } }
|
||
仅在 data.complex_detail 存在时注入 extracted
|
||
"""
|
||
warnings: list[str] = []
|
||
root = as_dict(payload, "$", warnings)
|
||
data, base_path = locate_core_data(root, warnings)
|
||
|
||
complex_detail = as_dict(
|
||
data.get("complex_detail"), f"{base_path}.complex_detail", warnings
|
||
)
|
||
brand_info = as_dict(
|
||
complex_detail.get("brand_info"),
|
||
f"{base_path}.complex_detail.brand_info",
|
||
warnings,
|
||
)
|
||
brand_sub_title = field(
|
||
brand_info, "sub_title",
|
||
f"{base_path}.complex_detail.brand_info", warnings,
|
||
)
|
||
|
||
# ---------- 榜单 ----------
|
||
components = as_list(
|
||
complex_detail.get("component_list"),
|
||
f"{base_path}.complex_detail.component_list", warnings,
|
||
)
|
||
ranking: list[dict[str, Any]] = []
|
||
for index, raw_item in enumerate(components):
|
||
item_path = f"{base_path}.complex_detail.component_list[{index}]"
|
||
item = as_dict(raw_item, item_path, warnings)
|
||
ranking.append({
|
||
"产品名称": field(item, "content", item_path, warnings),
|
||
"推荐比例": field(item, "desc", item_path, warnings),
|
||
"当前排名": field(item, "index", item_path, warnings),
|
||
})
|
||
|
||
# ---------- 产品详情 ----------
|
||
data_list = as_list(
|
||
complex_detail.get("data_list"),
|
||
f"{base_path}.complex_detail.data_list", warnings,
|
||
)
|
||
recall_lookup, has_recall_details = build_recall_lookup(data, base_path, warnings)
|
||
matched_recall_keys: set[tuple[str, str]] = set()
|
||
count_mismatch_total = 0
|
||
recall_total = 0
|
||
products: list[dict[str, Any]] = []
|
||
for index, raw_item in enumerate(data_list):
|
||
item_path = f"{base_path}.complex_detail.data_list[{index}]"
|
||
item = as_dict(raw_item, item_path, warnings)
|
||
product_name = field(item, "text", item_path, warnings)
|
||
tag_list = as_list(
|
||
item.get("onepage_tag_list"),
|
||
f"{item_path}.onepage_tag_list", warnings,
|
||
)
|
||
product_tags: list[dict[str, Any]] = []
|
||
for tag_index, raw_tag in enumerate(tag_list):
|
||
tag_path = f"{item_path}.onepage_tag_list[{tag_index}]"
|
||
tag = as_dict(raw_tag, tag_path, warnings)
|
||
tag_name = field(tag, "content", tag_path, warnings)
|
||
experience_count = field(tag, "count", tag_path, warnings)
|
||
tag_output: dict[str, Any] = {
|
||
"内容标签": tag_name,
|
||
"经验数量": experience_count,
|
||
}
|
||
if (
|
||
has_recall_details
|
||
and isinstance(product_name, str)
|
||
and isinstance(tag_name, str)
|
||
):
|
||
key = (product_name, tag_name)
|
||
detail = recall_lookup.get(key)
|
||
if detail is None:
|
||
warnings.append(f"未找到召回明细：{product_name} / {tag_name}")
|
||
tag_output["实际召回数量"] = 0
|
||
tag_output["召回内容"] = []
|
||
else:
|
||
matched_recall_keys.add(key)
|
||
recall_notes = extract_recall_notes(detail, warnings)
|
||
actual_count = len(recall_notes)
|
||
recall_total += actual_count
|
||
if experience_count != actual_count:
|
||
count_mismatch_total += 1
|
||
tag_output["实际召回数量"] = actual_count
|
||
tag_output["召回内容"] = recall_notes
|
||
product_tags.append(tag_output)
|
||
|
||
ranked_tags = sorted(
|
||
enumerate(product_tags),
|
||
key=lambda pair: (
|
||
numeric_sort_value(pair[1].get("经验数量")) is None,
|
||
-(numeric_sort_value(pair[1].get("经验数量")) or 0),
|
||
pair[0],
|
||
),
|
||
)
|
||
tag_top3 = [
|
||
{
|
||
"排名": rank,
|
||
"内容标签": tag["内容标签"],
|
||
"经验数量": tag["经验数量"],
|
||
}
|
||
for rank, (_, tag) in enumerate(ranked_tags[:3], start=1)
|
||
]
|
||
products.append({
|
||
"当前排行": index + 1,
|
||
"参考经验人数": field(item, "desc", item_path, warnings),
|
||
"问点点输出": field(item, "recommend_words", item_path, warnings),
|
||
"推荐比例": field(item, "sub_title", item_path, warnings),
|
||
"产品名称": product_name,
|
||
"标签数量Top3": tag_top3,
|
||
"内容标签": product_tags,
|
||
})
|
||
|
||
if has_recall_details:
|
||
unmatched_keys = set(recall_lookup) - matched_recall_keys
|
||
for product_name, tag_name in sorted(unmatched_keys):
|
||
warnings.append(
|
||
f"召回明细未匹配到榜单标签：{product_name} / {tag_name}"
|
||
)
|
||
|
||
# ---------- 召回统计 ----------
|
||
all_recall_notes = [
|
||
note
|
||
for product in products
|
||
for tag in product["内容标签"]
|
||
for note in tag.get("召回内容", [])
|
||
]
|
||
recall_type_counts: dict[str, int] = {}
|
||
for note in all_recall_notes:
|
||
recall_type = note.get("召回类型")
|
||
recall_type_counts[recall_type] = (
|
||
recall_type_counts.get(recall_type, 0) + 1
|
||
)
|
||
filter_tags = as_list(
|
||
data.get("filter_tags"), f"{base_path}.filter_tags", warnings
|
||
)
|
||
normal_names: list[Any] = []
|
||
for index, raw_tag in enumerate(filter_tags):
|
||
tag_path = f"{base_path}.filter_tags[{index}]"
|
||
tag = as_dict(raw_tag, tag_path, warnings)
|
||
if tag.get("word_type") == "normal":
|
||
normal_names.append(field(tag, "name", tag_path, warnings))
|
||
|
||
extracted = {
|
||
"schema_version": "1.17",
|
||
"提取元数据": {
|
||
"包含召回明细": has_recall_details,
|
||
"召回统计": {
|
||
"标签总数": sum(len(product["内容标签"]) for product in products),
|
||
"实际召回内容总数": recall_total,
|
||
"数量不一致标签数": count_mismatch_total,
|
||
"召回类型统计": recall_type_counts,
|
||
},
|
||
},
|
||
"参考来源笔记总量": {
|
||
"原始文本": brand_sub_title,
|
||
"提取数量": first_integer(brand_sub_title),
|
||
},
|
||
"榜单": ranking,
|
||
"产品详情": products,
|
||
"普通筛选标签": normal_names,
|
||
"提取警告": warnings,
|
||
}
|
||
return extracted
|
||
|
||
|
||
def transform_result(result: dict[str, Any]) -> dict[str, Any]:
|
||
"""
|
||
对任务结果调用 extract 并返回合并后的结果。
|
||
不修改原始 result，追加 result.extracted（当爬虫返回了 complex_detail 时）。
|
||
|
||
用法：
|
||
saved_result = json_loads(row["result_json"]) or {}
|
||
enriched = transform_result(saved_result)
|
||
data["result"] = enriched
|
||
"""
|
||
if not isinstance(result, dict):
|
||
return result
|
||
if "complex_detail" not in str(result.get("data", {})):
|
||
# 尝试 data.data.complex_detail（两级嵌套场景）
|
||
inner = result.get("data", {})
|
||
if isinstance(inner, dict) and "complex_detail" not in inner:
|
||
inner = inner.get("data", {})
|
||
if not isinstance(inner, dict) or "complex_detail" not in inner:
|
||
return result
|
||
extracted = extract(result)
|
||
return {**result, "extracted": extracted}
|