#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 按【国家 + 品类】爬取 Inkreach 商城商品款号（如 DG505、JPTM001、PLTK005），去重后输出。 数据源（均为公开只读接口）： - 国家列表 : GET {BASE}/products/pageOptionGroup -> shipmentCountryList (code / 中文名 / 商品数) - 品类树 : GET {BASE}/categories -> 品类层级（辅助，商品自带品类标签优先） - 商品列表 : GET {BASE}/products/page -> 分页商品（shipmentArea=overseas / CN） 说明： - 服务端 overseasArea / id 筛选参数实测不生效，因此采用「全量分页拉取 + 本地按国家/品类聚合」。 - 款号从商品 name 中提取：name 形如 "美国（不包邮）180g男士纯棉长袖-DG505-单面印花-美西洛杉矶一仓"， 取被 "-" 分隔、且满足「纯大写字母+数字、长度 3~8、同时含字母和数字」的段。 - 国家由商品 categories[0]（如"美国工厂直发"/"巴西本地工厂直发"）与 name 前缀匹配中文国家名得到。 品类白名单： - 只输出品类名含 T恤 / 卫衣 / 长袖 / 背心 / 夹克 的款号（见 KEEP_CATEGORY_KEYWORDS）； 下装（短裤/长裤）、家居、家居配饰、内衣、配饰、印花烫贴、桌面用品等品类连同其款号一并剔除。 - 恢复全量：把 KEEP_CATEGORY_KEYWORDS 改成空元组 () 重跑即可。 输出（与脚本同目录）： - style_codes_by_country.json {国家代码: {country, categories: {品类: [款号...]}}} - style_codes_by_country.csv 国家代码,国家,品类,款号 """ import csv import json import re import sys import time import urllib.parse import urllib.request from collections import defaultdict try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass BASE = "https://mapi.sdspod.com" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Accept": "application/json", "Referer": "https://www.inkreach.cn/", } PAGE_SIZE = 100 # 实测服务端允许的最大分页大小 REQUEST_DELAY = 0.3 # 请求间隔（秒），礼貌爬取 TIMEOUT = 30 RETRIES = 3 OUT_JSON = "style_codes_by_country.json" OUT_CSV = "style_codes_by_country.csv" # 品类白名单：品类名包含任一关键词才保留，其余（下装/短裤/长裤/家居/家居配饰/内衣/ # 配饰/印花烫贴/桌面用品 等）连同其款号一并剔除。想恢复全量，把这里改成空元组即可。 KEEP_CATEGORY_KEYWORDS = ("T恤", "卫衣", "长袖", "背心", "夹克") # ---------------------------------------------------------------- 基础请求 def http_get_json(path, params=None): """GET 请求，返回 dict；失败自动重试。""" url = BASE + path if params: url += "?" + urllib.parse.urlencode(params) last_err = None for attempt in range(1, RETRIES + 1): try: req = urllib.request.Request(url, headers=HEADERS) with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return json.loads(resp.read().decode("utf-8")) except Exception as e: # noqa: BLE001 last_err = e if attempt < RETRIES: time.sleep(1.0 * attempt) raise RuntimeError(f"请求失败 {url} -> {last_err}") # ---------------------------------------------------------------- 数据抓取 def fetch_countries(): """返回 [(countryCode, 中文名, 商品数), ...]，按名称长度降序（便于长名优先匹配）。""" d = http_get_json("/products/pageOptionGroup") lst = d.get("shipmentCountryList") or [] out = [(c.get("countryCode") or "", c.get("country") or "", c.get("num") or 0) for c in lst] out.sort(key=lambda x: -len(x[1])) return out def fetch_all_products(): """分页拉取全部商品（海外 + 中国两个发货区域），按 id 去重。""" products, seen = [], set() for area in ("overseas", "CN"): page, got = 1, 0 while True: d = http_get_json("/products/page", {"shipmentArea": area, "page": page, "size": PAGE_SIZE}) items = d.get("items") or [] if not items: break for it in items: pid = it.get("id") if pid in seen: continue seen.add(pid) it["_shipmentArea"] = area products.append(it) total = int(d.get("totalCount") or 0) got += len(items) print(f" [{area}] page={page:<3} +{len(items):<4} 本区累计 {got}/{total}") if got >= total or len(items) < PAGE_SIZE: break page += 1 time.sleep(REQUEST_DELAY) return products # ---------------------------------------------------------------- 解析规则 CODE_SEG = re.compile(r"^[A-Z0-9]{2,8}$") CODE_LABEL = re.compile(r"^([A-Z0-9]{2,8})\s+") # "DG505 180g男士纯棉长袖" CODE_INLINE = re.compile(r"\b([A-Z]{1,5}\d{2,6}[A-Z]?|\d{2,5}[A-Z]{1,3})\b") # 尺码词：避免把 S / M / XL / 3XL 之类误判成款号 SIZE_WORDS = {"S", "M", "L", "XL", "XXL", "XXXL", "XXXXL", "XXXXXL", "XS", "XXS", "2XL", "3XL", "4XL", "5XL", "6XL", "SML", "ML", "OS"} # 国家别名：站点文案与 pageOptionGroup 的正式国名不一致时的映射 COUNTRY_ALIASES = { "中东": "SA", # "中东本地工厂直发" -> 沙特阿拉伯(SA) "国内工厂": "CN", # "国内工厂" -> 中国(CN) } def is_valid_code(s): """款号合法性：2~8 位大写字母/数字，含字母，且不是尺码词。""" s = (s or "").upper() if s in SIZE_WORDS: return False if not CODE_SEG.match(s): return False return bool(re.search(r"[A-Z]", s)) # 必须含字母，排除 "180" 之类纯数字 def extract_style_code(name, cats): """提取款号：优先 categories 的『款号 品名』标签，其次 name 的 '-' 分段，最后全局正则。""" # 1) categories 里的 "DG505 180g男士纯棉长袖" / "SFB 运动短裤"（最可靠，纯字母款号也能拿到） for c in cats: m = CODE_LABEL.match(c.strip()) if m and is_valid_code(m.group(1)): return m.group(1).upper() # 2) name 按 '-' 分段 for seg in (name or "").split("-"): s = seg.strip().upper() if is_valid_code(s): return s # 3) 兜底：全局正则 m = CODE_INLINE.search(name or "") if m and is_valid_code(m.group(1)): return m.group(1).upper() return None def match_country(text, head, countries, area=None): """匹配国家：别名 -> name 前段 -> 全文 -> shipmentArea 兜底。""" cn_by_code = {code: cn for code, cn, _ in countries} for alias, code in COUNTRY_ALIASES.items(): if alias in head or alias in text: return code, cn_by_code.get(code, alias) for scope in (head, text): for code, cn, _ in countries: if cn and cn in scope: return code, cn if area == "CN": return "CN", cn_by_code.get("CN", "中国") return "UNKNOWN", "未识别" def pick_category(cats, cn_name, style_code): """品类 = categories 中去掉『国家/工厂标签』与『款号+品名标签』后剩下的部分。""" rest = [] for c in cats: c_stripped = c.strip() if not c_stripped: continue if cn_name and cn_name in c_stripped: # 国家标签，如"美国工厂直发" continue if "直发" in c_stripped or "工厂" in c_stripped: continue head = c_stripped.split(" ")[0] if style_code and head.upper() == style_code.upper(): # "DG505 180g男士纯棉长袖" continue rest.append(c_stripped) return " / ".join(rest) if rest else "未分类" def guess_sleeve_length(name, category): """推断袖长，返回 (袖长, 来源)。 商城没有袖长分类字段，只能从商品名/品类推断，来源用于审计： - 商品名明确写「长袖/短袖/无袖」 -> 来源 "name" - 品类是背心/吊带 -> 无袖，来源 "品类" - 品类是卫衣/夹克 -> 长袖，来源 "品类推断"（卫衣、夹克默认长袖） - 其余（普通 T 恤，名里没写） -> 短袖，来源 "默认(T恤)" """ n, c = name or "", category or "" if "无袖" in n or "背心" in n or "吊带" in n or "背心" in c: return "无袖", "品类" if ("背心" in c and "背心" not in n and "无袖" not in n) else "name" if "长袖" in n: return "长袖", "name" if "短袖" in n: return "短袖", "name" if "卫衣" in c or "夹克" in c: return "长袖", "品类推断" return "短袖", "默认(T恤)" # ---------------------------------------------------------------- 主流程 def main(): print("=" * 66) print("Inkreach 款号爬取：按国家 + 品类，去重") print("=" * 66) print("\n[1/4] 获取国家列表 ...") countries = fetch_countries() for code, cn, num in countries: print(f" {code:<8} {cn:<10} {num:>4} 个商品") print(f" 共 {len(countries)} 个国家/地区") print("\n[2/4] 获取品类树 ...") try: cat_tree = http_get_json("/categories").get("items") or [] leaf = sum(len(c.get("subcategory") or []) for c in cat_tree) print(f" 顶层品类 {len(cat_tree)} 个，二级品类 {leaf} 个（商品自带品类标签优先，此树仅参考）") except Exception as e: # noqa: BLE001 print(f" 品类树获取失败（不影响主流程）: {e}") print("\n[3/4] 分页拉取全部商品 ...") products = fetch_all_products() print(f" 商品总数（去重后）: {len(products)}") print("\n[4/4] 解析并按 国家 / 品类 聚合款号 ...") # country_code -> category -> {style_code} data = dict() country_cn = {} no_code = [] dropped = defaultdict(int) # 被品类白名单剔除的品类 -> 商品条数 sleeve_map = {} # 款号 -> (袖长, 来源) for it in products: name = it.get("name") or "" cats = [c.get("name", "") for c in (it.get("categories") or [])] text = " ".join([name] + cats) head = name[:12] # 国家一般写在 name 最前面 code, cn = match_country(text, head, countries, it.get("_shipmentArea")) style = extract_style_code(name, cats) if not style: no_code.append(name) continue style = style.upper() category = pick_category(cats, cn, style) # 品类白名单过滤：只保留 T恤/卫衣/长袖/背心/夹克，其余品类及其款号一并剔除 if KEEP_CATEGORY_KEYWORDS and not any(k in category for k in KEEP_CATEGORY_KEYWORDS): dropped[category] += 1 continue country_cn.setdefault(code, cn) data.setdefault(code, {}).setdefault(category, set()).add(style) if style not in sleeve_map: sleeve_map[style] = guess_sleeve_length(name, category) if no_code: print(f" ⚠ {len(no_code)} 条商品未提取到款号（见控制台样例）:") for n in no_code[:5]: print(f" - {n}") if dropped: print(f" 已按品类白名单剔除 {sum(dropped.values())} 条商品，涉及品类：") for c, n in sorted(dropped.items(), key=lambda x: -x[1]): print(f" - {c}（{n} 条）") # ---------------- 汇总输出 ---------------- print("\n" + "=" * 66) print("汇总：各国品类数 / 款号数（去重）") print("=" * 66) print(f"{'国家':<12}{'代码':<8}{'品类数':>8}{'款号数':>10}") print("-" * 66) total_cats = total_codes = 0 all_codes = set() for code in sorted(data, key=lambda c: -sum(len(v) for v in data[c].values())): cats_map = data[code] n_codes = sum(len(v) for v in cats_map.values()) total_cats += len(cats_map) total_codes += n_codes for v in cats_map.values(): all_codes |= v print(f"{country_cn.get(code,''):<12}{code:<8}{len(cats_map):>8}{n_codes:>10}") print("-" * 66) print(f"{'合计':<20}{total_cats:>8}{total_codes:>10} (全局去重款号 {len(all_codes)} 个)") # ---------------- 袖长分布 ---------------- if sleeve_map: print("\n=== 袖长分布（按款号去重）===") sl_cnt, src_cnt = defaultdict(int), defaultdict(int) for v, s in sleeve_map.values(): sl_cnt[v] += 1 src_cnt[s] += 1 for k, v in sorted(sl_cnt.items(), key=lambda x: -x[1]): print(f" {k:<6} {v:>4} 个款号") print(" 判定来源:", dict(sorted(src_cnt.items(), key=lambda x: -x[1]))) # ---------------- 写 JSON ---------------- out = {} for code, cats_map in data.items(): codes_here = sorted({c for v in cats_map.values() for c in v}) out[code] = { "country": country_cn.get(code, ""), "categories": {c: sorted(v) for c, v in sorted(cats_map.items())}, "sleeve_length": {c: sleeve_map.get(c, ("", ""))[0] for c in codes_here}, "sleeve_source": {c: sleeve_map.get(c, ("", ""))[1] for c in codes_here}, } with open(OUT_JSON, "w", encoding="utf-8") as f: json.dump(out, f, ensure_ascii=False, indent=2) print(f"\n✅ 已写出 {OUT_JSON}") # ---------------- 写 CSV ---------------- rows = [] for code in sorted(out): for cat, codes in out[code]["categories"].items(): for c in codes: rows.append([code, country_cn.get(code, ""), cat, c, sleeve_map.get(c, ("", ""))[0]]) with open(OUT_CSV, "w", encoding="utf-8-sig", newline="") as f: w = csv.writer(f) w.writerow(["country_code", "country", "category", "style_code", "sleeve_length"]) w.writerows(rows) print(f"✅ 已写出 {OUT_CSV}（{len(rows)} 行）") # ---------------- 样例预览 ---------------- print("\n样例预览（每个国家取 1 个品类，最多 8 个款号）：") for code in list(sorted(out))[:6]: cats_map = out[code]["categories"] if not cats_map: continue cat0 = sorted(cats_map)[0] codes = cats_map[cat0][:8] print(f" [{code}] {country_cn.get(code,'')} / {cat0}: {', '.join(codes)}" + (" ..." if len(cats_map[cat0]) > 8 else "")) if __name__ == "__main__": main()