mirror of
https://github.com/primedigitaltech/market-assistant.git
synced 2026-07-21 23:41:39 +08:00
refactor(crawler): 抽取 competitor_report 常量与 CSV 辅助
- 新增 jd_pc_search/competitor_report：constants、config（resolve_report_tuning）、csv_io。 - jd_competitor_report 通过 import * 保持原有符号；KEYWORD/EXISTING_RUN_DIR/OVERRIDE 仍在本脚本修改。 - 修正 constants.__all__ 中评价列名为 _COMMENT_CSV_*，避免 star import 漏绑。 Made-with: Cursor
This commit is contained in:
hub-gif
parent
2a27a611ca
commit
4687bcb907
@ -0,0 +1 @@
|
||||
"""竞品报告脚本共享：表头常量、CSV 辅助、报告调参解析。"""
|
||||
109
backend/crawler_copy/jd_pc_search/competitor_report/config.py
Normal file
109
backend/crawler_copy/jd_pc_search/competitor_report/config.py
Normal file
@ -0,0 +1,109 @@
|
||||
"""``report_config`` JSON → 关注词、场景组、外部市场表行。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from .constants import (
|
||||
COMMENT_FOCUS_WORDS,
|
||||
COMMENT_SCENARIO_GROUPS,
|
||||
EXTERNAL_MARKET_TABLE_ROWS,
|
||||
)
|
||||
|
||||
|
||||
def _normalize_focus_words(raw: Any) -> tuple[str, ...]:
|
||||
if not isinstance(raw, list) or not raw:
|
||||
return COMMENT_FOCUS_WORDS
|
||||
out: list[str] = []
|
||||
for x in raw[:120]:
|
||||
s = str(x).strip()
|
||||
if len(s) > 48:
|
||||
s = s[:48]
|
||||
if s:
|
||||
out.append(s)
|
||||
return tuple(out) if out else COMMENT_FOCUS_WORDS
|
||||
|
||||
|
||||
def _normalize_scenario_groups(
|
||||
raw: Any,
|
||||
) -> tuple[tuple[str, tuple[str, ...]], ...]:
|
||||
if not isinstance(raw, list) or not raw:
|
||||
return COMMENT_SCENARIO_GROUPS
|
||||
parsed: list[tuple[str, tuple[str, ...]]] = []
|
||||
for item in raw[:40]:
|
||||
label = ""
|
||||
triggers: list[str] = []
|
||||
if isinstance(item, dict):
|
||||
label = str(item.get("label") or "").strip()[:80]
|
||||
tr = item.get("triggers")
|
||||
if isinstance(tr, list):
|
||||
for t in tr[:48]:
|
||||
s = str(t).strip()
|
||||
if len(s) > 48:
|
||||
s = s[:48]
|
||||
if s:
|
||||
triggers.append(s)
|
||||
elif isinstance(item, (list, tuple)) and len(item) >= 2:
|
||||
label = str(item[0]).strip()[:80]
|
||||
tr = item[1]
|
||||
if isinstance(tr, (list, tuple)):
|
||||
for t in tr[:48]:
|
||||
s = str(t).strip()
|
||||
if len(s) > 48:
|
||||
s = s[:48]
|
||||
if s:
|
||||
triggers.append(s)
|
||||
if label and triggers:
|
||||
parsed.append((label, tuple(triggers)))
|
||||
return tuple(parsed) if parsed else COMMENT_SCENARIO_GROUPS
|
||||
|
||||
|
||||
def _normalize_external_market_rows(
|
||||
raw: Any,
|
||||
) -> tuple[tuple[str, str, str, str], ...]:
|
||||
if not isinstance(raw, list) or not raw:
|
||||
return EXTERNAL_MARKET_TABLE_ROWS
|
||||
rows: list[tuple[str, str, str, str]] = []
|
||||
|
||||
def _four_cells(x: Any) -> tuple[str, str, str, str] | None:
|
||||
if isinstance(x, (list, tuple)) and len(x) >= 4:
|
||||
return tuple(str(c)[:500] for c in x[:4])
|
||||
if isinstance(x, dict):
|
||||
a = str(x.get("indicator") or x.get("a") or "").strip()[:500]
|
||||
b = str(x.get("value_and_scope") or x.get("b") or "").strip()[:500]
|
||||
c = str(x.get("source") or x.get("c") or "").strip()[:500]
|
||||
d = str(x.get("year") or x.get("d") or "").strip()[:500]
|
||||
if any((a, b, c, d)):
|
||||
return (a, b, c, d)
|
||||
return None
|
||||
|
||||
for item in raw[:24]:
|
||||
r = _four_cells(item)
|
||||
if r:
|
||||
rows.append(r)
|
||||
return tuple(rows) if rows else EXTERNAL_MARKET_TABLE_ROWS
|
||||
|
||||
|
||||
def resolve_report_tuning(
|
||||
report_config: dict[str, Any] | None,
|
||||
) -> tuple[
|
||||
tuple[str, ...],
|
||||
tuple[tuple[str, tuple[str, ...]], ...],
|
||||
tuple[tuple[str, str, str, str], ...],
|
||||
]:
|
||||
if not report_config:
|
||||
return COMMENT_FOCUS_WORDS, COMMENT_SCENARIO_GROUPS, EXTERNAL_MARKET_TABLE_ROWS
|
||||
return (
|
||||
_normalize_focus_words(report_config.get("comment_focus_words")),
|
||||
_normalize_scenario_groups(report_config.get("comment_scenario_groups")),
|
||||
_normalize_external_market_rows(
|
||||
report_config.get("external_market_table_rows")
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"resolve_report_tuning",
|
||||
"_normalize_external_market_rows",
|
||||
"_normalize_focus_words",
|
||||
"_normalize_scenario_groups",
|
||||
]
|
||||
141
backend/crawler_copy/jd_pc_search/competitor_report/constants.py
Normal file
141
backend/crawler_copy/jd_pc_search/competitor_report/constants.py
Normal file
@ -0,0 +1,141 @@
|
||||
"""CSV 表头键、运行默认调参与关注词/场景配置（与 ``jd_competitor_report`` 顶层一致）。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pipeline.csv_schema import (
|
||||
COMMENT_CSV_COLUMNS,
|
||||
JD_SEARCH_CSV_HEADERS,
|
||||
MERGED_FIELD_TO_CSV_HEADER,
|
||||
)
|
||||
|
||||
_JD_LIST_PRICE_KEY = JD_SEARCH_CSV_HEADERS["price"]
|
||||
_COUPON_SHOW_PRICE_KEY = JD_SEARCH_CSV_HEADERS["coupon_price"]
|
||||
_ORIGINAL_LIST_PRICE_KEY = JD_SEARCH_CSV_HEADERS["original_price"]
|
||||
_SELLING_POINT_KEY = JD_SEARCH_CSV_HEADERS["selling_point"]
|
||||
_RANK_TAGLINE_KEY = JD_SEARCH_CSV_HEADERS["hot_list_rank"]
|
||||
|
||||
# 历史批次 CSV 表头（括号英文）；新批次为纯中文，读取时新键优先
|
||||
_LEGACY_JD_LIST_PRICE_KEY = "标价(jdPrice,jdPriceText,realPrice)"
|
||||
_LEGACY_COUPON_SHOW_PRICE_KEY = (
|
||||
"券后到手价(couponPrice,subsidyPrice,finalPrice.estimatedPrice,priceShow)"
|
||||
)
|
||||
_LEGACY_SHOP_NAME_KEY = "店铺名(shopName)"
|
||||
_LEGACY_RANK_TAGLINE_KEY = "榜单类文案(标签/腰带/标题数组中的榜、TOP 等)"
|
||||
_LEGACY_COMMENT_FUZZ_KEY = "评价量(commentFuzzy)"
|
||||
_LEGACY_SELLING_POINT_KEY = "卖点(sellingPoint)"
|
||||
_LIST_BRAND_TITLE_HEADER = "店铺信息标题"
|
||||
_LEGACY_LIST_BRAND_TITLE_KEY = "店铺信息标题(shopInfoTitle,brandName)"
|
||||
|
||||
_DETAIL_PRICE_FINAL_CSV_KEYS: tuple[str, ...] = (
|
||||
MERGED_FIELD_TO_CSV_HEADER["detail_price_final"],
|
||||
"detail_price_final",
|
||||
)
|
||||
_LIST_PRICE_AND_COUPON_KEYS: tuple[str, ...] = (
|
||||
*_DETAIL_PRICE_FINAL_CSV_KEYS,
|
||||
_JD_LIST_PRICE_KEY,
|
||||
_LEGACY_JD_LIST_PRICE_KEY,
|
||||
_COUPON_SHOW_PRICE_KEY,
|
||||
_LEGACY_COUPON_SHOW_PRICE_KEY,
|
||||
)
|
||||
|
||||
# 报告摘录「标价」：列表标价优先，缺省时用商详到手价列兜底
|
||||
_LIST_SHOW_PRICE_CELL_KEYS: tuple[str, ...] = (
|
||||
_JD_LIST_PRICE_KEY,
|
||||
_LEGACY_JD_LIST_PRICE_KEY,
|
||||
MERGED_FIELD_TO_CSV_HEADER["detail_price_final"],
|
||||
"detail_price_final",
|
||||
)
|
||||
|
||||
_MERGED_SHOP_CELL_KEYS: tuple[str, ...] = (
|
||||
MERGED_FIELD_TO_CSV_HEADER["detail_shop_name"],
|
||||
"detail_shop_name",
|
||||
JD_SEARCH_CSV_HEADERS["shop_name"],
|
||||
_LEGACY_SHOP_NAME_KEY,
|
||||
)
|
||||
|
||||
_COMMENT_FUZZ_KEYS: tuple[str, ...] = (
|
||||
MERGED_FIELD_TO_CSV_HEADER["comment_fuzzy"],
|
||||
_LEGACY_COMMENT_FUZZ_KEY,
|
||||
)
|
||||
|
||||
_COMMENT_CSV_SKU = COMMENT_CSV_COLUMNS[0]
|
||||
_COMMENT_CSV_BODY = COMMENT_CSV_COLUMNS[3]
|
||||
_COMMENT_CSV_SCORE = COMMENT_CSV_COLUMNS[7] # 「评分」→ commentScore
|
||||
|
||||
# 评价星级与 §8.2 分桶：先按评分筛正负，再在对应子集内统计口语短语（无评分时回退关键词）
|
||||
_COMMENT_SCORE_NEG_MAX = 2 # 1～2 星 → 偏负向
|
||||
_COMMENT_SCORE_POS_MIN = 4 # 4～5 星 → 偏正向（3 星为中评，归入中性）
|
||||
|
||||
_DETAIL_CATEGORY_PATH_KEY = MERGED_FIELD_TO_CSV_HEADER["detail_category_path"]
|
||||
_K_CAT_COL = JD_SEARCH_CSV_HEADERS["leaf_category"]
|
||||
_K_PROP_COL = JD_SEARCH_CSV_HEADERS["attributes"]
|
||||
|
||||
EXTERNAL_MARKET_TABLE_ROWS: tuple[tuple[str, str, str, str], ...] = ()
|
||||
|
||||
COMMENT_FOCUS_WORDS: tuple[str, ...] = (
|
||||
"口感",
|
||||
"甜",
|
||||
"糖",
|
||||
"血糖",
|
||||
"控糖",
|
||||
"低糖",
|
||||
"无糖",
|
||||
"饱腹",
|
||||
"升糖",
|
||||
"GI",
|
||||
"gi",
|
||||
"孕妇",
|
||||
"老人",
|
||||
"糖尿病",
|
||||
"价格",
|
||||
"贵",
|
||||
"便宜",
|
||||
"回购",
|
||||
"包装",
|
||||
"物流",
|
||||
"分量",
|
||||
"量少",
|
||||
"克重",
|
||||
)
|
||||
|
||||
COMMENT_SCENARIO_GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||||
("早餐/代餐", ("早餐", "代餐", "早饭", "当早餐", "当早饭", "早上吃", "晨起")),
|
||||
("零食/加餐/解馋", ("零食", "加餐", "嘴馋", "小零食", "解馋", "垫肚子", "饿了", "肚子饿", "两餐之间", "间食")),
|
||||
("控糖/血糖相关", ("控糖", "血糖高", "升糖", "糖友", "糖尿病", "孕期控糖", "妊娠糖", "血糖")),
|
||||
("孕期/育儿", ("孕期", "孕妇", "怀孕", "产妇", "坐月子", "哺乳", "给宝宝", "给娃", "孩子吃", "小孩吃", "宝宝吃")),
|
||||
("健身/减脂", ("减肥", "减脂", "瘦身", "健身", "卡路里", "热量低", "低脂")),
|
||||
("长辈/家庭", ("老人", "爸妈", "父母", "长辈", "爷爷奶奶", "给家里")),
|
||||
("办公/外出", ("办公室", "上班吃", "出门", "外出", "随身带", "包里", "便携")),
|
||||
("送礼/囤货", ("送礼", "送人", "囤货", "年货")),
|
||||
("夜宵/熬夜", ("夜宵", "熬夜", "晚上饿")),
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"_COMMENT_CSV_BODY",
|
||||
"_COMMENT_CSV_SCORE",
|
||||
"_COMMENT_CSV_SKU",
|
||||
"COMMENT_FOCUS_WORDS",
|
||||
"COMMENT_SCENARIO_GROUPS",
|
||||
"EXTERNAL_MARKET_TABLE_ROWS",
|
||||
"_COMMENT_FUZZ_KEYS",
|
||||
"_COMMENT_SCORE_NEG_MAX",
|
||||
"_COMMENT_SCORE_POS_MIN",
|
||||
"_COUPON_SHOW_PRICE_KEY",
|
||||
"_DETAIL_CATEGORY_PATH_KEY",
|
||||
"_DETAIL_PRICE_FINAL_CSV_KEYS",
|
||||
"_JD_LIST_PRICE_KEY",
|
||||
"_K_CAT_COL",
|
||||
"_K_PROP_COL",
|
||||
"_LEGACY_COUPON_SHOW_PRICE_KEY",
|
||||
"_LEGACY_JD_LIST_PRICE_KEY",
|
||||
"_LEGACY_LIST_BRAND_TITLE_KEY",
|
||||
"_LEGACY_RANK_TAGLINE_KEY",
|
||||
"_LEGACY_SELLING_POINT_KEY",
|
||||
"_LEGACY_SHOP_NAME_KEY",
|
||||
"_LIST_BRAND_TITLE_HEADER",
|
||||
"_LIST_PRICE_AND_COUPON_KEYS",
|
||||
"_LIST_SHOW_PRICE_CELL_KEYS",
|
||||
"_MERGED_SHOP_CELL_KEYS",
|
||||
"_ORIGINAL_LIST_PRICE_KEY",
|
||||
"_RANK_TAGLINE_KEY",
|
||||
"_SELLING_POINT_KEY",
|
||||
]
|
||||
@ -0,0 +1,99 @@
|
||||
"""CSV 行读取与单元格、价格抽取等通用辅助。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from .constants import (
|
||||
_DETAIL_CATEGORY_PATH_KEY,
|
||||
_K_CAT_COL,
|
||||
_K_PROP_COL,
|
||||
_LIST_PRICE_AND_COUPON_KEYS,
|
||||
)
|
||||
|
||||
|
||||
def _cell(row: dict[str, str], *keys: str) -> str:
|
||||
for k in keys:
|
||||
v = str(row.get(k) or "").strip()
|
||||
if v:
|
||||
return v
|
||||
return ""
|
||||
|
||||
|
||||
def _shortname_from_prop(prop: str) -> str:
|
||||
m = re.search(r"简称[:：]\s*([^|]+)", prop or "")
|
||||
return m.group(1).strip()[:120] if m else ""
|
||||
|
||||
|
||||
def _detail_category_path_cell(row: dict[str, str]) -> str:
|
||||
"""细类矩阵与按细类评价统计仅以该列为准；空则视为商详类目不完整。"""
|
||||
return _cell(row, _DETAIL_CATEGORY_PATH_KEY, "detail_category_path")
|
||||
|
||||
|
||||
def _search_export_catid_to_shortname_map(rows: list[dict[str, str]]) -> dict[str, str]:
|
||||
"""列表导出中叶子类目列常为纯数字 ID：用同行规格属性「简称」映射为可读名称。"""
|
||||
m: dict[str, str] = {}
|
||||
for r in rows:
|
||||
cid = _cell(r, _K_CAT_COL).strip()
|
||||
if not cid.isdigit():
|
||||
continue
|
||||
if cid in m:
|
||||
continue
|
||||
sn = _shortname_from_prop(_cell(r, _K_PROP_COL))
|
||||
if sn:
|
||||
m[cid] = sn
|
||||
return m
|
||||
|
||||
|
||||
def _md_cell(s: str, max_len: int = 120) -> str:
|
||||
t = (s or "").replace("\r\n", " ").replace("\n", " ").replace("|", "/")
|
||||
t = " ".join(t.split())
|
||||
return (t[:max_len] + "…") if max_len > 0 and len(t) > max_len else t
|
||||
|
||||
|
||||
def _read_csv_rows(path: Path) -> tuple[list[str], list[dict[str, str]]]:
|
||||
if not path.is_file():
|
||||
return [], []
|
||||
raw = path.read_text(encoding="utf-8-sig")
|
||||
lines = raw.splitlines()
|
||||
if not lines:
|
||||
return [], []
|
||||
rdr = csv.DictReader(lines)
|
||||
fn = rdr.fieldnames or []
|
||||
return list(fn), list(rdr)
|
||||
|
||||
|
||||
def _float_price(s: str) -> float | None:
|
||||
if not (s or "").strip():
|
||||
return None
|
||||
m = re.search(r"(\d+(?:\.\d+)?)", str(s).replace(",", ""))
|
||||
if not m:
|
||||
return None
|
||||
try:
|
||||
return float(m.group(1))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _collect_prices(rows: list[dict[str, str]]) -> list[float]:
|
||||
out: list[float] = []
|
||||
for row in rows:
|
||||
for k in _LIST_PRICE_AND_COUPON_KEYS:
|
||||
p = _float_price(_cell(row, k))
|
||||
if p is not None and 0 < p < 1_000_000:
|
||||
out.append(p)
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
__all__ = [
|
||||
"_cell",
|
||||
"_collect_prices",
|
||||
"_detail_category_path_cell",
|
||||
"_float_price",
|
||||
"_md_cell",
|
||||
"_read_csv_rows",
|
||||
"_search_export_catid_to_shortname_map",
|
||||
"_shortname_from_prop",
|
||||
]
|
||||
@ -23,7 +23,6 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
@ -51,66 +50,12 @@ from pipeline.csv_schema import ( # noqa: E402
|
||||
merged_csv_effective_total_sales,
|
||||
)
|
||||
|
||||
_JD_LIST_PRICE_KEY = JD_SEARCH_CSV_HEADERS["price"]
|
||||
_COUPON_SHOW_PRICE_KEY = JD_SEARCH_CSV_HEADERS["coupon_price"]
|
||||
_ORIGINAL_LIST_PRICE_KEY = JD_SEARCH_CSV_HEADERS["original_price"]
|
||||
_SELLING_POINT_KEY = JD_SEARCH_CSV_HEADERS["selling_point"]
|
||||
_RANK_TAGLINE_KEY = JD_SEARCH_CSV_HEADERS["hot_list_rank"]
|
||||
|
||||
# 历史批次 CSV 表头（括号英文）；新批次为纯中文，读取时新键优先
|
||||
_LEGACY_JD_LIST_PRICE_KEY = "标价(jdPrice,jdPriceText,realPrice)"
|
||||
_LEGACY_COUPON_SHOW_PRICE_KEY = (
|
||||
"券后到手价(couponPrice,subsidyPrice,finalPrice.estimatedPrice,priceShow)"
|
||||
)
|
||||
_LEGACY_SHOP_NAME_KEY = "店铺名(shopName)"
|
||||
_LEGACY_RANK_TAGLINE_KEY = "榜单类文案(标签/腰带/标题数组中的榜、TOP 等)"
|
||||
_LEGACY_COMMENT_FUZZ_KEY = "评价量(commentFuzzy)"
|
||||
_LEGACY_SELLING_POINT_KEY = "卖点(sellingPoint)"
|
||||
_LIST_BRAND_TITLE_HEADER = "店铺信息标题"
|
||||
_LEGACY_LIST_BRAND_TITLE_KEY = "店铺信息标题(shopInfoTitle,brandName)"
|
||||
|
||||
_DETAIL_PRICE_FINAL_CSV_KEYS: tuple[str, ...] = (
|
||||
MERGED_FIELD_TO_CSV_HEADER["detail_price_final"],
|
||||
"detail_price_final",
|
||||
)
|
||||
_LIST_PRICE_AND_COUPON_KEYS: tuple[str, ...] = (
|
||||
*_DETAIL_PRICE_FINAL_CSV_KEYS,
|
||||
_JD_LIST_PRICE_KEY,
|
||||
_LEGACY_JD_LIST_PRICE_KEY,
|
||||
_COUPON_SHOW_PRICE_KEY,
|
||||
_LEGACY_COUPON_SHOW_PRICE_KEY,
|
||||
)
|
||||
|
||||
# 报告摘录「标价」：列表标价优先，缺省时用商详到手价列兜底
|
||||
_LIST_SHOW_PRICE_CELL_KEYS: tuple[str, ...] = (
|
||||
_JD_LIST_PRICE_KEY,
|
||||
_LEGACY_JD_LIST_PRICE_KEY,
|
||||
MERGED_FIELD_TO_CSV_HEADER["detail_price_final"],
|
||||
"detail_price_final",
|
||||
)
|
||||
|
||||
_MERGED_SHOP_CELL_KEYS: tuple[str, ...] = (
|
||||
MERGED_FIELD_TO_CSV_HEADER["detail_shop_name"],
|
||||
"detail_shop_name",
|
||||
JD_SEARCH_CSV_HEADERS["shop_name"],
|
||||
_LEGACY_SHOP_NAME_KEY,
|
||||
)
|
||||
|
||||
_COMMENT_FUZZ_KEYS: tuple[str, ...] = (
|
||||
MERGED_FIELD_TO_CSV_HEADER["comment_fuzzy"],
|
||||
_LEGACY_COMMENT_FUZZ_KEY,
|
||||
)
|
||||
|
||||
_COMMENT_CSV_SKU = COMMENT_CSV_COLUMNS[0]
|
||||
_COMMENT_CSV_BODY = COMMENT_CSV_COLUMNS[3]
|
||||
_COMMENT_CSV_SCORE = COMMENT_CSV_COLUMNS[7] # 「评分」→ commentScore
|
||||
|
||||
# 评价星级与 §8.2 分桶：先按评分筛正负，再在对应子集内统计口语短语（无评分时回退关键词）
|
||||
_COMMENT_SCORE_NEG_MAX = 2 # 1～2 星 → 偏负向
|
||||
_COMMENT_SCORE_POS_MIN = 4 # 4～5 星 → 偏正向（3 星为中评，归入中性）
|
||||
from competitor_report.config import * # noqa: F403
|
||||
from competitor_report.constants import * # noqa: F403
|
||||
from competitor_report.csv_io import * # noqa: F403
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 运行配置（按需改这里）
|
||||
# 运行配置（按需改这里；与 competitor_report.constants 中默认关注词等配合使用）
|
||||
# ---------------------------------------------------------------------------
|
||||
# KEYWORD：京东 PC 搜索词；全量抓取时必填。「仅已有目录」模式下可留空，改从 run_meta / 目录名推断。
|
||||
KEYWORD = "低GI"
|
||||
@ -122,220 +67,6 @@ OVERRIDE_MAX_SKUS: int | None = None
|
||||
OVERRIDE_PAGE_START: int | None = None
|
||||
OVERRIDE_PAGE_TO: int | None = None
|
||||
|
||||
# 评价/预览文本中可统计的「低 GI / 控糖」语境词（命中次数供侧写，非严谨 NLP）
|
||||
# 可选：第三方市场规模 / 行业增速等（每行四列：指标 | 数值与说明 | 来源 | 年份）。留空则不生成该小节。
|
||||
EXTERNAL_MARKET_TABLE_ROWS: tuple[tuple[str, str, str, str], ...] = ()
|
||||
|
||||
COMMENT_FOCUS_WORDS: tuple[str, ...] = (
|
||||
"口感",
|
||||
"甜",
|
||||
"糖",
|
||||
"血糖",
|
||||
"控糖",
|
||||
"低糖",
|
||||
"无糖",
|
||||
"饱腹",
|
||||
"升糖",
|
||||
"GI",
|
||||
"gi",
|
||||
"孕妇",
|
||||
"老人",
|
||||
"糖尿病",
|
||||
"价格",
|
||||
"贵",
|
||||
"便宜",
|
||||
"回购",
|
||||
"包装",
|
||||
"物流",
|
||||
# 规格/分量（与「硬」等质地问题并列的常见抱怨维度）
|
||||
"分量",
|
||||
"量少",
|
||||
"克重",
|
||||
)
|
||||
|
||||
# 用途/场景：每组 (展示名, 触发子串…)。每条评价若命中组内任一子串则该组 +1；同一条可属多组。
|
||||
COMMENT_SCENARIO_GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||||
("早餐/代餐", ("早餐", "代餐", "早饭", "当早餐", "当早饭", "早上吃", "晨起")),
|
||||
("零食/加餐/解馋", ("零食", "加餐", "嘴馋", "小零食", "解馋", "垫肚子", "饿了", "肚子饿", "两餐之间", "间食")),
|
||||
("控糖/血糖相关", ("控糖", "血糖高", "升糖", "糖友", "糖尿病", "孕期控糖", "妊娠糖", "血糖")),
|
||||
("孕期/育儿", ("孕期", "孕妇", "怀孕", "产妇", "坐月子", "哺乳", "给宝宝", "给娃", "孩子吃", "小孩吃", "宝宝吃")),
|
||||
("健身/减脂", ("减肥", "减脂", "瘦身", "健身", "卡路里", "热量低", "低脂")),
|
||||
("长辈/家庭", ("老人", "爸妈", "父母", "长辈", "爷爷奶奶", "给家里")),
|
||||
("办公/外出", ("办公室", "上班吃", "出门", "外出", "随身带", "包里", "便携")),
|
||||
("送礼/囤货", ("送礼", "送人", "囤货", "年货")),
|
||||
("夜宵/熬夜", ("夜宵", "熬夜", "晚上饿")),
|
||||
)
|
||||
|
||||
|
||||
def _normalize_focus_words(raw: Any) -> tuple[str, ...]:
|
||||
if not isinstance(raw, list) or not raw:
|
||||
return COMMENT_FOCUS_WORDS
|
||||
out: list[str] = []
|
||||
for x in raw[:120]:
|
||||
s = str(x).strip()
|
||||
if len(s) > 48:
|
||||
s = s[:48]
|
||||
if s:
|
||||
out.append(s)
|
||||
return tuple(out) if out else COMMENT_FOCUS_WORDS
|
||||
|
||||
|
||||
def _normalize_scenario_groups(
|
||||
raw: Any,
|
||||
) -> tuple[tuple[str, tuple[str, ...]], ...]:
|
||||
if not isinstance(raw, list) or not raw:
|
||||
return COMMENT_SCENARIO_GROUPS
|
||||
parsed: list[tuple[str, tuple[str, ...]]] = []
|
||||
for item in raw[:40]:
|
||||
label = ""
|
||||
triggers: list[str] = []
|
||||
if isinstance(item, dict):
|
||||
label = str(item.get("label") or "").strip()[:80]
|
||||
tr = item.get("triggers")
|
||||
if isinstance(tr, list):
|
||||
for t in tr[:48]:
|
||||
s = str(t).strip()
|
||||
if len(s) > 48:
|
||||
s = s[:48]
|
||||
if s:
|
||||
triggers.append(s)
|
||||
elif isinstance(item, (list, tuple)) and len(item) >= 2:
|
||||
label = str(item[0]).strip()[:80]
|
||||
tr = item[1]
|
||||
if isinstance(tr, (list, tuple)):
|
||||
for t in tr[:48]:
|
||||
s = str(t).strip()
|
||||
if len(s) > 48:
|
||||
s = s[:48]
|
||||
if s:
|
||||
triggers.append(s)
|
||||
if label and triggers:
|
||||
parsed.append((label, tuple(triggers)))
|
||||
return tuple(parsed) if parsed else COMMENT_SCENARIO_GROUPS
|
||||
|
||||
|
||||
def _normalize_external_market_rows(
|
||||
raw: Any,
|
||||
) -> tuple[tuple[str, str, str, str], ...]:
|
||||
if not isinstance(raw, list) or not raw:
|
||||
return EXTERNAL_MARKET_TABLE_ROWS
|
||||
rows: list[tuple[str, str, str, str]] = []
|
||||
|
||||
def _four_cells(x: Any) -> tuple[str, str, str, str] | None:
|
||||
if isinstance(x, (list, tuple)) and len(x) >= 4:
|
||||
return tuple(str(c)[:500] for c in x[:4])
|
||||
if isinstance(x, dict):
|
||||
a = str(x.get("indicator") or x.get("a") or "").strip()[:500]
|
||||
b = str(x.get("value_and_scope") or x.get("b") or "").strip()[:500]
|
||||
c = str(x.get("source") or x.get("c") or "").strip()[:500]
|
||||
d = str(x.get("year") or x.get("d") or "").strip()[:500]
|
||||
if any((a, b, c, d)):
|
||||
return (a, b, c, d)
|
||||
return None
|
||||
|
||||
for item in raw[:24]:
|
||||
r = _four_cells(item)
|
||||
if r:
|
||||
rows.append(r)
|
||||
return tuple(rows) if rows else EXTERNAL_MARKET_TABLE_ROWS
|
||||
|
||||
|
||||
def resolve_report_tuning(
|
||||
report_config: dict[str, Any] | None,
|
||||
) -> tuple[
|
||||
tuple[str, ...],
|
||||
tuple[tuple[str, tuple[str, ...]], ...],
|
||||
tuple[tuple[str, str, str, str], ...],
|
||||
]:
|
||||
if not report_config:
|
||||
return COMMENT_FOCUS_WORDS, COMMENT_SCENARIO_GROUPS, EXTERNAL_MARKET_TABLE_ROWS
|
||||
return (
|
||||
_normalize_focus_words(report_config.get("comment_focus_words")),
|
||||
_normalize_scenario_groups(report_config.get("comment_scenario_groups")),
|
||||
_normalize_external_market_rows(
|
||||
report_config.get("external_market_table_rows")
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _cell(row: dict[str, str], *keys: str) -> str:
|
||||
for k in keys:
|
||||
v = str(row.get(k) or "").strip()
|
||||
if v:
|
||||
return v
|
||||
return ""
|
||||
|
||||
|
||||
_DETAIL_CATEGORY_PATH_KEY = MERGED_FIELD_TO_CSV_HEADER["detail_category_path"]
|
||||
_K_CAT_COL = JD_SEARCH_CSV_HEADERS["leaf_category"]
|
||||
_K_PROP_COL = JD_SEARCH_CSV_HEADERS["attributes"]
|
||||
|
||||
|
||||
def _shortname_from_prop(prop: str) -> str:
|
||||
m = re.search(r"简称[:：]\s*([^|]+)", prop or "")
|
||||
return m.group(1).strip()[:120] if m else ""
|
||||
|
||||
|
||||
def _detail_category_path_cell(row: dict[str, str]) -> str:
|
||||
"""细类矩阵与按细类评价统计仅以该列为准；空则视为商详类目不完整。"""
|
||||
return _cell(row, _DETAIL_CATEGORY_PATH_KEY, "detail_category_path")
|
||||
|
||||
|
||||
def _search_export_catid_to_shortname_map(rows: list[dict[str, str]]) -> dict[str, str]:
|
||||
"""列表导出中叶子类目列常为纯数字 ID：用同行规格属性「简称」映射为可读名称。"""
|
||||
m: dict[str, str] = {}
|
||||
for r in rows:
|
||||
cid = _cell(r, _K_CAT_COL).strip()
|
||||
if not cid.isdigit():
|
||||
continue
|
||||
if cid in m:
|
||||
continue
|
||||
sn = _shortname_from_prop(_cell(r, _K_PROP_COL))
|
||||
if sn:
|
||||
m[cid] = sn
|
||||
return m
|
||||
|
||||
|
||||
def _md_cell(s: str, max_len: int = 120) -> str:
|
||||
t = (s or "").replace("\r\n", " ").replace("\n", " ").replace("|", "/")
|
||||
t = " ".join(t.split())
|
||||
return (t[:max_len] + "…") if max_len > 0 and len(t) > max_len else t
|
||||
|
||||
|
||||
def _read_csv_rows(path: Path) -> tuple[list[str], list[dict[str, str]]]:
|
||||
if not path.is_file():
|
||||
return [], []
|
||||
raw = path.read_text(encoding="utf-8-sig")
|
||||
lines = raw.splitlines()
|
||||
if not lines:
|
||||
return [], []
|
||||
rdr = csv.DictReader(lines)
|
||||
fn = rdr.fieldnames or []
|
||||
return list(fn), list(rdr)
|
||||
|
||||
|
||||
def _float_price(s: str) -> float | None:
|
||||
if not (s or "").strip():
|
||||
return None
|
||||
m = re.search(r"(\d+(?:\.\d+)?)", str(s).replace(",", ""))
|
||||
if not m:
|
||||
return None
|
||||
try:
|
||||
return float(m.group(1))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _collect_prices(rows: list[dict[str, str]]) -> list[float]:
|
||||
out: list[float] = []
|
||||
for row in rows:
|
||||
for k in _LIST_PRICE_AND_COUPON_KEYS:
|
||||
p = _float_price(_cell(row, k))
|
||||
if p is not None and 0 < p < 1_000_000:
|
||||
out.append(p)
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def _analyze_price_promotions(rows: list[dict[str, str]]) -> dict[str, Any]:
|
||||
"""
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user