mirror of
https://github.com/primedigitaltech/market-assistant.git
synced 2026-07-22 08:01:34 +08:00
6280e436d8
- 统一 csv_schema 与搜索/合并/评价/商详导出为纯中文列名，入库与视图按中文表头解析 - 竞品分析报告兼容新旧表头；新增 csv_header_rewrite 与 rewrite_pipeline_csv_headers 管理命令 - 调整 pipeline 至 jd、llm、reporting、demos 子包并更新任务与测试引用 - 新增购买者优惠摘要抽取、合并表 regen/ingest 命令、0016 迁移及相关测试 Made-with: Cursor
237 lines
7.4 KiB
Python
237 lines
7.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
将历史 JD 流水线 CSV 表头规范为 ``csv_schema`` 中的纯中文表头（仅重命名与列序，不改单元格内容逻辑）。
|
||
|
||
用于已落盘的 ``pipeline_runs/...`` 目录；新跑批次由爬虫直接写出新表头。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
|
||
from .csv_schema import (
|
||
COMMENT_CSV_COLUMNS,
|
||
COMMENT_ROW_DICT_KEYS,
|
||
DETAIL_CSV_COLUMNS,
|
||
JD_SEARCH_CSV_HEADERS,
|
||
JD_SEARCH_INTERNAL_KEYS,
|
||
MERGED_CSV_COLUMNS,
|
||
MERGED_LEAN_DETAIL_INTERNAL_KEYS,
|
||
LEAN_DETAIL_CSV_HEADERS,
|
||
)
|
||
|
||
|
||
def _search_fieldnames_canonical() -> list[str]:
|
||
return [JD_SEARCH_CSV_HEADERS[k] for k in JD_SEARCH_INTERNAL_KEYS]
|
||
|
||
|
||
def _merged_fieldnames_canonical() -> list[str]:
|
||
return list(MERGED_CSV_COLUMNS)
|
||
|
||
|
||
def _detail_fieldnames_canonical() -> list[str]:
|
||
return list(DETAIL_CSV_COLUMNS)
|
||
|
||
|
||
def _comment_fieldnames_canonical() -> list[str]:
|
||
return list(COMMENT_CSV_COLUMNS)
|
||
|
||
|
||
def build_legacy_header_map() -> dict[str, str]:
|
||
"""
|
||
旧表头字符串 → 新表头（纯中文或与 schema 一致）。
|
||
|
||
覆盖：带英文括号的搜索/合并表、英文 ``pipeline_keyword`` / ``detail_*``、
|
||
评价 CSV 的接口字段名、商详 ``skuId`` 等。
|
||
"""
|
||
m: dict[str, str] = {}
|
||
|
||
# --- 合并表 / pc_search 曾用的「括号英文」列名 ---
|
||
legacy_search_pairs: tuple[tuple[str, str], ...] = (
|
||
("主商品ID(wareId)", JD_SEARCH_CSV_HEADERS["item_id"]),
|
||
("SKU(skuId)", JD_SEARCH_CSV_HEADERS["sku_id"]),
|
||
("标题(wareName)", JD_SEARCH_CSV_HEADERS["title"]),
|
||
(
|
||
"标价(jdPrice,jdPriceText,realPrice)",
|
||
JD_SEARCH_CSV_HEADERS["price"],
|
||
),
|
||
(
|
||
"券后到手价(couponPrice,subsidyPrice,finalPrice.estimatedPrice,priceShow)",
|
||
JD_SEARCH_CSV_HEADERS["coupon_price"],
|
||
),
|
||
(
|
||
"原价(oriPrice,originalPrice,marketPrice)",
|
||
JD_SEARCH_CSV_HEADERS["original_price"],
|
||
),
|
||
("卖点(sellingPoint)", JD_SEARCH_CSV_HEADERS["selling_point"]),
|
||
("销量楼层(commentSalesFloor)", JD_SEARCH_CSV_HEADERS["comment_sales_floor"]),
|
||
("销量口径(totalSales)", JD_SEARCH_CSV_HEADERS["total_sales"]),
|
||
(
|
||
"榜单类文案(标签/腰带/标题数组中的榜、TOP 等)",
|
||
JD_SEARCH_CSV_HEADERS["hot_list_rank"],
|
||
),
|
||
("评价量(commentFuzzy)", JD_SEARCH_CSV_HEADERS["comment_count"]),
|
||
("店铺名(shopName)", JD_SEARCH_CSV_HEADERS["shop_name"]),
|
||
("店铺链接(shopUrl,shopId)", JD_SEARCH_CSV_HEADERS["shop_url"]),
|
||
(
|
||
"店铺信息链接(shopInfoUrl,brandUrl)",
|
||
JD_SEARCH_CSV_HEADERS["shop_info_url"],
|
||
),
|
||
("地域(deliveryAddress,area,procity)", JD_SEARCH_CSV_HEADERS["location"]),
|
||
(
|
||
"商品链接(toUrl,clickUrl,item.m.jd.com)",
|
||
JD_SEARCH_CSV_HEADERS["detail_url"],
|
||
),
|
||
("主图(imageurl,imageUrl)", JD_SEARCH_CSV_HEADERS["image"]),
|
||
("秒杀(seckillInfo,secKill)", JD_SEARCH_CSV_HEADERS["seckill_info"]),
|
||
(
|
||
"规格属性(propertyList,color,catid,shortName)",
|
||
JD_SEARCH_CSV_HEADERS["attributes"],
|
||
),
|
||
("类目(leafCategory,cid3Name,catid)", JD_SEARCH_CSV_HEADERS["leaf_category"]),
|
||
("平台(platform)", JD_SEARCH_CSV_HEADERS["platform"]),
|
||
("搜索词(keyword)", JD_SEARCH_CSV_HEADERS["keyword"]),
|
||
("页码(page)", JD_SEARCH_CSV_HEADERS["page"]),
|
||
)
|
||
for old, new in legacy_search_pairs:
|
||
m[old] = new
|
||
|
||
# 合并表首列
|
||
m["pipeline_keyword"] = "流水线关键词"
|
||
|
||
# 商详块：历史合并表曾直接写英文内部键
|
||
for ik, zh in zip(MERGED_LEAN_DETAIL_INTERNAL_KEYS, LEAN_DETAIL_CSV_HEADERS):
|
||
m[ik] = zh
|
||
m["comment_count"] = "评论条数"
|
||
m["comment_preview"] = "评价摘要"
|
||
|
||
# --- comments_flat：接口字段 / 小写 sku ---
|
||
for api_k, zh in zip(COMMENT_ROW_DICT_KEYS, COMMENT_CSV_COLUMNS):
|
||
m[api_k] = zh
|
||
m["sku"] = "SKU"
|
||
|
||
# --- detail_ware_export ---
|
||
m["skuId"] = "SKU"
|
||
|
||
return m
|
||
|
||
|
||
LEGACY_HEADER_MAP: dict[str, str] = build_legacy_header_map()
|
||
|
||
|
||
def _normalize_field_key(k: str | None) -> str:
|
||
return (k or "").strip().lstrip("\ufeff")
|
||
|
||
|
||
def _row_with_canonical_keys(
|
||
row: dict[str, str],
|
||
legacy_map: dict[str, str],
|
||
) -> dict[str, str]:
|
||
"""将一行从任意旧表头映射为 canonical 列名；同列多旧键时取非空优先。"""
|
||
out: dict[str, str] = {}
|
||
for raw_k, v in row.items():
|
||
k = _normalize_field_key(raw_k)
|
||
if not k:
|
||
continue
|
||
nk = legacy_map.get(k, k)
|
||
sv = "" if v is None else str(v)
|
||
if nk not in out or (sv.strip() and not str(out.get(nk, "")).strip()):
|
||
out[nk] = sv
|
||
return out
|
||
|
||
|
||
def rewrite_csv_inplace(
|
||
path: Path,
|
||
fieldnames: list[str],
|
||
legacy_map: dict[str, str],
|
||
*,
|
||
dry_run: bool = False,
|
||
) -> tuple[bool, str]:
|
||
"""
|
||
读 UTF-8 BOM CSV，按 ``fieldnames`` 写出（缺列填空）。
|
||
|
||
返回 (是否执行写入, 说明)。
|
||
"""
|
||
path = path.expanduser().resolve()
|
||
if not path.is_file():
|
||
return False, f"跳过（不存在）: {path}"
|
||
|
||
with path.open(encoding="utf-8-sig", newline="") as f:
|
||
rdr = csv.DictReader(f)
|
||
rows_in = list(rdr)
|
||
|
||
if not rows_in:
|
||
return False, f"空文件: {path}"
|
||
|
||
rows_out: list[dict[str, str]] = []
|
||
extras: set[str] = set()
|
||
for row in rows_in:
|
||
canon = _row_with_canonical_keys(row, legacy_map)
|
||
for k in canon:
|
||
if k not in fieldnames:
|
||
extras.add(k)
|
||
rows_out.append(canon)
|
||
|
||
fieldnames_out = list(fieldnames) + sorted(extras)
|
||
|
||
if dry_run:
|
||
return True, f"[dry-run] 将写 {len(rows_out)} 行 -> {path.name}"
|
||
|
||
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
||
w = csv.DictWriter(
|
||
f,
|
||
fieldnames=fieldnames_out,
|
||
extrasaction="ignore",
|
||
)
|
||
w.writeheader()
|
||
w.writerows(
|
||
[{fn: str(r.get(fn, "") or "") for fn in fieldnames_out} for r in rows_out]
|
||
)
|
||
|
||
return True, f"已写 {len(rows_out)} 行 -> {path}"
|
||
|
||
|
||
def rewrite_run_dir_csv_headers(
|
||
run_dir: Path,
|
||
*,
|
||
dry_run: bool = False,
|
||
only: Iterable[str] | None = None,
|
||
) -> list[str]:
|
||
"""
|
||
处理单个 run 目录下四种 CSV（存在则处理）。
|
||
|
||
``only`` 为文件名子集，例如 ``("keyword_pipeline_merged.csv",)``。
|
||
"""
|
||
from .ingest import (
|
||
FILE_COMMENTS_FLAT_CSV,
|
||
FILE_DETAIL_WARE_CSV,
|
||
FILE_MERGED_CSV,
|
||
FILE_PC_SEARCH_CSV,
|
||
)
|
||
|
||
run_dir = run_dir.expanduser().resolve()
|
||
lm = LEGACY_HEADER_MAP
|
||
only_set = {x.strip() for x in only} if only else None
|
||
|
||
tasks: list[tuple[str, list[str]]] = [
|
||
(FILE_MERGED_CSV, _merged_fieldnames_canonical()),
|
||
(FILE_PC_SEARCH_CSV, _search_fieldnames_canonical()),
|
||
(FILE_COMMENTS_FLAT_CSV, _comment_fieldnames_canonical()),
|
||
(FILE_DETAIL_WARE_CSV, _detail_fieldnames_canonical()),
|
||
]
|
||
|
||
messages: list[str] = []
|
||
for fname, fieldnames in tasks:
|
||
if only_set is not None and fname not in only_set:
|
||
continue
|
||
ok, msg = rewrite_csv_inplace(
|
||
run_dir / fname,
|
||
fieldnames,
|
||
lm,
|
||
dry_run=dry_run,
|
||
)
|
||
if ok or "空文件" in msg or "不存在" in msg:
|
||
messages.append(msg)
|
||
return messages
|