diff --git a/backend/pipeline/csv_schema.py b/backend/pipeline/csv_schema.py index 0d90bfd..e3daf97 100644 --- a/backend/pipeline/csv_schema.py +++ b/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 "")) diff --git a/backend/pipeline/ingest.py b/backend/pipeline/ingest.py index 5549552..34023dd 100644 --- a/backend/pipeline/ingest.py +++ b/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) diff --git a/backend/pipeline/management/commands/refresh_jd_merged_total_sales.py b/backend/pipeline/management/commands/refresh_jd_merged_total_sales.py index 77b9858..da27599 100644 --- a/backend/pipeline/management/commands/refresh_jd_merged_total_sales.py +++ b/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 diff --git a/backend/pipeline/migrations/0015_jdjobsearchrow_total_sales.py b/backend/pipeline/migrations/0015_jdjobsearchrow_total_sales.py new file mode 100644 index 0000000..ef78c8f --- /dev/null +++ b/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=""), + ), + ] diff --git a/backend/pipeline/models.py b/backend/pipeline/models.py index bd6a407..ce36494 100644 --- a/backend/pipeline/models.py +++ b/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="")