devops-pdt613/market-assistant
1
0
Fork 0
mirror of https://github.com/primedigitaltech/market-assistant.git synced 2026-07-22 08:01:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(dataset): 类目筛选改为报告矩阵细类名（report_group）

Browse Source 
新增 matrix_group_label 与 jd_competitor_report 同源路径解析；API 使用 report_group 字符串；摘要返回 report_group_options；搜索行按 SKU 从合并表回填细类；导出含报告细类列；前端下拉展示饼干/米等名称。

Made-with: Cursor
This commit is contained in:
hub-gif 2026-04-16 09:58:22 +08:00
parent f915653104
commit d95189723c
11 changed files with 418 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
backend/pipeline/dataset_api.py
View File

@ -1,18 +1,35 @@
"""库内数据浏览 API：排序、价格与类目筛选（查询参数解析与 QuerySet 变换）。"""
"""库内数据浏览 API：排序、价格与报告细类筛选（查询参数解析与 QuerySet 变换）。"""
from __future__ import annotations
from typing import Any
from django.db.models import F, QuerySet
from django.db.models import F, Q, QuerySet
from django.db.models.expressions import OrderBy
from rest_framework.request import Request
SEARCH_SORT_FIELDS = frozenset({"row_index", "price", "sku_id", "title", "leaf_category"})
SEARCH_SORT_FIELDS = frozenset(
{"row_index", "price", "sku_id", "title", "leaf_category", "matrix_group_label"}
)
DETAIL_SORT_FIELDS = frozenset(
{"row_index", "price", "sku_id", "detail_category_path", "detail_brand"}
{
"row_index",
"price",
"sku_id",
"detail_category_path",
"detail_brand",
"matrix_group_label",
}
)
MERGED_SORT_FIELDS = frozenset(
{"row_index", "price", "sku_id", "title", "leaf_category", "detail_category_path"}
{
"row_index",
"price",
"sku_id",
"title",
"leaf_category",
"detail_category_path",
"matrix_group_label",
}
)
@ -39,11 +56,9 @@ def price_bounds_from_request(request: Request) -> tuple[float | None, float | N
)
def category_norm_id_from_request(request: Request) -> int | None:
raw = (request.query_params.get("category_norm_id") or "").strip()
if raw.isdigit():
return int(raw)
return None
def report_group_from_request(request: Request) -> str:
"""与 §5 矩阵一致的细类名（如「饼干」「米」）；对应查询参数 ``report_group``。"""
return (request.query_params.get("report_group") or "").strip()
def detail_category_q_from_request(request: Request) -> str:
@ -52,7 +67,7 @@ def detail_category_q_from_request(request: Request) -> str:
def filter_echo(
*,
category_norm_id: int | None,
report_group: str,
price_min: float | None,
price_max: float | None,
detail_category_q: str,
@ -60,7 +75,7 @@ def filter_echo(
desc: bool,
) -> dict[str, Any]:
return {
"category_norm_id": category_norm_id,
"report_group": report_group or None,
"price_min": price_min,
"price_max": price_max,
"detail_category_q": detail_category_q or None,
@ -70,9 +85,9 @@ def filter_echo(
def apply_search_filters(qs: QuerySet, request: Request) -> QuerySet:
cid = category_norm_id_from_request(request)
if cid is not None:
qs = qs.filter(leaf_category_norm_id=cid)
rg = report_group_from_request(request)
if rg:
qs = qs.filter(Q(matrix_group_label=rg) | Q(leaf_category=rg))
pmin, pmax = price_bounds_from_request(request)
if pmin is not None:
qs = qs.filter(price_value__gte=pmin)
@ -94,6 +109,7 @@ def apply_search_order(qs: QuerySet, sort: str, desc: bool) -> QuerySet:
"sku_id": "sku_id",
"title": "title",
"leaf_category": "leaf_category",
"matrix_group_label": "matrix_group_label",
}[sort]
return qs.order_by(
OrderBy(F(field), descending=desc, nulls_last=True),
@ -102,6 +118,9 @@ def apply_search_order(qs: QuerySet, sort: str, desc: bool) -> QuerySet:
def apply_detail_filters(qs: QuerySet, request: Request) -> QuerySet:
rg = report_group_from_request(request)
if rg:
qs = qs.filter(matrix_group_label=rg)
q = detail_category_q_from_request(request)
if q:
qs = qs.filter(detail_category_path__icontains=q)
@ -126,6 +145,7 @@ def apply_detail_order(qs: QuerySet, sort: str, desc: bool) -> QuerySet:
"sku_id": "sku_id",
"detail_category_path": "detail_category_path",
"detail_brand": "detail_brand",
"matrix_group_label": "matrix_group_label",
}[sort]
return qs.order_by(
OrderBy(F(field), descending=desc, nulls_last=True),
@ -134,9 +154,9 @@ def apply_detail_order(qs: QuerySet, sort: str, desc: bool) -> QuerySet:
def apply_merged_filters(qs: QuerySet, request: Request) -> QuerySet:
cid = category_norm_id_from_request(request)
if cid is not None:
qs = qs.filter(leaf_category_norm_id=cid)
rg = report_group_from_request(request)
if rg:
qs = qs.filter(matrix_group_label=rg)
q = detail_category_q_from_request(request)
if q:
qs = qs.filter(detail_category_path__icontains=q)
@ -162,6 +182,7 @@ def apply_merged_order(qs: QuerySet, sort: str, desc: bool) -> QuerySet:
"title": "title",
"leaf_category": "leaf_category",
"detail_category_path": "detail_category_path",
"matrix_group_label": "matrix_group_label",
}[sort]
return qs.order_by(
OrderBy(F(field), descending=desc, nulls_last=True),

35
backend/pipeline/dataset_nonempty.py
View File

@ -14,6 +14,8 @@ from .csv_schema import (
from .models import JdJobCommentRow, JdJobDetailRow, JdJobMergedRow, JdJobSearchRow, PipelineJob
from .row_serialize import COMMENT_FIELDS_ORDER, DETAIL_FIELDS_ORDER
MATRIX_GROUP_COLUMN = {"key": "matrix_group_label", "label": "报告细类（§5矩阵）"}
def _is_nonempty(val) -> bool:
if val is None:
@ -62,7 +64,13 @@ def nonempty_merged_fields_for_job(job: PipelineJob) -> list[str]:
def search_columns_for_api(job: PipelineJob) -> list[dict[str, str]]:
return [{"key": k, "label": JD_SEARCH_CSV_HEADERS[k]} for k in nonempty_search_keys_for_job(job)]
cols = [
{"key": k, "label": JD_SEARCH_CSV_HEADERS[k]}
for k in nonempty_search_keys_for_job(job)
]
if JdJobSearchRow.objects.filter(job=job).exclude(matrix_group_label="").exists():
cols.append(dict(MATRIX_GROUP_COLUMN))
return cols
def _detail_field_to_csv_col(field: str) -> str:
@ -73,10 +81,13 @@ def _detail_field_to_csv_col(field: str) -> str:
def detail_columns_for_api(job: PipelineJob) -> list[dict[str, str]]:
return [
cols = [
{"key": f, "label": _detail_field_to_csv_col(f)}
for f in nonempty_detail_fields_for_job(job)
]
if JdJobDetailRow.objects.filter(job=job).exclude(matrix_group_label="").exists():
cols.append(dict(MATRIX_GROUP_COLUMN))
return cols
def _comment_field_to_csv_col(field: str) -> str:
@ -94,21 +105,30 @@ def comment_columns_for_api(job: PipelineJob) -> list[dict[str, str]]:
def merged_columns_for_api(job: PipelineJob) -> list[dict[str, str]]:
return [
cols = [
{"key": k, "label": MERGED_FIELD_TO_CSV_HEADER[k]}
for k in nonempty_merged_fields_for_job(job)
]
if JdJobMergedRow.objects.filter(job=job).exclude(matrix_group_label="").exists():
cols.append(dict(MATRIX_GROUP_COLUMN))
return cols
def search_export_headers(job: PipelineJob) -> list[str]:
keys = nonempty_search_keys_for_job(job)
return ["id", "row_index"] + [JD_SEARCH_CSV_HEADERS[k] for k in keys]
h = ["id", "row_index"] + [JD_SEARCH_CSV_HEADERS[k] for k in keys]
if JdJobSearchRow.objects.filter(job=job).exclude(matrix_group_label="").exists():
h.append(MATRIX_GROUP_COLUMN["label"])
return h
def detail_export_headers(job: PipelineJob) -> list[str]:
fields = set(nonempty_detail_fields_for_job(job))
cols = [c for c in DETAIL_CSV_COLUMNS if DETAIL_CSV_TO_FIELD[c] in fields]
return ["id", "row_index"] + cols
h = ["id", "row_index"] + cols
if JdJobDetailRow.objects.filter(job=job).exclude(matrix_group_label="").exists():
h.append(MATRIX_GROUP_COLUMN["label"])
return h
def comment_export_headers(job: PipelineJob) -> list[str]:
@ -119,4 +139,7 @@ def comment_export_headers(job: PipelineJob) -> list[str]:
def merged_export_headers(job: PipelineJob) -> list[str]:
keys = nonempty_merged_fields_for_job(job)
return ["id", "row_index"] + [MERGED_FIELD_TO_CSV_HEADER[k] for k in keys]
h = ["id", "row_index"] + [MERGED_FIELD_TO_CSV_HEADER[k] for k in keys]
if JdJobMergedRow.objects.filter(job=job).exclude(matrix_group_label="").exists():
h.append(MATRIX_GROUP_COLUMN["label"])
return h

135
backend/pipeline/export_job.py
View File

@ -16,6 +16,7 @@ from .csv_schema import (
MERGED_FIELD_TO_CSV_HEADER,
)
from .dataset_nonempty import (
MATRIX_GROUP_COLUMN,
comment_export_headers,
detail_export_headers,
merged_export_headers,
@ -34,20 +35,30 @@ from .row_serialize import (
)
def _search_row_csv_dict(r: JdJobSearchRow, internal_keys: list[str]) -> dict[str, Any]:
def _search_row_csv_dict(
r: JdJobSearchRow, internal_keys: list[str], headers: list[str]
) -> dict[str, Any]:
d = search_row_to_dict(r)
out: dict[str, Any] = {"id": d["id"], "row_index": d["row_index"]}
for k in internal_keys:
out[JD_SEARCH_CSV_HEADERS[k]] = d.get(k, "")
zh = MATRIX_GROUP_COLUMN["label"]
if zh in headers:
out[zh] = d.get("matrix_group_label", "")
return out
def _detail_row_csv_dict(r: JdJobDetailRow, csv_cols: list[str]) -> dict[str, Any]:
def _detail_row_csv_dict(
r: JdJobDetailRow, csv_cols: list[str], headers: list[str]
) -> dict[str, Any]:
d = detail_row_to_dict(r)
out: dict[str, Any] = {"id": d["id"], "row_index": d["row_index"]}
for col in csv_cols:
fn = DETAIL_CSV_TO_FIELD[col]
out[col] = d.get(fn, "")
zh = MATRIX_GROUP_COLUMN["label"]
if zh in headers:
out[zh] = d.get("matrix_group_label", "")
return out
@ -60,11 +71,16 @@ def _comment_row_csv_dict(r: JdJobCommentRow, csv_cols: list[str]) -> dict[str,
return out
def _merged_row_csv_dict(r: JdJobMergedRow, internal_keys: list[str]) -> dict[str, Any]:
def _merged_row_csv_dict(
r: JdJobMergedRow, internal_keys: list[str], headers: list[str]
) -> dict[str, Any]:
d = merged_row_to_dict(r)
out: dict[str, Any] = {"id": d["id"], "row_index": d["row_index"]}
for k in internal_keys:
out[MERGED_FIELD_TO_CSV_HEADER[k]] = d.get(k, "")
zh = MATRIX_GROUP_COLUMN["label"]
if zh in headers:
out[zh] = d.get("matrix_group_label", "")
return out
@ -98,20 +114,30 @@ def _prune_merged_dict(d: dict[str, Any], fields: list[str]) -> dict[str, Any]:
def _rows_as_list_search(job: PipelineJob) -> list[dict[str, Any]]:
keys = nonempty_search_keys_for_job(job)
extra_mg = JdJobSearchRow.objects.filter(job=job).exclude(matrix_group_label="").exists()
qs = JdJobSearchRow.objects.filter(job=job)
return [
_prune_search_dict(search_row_to_dict(obj), keys)
for obj in qs.order_by("row_index").iterator(chunk_size=400)
]
out: list[dict[str, Any]] = []
for obj in qs.order_by("row_index").iterator(chunk_size=400):
d = search_row_to_dict(obj)
row = _prune_search_dict(d, keys)
if extra_mg:
row["matrix_group_label"] = d.get("matrix_group_label", "")
out.append(row)
return out
def _rows_as_list_detail(job: PipelineJob) -> list[dict[str, Any]]:
fields = nonempty_detail_fields_for_job(job)
extra_mg = JdJobDetailRow.objects.filter(job=job).exclude(matrix_group_label="").exists()
qs = JdJobDetailRow.objects.filter(job=job)
return [
_prune_detail_dict(detail_row_to_dict(obj), fields)
for obj in qs.order_by("row_index").iterator(chunk_size=400)
]
out: list[dict[str, Any]] = []
for obj in qs.order_by("row_index").iterator(chunk_size=400):
d = detail_row_to_dict(obj)
row = _prune_detail_dict(d, fields)
if extra_mg:
row["matrix_group_label"] = d.get("matrix_group_label", "")
out.append(row)
return out
def _rows_as_list_comment(job: PipelineJob) -> list[dict[str, Any]]:
@ -125,11 +151,16 @@ def _rows_as_list_comment(job: PipelineJob) -> list[dict[str, Any]]:
def _rows_as_list_merged(job: PipelineJob) -> list[dict[str, Any]]:
fields = nonempty_merged_fields_for_job(job)
extra_mg = JdJobMergedRow.objects.filter(job=job).exclude(matrix_group_label="").exists()
qs = JdJobMergedRow.objects.filter(job=job)
return [
_prune_merged_dict(merged_row_to_dict(obj), fields)
for obj in qs.order_by("row_index").iterator(chunk_size=400)
]
out: list[dict[str, Any]] = []
for obj in qs.order_by("row_index").iterator(chunk_size=400):
d = merged_row_to_dict(obj)
row = _prune_merged_dict(d, fields)
if extra_mg:
row["matrix_group_label"] = d.get("matrix_group_label", "")
out.append(row)
return out
def build_json_bytes(*, job: PipelineJob, kind: str) -> tuple[bytes, str]:
@ -178,18 +209,21 @@ def _write_csv_from_qs(
def build_csv_bytes(*, job: PipelineJob, kind: str) -> tuple[bytes, str]:
if kind == "search":
sk = nonempty_search_keys_for_job(job)
headers = search_export_headers(job)
text = _write_csv_from_qs(
qs=JdJobSearchRow.objects.filter(job=job),
headers=search_export_headers(job),
row_fn=lambda o, _sk=sk: _search_row_csv_dict(o, _sk),
headers=headers,
row_fn=lambda o, _sk=sk, _h=headers: _search_row_csv_dict(o, _sk, _h),
)
name = f"job_{job.id}_search.csv"
elif kind == "detail":
dcols = [c for c in detail_export_headers(job) if c not in ("id", "row_index")]
headers = detail_export_headers(job)
zh = MATRIX_GROUP_COLUMN["label"]
dcols = [c for c in headers if c not in ("id", "row_index", zh)]
text = _write_csv_from_qs(
qs=JdJobDetailRow.objects.filter(job=job),
headers=detail_export_headers(job),
row_fn=lambda o, _dc=dcols: _detail_row_csv_dict(o, _dc),
headers=headers,
row_fn=lambda o, _dc=dcols, _h=headers: _detail_row_csv_dict(o, _dc, _h),
)
name = f"job_{job.id}_detail.csv"
elif kind == "comments":
@ -202,22 +236,28 @@ def build_csv_bytes(*, job: PipelineJob, kind: str) -> tuple[bytes, str]:
name = f"job_{job.id}_comments.csv"
elif kind == "all":
sk = nonempty_search_keys_for_job(job)
dcols = [c for c in detail_export_headers(job) if c not in ("id", "row_index")]
sheaders = search_export_headers(job)
dheaders = detail_export_headers(job)
zh = MATRIX_GROUP_COLUMN["label"]
dcols = [c for c in dheaders if c not in ("id", "row_index", zh)]
ccols = [c for c in comment_export_headers(job) if c not in ("id", "row_index")]
mk = nonempty_merged_fields_for_job(job)
mheaders = merged_export_headers(job)
parts = [
"# search",
_write_csv_from_qs(
qs=JdJobSearchRow.objects.filter(job=job),
headers=search_export_headers(job),
row_fn=lambda o, _sk=sk: _search_row_csv_dict(o, _sk),
headers=sheaders,
row_fn=lambda o, _sk=sk, _h=sheaders: _search_row_csv_dict(o, _sk, _h),
),
"",
"# detail",
_write_csv_from_qs(
qs=JdJobDetailRow.objects.filter(job=job),
headers=detail_export_headers(job),
row_fn=lambda o, _dc=dcols: _detail_row_csv_dict(o, _dc),
headers=dheaders,
row_fn=lambda o, _dc=dcols, _h=dheaders: _detail_row_csv_dict(
o, _dc, _h
),
),
"",
"# comments",
@ -230,18 +270,19 @@ def build_csv_bytes(*, job: PipelineJob, kind: str) -> tuple[bytes, str]:
"# merged",
_write_csv_from_qs(
qs=JdJobMergedRow.objects.filter(job=job),
headers=merged_export_headers(job),
row_fn=lambda o, _mk=mk: _merged_row_csv_dict(o, _mk),
headers=mheaders,
row_fn=lambda o, _mk=mk, _h=mheaders: _merged_row_csv_dict(o, _mk, _h),
),
]
text = "\n".join(parts)
name = f"job_{job.id}_all.csv"
elif kind == "merged":
mk = nonempty_merged_fields_for_job(job)
headers = merged_export_headers(job)
text = _write_csv_from_qs(
qs=JdJobMergedRow.objects.filter(job=job),
headers=merged_export_headers(job),
row_fn=lambda o, _mk=mk: _merged_row_csv_dict(o, _mk),
headers=headers,
row_fn=lambda o, _mk=mk, _h=headers: _merged_row_csv_dict(o, _mk, _h),
)
name = f"job_{job.id}_merged.csv"
else:
@ -262,22 +303,25 @@ def build_xlsx_bytes(*, job: PipelineJob, kind: str) -> tuple[bytes, str]:
ws = wb.active
ws.title = "search"[:31]
sk = nonempty_search_keys_for_job(job)
sheaders = search_export_headers(job)
_append_sheet(
ws,
search_export_headers(job),
sheaders,
JdJobSearchRow.objects.filter(job=job),
lambda o, _sk=sk: _search_row_csv_dict(o, _sk),
lambda o, _sk=sk, _h=sheaders: _search_row_csv_dict(o, _sk, _h),
)
name = f"job_{job.id}_search.xlsx"
elif kind == "detail":
ws = wb.active
ws.title = "detail"[:31]
dcols = [c for c in detail_export_headers(job) if c not in ("id", "row_index")]
dheaders = detail_export_headers(job)
zh = MATRIX_GROUP_COLUMN["label"]
dcols = [c for c in dheaders if c not in ("id", "row_index", zh)]
_append_sheet(
ws,
detail_export_headers(job),
dheaders,
JdJobDetailRow.objects.filter(job=job),
lambda o, _dc=dcols: _detail_row_csv_dict(o, _dc),
lambda o, _dc=dcols, _h=dheaders: _detail_row_csv_dict(o, _dc, _h),
)
name = f"job_{job.id}_detail.xlsx"
elif kind == "comments":
@ -293,23 +337,27 @@ def build_xlsx_bytes(*, job: PipelineJob, kind: str) -> tuple[bytes, str]:
name = f"job_{job.id}_comments.xlsx"
elif kind == "all":
sk = nonempty_search_keys_for_job(job)
dcols = [c for c in detail_export_headers(job) if c not in ("id", "row_index")]
sheaders = search_export_headers(job)
dheaders = detail_export_headers(job)
zh = MATRIX_GROUP_COLUMN["label"]
dcols = [c for c in dheaders if c not in ("id", "row_index", zh)]
ccols = [c for c in comment_export_headers(job) if c not in ("id", "row_index")]
mk = nonempty_merged_fields_for_job(job)
mheaders = merged_export_headers(job)
ws1 = wb.active
ws1.title = "search"[:31]
_append_sheet(
ws1,
search_export_headers(job),
sheaders,
JdJobSearchRow.objects.filter(job=job),
lambda o, _sk=sk: _search_row_csv_dict(o, _sk),
lambda o, _sk=sk, _h=sheaders: _search_row_csv_dict(o, _sk, _h),
)
ws2 = wb.create_sheet("detail"[:31])
_append_sheet(
ws2,
detail_export_headers(job),
dheaders,
JdJobDetailRow.objects.filter(job=job),
lambda o, _dc=dcols: _detail_row_csv_dict(o, _dc),
lambda o, _dc=dcols, _h=dheaders: _detail_row_csv_dict(o, _dc, _h),
)
ws3 = wb.create_sheet("comments"[:31])
_append_sheet(
@ -321,20 +369,21 @@ def build_xlsx_bytes(*, job: PipelineJob, kind: str) -> tuple[bytes, str]:
ws4 = wb.create_sheet("merged"[:31])
_append_sheet(
ws4,
merged_export_headers(job),
mheaders,
JdJobMergedRow.objects.filter(job=job),
lambda o, _mk=mk: _merged_row_csv_dict(o, _mk),
lambda o, _mk=mk, _h=mheaders: _merged_row_csv_dict(o, _mk, _h),
)
name = f"job_{job.id}_all.xlsx"
elif kind == "merged":
mk = nonempty_merged_fields_for_job(job)
ws = wb.active
ws.title = "merged"[:31]
mheaders = merged_export_headers(job)
_append_sheet(
ws,
merged_export_headers(job),
mheaders,
JdJobMergedRow.objects.filter(job=job),
lambda o, _mk=mk: _merged_row_csv_dict(o, _mk),
lambda o, _mk=mk, _h=mheaders: _merged_row_csv_dict(o, _mk, _h),
)
name = f"job_{job.id}_merged.xlsx"
else:

84
backend/pipeline/ingest.py
View File

@ -29,12 +29,12 @@ from .csv_schema import (
search_csv_effective_total_sales,
strip_buyer_ranking_line_prefix,
)
from .matrix_group_label import matrix_group_label_from_detail_path
from .models import (
JdJobCommentRow,
JdJobDetailRow,
JdJobMergedRow,
JdJobSearchRow,
JdLeafCategoryNorm,
JdProduct,
JdProductSnapshot,
PipelineJob,
@ -119,6 +119,33 @@ def _bulk_create_in_chunks(model, objects: list[Any]) -> None:
model.objects.bulk_create(objects[i : i + BULK_CHUNK])
def _sync_search_rows_matrix_labels(
job: PipelineJob, merged_kw_list: list[tuple[int, dict[str, str]]]
) -> None:
"""按 SKU 将合并表解析出的报告细类回填到搜索行（与 §5 矩阵口径一致）。"""
sku_to_mg: dict[str, str] = {}
for _, kw in merged_kw_list:
sk = (kw.get("sku_id") or "").strip()
if not sk:
continue
mg = matrix_group_label_from_detail_path(kw.get("detail_category_path") or "")
if mg:
sku_to_mg[sk] = mg
if not sku_to_mg:
return
chunk: list[JdJobSearchRow] = []
for r in JdJobSearchRow.objects.filter(job=job).iterator(chunk_size=400):
sk = (r.sku_id or "").strip()
if sk and sk in sku_to_mg:
r.matrix_group_label = sku_to_mg[sk]
chunk.append(r)
if len(chunk) >= 400:
JdJobSearchRow.objects.bulk_update(chunk, ["matrix_group_label"])
chunk.clear()
if chunk:
JdJobSearchRow.objects.bulk_update(chunk, ["matrix_group_label"])
def _run_dir(job: PipelineJob) -> Path:
return Path(job.run_dir or "").expanduser().resolve()
@ -174,32 +201,12 @@ def ingest_job_dataset_rows(job: PipelineJob) -> dict[str, Any]:
if not search_rows and search_path.is_file() is False:
pass
search_kw_list: list[tuple[int, dict[str, str]]] = []
leaf_labels: set[str] = set()
for i, row in enumerate(search_rows):
_normalize_search_csv_total_sales(row)
kw = _search_row_kwargs(row)
search_kw_list.append((i, kw))
lc = (kw.get("leaf_category") or "").strip()[:512]
if lc:
leaf_labels.add(lc)
norm_map: dict[str, JdLeafCategoryNorm] = {}
if leaf_labels:
have = set(
JdLeafCategoryNorm.objects.filter(label__in=leaf_labels).values_list(
"label", flat=True
)
)
missing = [JdLeafCategoryNorm(label=l) for l in leaf_labels if l not in have]
if missing:
JdLeafCategoryNorm.objects.bulk_create(missing, ignore_conflicts=True)
norm_map = {
n.label: n
for n in JdLeafCategoryNorm.objects.filter(label__in=leaf_labels)
}
s_objs: list[JdJobSearchRow] = []
for i, kw in search_kw_list:
lc = (kw.get("leaf_category") or "").strip()[:512]
norm = norm_map.get(lc) if lc else None
pv = effective_list_price_value(
kw.get("coupon_price"), kw.get("price"), kw.get("original_price")
)
@ -207,7 +214,7 @@ def ingest_job_dataset_rows(job: PipelineJob) -> dict[str, Any]:
JdJobSearchRow(
job=job,
row_index=i,
leaf_category_norm=norm,
matrix_group_label="",
price_value=pv,
**kw,
)
@ -221,9 +228,14 @@ def ingest_job_dataset_rows(job: PipelineJob) -> dict[str, Any]:
for i, row in enumerate(detail_rows):
kw = _detail_row_kwargs(row)
dpv = float_price_from_cell(kw.get("detail_price_final"))
mg = matrix_group_label_from_detail_path(kw.get("detail_category_path") or "")
d_objs.append(
JdJobDetailRow(
job=job, row_index=i, detail_price_value=dpv, **kw
job=job,
row_index=i,
matrix_group_label=mg,
detail_price_value=dpv,
**kw,
)
)
_bulk_create_in_chunks(JdJobDetailRow, d_objs)
@ -241,34 +253,13 @@ def ingest_job_dataset_rows(job: PipelineJob) -> dict[str, Any]:
merged_path = run_dir / FILE_MERGED_CSV
merged_rows = _read_csv_rows(merged_path) if merged_path.is_file() else []
merged_kw_list: list[tuple[int, dict[str, str]]] = []
leaf_labels_m: set[str] = set()
for i, row in enumerate(merged_rows):
_normalize_merged_csv_total_sales(row)
kw = _merged_row_kwargs(row)
merged_kw_list.append((i, kw))
lc = (kw.get("leaf_category") or "").strip()[:512]
if lc:
leaf_labels_m.add(lc)
norm_map_m: dict[str, JdLeafCategoryNorm] = {}
if leaf_labels_m:
have_m = set(
JdLeafCategoryNorm.objects.filter(label__in=leaf_labels_m).values_list(
"label", flat=True
)
)
missing_m = [
JdLeafCategoryNorm(label=l) for l in leaf_labels_m if l not in have_m
]
if missing_m:
JdLeafCategoryNorm.objects.bulk_create(missing_m, ignore_conflicts=True)
norm_map_m = {
n.label: n
for n in JdLeafCategoryNorm.objects.filter(label__in=leaf_labels_m)
}
m_objs: list[JdJobMergedRow] = []
for i, kw in merged_kw_list:
lc = (kw.get("leaf_category") or "").strip()[:512]
norm = norm_map_m.get(lc) if lc else None
mg = matrix_group_label_from_detail_path(kw.get("detail_category_path") or "")
pv = effective_list_price_value(
kw.get("coupon_price"), kw.get("price"), kw.get("original_price")
)
@ -276,13 +267,14 @@ def ingest_job_dataset_rows(job: PipelineJob) -> dict[str, Any]:
JdJobMergedRow(
job=job,
row_index=i,
leaf_category_norm=norm,
matrix_group_label=mg,
price_value=pv,
**kw,
)
)
_bulk_create_in_chunks(JdJobMergedRow, m_objs)
stats["merged_table_rows"] = len(m_objs)
_sync_search_rows_matrix_labels(job, merged_kw_list)
return stats

63
backend/pipeline/matrix_group_label.py Normal file
View File

@ -0,0 +1,63 @@
"""
与 ``jd_competitor_report._matrix_group_label_from_path`` 同源：
从商详 ``detail_category_path`` 解析 §5 竞品矩阵用的细类展示名（如「饼干」「米」）。
"""
from __future__ import annotations
import re
def category_token_meaningless(seg: str) -> bool:
"""纯数字类目 ID、空串或疑似内部编码的段，不宜直接作为矩阵分组展示名。"""
t = (seg or "").strip()
if not t:
return True
if t.isdigit():
return True
if len(t) >= 14 and re.fullmatch(r"[A-Za-z0-9_\-]+", t):
return True
return False
def matrix_display_segment_from_parts(parts: list[str]) -> str | None:
"""
与历史逻辑一致的主选段；若该段无意义则自右向左找第一段可读文本
（避免「仅类目码」或中间段为数字 ID 时整组成品名式乱桶）。
"""
if not parts:
return None
if len(parts) >= 4:
preferred = parts[-2]
elif len(parts) >= 3:
preferred = parts[1]
elif len(parts) >= 2:
preferred = parts[1]
else:
preferred = parts[0]
order: list[str] = []
if preferred:
order.append(preferred)
if len(parts) >= 2:
order.append(parts[-2])
order.append(parts[-1])
order.extend(reversed(parts))
seen: set[str] = set()
for cand in order:
if not cand or cand in seen:
continue
seen.add(cand)
if not category_token_meaningless(cand):
return cand.strip()
return None
def matrix_group_label_from_detail_path(path: str) -> str:
"""由 ``detail_category_path`` 文本解析细类展示名；空或无可读段则返回空串。"""
t = (path or "").strip()
if not t:
return ""
parts = [p.strip() for p in t.replace("＞", ">").split(">") if p.strip()]
if not parts:
return ""
key = matrix_display_segment_from_parts(parts)
return (key[:80] if key else "")

92
backend/pipeline/migrations/0018_report_group_matrix_label.py Normal file
View File

@ -0,0 +1,92 @@
# Generated by Django 5.2.1 on 2026-04-16 01:55
from django.db import migrations, models
def backfill_matrix_group_labels(apps, schema_editor):
from pipeline.matrix_group_label import matrix_group_label_from_detail_path
JdJobDetailRow = apps.get_model("pipeline", "JdJobDetailRow")
JdJobMergedRow = apps.get_model("pipeline", "JdJobMergedRow")
JdJobSearchRow = apps.get_model("pipeline", "JdJobSearchRow")
chunk: list = []
for r in JdJobDetailRow.objects.all().iterator(chunk_size=400):
r.matrix_group_label = matrix_group_label_from_detail_path(
r.detail_category_path or ""
)
chunk.append(r)
if len(chunk) >= 400:
JdJobDetailRow.objects.bulk_update(chunk, ["matrix_group_label"])
chunk.clear()
if chunk:
JdJobDetailRow.objects.bulk_update(chunk, ["matrix_group_label"])
chunk = []
for r in JdJobMergedRow.objects.all().iterator(chunk_size=400):
r.matrix_group_label = matrix_group_label_from_detail_path(
r.detail_category_path or ""
)
chunk.append(r)
if len(chunk) >= 400:
JdJobMergedRow.objects.bulk_update(chunk, ["matrix_group_label"])
chunk.clear()
if chunk:
JdJobMergedRow.objects.bulk_update(chunk, ["matrix_group_label"])
sku_to_mg: dict[str, str] = {}
for r in JdJobMergedRow.objects.exclude(matrix_group_label="").iterator(
chunk_size=400
):
sk = str(r.sku_id or "").strip()
if sk:
sku_to_mg[sk] = r.matrix_group_label
chunk = []
for r in JdJobSearchRow.objects.all().iterator(chunk_size=400):
sk = str(r.sku_id or "").strip()
r.matrix_group_label = sku_to_mg.get(sk, "")
chunk.append(r)
if len(chunk) >= 400:
JdJobSearchRow.objects.bulk_update(chunk, ["matrix_group_label"])
chunk.clear()
if chunk:
JdJobSearchRow.objects.bulk_update(chunk, ["matrix_group_label"])
class Migration(migrations.Migration):
dependencies = [
('pipeline', '0017_dataset_browse_filters'),
]
operations = [
migrations.AddField(
model_name='jdjobdetailrow',
name='matrix_group_label',
field=models.CharField(blank=True, db_index=True, default='', help_text='与 §5 矩阵同源：由 detail_category_path 解析', max_length=80, verbose_name='报告细类'),
),
migrations.AddField(
model_name='jdjobmergedrow',
name='matrix_group_label',
field=models.CharField(blank=True, db_index=True, default='', help_text='与 §5 矩阵同源：由 detail_category_path 解析', max_length=80, verbose_name='报告细类'),
),
migrations.AddField(
model_name='jdjobsearchrow',
name='matrix_group_label',
field=models.CharField(blank=True, db_index=True, default='', help_text='与 §5 矩阵同源：由合并表商详路径解析；可按 SKU 从合并表回填', max_length=80, verbose_name='报告细类'),
),
migrations.AddIndex(
model_name='jdjobdetailrow',
index=models.Index(fields=['job', 'matrix_group_label'], name='pipeline_jd_job_id_5595d3_idx'),
),
migrations.AddIndex(
model_name='jdjobmergedrow',
index=models.Index(fields=['job', 'matrix_group_label'], name='pipeline_jd_job_id_163e3f_idx'),
),
migrations.AddIndex(
model_name='jdjobsearchrow',
index=models.Index(fields=['job', 'matrix_group_label'], name='pipeline_jd_job_id_38fae5_idx'),
),
migrations.RunPython(backfill_matrix_group_labels, migrations.RunPython.noop),
]

27
backend/pipeline/models.py
View File

@ -190,6 +190,14 @@ class JdJobSearchRow(models.Model):
on_delete=models.SET_NULL,
related_name="search_rows",
)
matrix_group_label = models.CharField(
max_length=80,
blank=True,
default="",
db_index=True,
verbose_name="报告细类",
help_text="与 §5 矩阵同源：由合并表商详路径解析；可按 SKU 从合并表回填",
)
price_value = models.FloatField(null=True, blank=True, db_index=True)
platform = models.TextField(blank=True, default="")
keyword = models.TextField(blank=True, default="")
@ -206,6 +214,7 @@ class JdJobSearchRow(models.Model):
indexes = [
models.Index(fields=["job", "sku_id"]),
models.Index(fields=["job", "leaf_category_norm"]),
models.Index(fields=["job", "matrix_group_label"]),
models.Index(fields=["job", "price_value"]),
]
@ -232,6 +241,14 @@ class JdJobDetailRow(models.Model):
buyer_ranking_line = models.TextField(blank=True, default="")
buyer_promo_text = models.TextField(blank=True, default="")
detail_price_value = models.FloatField(null=True, blank=True, db_index=True)
matrix_group_label = models.CharField(
max_length=80,
blank=True,
default="",
db_index=True,
verbose_name="报告细类",
help_text="与 §5 矩阵同源：由 detail_category_path 解析",
)
class Meta:
ordering = ["row_index"]
@ -244,6 +261,7 @@ class JdJobDetailRow(models.Model):
indexes = [
models.Index(fields=["job", "sku_id"]),
models.Index(fields=["job", "detail_price_value"]),
models.Index(fields=["job", "matrix_group_label"]),
]
def __str__(self) -> str:
@ -317,6 +335,14 @@ class JdJobMergedRow(models.Model):
on_delete=models.SET_NULL,
related_name="merged_rows",
)
matrix_group_label = models.CharField(
max_length=80,
blank=True,
default="",
db_index=True,
verbose_name="报告细类",
help_text="与 §5 矩阵同源：由 detail_category_path 解析",
)
price_value = models.FloatField(null=True, blank=True, db_index=True)
keyword = models.TextField(blank=True, default="")
page = models.TextField(blank=True, default="")
@ -342,6 +368,7 @@ class JdJobMergedRow(models.Model):
indexes = [
models.Index(fields=["job", "sku_id"]),
models.Index(fields=["job", "leaf_category_norm"]),
models.Index(fields=["job", "matrix_group_label"]),
models.Index(fields=["job", "price_value"]),
]

3
backend/pipeline/row_serialize.py
View File

@ -22,6 +22,7 @@ def search_row_to_dict(r: JdJobSearchRow) -> dict[str, Any]:
out: dict[str, Any] = {"id": r.id, "row_index": r.row_index}
for k in JD_SEARCH_INTERNAL_KEYS:
out[k] = getattr(r, k) or ""
out["matrix_group_label"] = r.matrix_group_label or ""
return out
@ -29,6 +30,7 @@ def detail_row_to_dict(r: JdJobDetailRow) -> dict[str, Any]:
out: dict[str, Any] = {"id": r.id, "row_index": r.row_index}
for k in DETAIL_FIELDS_ORDER:
out[k] = getattr(r, k) or ""
out["matrix_group_label"] = r.matrix_group_label or ""
return out
@ -43,4 +45,5 @@ def merged_row_to_dict(r: JdJobMergedRow) -> dict[str, Any]:
out: dict[str, Any] = {"id": r.id, "row_index": r.row_index}
for k in MERGED_FIELDS_ORDER:
out[k] = getattr(r, k) or ""
out["matrix_group_label"] = r.matrix_group_label or ""
return out

38
backend/pipeline/views.py
View File

@ -27,11 +27,11 @@ from .dataset_api import (
apply_merged_order,
apply_search_filters,
apply_search_order,
category_norm_id_from_request,
detail_category_q_from_request,
filter_echo,
parse_sort_meta,
price_bounds_from_request,
report_group_from_request,
)
from .dataset_nonempty import (
comment_columns_for_api,
@ -66,7 +66,6 @@ from .models import (
JdJobDetailRow,
JdJobMergedRow,
JdJobSearchRow,
JdLeafCategoryNorm,
JdProduct,
JdProductSnapshot,
JobStatus,
@ -756,17 +755,15 @@ def _read_page_params(request) -> tuple[int, int]:
return page, page_size
def _category_norm_options_for_job(job: PipelineJob, RowModel: type) -> list[dict[str, Any]]:
ids = (
RowModel.objects.filter(job=job, leaf_category_norm_id__isnull=False)
.values_list("leaf_category_norm_id", flat=True)
def _report_group_options_for_job(job: PipelineJob) -> list[str]:
"""与 §5 矩阵一致的细类名列表（来自合并表 ``detail_category_path`` 解析）。"""
qs = (
JdJobMergedRow.objects.filter(job=job)
.exclude(matrix_group_label="")
.values_list("matrix_group_label", flat=True)
.distinct()
)
return list(
JdLeafCategoryNorm.objects.filter(id__in=ids)
.order_by("label")
.values("id", "label")
)
return sorted({str(x) for x in qs if x})
def _detail_category_path_options(job: PipelineJob) -> list[str]:
@ -797,12 +794,7 @@ class JobDatasetSummaryView(APIView):
"detail_columns": detail_columns_for_api(job),
"comment_columns": comment_columns_for_api(job),
"merged_columns": merged_columns_for_api(job),
"search_category_options": _category_norm_options_for_job(
job, JdJobSearchRow
),
"merged_category_options": _category_norm_options_for_job(
job, JdJobMergedRow
),
"report_group_options": _report_group_options_for_job(job),
"detail_category_path_options": _detail_category_path_options(job),
"dataset_sort_help": {
"search": sorted(SEARCH_SORT_FIELDS),
@ -820,7 +812,7 @@ class JobDatasetSearchView(APIView):
page, page_size = _read_page_params(request)
sort, desc = parse_sort_meta(request)
sort_eff = sort if sort in SEARCH_SORT_FIELDS else "row_index"
cid = category_norm_id_from_request(request)
rg = report_group_from_request(request)
pmin, pmax = price_bounds_from_request(request)
dcq = detail_category_q_from_request(request)
qs = JdJobSearchRow.objects.filter(job=job)
@ -835,7 +827,7 @@ class JobDatasetSearchView(APIView):
"page": page,
"page_size": page_size,
"filters": filter_echo(
category_norm_id=cid,
report_group=rg,
price_min=pmin,
price_max=pmax,
detail_category_q=dcq,
@ -853,7 +845,7 @@ class JobDatasetDetailView(APIView):
page, page_size = _read_page_params(request)
sort, desc = parse_sort_meta(request)
sort_eff = sort if sort in DETAIL_SORT_FIELDS else "row_index"
cid = category_norm_id_from_request(request)
rg = report_group_from_request(request)
pmin, pmax = price_bounds_from_request(request)
dcq = detail_category_q_from_request(request)
qs = JdJobDetailRow.objects.filter(job=job)
@ -868,7 +860,7 @@ class JobDatasetDetailView(APIView):
"page": page,
"page_size": page_size,
"filters": filter_echo(
category_norm_id=cid,
report_group=rg,
price_min=pmin,
price_max=pmax,
detail_category_q=dcq,
@ -908,7 +900,7 @@ class JobDatasetMergedView(APIView):
page, page_size = _read_page_params(request)
sort, desc = parse_sort_meta(request)
sort_eff = sort if sort in MERGED_SORT_FIELDS else "row_index"
cid = category_norm_id_from_request(request)
rg = report_group_from_request(request)
pmin, pmax = price_bounds_from_request(request)
dcq = detail_category_q_from_request(request)
qs = JdJobMergedRow.objects.filter(job=job)
@ -923,7 +915,7 @@ class JobDatasetMergedView(APIView):
"page": page,
"page_size": page_size,
"filters": filter_echo(
category_norm_id=cid,
report_group=rg,
price_min=pmin,
price_max=pmax,
detail_category_q=dcq,

39
frontend/src/components/JobDatasetModal.vue
View File

@ -24,6 +24,7 @@ const SORT_LABELS = {
sku_id: 'SKU',
title: '标题',
leaf_category: '叶类目',
matrix_group_label: '报告细类',
detail_category_path: '类目路径',
detail_brand: '品牌',
}
@ -39,7 +40,8 @@ const err = ref('')
const commentSkuFilter = ref('')
const sortField = ref('row_index')
const sortOrder = ref('asc')
const categoryNormId = ref('')
/** 与 §5 矩阵一致的细类名（如饼干、米），对应接口参数 report_group */
const reportGroup = ref('')
const priceMin = ref('')
const priceMax = ref('')
const detailCategoryQ = ref('')
@ -68,12 +70,7 @@ const sortOptions = computed(() => {
return keys.map((k) => ({ value: k, label: SORT_LABELS[k] || k }))
})
const categoryNormOptions = computed(() => {
if (!summary.value) return []
if (tab.value === 'search') return summary.value.search_category_options || []
if (tab.value === 'merged') return summary.value.merged_category_options || []
return []
})
const reportGroupOptions = computed(() => summary.value?.report_group_options || [])
const displayColumns = computed(() => {
const s = summary.value
@ -164,7 +161,7 @@ async function refreshList() {
: {
sort: sortField.value,
order: sortOrder.value,
categoryNormId: categoryNormId.value.trim(),
reportGroup: reportGroup.value.trim(),
priceMin: priceMin.value,
priceMax: priceMax.value,
detailCategoryQ: detailCategoryQ.value.trim(),
@ -198,7 +195,7 @@ watch(
page.value = 1
sortField.value = 'row_index'
sortOrder.value = 'asc'
categoryNormId.value = ''
reportGroup.value = ''
priceMin.value = ''
priceMax.value = ''
detailCategoryQ.value = ''
@ -216,7 +213,7 @@ watch(tab, () => {
exportPanelOpen.value = false
sortField.value = 'row_index'
sortOrder.value = 'asc'
categoryNormId.value = ''
reportGroup.value = ''
priceMin.value = ''
priceMax.value = ''
detailCategoryQ.value = ''
@ -226,7 +223,7 @@ watch(
[
sortField,
sortOrder,
categoryNormId,
reportGroup,
priceMin,
priceMax,
detailCategoryQ,
@ -245,7 +242,7 @@ watch(
commentSkuFilter,
sortField,
sortOrder,
categoryNormId,
reportGroup,
priceMin,
priceMax,
detailCategoryQ,
@ -430,17 +427,13 @@ async function runExport(format) {
<option value="desc">降序</option>
</select>
</label>
<template v-if="tab === 'search' || tab === 'merged'">
<label class="filter-item">
叶类目
<select v-model="categoryNormId" class="filter-select wide">
<option value="">全部</option>
<option v-for="c in categoryNormOptions" :key="c.id" :value="String(c.id)">
{{ c.label }}
</option>
</select>
</label>
</template>
<label class="filter-item">
报告细类（与矩阵一致）
<select v-model="reportGroup" class="filter-select wide">
<option value="">全部</option>
<option v-for="g in reportGroupOptions" :key="g" :value="g">{{ g }}</option>
</select>
</label>
<template v-if="tab === 'detail' || tab === 'merged'">
<label class="filter-item">
类目路径包含

7
frontend/src/composables/useJobs.js
View File

@ -205,9 +205,10 @@ export function jobDatasetPageUrl(jobId, kind, page = 1, pageSize = 50, opts = {
if (sku) p.set('sku_id', String(sku))
if (o.sort) p.set('sort', String(o.sort))
if (o.order) p.set('order', String(o.order))
const cn = o.categoryNormId ?? o.category_norm_id
if (cn !== undefined && cn !== null && String(cn).trim() !== '')
p.set('category_norm_id', String(cn).trim())
const rg =
o.reportGroup ?? o.report_group ?? o.categoryNormId ?? o.category_norm_id
if (rg !== undefined && rg !== null && String(rg).trim() !== '')
p.set('report_group', String(rg).trim())
const pmin = o.priceMin ?? o.price_min
if (pmin !== undefined && pmin !== null && String(pmin).trim() !== '')
p.set('price_min', String(pmin).trim())