From b6a648af9cbce2b0cb1cf9fb8b69b6ca7ba543e3 Mon Sep 17 00:00:00 2001 From: hub-gif <2487812171@qq.com> Date: Fri, 17 Apr 2026 13:53:28 +0800 Subject: [PATCH] =?UTF-8?q?refactor(jd=5Fpc=5Fsearch):=20=E5=B0=86=20CSV?= =?UTF-8?q?=20=E8=90=BD=E7=9B=98=E4=B8=8E=E9=87=87=E9=9B=86=E6=B5=81?= =?UTF-8?q?=E6=B0=B4=E7=BA=BF=E8=A7=A3=E8=80=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 jd_pipeline_export 模块承载合并表、PC 搜索导出、详情表与 run_meta 的列名规范化与写入；jd_keyword_pipeline 仅保留采集编排与调用。 Made-with: Cursor --- .../jd_pc_search/jd_keyword_pipeline.py | 155 ++++------------ .../jd_pc_search/jd_pipeline_export.py | 171 ++++++++++++++++++ 2 files changed, 205 insertions(+), 121 deletions(-) create mode 100644 backend/crawler_copy/jd_pc_search/jd_pipeline_export.py diff --git a/backend/crawler_copy/jd_pc_search/jd_keyword_pipeline.py b/backend/crawler_copy/jd_pc_search/jd_keyword_pipeline.py index 20af7dc..e83de89 100644 --- a/backend/crawler_copy/jd_pc_search/jd_keyword_pipeline.py +++ b/backend/crawler_copy/jd_pc_search/jd_keyword_pipeline.py @@ -30,12 +30,10 @@ PC 搜索导出 CSV、评价扁平 CSV、详情汇总 CSV（``detail_ware_export from __future__ import annotations -import csv import json import random import sys import time -from io import StringIO from pathlib import Path from types import SimpleNamespace from typing import Any @@ -142,10 +140,6 @@ for _p in (_SEARCH_DIR, _COMMENT_DIR, _DETAIL_DIR): _BACKEND_ROOT = Path(__file__).resolve().parents[2] if str(_BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(_BACKEND_ROOT)) -from pipeline.csv_schema import ( # noqa: E402 - MERGED_CSV_COLUMNS, - remap_merged_row_english_detail_keys_to_csv_headers, -) from collect_pc_search_items import ( # noqa: E402 SearchCollectionCancelled, @@ -180,58 +174,20 @@ from jd_h5_item_comment_requests import ( # noqa: E402 ) from jd_h5_search_requests import ( # noqa: E402 CSV_FIELDS, - JD_EXPORT_COLUMN_HEADERS, jd_row_to_export, ) - - -_SKU_CSV_HEADER = JD_EXPORT_COLUMN_HEADERS["sku_id"] - -_MERGED_EXTRA_FIELDS = ( - ["pipeline_keyword"] - + list(WARE_BUSINESS_MERGE_FIELDNAMES) - + ["comment_count", "comment_preview"] +from jd_pipeline_export import ( # noqa: E402 + SKU_CSV_HEADER, + comment_fields_from_rows, + dedupe_comment_rows, + finalize_merged_row_for_disk, + write_detail_ware_csv, + write_merged_csv, + write_pc_search_export_csv, + write_run_meta_json, ) -def _finalize_merged_row_for_disk(merged: dict[str, str]) -> None: - """英文内部键 → 中文 CSV 列名；评论摘要列名。""" - remap_merged_row_english_detail_keys_to_csv_headers(merged) - if "comment_count" in merged: - merged["评论条数"] = str(merged.pop("comment_count") or "") - if "comment_preview" in merged: - merged["评价摘要"] = str(merged.pop("comment_preview") or "") - - -def _merged_csv_fieldnames() -> list[str]: - if (MERGED_CSV_MODE or "lean").strip().lower() == "full": - return list(CSV_FIELDS) + [ - f for f in _MERGED_EXTRA_FIELDS if f not in CSV_FIELDS - ] - return list(MERGED_CSV_COLUMNS) - - -def _normalize_merged_rows_for_export(rows: list[dict[str, str]]) -> None: - """ - 整合表落盘前：搜索侧「榜单类文案」与「榜单排名」去掉 ``榜单/曝光：`` 前缀， - 与 ``strip_buyer_ranking_line_prefix`` / 入库规则一致。 - """ - from pipeline.csv_schema import strip_buyer_ranking_line_prefix # noqa: WPS433 - - hot_key = "榜单类文案" - rank_key = "榜单排名" - for merged in rows: - if merged.get(hot_key): - merged[hot_key] = strip_buyer_ranking_line_prefix(merged[hot_key]) - merged[rank_key] = strip_buyer_ranking_line_prefix(merged.get(rank_key) or "") - - -def _detail_ware_csv_fieldnames() -> list[str]: - if (DETAIL_WARE_CSV_MODE or "lean").strip().lower() == "full": - return list(WARE_PARSED_CSV_FIELDNAMES) - return list(DETAIL_WARE_LEAN_CSV_FIELDNAMES) - - def _sleep_range(spec: str, label: str) -> None: try: lo, hi = parse_request_delay_range(spec) @@ -244,33 +200,6 @@ def _sleep_range(spec: str, label: str) -> None: time.sleep(t) -def _dedupe_comment_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: - """按 commentId 去重（跨首屏 + 多页列表）。""" - seen: set[str] = set() - out: list[dict[str, Any]] = [] - for r in rows: - cid = str(r.get("commentId") or "").strip() - if cid: - if cid in seen: - continue - seen.add(cid) - out.append(r) - return out - - -def _comment_fields_from_rows(rows: list[dict[str, Any]]) -> dict[str, str]: - previews: list[str] = [] - for r in rows[:8]: - t = str(r.get("tagCommentContent") or "").strip() - if t: - previews.append(t[:400]) - joined = " | ".join(previews)[:4000] - return { - "comment_count": str(len(rows)), - "comment_preview": joined, - } - - def _loads_json(text: str) -> Any: try: return json.loads(text) @@ -479,7 +408,7 @@ def main(keyword: str | None = None) -> Path: skus_ordered: list[str] = [] seen: set[str] = set() for row in export_rows_for_skus: - sid = str(row.get(_SKU_CSV_HEADER) or "").strip() + sid = str(row.get(SKU_CSV_HEADER) or "").strip() if not sid or sid in seen: continue seen.add(sid) @@ -496,13 +425,7 @@ def main(keyword: str | None = None) -> Path: stop_pipeline = True search_csv_path = run_dir / FILE_PC_SEARCH_CSV - sbuf = StringIO() - sw = csv.DictWriter( - sbuf, fieldnames=list(CSV_FIELDS), extrasaction="ignore" - ) - sw.writeheader() - sw.writerows(rows_for_search_csv) - search_csv_path.write_text("\ufeff" + sbuf.getvalue(), encoding="utf-8") + write_pc_search_export_csv(search_csv_path, rows_for_search_csv) print( f"[流水线] 已写 PC 搜索导出 {search_csv_path}", file=sys.stderr, @@ -535,7 +458,7 @@ def main(keyword: str | None = None) -> Path: ( r for r in export_rows_full - if str(r.get(_SKU_CSV_HEADER) or "").strip() == sku + if str(r.get(SKU_CSV_HEADER) or "").strip() == sku ), {}, ) @@ -682,7 +605,7 @@ def main(keyword: str | None = None) -> Path: stop_pipeline = True merged["comment_count"] = "0" merged["comment_preview"] = "" - _finalize_merged_row_for_disk(merged) + finalize_merged_row_for_disk(merged) merged_rows.append(merged) break @@ -692,7 +615,7 @@ def main(keyword: str | None = None) -> Path: stop_pipeline = True merged["comment_count"] = "0" merged["comment_preview"] = "" - _finalize_merged_row_for_disk(merged) + finalize_merged_row_for_disk(merged) merged_rows.append(merged) break @@ -706,7 +629,7 @@ def main(keyword: str | None = None) -> Path: except SystemExit: merged["comment_count"] = "0" merged["comment_preview"] = "" - _finalize_merged_row_for_disk(merged) + finalize_merged_row_for_disk(merged) merged_rows.append(merged) continue @@ -714,7 +637,7 @@ def main(keyword: str | None = None) -> Path: stop_pipeline = True merged["comment_count"] = "0" merged["comment_preview"] = "" - _finalize_merged_row_for_disk(merged) + finalize_merged_row_for_disk(merged) merged_rows.append(merged) break @@ -813,8 +736,8 @@ def main(keyword: str | None = None) -> Path: "firstCommentGuid，仅保留首屏评价", file=sys.stderr, ) - comment_rows = _dedupe_comment_rows(comment_rows) - merged.update(_comment_fields_from_rows(comment_rows)) + comment_rows = dedupe_comment_rows(comment_rows) + merged.update(comment_fields_from_rows(comment_rows)) all_comment_rows.extend(comment_rows) except Exception as e: print( @@ -824,7 +747,7 @@ def main(keyword: str | None = None) -> Path: merged["comment_count"] = "0" merged["comment_preview"] = "" - _finalize_merged_row_for_disk(merged) + finalize_merged_row_for_disk(merged) merged_rows.append(merged) print(f"[流水线] [{idx + 1}/{len(skus_ordered)}] sku={sku} OK", file=sys.stderr) if stop_pipeline: @@ -841,33 +764,26 @@ def main(keyword: str | None = None) -> Path: browser.close() out_path = run_dir / FILE_MERGED_CSV - fieldnames = _merged_csv_fieldnames() - _normalize_merged_rows_for_export(merged_rows) - buf = StringIO() - w = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore") - w.writeheader() - w.writerows(merged_rows) - out_path.write_text("\ufeff" + buf.getvalue(), encoding="utf-8") + _, merged_col_count = write_merged_csv( + out_path, + merged_rows, + merged_csv_mode=MERGED_CSV_MODE, + ) print( f"[流水线] 已写合并表 {out_path} 共 {len(merged_rows)} 行 " - f"（MERGED_CSV_MODE={MERGED_CSV_MODE!r}，{len(fieldnames)} 列）", + f"（MERGED_CSV_MODE={MERGED_CSV_MODE!r}，{merged_col_count} 列）", file=sys.stderr, ) detail_csv_path = run_dir / FILE_DETAIL_WARE_CSV - detail_csv_path.parent.mkdir(parents=True, exist_ok=True) - detail_fn = _detail_ware_csv_fieldnames() - with detail_csv_path.open("w", encoding="utf-8-sig", newline="") as dcf: - dw = csv.DictWriter( - dcf, - fieldnames=detail_fn, - extrasaction="ignore", - ) - dw.writeheader() - dw.writerows(detail_csv_rows) + _, detail_col_count = write_detail_ware_csv( + detail_csv_path, + detail_csv_rows, + detail_ware_csv_mode=DETAIL_WARE_CSV_MODE, + ) print( f"[流水线] 已写详情扁平表 {detail_csv_path} 共 {len(detail_csv_rows)} 行 " - f"（DETAIL_WARE_CSV_MODE={DETAIL_WARE_CSV_MODE!r}，{len(detail_fn)} 列）", + f"（DETAIL_WARE_CSV_MODE={DETAIL_WARE_CSV_MODE!r}，{detail_col_count} 列）", file=sys.stderr, ) @@ -916,19 +832,16 @@ def main(keyword: str | None = None) -> Path: "pc_search_export_rows_full": len(export_rows_full), "merged_rows": len(merged_rows), "merged_csv_mode": (MERGED_CSV_MODE or "lean").strip().lower(), - "merged_csv_column_count": len(fieldnames), + "merged_csv_column_count": merged_col_count, "detail_ware_csv_mode": (DETAIL_WARE_CSV_MODE or "lean").strip().lower(), - "detail_ware_csv_column_count": len(detail_fn), + "detail_ware_csv_column_count": detail_col_count, "comment_flat_rows": len(all_comment_rows), "detail_ware_csv_rows": len(detail_csv_rows), "buyer_offer_profiles_dir": DIR_BUYER_OFFER_PROFILES, "with_comment_list": bool(WITH_COMMENT_LIST), "list_pages": (LIST_PAGES or "").strip(), } - (run_dir / FILE_RUN_META_JSON).write_text( - json.dumps(meta, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8", - ) + write_run_meta_json(run_dir / FILE_RUN_META_JSON, meta) if stop_pipeline: print("[流水线] 已按请求终止（已写出当前进度）", file=sys.stderr) raise PipelineCancelled(run_dir) diff --git a/backend/crawler_copy/jd_pc_search/jd_pipeline_export.py b/backend/crawler_copy/jd_pc_search/jd_pipeline_export.py new file mode 100644 index 0000000..8dc0c68 --- /dev/null +++ b/backend/crawler_copy/jd_pc_search/jd_pipeline_export.py @@ -0,0 +1,171 @@ +# -*- coding: utf-8 -*- +""" +流水线**落盘层**：合并表 / PC 搜索导出 / 详情扁平 CSV 的列名、行规范化与 UTF-8 BOM 写入。 + +与 ``jd_keyword_pipeline`` 中的 **采集编排**（Playwright、请求、合并内存行）分离，便于单独阅读与单测。 +""" +from __future__ import annotations + +import csv +import json +from io import StringIO +from pathlib import Path +from typing import Any + +from pipeline.csv_schema import ( # noqa: E402 + MERGED_CSV_COLUMNS, + remap_merged_row_english_detail_keys_to_csv_headers, +) +from jd_detail_ware_business_requests import ( # noqa: E402 + DETAIL_WARE_LEAN_CSV_FIELDNAMES, + WARE_BUSINESS_MERGE_FIELDNAMES, + WARE_PARSED_CSV_FIELDNAMES, +) +from jd_h5_search_requests import CSV_FIELDS, JD_EXPORT_COLUMN_HEADERS # noqa: E402 + +SKU_CSV_HEADER = JD_EXPORT_COLUMN_HEADERS["sku_id"] + +_MERGED_EXTRA_FIELDS = ( + ["pipeline_keyword"] + + list(WARE_BUSINESS_MERGE_FIELDNAMES) + + ["comment_count", "comment_preview"] +) + + +def finalize_merged_row_for_disk(merged: dict[str, str]) -> None: + """英文内部键 → 中文 CSV 列名；评论摘要列名。""" + remap_merged_row_english_detail_keys_to_csv_headers(merged) + if "comment_count" in merged: + merged["评论条数"] = str(merged.pop("comment_count") or "") + if "comment_preview" in merged: + merged["评价摘要"] = str(merged.pop("comment_preview") or "") + + +def merged_csv_fieldnames(merged_csv_mode: str) -> list[str]: + if (merged_csv_mode or "lean").strip().lower() == "full": + return list(CSV_FIELDS) + [ + f for f in _MERGED_EXTRA_FIELDS if f not in CSV_FIELDS + ] + return list(MERGED_CSV_COLUMNS) + + +def normalize_merged_rows_for_export(rows: list[dict[str, str]]) -> None: + """ + 整合表落盘前：搜索侧「榜单类文案」与「榜单排名」去掉 ``榜单/曝光：`` 前缀， + 与 ``strip_buyer_ranking_line_prefix`` / 入库规则一致。 + """ + from pipeline.csv_schema import strip_buyer_ranking_line_prefix # noqa: WPS433 + + hot_key = "榜单类文案" + rank_key = "榜单排名" + for merged in rows: + if merged.get(hot_key): + merged[hot_key] = strip_buyer_ranking_line_prefix(merged[hot_key]) + merged[rank_key] = strip_buyer_ranking_line_prefix(merged.get(rank_key) or "") + + +def detail_ware_csv_fieldnames(detail_ware_csv_mode: str) -> list[str]: + if (detail_ware_csv_mode or "lean").strip().lower() == "full": + return list(WARE_PARSED_CSV_FIELDNAMES) + return list(DETAIL_WARE_LEAN_CSV_FIELDNAMES) + + +def dedupe_comment_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """按 commentId 去重（跨首屏 + 多页列表）。""" + seen: set[str] = set() + out: list[dict[str, Any]] = [] + for r in rows: + cid = str(r.get("commentId") or "").strip() + if cid: + if cid in seen: + continue + seen.add(cid) + out.append(r) + return out + + +def comment_fields_from_rows(rows: list[dict[str, Any]]) -> dict[str, str]: + previews: list[str] = [] + for r in rows[:8]: + t = str(r.get("tagCommentContent") or "").strip() + if t: + previews.append(t[:400]) + joined = " | ".join(previews)[:4000] + return { + "comment_count": str(len(rows)), + "comment_preview": joined, + } + + +def write_pc_search_export_csv( + path: Path, rows: list[dict[str, str]] +) -> None: + """写入 ``pc_search_export.csv``（UTF-8 BOM + 全列）。""" + sbuf = StringIO() + sw = csv.DictWriter( + sbuf, fieldnames=list(CSV_FIELDS), extrasaction="ignore" + ) + sw.writeheader() + sw.writerows(rows) + path.write_text("\ufeff" + sbuf.getvalue(), encoding="utf-8") + + +def write_merged_csv( + path: Path, + merged_rows: list[dict[str, str]], + *, + merged_csv_mode: str, +) -> tuple[list[str], int]: + """ + 写入合并表；返回 (fieldnames, 列数) 供 ``run_meta`` 使用。 + """ + fieldnames = merged_csv_fieldnames(merged_csv_mode) + normalize_merged_rows_for_export(merged_rows) + buf = StringIO() + w = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore") + w.writeheader() + w.writerows(merged_rows) + path.write_text("\ufeff" + buf.getvalue(), encoding="utf-8") + return fieldnames, len(fieldnames) + + +def write_detail_ware_csv( + path: Path, + detail_csv_rows: list[dict[str, str]], + *, + detail_ware_csv_mode: str, +) -> tuple[list[str], int]: + """写入 ``detail_ware_export.csv``；返回 (fieldnames, 列数)。""" + path.parent.mkdir(parents=True, exist_ok=True) + detail_fn = detail_ware_csv_fieldnames(detail_ware_csv_mode) + with path.open("w", encoding="utf-8-sig", newline="") as dcf: + dw = csv.DictWriter( + dcf, + fieldnames=detail_fn, + extrasaction="ignore", + ) + dw.writeheader() + dw.writerows(detail_csv_rows) + return detail_fn, len(detail_fn) + + +def write_run_meta_json(path: Path, meta: dict[str, Any]) -> None: + path.write_text( + json.dumps(meta, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + + +__all__ = [ + "SKU_CSV_HEADER", + "comment_fields_from_rows", + "dedupe_comment_rows", + "detail_ware_csv_fieldnames", + "finalize_merged_row_for_disk", + "merged_csv_fieldnames", + "normalize_merged_rows_for_export", + "write_detail_ware_csv", + "write_merged_csv", + "write_pc_search_export_csv", + "write_run_meta_json", +]