diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/__init__.py b/backend/crawler_copy/jd_pc_search/competitor_report/__init__.py new file mode 100644 index 0000000..96c4459 --- /dev/null +++ b/backend/crawler_copy/jd_pc_search/competitor_report/__init__.py @@ -0,0 +1 @@ +"""竞品报告脚本共享：表头常量、CSV 辅助、报告调参解析。""" diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/config.py b/backend/crawler_copy/jd_pc_search/competitor_report/config.py new file mode 100644 index 0000000..3053c0c --- /dev/null +++ b/backend/crawler_copy/jd_pc_search/competitor_report/config.py @@ -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", +] diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/constants.py b/backend/crawler_copy/jd_pc_search/competitor_report/constants.py new file mode 100644 index 0000000..81fcf78 --- /dev/null +++ b/backend/crawler_copy/jd_pc_search/competitor_report/constants.py @@ -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", +] diff --git a/backend/crawler_copy/jd_pc_search/competitor_report/csv_io.py b/backend/crawler_copy/jd_pc_search/competitor_report/csv_io.py new file mode 100644 index 0000000..7f0c7ed --- /dev/null +++ b/backend/crawler_copy/jd_pc_search/competitor_report/csv_io.py @@ -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", +] diff --git a/backend/crawler_copy/jd_pc_search/jd_competitor_report.py b/backend/crawler_copy/jd_pc_search/jd_competitor_report.py index edf464f..01bb2d1 100644 --- a/backend/crawler_copy/jd_pc_search/jd_competitor_report.py +++ b/backend/crawler_copy/jd_pc_search/jd_competitor_report.py @@ -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]: """