fix(pipeline): add JdJobSearchRow.total_sales for API dataset export

Made-with: Cursor

backend/pipeline/csv_schema.py

@@ -209,3 +209,13 @@ def merged_csv_effective_total_sales(row: dict[str, str]) -> str:
     if direct:
         return direct
     return infer_total_sales_from_sales_floor(str(row.get(h_fl) or ""))
+
+
+def search_csv_effective_total_sales(row: dict[str, str]) -> str:
+    """PC 搜索导出表一行：与 ``merged_csv_effective_total_sales`` 口径一致（中文表头）。"""
+    h_ts = JD_SEARCH_CSV_HEADERS["total_sales"]
+    h_fl = JD_SEARCH_CSV_HEADERS["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

@@ -25,6 +25,7 @@ from .csv_schema import (
     MERGED_FIELD_TO_CSV_HEADER,
     SEARCH_CSV_HEADER_TO_FIELD,
     merged_csv_effective_total_sales,
+    search_csv_effective_total_sales,
 )
 from .models import (
     JdJobCommentRow,
@@ -64,6 +65,11 @@ def _payload_as_json(row: dict[str, str]) -> dict[str, str]:
     return {str(k): str(v) if v is not None else "" for k, v in row.items()}
 
 
+def _normalize_search_csv_total_sales(row: dict[str, str]) -> None:
+    h = JD_SEARCH_CSV_HEADERS["total_sales"]
+    row[h] = search_csv_effective_total_sales(row)
+
+
 def _search_row_kwargs(row: dict[str, str]) -> dict[str, str]:
     vals = {k: "" for k in JD_SEARCH_INTERNAL_KEYS}
     for csv_header, cell in row.items():
@@ -133,6 +139,7 @@ def ingest_job_dataset_rows(job: PipelineJob) -> dict[str, Any]:
         pass
     s_objs: list[JdJobSearchRow] = []
     for i, row in enumerate(search_rows):
+        _normalize_search_csv_total_sales(row)
         kw = _search_row_kwargs(row)
         s_objs.append(JdJobSearchRow(job=job, row_index=i, **kw))
     _bulk_create_in_chunks(JdJobSearchRow, s_objs)

backend/pipeline/management/commands/refresh_jd_merged_total_sales.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 """
-合并表入库后若 ``total_sales`` 为空，可按与入库相同的规则从 ``comment_sales_floor`` 补全。
+合并表 / 搜索导出表入库后若 ``total_sales`` 为空，可按与 ingest 相同规则从 ``comment_sales_floor`` 补全。
 
   python manage.py refresh_jd_merged_total_sales
   python manage.py refresh_jd_merged_total_sales --job-id 42
@@ -9,23 +9,38 @@ 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
+from pipeline.csv_schema import (
+    JD_SEARCH_CSV_HEADERS,
+    MERGED_FIELD_TO_CSV_HEADER,
+    merged_csv_effective_total_sales,
+    search_csv_effective_total_sales,
+)
+from pipeline.models import JdJobMergedRow, JdJobSearchRow
 
 
 class Command(BaseCommand):
-    help = "从销量楼层推断并回填 JdJobMergedRow.total_sales（与 ingest 口径一致）。"
+    help = "从销量楼层推断并回填 JdJobMergedRow / JdJobSearchRow 的 total_sales（与 ingest 一致）。"
 
     def add_arguments(self, parser) -> None:
         parser.add_argument(
             "--job-id",
             type=int,
             default=None,
-            help="仅处理该 PipelineJob；默认处理全部任务下的合并行",
+            help="仅处理该 PipelineJob；默认处理全部任务下的行",
         )
 
     def handle(self, *args, **options) -> None:
         job_id = options.get("job_id")
+        n_merged = self._refresh_merged(job_id)
+        n_search = self._refresh_search(job_id)
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"完成：JdJobMergedRow 更新约 {n_merged} 行，"
+                f"JdJobSearchRow 更新约 {n_search} 行"
+            )
+        )
+
+    def _refresh_merged(self, job_id: int | None) -> int:
         qs = JdJobMergedRow.objects.all().order_by("id")
         if job_id is not None:
             qs = qs.filter(job_id=job_id)
@@ -46,9 +61,27 @@ class Command(BaseCommand):
                 updates.clear()
         if updates:
             JdJobMergedRow.objects.bulk_update(updates, ["total_sales"])
+        return n_changed
 
-        self.stdout.write(
-            self.style.SUCCESS(
-                f"refresh_jd_merged_total_sales 完成，更新行数约 {n_changed}"
-            )
-        )
+    def _refresh_search(self, job_id: int | None) -> int:
+        qs = JdJobSearchRow.objects.all().order_by("id")
+        if job_id is not None:
+            qs = qs.filter(job_id=job_id)
+
+        h_ts = JD_SEARCH_CSV_HEADERS["total_sales"]
+        h_fl = JD_SEARCH_CSV_HEADERS["comment_sales_floor"]
+        updates: list[JdJobSearchRow] = []
+        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 = search_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:
+                JdJobSearchRow.objects.bulk_update(updates, ["total_sales"])
+                updates.clear()
+        if updates:
+            JdJobSearchRow.objects.bulk_update(updates, ["total_sales"])
+        return n_changed

backend/pipeline/migrations/0015_jdjobsearchrow_total_sales.py

@@ -0,0 +1,18 @@
+# Generated manually: align JdJobSearchRow with JD_SEARCH_INTERNAL_KEYS / pc_search_export.
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("pipeline", "0014_jdjobmergedrow_total_sales"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="jdjobsearchrow",
+            name="total_sales",
+            field=models.TextField(blank=True, default=""),
+        ),
+    ]

backend/pipeline/models.py

@@ -159,6 +159,7 @@ class JdJobSearchRow(models.Model):
     original_price = models.TextField(blank=True, default="")
     selling_point = models.TextField(blank=True, default="")
     comment_sales_floor = models.TextField(blank=True, default="")
+    total_sales = models.TextField(blank=True, default="")
     hot_list_rank = models.TextField(blank=True, default="")
     comment_count = models.TextField(blank=True, default="")
     shop_name = models.TextField(blank=True, default="")