refactor(pipeline): inline total_sales fallback; remove temp backfill script; add refresh command

Made-with: Cursor

backend/crawler_copy/jd_pc_search/jd_competitor_report.py

@@ -37,7 +37,12 @@ _ROOT = Path(__file__).resolve().parent
 if str(_ROOT) not in sys.path:
     sys.path.insert(0, str(_ROOT))
 
+_BACKEND_ROOT = Path(__file__).resolve().parents[2]
+if str(_BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(_BACKEND_ROOT))
+
 import jd_keyword_pipeline as kpl  # noqa: E402
+from pipeline.csv_schema import merged_csv_effective_total_sales  # noqa: E402
 
 # ---------------------------------------------------------------------------
 # 运行配置（按需改这里）
@@ -1578,7 +1583,8 @@ def _competitor_matrix_md_line(
     )
     cat = _md_cell(_detail_category_path_cell(row), 24)
     ing = _matrix_ingredients_cell(row)
-    cc = _md_cell(_cell(row, "销量口径(totalSales)", "评价量(commentFuzzy)"), 14)
+    ts_eff = merged_csv_effective_total_sales(row)
+    cc = _md_cell(ts_eff or _cell(row, "评价量(commentFuzzy)"), 14)
     prev = _md_cell(_cell(row, "comment_preview"), 72)
     return (
         f"| {sku} | {title} | {brand} | {pj} | {df} | {shop} | {sell} | {rank} | "
@@ -2665,7 +2671,7 @@ def build_competitor_brief(
                     "category": _detail_category_path_cell(row),
                     "selling_point": _cell(row, "卖点(sellingPoint)")[:240],
                     "comment_fuzzy": _cell(row, "评价量(commentFuzzy)"),
-                    "total_sales": _cell(row, "销量口径(totalSales)"),
+                    "total_sales": merged_csv_effective_total_sales(row),
                 }
             )
         matrix_groups.append(

backend/pipeline/backfill_merged_total_sales.py

@@ -1,288 +0,0 @@
-"""
-不重新抓搜索：从已有 ``pc_search_raw/*.json``（及 ``.js``）解析 ``totalSales``，
-并写回 ``keyword_pipeline_merged.csv`` /可选 ``pc_search_export.csv``。
-若原始 JSON 无该字段，则尝试从「销量楼层(commentSalesFloor)」单元格中抽取「已售…」片段。
-
-用法（在 backend 目录下）::
-
-  python pipeline/backfill_merged_total_sales.py --run-dir "../data/JD/pipeline_runs/某批次"
-  python pipeline/backfill_merged_total_sales.py --merged "D:/path/keyword_pipeline_merged.csv" --dry-run
-
-"""
-from __future__ import annotations
-
-import argparse
-import csv
-import json
-import re
-import sys
-from pathlib import Path
-from typing import Any
-
-BACKEND_ROOT = Path(__file__).resolve().parent.parent
-if str(BACKEND_ROOT) not in sys.path:
-    sys.path.insert(0, str(BACKEND_ROOT))
-
-from pipeline.csv_schema import (  # noqa: E402
-    JD_SEARCH_CSV_HEADERS,
-    MERGED_FIELD_TO_CSV_HEADER,
-)
-
-COL_MERGED_TOTAL = MERGED_FIELD_TO_CSV_HEADER["total_sales"]
-COL_MERGED_FLOOR = MERGED_FIELD_TO_CSV_HEADER["comment_sales_floor"]
-COL_SKU_MERGED = MERGED_FIELD_TO_CSV_HEADER["sku_id"]
-
-COL_EXPORT_TOTAL = JD_SEARCH_CSV_HEADERS["total_sales"]
-COL_EXPORT_FLOOR = JD_SEARCH_CSV_HEADERS["comment_sales_floor"]
-COL_SKU_EXPORT = JD_SEARCH_CSV_HEADERS["sku_id"]
-
-# 仅列表响应文件，避免误扫 ``pc_request_*.json`` 请求元数据。
-_RAW_GLOBS = ("pc_search_*.json", "pc_search_*.js")
-
-
-def infer_total_sales_from_sales_floor(cell: str) -> str:
-    """
-    从「销量楼层」合并列文案中截取可作销量口径的片段（供图表解析件数）。
-    例：``good:99%好评 | 已售50万+`` → ``已售50万+``。
-    """
-    t = (cell or "").strip()
-    if not t:
-        return ""
-    m = re.search(r"已售\s*[\d,，.+]*\s*[万亿]?\s*\+?", t)
-    if m:
-        return m.group(0).strip()
-    m2 = re.search(r"已售\s*[\d,，.+\s万千亿]+", t)
-    return m2.group(0).strip() if m2 else ""
-
-
-def _load_json_payload(path: Path) -> Any | None:
-    try:
-        text = path.read_text(encoding="utf-8")
-    except OSError:
-        return None
-    try:
-        return json.loads(text)
-    except json.JSONDecodeError:
-        pass
-    jcr_root = BACKEND_ROOT / "crawler_copy" / "jd_pc_search"
-    if str(jcr_root) not in sys.path:
-        sys.path.insert(0, str(jcr_root))
-    try:
-        from search.jd_h5_search_requests import _loads_json_or_jsonp  # noqa: WPS433
-
-        return _loads_json_or_jsonp(text)
-    except Exception:
-        return None
-
-
-def collect_total_sales_from_pc_search_raw(raw_dir: Path) -> dict[str, str]:
-    """
-    遍历 ``pc_search_raw``下保存的列表响应，按 SKU 汇总 ``total_sales``（后者覆盖前者）。
-    """
-    if str(BACKEND_ROOT / "crawler_copy" / "jd_pc_search") not in sys.path:
-        sys.path.insert(0, str(BACKEND_ROOT / "crawler_copy" / "jd_pc_search"))
-    from search.jd_h5_search_requests import parse_items_from_jd_json_payload  # noqa: WPS433
-
-    out: dict[str, str] = {}
-    seen_paths: set[Path] = set()
-    for pattern in _RAW_GLOBS:
-        for p in sorted(raw_dir.glob(pattern)):
-            if p in seen_paths:
-                continue
-            seen_paths.add(p)
-            payload = _load_json_payload(p)
-            if payload is None:
-                continue
-            try:
-                rows = parse_items_from_jd_json_payload(
-                    payload, keyword="", page=1
-                )
-            except Exception:
-                continue
-            for r in rows:
-                if not isinstance(r, dict):
-                    continue
-                sku = str(r.get("sku_id") or "").strip()
-                ts = str(r.get("total_sales") or "").strip()
-                if sku and ts:
-                    out[sku] = ts
-    return out
-
-
-def _insert_column_after(fieldnames: list[str], col: str, after: str) -> list[str]:
-    fn = list(fieldnames)
-    if col in fn:
-        return fn
-    if after in fn:
-        i = fn.index(after) + 1
-        fn.insert(i, col)
-        return fn
-    # 极旧表：无销量楼层时插在评价量后
-    fallback_after = MERGED_FIELD_TO_CSV_HEADER["comment_fuzzy"]
-    if fallback_after in fn:
-        fn.insert(fn.index(fallback_after) + 1, col)
-        return fn
-    fn.append(col)
-    return fn
-
-
-def _backfill_rows(
-    rows: list[dict[str, str]],
-    sku_col: str,
-    total_col: str,
-    floor_col: str,
-    sku_to_total: dict[str, str],
-    *,
-    use_floor_fallback: bool,
-) -> int:
-    n = 0
-    for row in rows:
-        cur = str(row.get(total_col) or "").strip()
-        sku = str(row.get(sku_col) or "").strip()
-        if not cur and sku:
-            ts = sku_to_total.get(sku, "")
-            if ts:
-                row[total_col] = ts
-                cur = ts
-                n += 1
-        if not cur and use_floor_fallback:
-            floor = str(row.get(floor_col) or "").strip()
-            inf = infer_total_sales_from_sales_floor(floor)
-            if inf:
-                row[total_col] = inf
-                n += 1
-    return n
-
-
-def backfill_csv_file(
-    path: Path,
-    *,
-    sku_to_total: dict[str, str],
-    is_merged: bool,
-    use_floor_fallback: bool,
-    dry_run: bool,
-) -> tuple[int, list[str]]:
-    raw = path.read_text(encoding="utf-8-sig")
-    lines = raw.splitlines()
-    if not lines:
-        return 0, []
-    reader = csv.DictReader(lines)
-    old_fn = reader.fieldnames or []
-    if is_merged:
-        sku_c, tot_c, fl_c = (
-            COL_SKU_MERGED,
-            COL_MERGED_TOTAL,
-            COL_MERGED_FLOOR,
-        )
-        fieldnames = _insert_column_after(list(old_fn), tot_c, COL_MERGED_FLOOR)
-    else:
-        sku_c, tot_c, fl_c = COL_SKU_EXPORT, COL_EXPORT_TOTAL, COL_EXPORT_FLOOR
-        fieldnames = _insert_column_after(list(old_fn), tot_c, COL_EXPORT_FLOOR)
-    rows = list(reader)
-    for r in rows:
-        for h in fieldnames:
-            r.setdefault(h, "")
-    filled = _backfill_rows(
-        rows, sku_c, tot_c, fl_c, sku_to_total, use_floor_fallback=use_floor_fallback
-    )
-    if dry_run:
-        return filled, fieldnames
-    from io import StringIO
-
-    sio = StringIO()
-    w2 = csv.DictWriter(sio, fieldnames=fieldnames, lineterminator="\n")
-    w2.writeheader()
-    w2.writerows(rows)
-    path.write_text("\ufeff" + sio.getvalue(), encoding="utf-8")
-    return filled, fieldnames
-
-
-def _resolve_raw_dir(
-    run_dir: Path | None, merged_path: Path | None
-) -> Path | None:
-    if run_dir is not None:
-        rd = (run_dir / "pc_search_raw").resolve()
-        if rd.is_dir():
-            return rd
-    if merged_path is not None:
-        rd = (merged_path.parent / "pc_search_raw").resolve()
-        if rd.is_dir():
-            return rd
-    return None
-
-
-def main() -> None:
-    ap = argparse.ArgumentParser(description="从 pc_search_raw 补全销量口径列（不重新请求搜索）")
-    ap.add_argument("--run-dir", type=Path, default=None, help="批次目录（含 keyword_pipeline_merged.csv 与 pc_search_raw）")
-    ap.add_argument("--merged", type=Path, default=None, help="合并表路径（可单独指定）")
-    ap.add_argument(
-        "--raw-dir",
-        type=Path,
-        default=None,
-        help="原始搜索响应目录（默认：run-dir 或 merged 父目录下的 pc_search_raw）",
-    )
-    ap.add_argument(
-        "--also-pc-search-export",
-        action="store_true",
-        help="同时处理同目录下的 pc_search_export.csv",
-    )
-    ap.add_argument(
-        "--no-floor-fallback",
-        action="store_true",
-        help="禁用从销量楼层文案推断（仅用原始 JSON 中的 totalSales）",
-    )
-    ap.add_argument("--dry-run", action="store_true", help="只统计将补全条数，不写文件")
-    args = ap.parse_args()
-
-    run_dir = args.run_dir.resolve() if args.run_dir else None
-    merged_path = args.merged
-    if merged_path is None and run_dir is not None:
-        merged_path = run_dir / "keyword_pipeline_merged.csv"
-    if merged_path is None or not merged_path.is_file():
-        ap.error("请指定有效的 --merged 或含 keyword_pipeline_merged.csv 的 --run-dir")
-
-    merged_path = merged_path.resolve()
-    raw_dir = args.raw_dir.resolve() if args.raw_dir else _resolve_raw_dir(run_dir, merged_path)
-    sku_map: dict[str, str] = {}
-    if raw_dir is not None:
-        sku_map = collect_total_sales_from_pc_search_raw(raw_dir)
-        print(f"[backfill] 自 {raw_dir} 解析到带 totalSales 的 SKU：{len(sku_map)}", file=sys.stderr)
-    else:
-        print(
-            "[backfill] 未找到 pc_search_raw，将仅尝试销量楼层推断（若未加 --no-floor-fallback）",
-            file=sys.stderr,
-        )
-
-    use_floor = not args.no_floor_fallback
-    n_m, _ = backfill_csv_file(
-        merged_path,
-        sku_to_total=sku_map,
-        is_merged=True,
-        use_floor_fallback=use_floor,
-        dry_run=args.dry_run,
-    )
-    print(
-        f"[backfill] merged：补全单元格数 {n_m}（空列→有值；dry_run={args.dry_run}）",
-        file=sys.stderr,
-    )
-
-    if args.also_pc_search_export:
-        exp = merged_path.parent / "pc_search_export.csv"
-        if exp.is_file():
-            n_e, _ = backfill_csv_file(
-                exp,
-                sku_to_total=sku_map,
-                is_merged=False,
-                use_floor_fallback=use_floor,
-                dry_run=args.dry_run,
-            )
-            print(
-                f"[backfill] pc_search_export：补全单元格数 {n_e}",
-                file=sys.stderr,
-            )
-        else:
-            print(f"[backfill] 跳过：无 {exp}", file=sys.stderr)
-
-
-if __name__ == "__main__":
-    main()

backend/pipeline/csv_schema.py

@@ -4,6 +4,8 @@
 """
 from __future__ import annotations
 
+import re
+
 # --- 搜索导出 pc_search_export.csv（列名为中文，与 jd_h5_search_requests.JD_EXPORT_COLUMN_HEADERS 一致）---
 JD_SEARCH_INTERNAL_KEYS: tuple[str, ...] = (
     "item_id",
@@ -183,3 +185,27 @@ MERGED_CSV_TO_FIELD: dict[str, str] = dict(zip(MERGED_CSV_COLUMNS, MERGED_INTERN
 MERGED_FIELD_TO_CSV_HEADER: dict[str, str] = {
     internal: csv_h for csv_h, internal in MERGED_CSV_TO_FIELD.items()
 }
+
+
+def infer_total_sales_from_sales_floor(cell: str) -> str:
+    """
+    从「销量楼层(commentSalesFloor)」列文案截取可作 ``销量口径(totalSales)`` 的片段（与列表接口未单独落 totalSales 列时的兜底一致）。
+    """
+    t = (cell or "").strip()
+    if not t:
+        return ""
+    m = re.search(r"已售\s*[\d,，.+]*\s*[万亿]?\s*\+?", t)
+    if m:
+        return m.group(0).strip()
+    m2 = re.search(r"已售\s*[\d,，.+\s万千亿]+", t)
+    return m2.group(0).strip() if m2 else ""
+
+
+def merged_csv_effective_total_sales(row: dict[str, str]) -> str:
+    """合并表一行：优先已有 ``销量口径(totalSales)``，否则从销量楼层推断。"""
+    h_ts = MERGED_FIELD_TO_CSV_HEADER["total_sales"]
+    h_fl = MERGED_FIELD_TO_CSV_HEADER["comment_sales_floor"]
+    direct = str(row.get(h_ts) or "").strip()
+    if direct:
+        return direct
+    return infer_total_sales_from_sales_floor(str(row.get(h_fl) or ""))

backend/pipeline/ingest.py

@@ -22,7 +22,9 @@ from .csv_schema import (
     JD_SEARCH_INTERNAL_KEYS,
     MERGED_CSV_COLUMNS,
     MERGED_CSV_TO_FIELD,
+    MERGED_FIELD_TO_CSV_HEADER,
     SEARCH_CSV_HEADER_TO_FIELD,
+    merged_csv_effective_total_sales,
 )
 from .models import (
     JdJobCommentRow,
@@ -84,6 +86,12 @@ def _comment_row_kwargs(row: dict[str, str]) -> dict[str, str]:
     }
 
 
+def _normalize_merged_csv_total_sales(row: dict[str, str]) -> None:
+    """列表未写 totalSales 列时，用销量楼层推断，保证入库与快照与报告口径一致。"""
+    h = MERGED_FIELD_TO_CSV_HEADER["total_sales"]
+    row[h] = merged_csv_effective_total_sales(row)
+
+
 def _merged_row_kwargs(row: dict[str, str]) -> dict[str, str]:
     return {
         MERGED_CSV_TO_FIELD[col]: str(row.get(col) or "").strip() for col in MERGED_CSV_COLUMNS
@@ -152,6 +160,7 @@ def ingest_job_dataset_rows(job: PipelineJob) -> dict[str, Any]:
     merged_rows = _read_csv_rows(merged_path) if merged_path.is_file() else []
     m_objs: list[JdJobMergedRow] = []
     for i, row in enumerate(merged_rows):
+        _normalize_merged_csv_total_sales(row)
         kw = _merged_row_kwargs(row)
         m_objs.append(JdJobMergedRow(job=job, row_index=i, **kw))
     _bulk_create_in_chunks(JdJobMergedRow, m_objs)
@@ -185,6 +194,7 @@ def ingest_job_merged_csv(job: PipelineJob) -> dict[str, Any]:
         sku = (row.get(SKU_FIELD_MERGED) or "").strip()
         if not sku:
             continue
+        _normalize_merged_csv_total_sales(row)
         payload = _payload_as_json(row)
         title = (row.get(TITLE_FIELD) or "")[:2000]
         ware = (row.get(WARE_FIELD) or "").strip()[:64]

backend/pipeline/management/commands/refresh_jd_merged_total_sales.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+"""
+合并表入库后若 ``total_sales`` 为空，可按与入库相同的规则从 ``comment_sales_floor`` 补全。
+
+  python manage.py refresh_jd_merged_total_sales
+  python manage.py refresh_jd_merged_total_sales --job-id 42
+"""
+from __future__ import annotations
+
+from django.core.management.base import BaseCommand
+
+from pipeline.csv_schema import MERGED_FIELD_TO_CSV_HEADER, merged_csv_effective_total_sales
+from pipeline.models import JdJobMergedRow
+
+
+class Command(BaseCommand):
+    help = "从销量楼层推断并回填 JdJobMergedRow.total_sales（与 ingest 口径一致）。"
+
+    def add_arguments(self, parser) -> None:
+        parser.add_argument(
+            "--job-id",
+            type=int,
+            default=None,
+            help="仅处理该 PipelineJob；默认处理全部任务下的合并行",
+        )
+
+    def handle(self, *args, **options) -> None:
+        job_id = options.get("job_id")
+        qs = JdJobMergedRow.objects.all().order_by("id")
+        if job_id is not None:
+            qs = qs.filter(job_id=job_id)
+
+        h_ts = MERGED_FIELD_TO_CSV_HEADER["total_sales"]
+        h_fl = MERGED_FIELD_TO_CSV_HEADER["comment_sales_floor"]
+        updates: list[JdJobMergedRow] = []
+        n_changed = 0
+        for r in qs.iterator(chunk_size=800):
+            row = {h_ts: r.total_sales or "", h_fl: r.comment_sales_floor or ""}
+            eff = merged_csv_effective_total_sales(row)
+            if eff and eff != (r.total_sales or "").strip():
+                r.total_sales = eff
+                updates.append(r)
+                n_changed += 1
+            if len(updates) >= 500:
+                JdJobMergedRow.objects.bulk_update(updates, ["total_sales"])
+                updates.clear()
+        if updates:
+            JdJobMergedRow.objects.bulk_update(updates, ["total_sales"])
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"refresh_jd_merged_total_sales 完成，更新行数约 {n_changed}"
+            )
+        )

backend/pipeline/tests/test_competitor_brief.py

@@ -8,7 +8,7 @@ from pathlib import Path
 from django.conf import settings
 from django.test import SimpleTestCase
 
-from pipeline.backfill_merged_total_sales import infer_total_sales_from_sales_floor
+from pipeline.csv_schema import infer_total_sales_from_sales_floor
 from pipeline.report_charts import _cn_volume_int