liqing/holy-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
fd83c2d1b2730b589725c8594cee90f64069fc5a
holy-python/extract_service.py

471 lines
18 KiB
Python
Raw Normal View History

holy后端
2026-08-04 14:02:45 +08:00
"""
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}
Reference in New Issue Copy Permalink