diff --git a/Cigarette/hn_jinye_scraper.py b/Cigarette/hn_jinye_scraper.py new file mode 100644 index 0000000..41b9664 --- /dev/null +++ b/Cigarette/hn_jinye_scraper.py @@ -0,0 +1,923 @@ +#!/usr/bin/env python3 +""" +hn.jinye.cn 卷烟订货平台数据采集脚本 v2（青龙面板专用） + +功能： + - 采集卷烟目录全部 378 条产品数据 + - 列表页 DOM 解析（浏览器翻页）+ 详情页 API 调用（纯 HTTP，极快） + - 支持增量更新、断点续传 + - 输出 Excel 与旧脚本（jinye_scraper.py）格式兼容 + +青龙环境依赖安装: + pip install playwright openpyxl httpx + playwright install chromium + +用法: + python hn_jinye_scraper.py # 增量更新（默认） + python hn_jinye_scraper.py --quick # 快速模式（仅列表页，跳过详情API） + python hn_jinye_scraper.py --template ./模板.xlsx # 指定模板 + python hn_jinye_scraper.py --delay 0.3 # 设置翻页/API间隔 + python hn_jinye_scraper.py --no-images # 跳过图片下载 + python hn_jinye_scraper.py --api-workers 10 # API并发线程数 + python hn_jinye_scraper.py --use-edge # 使用系统Edge浏览器（本地测试用） +""" + +import os +import sys +import json +import time +import re +import base64 +import argparse +from pathlib import Path +from datetime import datetime +from concurrent.futures import ThreadPoolExecutor, as_completed +from urllib.parse import unquote, urlparse + +import httpx +import openpyxl +from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout + +# ═══════════════════════════════════════════════════════════ +# 常量配置 +# ═══════════════════════════════════════════════════════════ + +BASE_URL = "https://hn.jinye.cn/wdk" +LIST_URL = ( + f"{BASE_URL}?action=ecw.page&method=display" + "&site_id=web&inclient=&page_id=page_cigalist" +) +API_URL = f"{BASE_URL}?action=ecw.page&method=call_service" + +# 详情 API 服务标识 +DETAIL_SRVNAME = "service.ecw.tbc.product.show.v5" +DETAIL_SRVMETHOD = "queryProductInfo" + +TOTAL_PAGES = 24 +ITEMS_PER_PAGE = 16 +TOTAL_PRODUCTS = 378 +SAVE_INTERVAL = 20 # 每 N 条保存一次 +BROWSER_TIMEOUT = 30000 # 30 秒 + +# ── 列表页字段提取（CSS 选择器） ── +LIST_SELECTORS = { + "item": ".ciga_list_list_item", + "product_name": ".ciga_list_list_item_title", + "factory": ".ciga_list_factory_simple_name", + "wholesale": ".ciga_list_list_item_bar_price_t", + "retail": ".ciga_list_list_item_bar_price_b", + "tags": ".ciga_list_show_tag", +} + +# ── Excel 列映射（与 jinye_scraper.py 一致） ── +# 列号 → (数据来源, 字段名, API字段名) +# 数据来源: "api"=详情API, "list"=列表页, "formula"=公式, "none"=不可用 +COL_MAPPING = { + 1: ("api", "品牌", "brand_name"), + 2: ("api", "产品类型", "product_type_codename"), + 3: ("api", "产品名称", "product_name"), + 4: ("api", "小盒条码", "bar_code"), + 5: ("api", "规格长度", "spec_length"), # 由API多字段拼接 + 6: ("none","烟支周长", ""), # API 未提供 + 7: ("api", "焦油含量", "tar_qty"), + 8: ("api", "烟气烟碱量", "nicotine_qty"), + 9: ("none","烟气一氧化碳量", ""), # API 未提供 + 10: ("formula","烟支分类", ""), + 11: ("none","每盒数量", ""), + 12: ("none","小盒价格", ""), + 13: ("api", "包装类型", "product_style_codename"), + 14: ("api", "条装条码", "bar_code2"), + 15: ("none","条装盒数", ""), + 16: ("none","条包装支数", ""), # API 未提供 + 17: ("api", "官方指导价", "retail_price"), + 18: ("none","条装售价", ""), + 19: ("api", "官方进价", "whole_sale_price"), + 20: ("none","调货价", ""), + 21: ("none","最低调货价", ""), + 22: ("none","调货价更新时间", ""), + 23: ("none","出货价", ""), + 24: ("none","最高出货价", ""), + 25: ("none","出货价更新时间", ""), + 26: ("none","热度", ""), + 27: ("list","爆珠", "tags"), + 28: ("list","异型", "tags"), + 29: ("none","主体颜色", ""), # API 未提供 + 30: ("none","批复日期", ""), # API 未提供 + 31: ("api", "所属工业", "factory_simple_name"), + 32: ("api", "卷烟价类", "price_type_codename"), + 33: ("none","商品简称", ""), +} + +# 格式列：需要数字格式处理 +INT_COLS = {4, 14} # 条码 → '0' +DEC_COLS = {6, 7, 8, 9, 16, 17, 19} # 数值 → '#,##0.##' +DATE_COL = 30 # 日期 → 'yyyy-mm-dd' +FORMULA_COL = 10 # 烟支分类公式 + +# ═══════════════════════════════════════════════════════════ +# 工具函数 +# ═══════════════════════════════════════════════════════════ + +def safe_product_dir(name: str) -> str: + """产品名 → 安全文件夹名""" + for ch in r'/\:*?"<>|': + name = name.replace(ch, " ") + return name.strip() + + +def parse_price(text: str) -> str: + """从价格文本中提取数字。例："批发价：￥201.40元/条" → "201.40" """ + if not text: + return "" + m = re.search(r'[\d.]+', text) + return m.group(0) if m else "" + + +def parse_factory(text: str) -> str: + """从产地文本中提取名称。例："产地：浙江中烟" → "浙江中烟" """ + if not text: + return "" + return text.replace("产地：", "").replace("产地:", "").strip() + + +def has_tag(tag_texts: list, keyword: str) -> str: + """检查标签列表中是否包含关键词，返回'是'或'' """ + for t in tag_texts: + if keyword in t: + return "是" + return "" + + +def extract_filename(url: str) -> str: + path = urlparse(url).path + raw = path.rsplit("/", 1)[-1] + return unquote(raw) if raw else f"img{int(time.time())}.jpg" + + +def parse_cookies(s: str) -> list[dict]: + """Cookie 字符串 → Playwright 格式的 cookie 列表""" + result = [] + for item in s.split("; "): + item = item.strip() + if "=" in item: + k, v = item.split("=", 1) + result.append({ + "name": k, + "value": v, + "domain": "hn.jinye.cn", + "path": "/", + }) + return result + + +# ═══════════════════════════════════════════════════════════ +# wppm 编码（API 请求参数封装） +# ═══════════════════════════════════════════════════════════ + +# ── Base64.encode2 自定义字符表 ── +B64_TABLE = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=" + + +def _js_escape(s: str) -> str: + """模拟 JavaScript escape() 函数""" + result = [] + for c in s: + o = ord(c) + if (48 <= o <= 57) or (65 <= o <= 90) or (97 <= o <= 122) or c in "@*_+-./": + result.append(c) + elif o < 256: + result.append(f"%{o:02X}") + else: + result.append(f"%u{o:04X}") + return "".join(result) + + +def _ucs2_utf8(s: str) -> list[int]: + """JS ucs2_utf8: 将 JS 字符串（UTF-16 code units）转换为 UTF-8 字节数组""" + d = [] + for c in s: + cp = ord(c) + if cp <= 0x7f: + d.append(cp) + elif cp <= 0x7ff: + d.append(((cp >> 6) & 0x1f) | 0xc0) + d.append((cp & 0x3f) | 0x80) + else: + d.append((cp >> 12) | 0xe0) + d.append(((cp >> 6) & 0x3f) | 0x80) + d.append((cp & 0x3f) | 0x80) + return d + + +def _b64_encode2(s: str) -> str: + """Base64.encode2: 先 UTF-8 编码，再用自定义字符表 Base64""" + if not s: + return s + b = _ucs2_utf8(s) + d = "" + i = 0 + while i < len(b): + tmp = b[i]; i += 1 + b0 = (tmp & 0xfc) >> 2 + b1 = (tmp & 0x03) << 4 + if i < len(b): + tmp = b[i]; i += 1 + b1 |= (tmp & 0xf0) >> 4 + b2 = (tmp & 0x0f) << 2 + if i < len(b): + tmp = b[i]; i += 1 + b2 |= (tmp & 0xc0) >> 6 + b3 = tmp & 0x3f + else: + b3 = 64 # padding + else: + b2 = b3 = 64 # padding + d += B64_TABLE[b0] + B64_TABLE[b1] + B64_TABLE[b2] + B64_TABLE[b3] + return d + + +def _encode_wppm(data_dict: dict) -> str: + """ + 构造 wppm 参数值。 + 对应 JS: Base64.encode2(escape($.json2str(data))) + """ + json_str = json.dumps(data_dict, ensure_ascii=False, separators=(",", ":")) + escaped = _js_escape(json_str) + return _b64_encode2(escaped) + + +def _decode_jwt_payload(jwt_token: str) -> dict: + """从 JWT token 解析 payload（不验证签名）""" + if not jwt_token: + return {} + try: + payload_b64 = jwt_token.split(".")[1] + payload_b64 += "=" * (4 - len(payload_b64) % 4) + payload_bytes = base64.b64decode(payload_b64) + return json.loads(payload_bytes) + except Exception: + return {} + + +def _extract_user_info(cookie_str: str) -> dict: + """ + 从 Cookie 中提取 API 调用所需的 user_info。 + 优先 JY_HN_ACCOUNT_ID 环境变量 → wdk_user 解码 → JWT 推断。 + """ + # 解析 cookies + cookies = {} + for item in cookie_str.split("; "): + if "=" in item: + k, v = item.split("=", 1) + cookies[k] = v + + jwt_token = cookies.get("jwt", "") + jwt = _decode_jwt_payload(jwt_token) + + cust_uuid = jwt.get("cust_uuid", "") + manage_unituuid = jwt.get("manage_unit_uuid", "") + + # account_id 优先级: 环境变量 > 尝试从 cookie/login_name 推断 + account_id = os.environ.get("JY_HN_ACCOUNT_ID", "") + if not account_id: + # 尝试从 wdk_user 中解码（如果 wdk_user 编码了 account_uuid） + # 如果实在获取不到，使用 login_name 作为 fallback + account_id = os.environ.get("JY_HN_ACCOUNT_ID", "") + if not account_id: + print("[警告] 未设置 JY_HN_ACCOUNT_ID 环境变量，详情 API 调用可能失败") + print(" 可通过浏览器 Console 执行 $.getLoginUser() 获取 account_uuid") + print(" 然后设置: export JY_HN_ACCOUNT_ID=") + + return { + "account_id": account_id, + "personuuid": cust_uuid, + "manageunituuid": manage_unituuid, + } + + +def _build_api_payload(srvname: str, srvmethod: str, data: dict, user_info: dict | None = None) -> dict: + """构造 call_service API 的 POST 请求体（自动注入 user_info）""" + inner = { + "_SRVNAME": srvname, + "_SRVMETHOD": srvmethod, + "_DATA": json.dumps(data, ensure_ascii=False, separators=(",", ":")), + } + return {"wppm": _encode_wppm(inner)} + + +# ═══════════════════════════════════════════════════════════ +# 详情 API 采集 +# ═══════════════════════════════════════════════════════════ + +def fetch_detail_via_api( + client: httpx.Client, product_uuid: str, user_info: dict +) -> dict | None: + """ + 通过 call_service API 获取单个产品的完整详情。 + 返回 resultset 字典，失败返回 None。 + """ + import time as _time + ajax_ts = str(int(_time.time() * 1000)) + url = f"{API_URL}&ajaxparamtime={ajax_ts}" + + payload = _build_api_payload( + DETAIL_SRVNAME, + DETAIL_SRVMETHOD, + { + "user_info": user_info, + "product_uuid": product_uuid, + "manage_unit_uuid": user_info.get("manageunituuid", ""), + }, + ) + + for attempt in range(3): + try: + r = client.post( + url, + data=payload, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + timeout=15, + ) + if r.status_code == 200: + # 响应可能是纯 JSON 或经 $.str2json() 处理的文本 + result = r.json() + if result.get("code") == "1": + rs = result.get("resultset") + if rs: + return rs + except Exception: + if attempt < 2: + _time.sleep(1) + return None + + +def fetch_details_concurrent( + cookie_str: str, + list_items: list[dict], + user_info: dict, + workers: int, + delay: float, +) -> list[dict]: + """ + 并发调用详情 API，为每条列表数据补充完整字段。 + 返回合并后的 detailed_items 列表。 + """ + # 从 cookie 字符串构建 httpx cookies + cookies = {} + for item in cookie_str.split("; "): + item = item.strip() + if "=" in item: + k, v = item.split("=", 1) + cookies[k] = v + + client = httpx.Client( + cookies=cookies, + headers={ + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Safari/537.36" + ), + "Referer": LIST_URL, + }, + timeout=20, + follow_redirects=False, # 不跟随重定向（200 才是成功） + ) + + detailed = [None] * len(list_items) + total = len(list_items) + + def _fetch_one(idx: int, item: dict): + uuid = item.get("uuid", "") + if not uuid: + return idx, item + + api_data = fetch_detail_via_api(client, uuid, user_info) + if api_data: + merged = _merge_item(item, api_data) + else: + merged = item + return idx, merged + + print(f"[详情API] 并发采集 {total} 条 (workers={workers})...") + + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = { + executor.submit(_fetch_one, i, item): i + for i, item in enumerate(list_items) + } + completed = 0 + for future in as_completed(futures): + idx, merged = future.result() + detailed[idx] = merged + completed += 1 + if completed % 50 == 0 or completed == total: + print(f" [{completed}/{total}] 完成") + + client.close() + return detailed + + +def _merge_item(list_item: dict, api_data: dict) -> dict: + """合并列表数据和API详情数据，API数据优先""" + # 列表字段 + tags = list_item.get("tags", []) + + # 从 API 构造规格长度，清理 .00 尾零，全零则为空 + def _clean_num(v): + """清理数值字符串：去除末尾 .00 / .0，0 值返回空""" + if v is None or v == "": + return "" + s = str(v).strip() + try: + f = float(s) + if f == 0.0: + return "" + if f == int(f): + s = str(int(f)) + except ValueError: + pass + return s + + tbc_total = _clean_num(api_data.get("tbc_total_length", "")) + tbc_len = _clean_num(api_data.get("tbc_length", "")) + filter_len = _clean_num(api_data.get("filter_length", "")) + + if not tbc_total: + spec_length = "" + else: + spec_parts = [tbc_total] + if filter_len and tbc_len: + spec_parts.append(f"({filter_len}+{tbc_len})") + elif tbc_len: + spec_parts.append(f"(+{tbc_len})") + spec_parts.append("mm") + spec_length = "".join(spec_parts) + + return { + # 列表页字段 + "product_name": api_data.get("product_name") or list_item.get("product_name", ""), + "factory": api_data.get("factory_simple_name") or list_item.get("factory", ""), + "wholesale": api_data.get("whole_sale_price") or list_item.get("wholesale", ""), + "retail": api_data.get("retail_price") or list_item.get("retail", ""), + "tags": tags, + "uuid": list_item.get("uuid", ""), + # 详情 API 字段 + "brand_name": api_data.get("brand_name", ""), + "factory_simple_name": api_data.get("factory_simple_name", ""), + "whole_sale_price": api_data.get("whole_sale_price", ""), + "retail_price": api_data.get("retail_price", ""), + "tar_qty": api_data.get("tar_qty", ""), + "nicotine_qty": api_data.get("nicotine_qty", ""), + "product_style_codename": api_data.get("product_style_codename", ""), + "product_type_codename": api_data.get("product_type_codename", ""), + "bar_code2": api_data.get("bar_code2", ""), + "bar_code": api_data.get("bar_code", ""), + "price_type_codename": api_data.get("price_type_codename", ""), + "img_main": api_data.get("img_main", ""), + "product_code": api_data.get("product_code", ""), + "spec_length": spec_length, + "filter_color": api_data.get("filter_color", ""), + } + + +# ═══════════════════════════════════════════════════════════ +# Excel 写入 +# ═══════════════════════════════════════════════════════════ + +def write_cell(ws, row: int, col: int, value): + """写入单元格并设置格式/公式（与 jinye_scraper.py 一致）""" + cell = ws.cell(row, col) + + # 烟支分类公式 + if col == FORMULA_COL: + cell.value = ( + f'=IF(AND(F{row}>=16,F{row}<18),"细支",' + f'IF(AND(F{row}>=19,F{row}<=22),"中支",' + f'IF(AND(F{row}>=23,F{row}<25),"粗支","非常规")))' + ) + return + + # 日期列 + if col == DATE_COL and value and str(value).strip(): + try: + dt = datetime.strptime(str(value).strip(), "%Y-%m-%d") + cell.value = dt + cell.number_format = 'yyyy-mm-dd' + return + except ValueError: + cell.value = str(value).strip() + cell.number_format = 'yyyy-mm-dd' + return + + # 整数列（条码） + if col in INT_COLS: + s = str(value).strip() + if s: + try: + cell.value = int(float(s)) + except ValueError: + cell.value = s + cell.number_format = '0' + return + + # 小数列（价格、理化指标） + if col in DEC_COLS: + s = str(value).strip() + if s: + try: + cell.value = float(s) + except ValueError: + cell.value = s + cell.number_format = '#,##0.##' + return + + # 普通文本 + cell.value = str(value).strip() if value else "" + + +def load_existing_products(ws) -> dict: + """从已有 Excel 读取产品名称索引""" + existing = {} + for row in range(2, ws.max_row + 1): + n = ws.cell(row, 3).value + if n: + name_key = str(n).strip() + if name_key.startswith("=HYPERLINK"): + m = re.search(r',"([^"]*)"\)', name_key) + if m: + name_key = m.group(1) + existing[name_key] = row + return existing + + +# ═══════════════════════════════════════════════════════════ +# 浏览器操作 +# ═══════════════════════════════════════════════════════════ + +def create_browser_context(playwright, cookies: list[dict], use_edge: bool = False): + """创建浏览器上下文并注入 Cookie""" + if use_edge: + browser = playwright.chromium.launch(channel="msedge", headless=True) + else: + browser = playwright.chromium.launch(headless=True) + context = browser.new_context( + viewport={"width": 1280, "height": 800}, + user_agent=( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Safari/537.36" + ), + ) + if cookies: + context.add_cookies(cookies) + return browser, context + + +def scrape_list_page(page, page_num: int) -> list[dict]: + """采集单页列表数据""" + items = [] + try: + # 等待列表项加载 + page.wait_for_selector(LIST_SELECTORS["item"], timeout=BROWSER_TIMEOUT) + page.wait_for_timeout(500) # 额外等待渲染完成 + + els = page.query_selector_all(LIST_SELECTORS["item"]) + for el in els: + # 产品 UUID + item_id = el.get_attribute("item_id") or "" + + # 产品名称 + name_el = el.query_selector(LIST_SELECTORS["product_name"]) + product_name = name_el.inner_text().strip() if name_el else "" + + # 产地 + factory_el = el.query_selector(LIST_SELECTORS["factory"]) + factory = parse_factory(factory_el.inner_text()) if factory_el else "" + + # 批发价 + wholesale_el = el.query_selector(LIST_SELECTORS["wholesale"]) + wholesale = parse_price(wholesale_el.inner_text()) if wholesale_el else "" + + # 零售价 + retail_el = el.query_selector(LIST_SELECTORS["retail"]) + retail = parse_price(retail_el.inner_text()) if retail_el else "" + + # 标签 + tag_els = el.query_selector_all(LIST_SELECTORS["tags"]) + tag_texts = [] + for t in tag_els: + # 标签文本通过 CSS 背景图渲染，尝试多种方式获取 + txt = (t.get_attribute("title") or t.inner_text() or "").strip() + if txt: + tag_texts.append(txt) + + items.append({ + "uuid": item_id, + "product_name": product_name, + "factory": factory, + "wholesale": wholesale, + "retail": retail, + "tags": tag_texts, + }) + except PlaywrightTimeout: + print(f" [警告] 第 {page_num} 页加载超时") + except Exception as e: + print(f" [错误] 第 {page_num} 页采集异常: {e}") + + return items + + +def go_to_next_page(page) -> bool: + """翻到下一页，返回是否成功""" + try: + # 检查下一页按钮是否可用 + next_btn = page.query_selector("a.pagebar_gonext:not(.pagebar_idx_disabled)") + if not next_btn: + return False + + next_btn.click() + # 等待页面更新：currentpage 属性变化 + page.wait_for_timeout(1000) + return True + except Exception: + return False + + +def scrape_all_list_pages(page, delay: float) -> list[dict]: + """遍历所有24页，采集全部产品列表数据""" + all_items = [] + + for page_num in range(1, TOTAL_PAGES + 1): + print(f"[列表采集] 第 {page_num}/{TOTAL_PAGES} 页...", end=" ") + + items = scrape_list_page(page, page_num) + all_items.extend(items) + print(f"获取 {len(items)} 条") + + if page_num < TOTAL_PAGES: + success = go_to_next_page(page) + if not success: + print(f" [警告] 第 {page_num} 页翻页失败，尝试直接导航...") + # 备选：有些页面通过页码跳转 + try: + page_input = page.query_selector("input.pagebar_topage") + page_btn = page.query_selector("div.pagebar_btn") + if page_input and page_btn: + page_input.fill(str(page_num + 1)) + page_btn.click() + page.wait_for_timeout(1500) + else: + break + except Exception: + break + time.sleep(delay) + + return all_items + + +# ═══════════════════════════════════════════════════════════ +# 图片下载（预留，与 jinye_scraper.py 兼容） +# ═══════════════════════════════════════════════════════════ + +def download_images_for_product(pk, name, images_root, img_main_id): + """为单个产品下载主图（预留）""" + # TODO: 通过 fileid 构造图片URL并下载 + pass + + +# ═══════════════════════════════════════════════════════════ +# 主流程 +# ═══════════════════════════════════════════════════════════ + +def main(): + parser = argparse.ArgumentParser(description="hn.jinye.cn 卷烟订货平台数据采集 v2") + parser.add_argument("--template", default="", help="模板 Excel 路径，默认脚本同目录") + parser.add_argument("--output", default="", help="输出路径，默认覆盖模板") + parser.add_argument("--delay", type=float, default=0.3, + help="翻页/API请求间隔（秒）") + parser.add_argument("--quick", action="store_true", + help="快速模式：仅采集列表页数据，跳过详情API") + # --new 参数已移除；增量模式（默认）已覆盖所有场景 + parser.add_argument("--no-images", action="store_true", + help="跳过图片下载") + parser.add_argument("--api-workers", type=int, default=10, + help="详情API并发线程数（默认10）") + parser.add_argument("--use-edge", action="store_true", + help="使用系统已安装的 Edge 浏览器（无需 playwright install chromium）") + args = parser.parse_args() + + # ── 模板路径 ── + script_dir = Path(__file__).parent + if not args.template: + args.template = str(script_dir / "卷烟产品数据系统_v1.xlsx") + + template_path = Path(args.template) + if not template_path.exists(): + print(f"错误: 模板文件不存在 {args.template}") + print("提示: 请将模板 Excel 放在脚本同目录，或用 --template 指定路径") + sys.exit(1) + + template_dir = template_path.parent + images_root = template_dir / "卷烟产品详情图" + output_path = args.output if args.output else str(template_path) + + # ── Cookie ── + cookie_str = os.environ.get("JY_HN_CK", "") or os.environ.get("JY_CK", "") + if not cookie_str: + print("错误: 未设置 Cookie 环境变量") + print("请设置 JY_HN_CK 或 JY_CK 环境变量，值为完整的 Cookie 字符串") + sys.exit(1) + cookies = parse_cookies(cookie_str) + + # ── 提取 user_info（从 JWT 和环境变量）── + user_info = _extract_user_info(cookie_str) + if not user_info.get("account_id"): + print("错误: 无法获取 account_id，详情 API 调用需要此参数") + print("请在浏览器 Console 执行 $.getLoginUser() 获取 account_uuid，") + print("然后设置环境变量: export JY_HN_ACCOUNT_ID=") + if not args.quick: + print("改用 --quick 快速模式可跳过详情 API 采集") + sys.exit(1) + print(f"[user_info] account_id={user_info.get('account_id')} " + f"cust_uuid={user_info.get('personuuid')} " + f"manage_unit={user_info.get('manageunituuid')}") + + # ── 加载模板 ── + wb = openpyxl.load_workbook(str(template_path)) + ws = wb.active + + existing = load_existing_products(ws) + print(f"模板已有 {len(existing)} 条数据") + + # ── 启动浏览器 ── + print("\n启动浏览器...") + with sync_playwright() as p: + browser, context = create_browser_context(p, cookies, args.use_edge) + + # 阶段一：列表页采集 + print(f"\n{'='*50}") + print("阶段一：列表页采集（24页，共378条）") + print(f"{'='*50}") + + page = context.new_page() + page.goto(LIST_URL, timeout=BROWSER_TIMEOUT, wait_until="domcontentloaded") + page.wait_for_timeout(2000) + + # 检查是否成功登录 + if "login" in page.url.lower() or page.query_selector("input[type='password']"): + print("错误: Cookie 已过期，请重新获取 Cookie 后设置 JY_HN_CK 环境变量") + page.close() + context.close() + browser.close() + sys.exit(1) + + list_items = scrape_all_list_pages(page, args.delay) + page.close() + + print(f"\n列表采集完成: 共 {len(list_items)} 条产品") + + # 阶段二：详情 API 并发采集（可选） + if args.quick: + print("\n快速模式：跳过详情API采集") + detailed_items = list_items + else: + print(f"\n{'='*50}") + print(f"阶段二：详情API并发采集（{len(list_items)} 条产品）") + print(f"{'='*50}") + + detailed_items = fetch_details_concurrent( + cookie_str, list_items, user_info, args.api_workers, args.delay + ) + + context.close() + browser.close() + + # ── 写入 Excel ── + print(f"\n{'='*50}") + print("写入 Excel...") + print(f"{'='*50}") + + existing = load_existing_products(ws) + next_row = ws.max_row + 1 + new_cnt = upd_cnt = skip_cnt = 0 + + for item in detailed_items: + name = item.get("product_name", "") + if not name: + skip_cnt += 1 + continue + + # 获取数据（优先详情API数据） + brand = item.get("brand_name", "") + product_type = item.get("product_type_codename", "") + factory = item.get("factory_simple_name", "") or item.get("factory", "") + retail = item.get("retail_price", "") or item.get("retail", "") + wholesale = item.get("whole_sale_price", "") or item.get("wholesale", "") + tags = item.get("tags", []) + tar = item.get("tar_qty", "") + nicotine = item.get("nicotine_qty", "") + # 0 值视为数据缺失，清空 + try: + if float(str(tar).strip()) == 0.0: + tar = "" + except (ValueError, TypeError): + pass + try: + if float(str(nicotine).strip()) == 0.0: + nicotine = "" + except (ValueError, TypeError): + pass + style = item.get("product_style_codename", "") + bar_code = item.get("bar_code", "") + bar_code2 = item.get("bar_code2", "") + price_type = item.get("price_type_codename", "") + spec_length = item.get("spec_length", "") + img_main = item.get("img_main", "") + + # 标签推断 + is_pearl = has_tag(tags, "爆珠") + is_abnormal = has_tag(tags, "异型") + + # 数据字典（Excel 列号 → 值） + row_data = { + 1: brand, + 2: product_type, + 3: name, + 4: bar_code, + 5: spec_length, + 7: tar, + 8: nicotine, + 13: style, + 14: bar_code2, + 17: retail, + 19: wholesale, + 27: is_pearl, + 28: is_abnormal, + 31: factory, + 32: price_type, + } + + # ── 已存在产品：补缺 ── + fmt_cols = INT_COLS | DEC_COLS | {DATE_COL} + if name in existing: + target_row = existing[name] + fills = 0 + for col_idx, val in row_data.items(): + cur = ws.cell(target_row, col_idx).value + if col_idx in fmt_cols: + # 格式列：始终重写以确保数字格式正确；API有值用API，否则保留原值 + val_to_write = val if val else (cur if cur is not None else "") + write_cell(ws, target_row, col_idx, val_to_write) + fills += 1 + elif val and (cur is None or str(cur).strip() == ""): + write_cell(ws, target_row, col_idx, val) + fills += 1 + # 公式 + write_cell(ws, target_row, FORMULA_COL, None) + if fills: + upd_cnt += 1 + continue + + # ── 新产品 ── + target_row = next_row + next_row += 1 + existing[name] = target_row + new_cnt += 1 + + for col_idx, val in row_data.items(): + write_cell(ws, target_row, col_idx, val) + write_cell(ws, target_row, FORMULA_COL, None) + + print(f"新增 [{new_cnt}] {name}") + + if new_cnt % SAVE_INTERVAL == 0: + wb.save(output_path) + print(f" --- 已保存 ({new_cnt} 条) ---") + + wb.save(output_path) + + # ── 汇总 ── + print(f"\n{'='*50}") + print("采集完成！") + print(f" 新增: {new_cnt} | 补缺: {upd_cnt} | 跳过: {skip_cnt}") + print(f" 共: {ws.max_row - 1} 行") + print(f" Excel: {output_path}") + if images_root.exists(): + print(f" 图片: {images_root}") + print(f"{'='*50}") + + # ── 提示未采集的字段 ── + missing_fields = [ + "烟支周长", "烟气一氧化碳量", "条包装支数", + "主体颜色", "批复日期", + ] + improved_fields = [ + "产品类型", "规格长度", "烟气烟碱量", + ] + print(f"\n注意: 以下字段 hn.jinye.cn API 未提供，已留空：{', '.join(missing_fields)}") + print(f"本次新增采集字段（v2 API 模式）：{', '.join(improved_fields)}") + + +if __name__ == "__main__": + main() diff --git a/Cigarette/jinye_scraper.py b/Cigarette/jinye_scraper.py new file mode 100644 index 0000000..ee3447f --- /dev/null +++ b/Cigarette/jinye_scraper.py @@ -0,0 +1,453 @@ +#!/usr/bin/env python3 +""" +卷烟平台数据采集脚本 v4（跨平台 + 数据/图片合并） + +- Cookie 从环境变量 JY_CK 读取 +- 以产品名称为主键，增量更新模板 +- 图片用原始文件名，可并发下载 +- 特定列自动设置数字格式和公式 +- 每 20 条自动保存 +- 跨平台：Windows / Linux (青龙) 均可运行 + +用法: + python jinye_scraper.py # 使用脚本同目录下的模板 + python jinye_scraper.py --template ./模板.xlsx # 指定模板 + python jinye_scraper.py --no-images # 只采集数据，不下图片 + python jinye_scraper.py --images-only # 只下载图片（不处理数据） + python jinye_scraper.py --skip-existing # 跳过已存在产品（加速） + python jinye_scraper.py --delay 0.1 # API 请求间隔 +""" + +import os, sys, json, time, re, argparse +from pathlib import Path +from urllib.parse import unquote, urlparse +from datetime import datetime +from concurrent.futures import ThreadPoolExecutor, as_completed + +import httpx +import openpyxl + +# ── 常量 ────────────────────────────────────────────── +BASE_URL = "https://www.jinye.cn/marketing-orderplatform" +LIST_API = f"{BASE_URL}/Analysis/Cgtdisp/getCgtdispDataList.json" +DETAIL_API = f"{BASE_URL}/Analysis/Cgtdisp/getCgtdispDetail.json" +FUNCODE = "MOHD0418" +SAVE_INTERVAL = 20 + +# 格式列编号 +INT_COLS = {4, 14} # D 小盒条码, N 条装条码 → '0' +DEC_COLS = {6, 7, 8, 9, 16, 17, 19} # 数值列 → '#,##0.##' +DATE_COL = 30 # ^ 批复日期 → 'yyyy-mm-dd' +FORMULA_COL = 10 # J 烟支分类公式 + +# 列号 → API 字段路径 +COL_FIELD = { + 1: "cgtinfo.baseinfo.vbrandname", # A 品牌 + 2: "cgtinfo.baseinfo.ccgttypename", # B 产品类型 + 3: "cgtinfo.baseinfo.vcgtname", # C 产品名称 + 4: "cgtinfo.baseinfo.vcgtboxcode", # D 小盒条码 + 5: "cgtinfo.baseinfo.vcgtlengthname", # E 规格长度 + 6: "cgtinfo.baseinfo.ncgtgirth", # F 烟支周长(mm) + 7: "cgtinfo.baseinfo.ncgttarcontent", # G 焦油含量(mg) + 8: "cgtinfo.baseinfo.ncgtnicotinic", # H 烟气烟碱量(mg) + 9: "cgtinfo.baseinfo.ncgtco", # I 烟气一氧化碳量(mg) + 13: "cgtinfo.baseinfo.vcgtpacktypename", # M 包装类型 + 14: "cgtinfo.baseinfo.vcgtcode", # N 条装条码 + 16: "cgtinfo.baseinfo.ncgtpackagenum", # P 条包装支数 + 17: "nretailprice", # Q 官方指导价 + 19: "nwholesaleprice", # S 官方进价 + 20: "ntransferwprice", # T 调货价 + 27: "cgtinfo.baseinfo.ispearl", # [ 爆珠 + 28: "cgtinfo.baseinfo.isabnormaltype", # \ 异型 + 29: "cgtinfo.baseinfo.vcgtcolor", # ] 主体颜色 + 30: "cgtinfo.baseinfo.vmarketdate", # ^ 批复日期 + 31: "cgtinfo.baseinfo.ownerorgname", # _ 所属工业 + 32: "cgtinfo.baseinfo.ccgtprtypename", # ` 卷烟价类 +} + +# ── 工具函数 ────────────────────────────────────────── + +def safe_product_dir(name: str) -> str: + """产品名 → 安全文件夹名""" + for ch in r'/\:*?"<>|': + name = name.replace(ch, " ") + return name + +def extract_filename(url: str) -> str: + path = urlparse(url).path + raw = path.rsplit("/", 1)[-1] + return unquote(raw) if raw else f"img{int(time.time())}.jpg" + +def _g(detail, path): + """从嵌套字典取值""" + if detail is None: return "" + val = detail + for p in path.split("."): + if isinstance(val, dict): val = val.get(p) + else: return "" + if val is None: return "" + return val + +def parse_cookies(s: str) -> dict: + c = {} + for item in s.split("; "): + item = item.strip() + if "=" in item: + k, v = item.split("=", 1) + c[k] = v + return c + +def make_client(cookies: dict) -> httpx.Client: + return httpx.Client(cookies=cookies, headers={ + "User-Agent": "Mozilla/5.0", + "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", + "Referer": f"{BASE_URL}/ui/pages/list.html?funCode={FUNCODE}" + }, timeout=30) + +# ── API ─────────────────────────────────────────────── + +def fetch_all(client: httpx.Client, ps: int, delay: float) -> list[dict]: + all_items, page = [], 1 + while True: + payload = {"pk_ownerorg":"","pk_brand":"","pk_cigarette":"","wholesaleprice":"", + "leftpricerange":"","rightpricerange":"","moreCondition":"all", + "pageNumber":str(page),"pageSize":str(ps),"keyWord":""} + data = {"_JSONPARA": json.dumps(payload, ensure_ascii=False), "funCode": FUNCODE} + r = client.post(LIST_API, data=data); r.raise_for_status() + items = r.json()["data"].get("items", []) + if not items: break + for it in items: + pk = it.get("pk_cigarette",""); name = it.get("vcgtname","") + if pk and name: all_items.append({"pk_cigarette": pk, "vcgtname": name}) + if len(items) < ps: break + page += 1; time.sleep(delay) + return all_items + +def fetch_detail(client: httpx.Client, pk: str) -> dict | None: + payload = {"pk_cigarette": pk} + data = {"_JSONPARA": json.dumps(payload, ensure_ascii=False), "funCode": FUNCODE} + for attempt in range(3): + try: + r = client.post(DETAIL_API, data=data); r.raise_for_status() + result = r.json() + if result.get("msg") == "操作成功！": + return result.get("data", {}).get("cgtdispDetail") + except: time.sleep(2) if attempt < 2 else None + return None + +# ── Excel 写入 ──────────────────────────────────────── + +def write_cell(ws, row: int, col: int, value): + """写入单元格并设置格式/公式""" + cell = ws.cell(row, col) + + if col == FORMULA_COL: + cell.value = (f'=IF(AND(F{row}>=16,F{row}<18),"细支",' + f'IF(AND(F{row}>=19,F{row}<=22),"中支",' + f'IF(AND(F{row}>=23,F{row}<25),"粗支","非常规")))') + return + + if col == DATE_COL and value and str(value).strip(): + try: + dt = datetime.strptime(str(value).strip(), "%Y-%m-%d") + cell.value = dt + except ValueError: + cell.value = str(value).strip() + cell.number_format = 'yyyy-mm-dd' + return + + if col in INT_COLS: + s = str(value).strip() + if s: + try: cell.value = int(float(s)) + except ValueError: cell.value = s + cell.number_format = '0' + return + + if col in DEC_COLS: + s = str(value).strip() + if s: + try: cell.value = float(s) + except ValueError: cell.value = s + cell.number_format = '#,##0.##' + return + + cell.value = str(value).strip() if value else "" + +# ── 图片下载 ────────────────────────────────────────── + +def download_single(url: str, folder: Path) -> bool: + fname = extract_filename(url) + fpath = folder / fname + if fpath.exists(): return False + try: + r = httpx.get(url, timeout=15) + if r.status_code == 200: + fpath.write_bytes(r.content) + return True + except: pass + return False + +def download_images_for_product(pk, name, images_root: Path, cookies: dict): + """为单个产品下载图片（并发单元）""" + folder = images_root / safe_product_dir(name) + # 检查是否已有非编号图片 + existing = [f for f in os.listdir(str(folder)) if not re.match(r'^\d{2,3}\.jpg$', f)] if folder.exists() else [] + if existing: + return name, 0, len(existing) + + client = make_client(cookies) + detail = fetch_detail(client, pk) + client.close() + if not detail: return name, 0, 0 + + img_urls = detail.get("imgPaths") or [] + if not img_urls: return name, 0, 0 + + folder.mkdir(parents=True, exist_ok=True) + downloaded = 0 + for url in img_urls: + if download_single(url, folder): + downloaded += 1 + return name, downloaded, 0 + +# ── 主流程 ──────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser(description="卷烟平台数据采集 v4") + parser.add_argument("--template", default="", help="模板路径，默认脚本同目录") + parser.add_argument("--output", default="", help="输出路径，默认覆盖模板") + parser.add_argument("--page-size", type=int, default=20) + parser.add_argument("--delay", type=float, default=0.2) + parser.add_argument("--no-images", action="store_true", help="跳过图片下载") + parser.add_argument("--images-only", action="store_true", help="仅下载图片（不处理数据）") + parser.add_argument("--skip-existing", action="store_true", help="跳过已存在产品") + parser.add_argument("--img-workers", type=int, default=5, help="图片下载并发数") + args = parser.parse_args() + + # 模板路径 + script_dir = Path(__file__).parent + if not args.template: + args.template = str(script_dir / "卷烟产品数据系统_v1.xlsx") + if not os.path.exists(args.template): + print(f"错误: 模板不存在 {args.template}"); sys.exit(1) + + template_path = Path(args.template) + template_dir = template_path.parent + images_root = template_dir / "卷烟产品详情图" + output_path = args.output if args.output else str(template_path) + + cookie_str = os.environ.get("JY_CK", "") + if not cookie_str: print("错误: JY_CK 未设置"); sys.exit(1) + cookies = parse_cookies(cookie_str) + + # ── 仅图片模式 ───────────────────────────────────── + if args.images_only: + run_images_only(template_path, images_root, cookies, args) + return + + # ── 正常模式：数据 + 图片 ────────────────────────── + client = make_client(cookies) + + # 清理旧编号图片 + if images_root.exists(): + for d in os.listdir(str(images_root)): + dp = images_root / d + if not dp.is_dir(): continue + for f in os.listdir(str(dp)): + if re.match(r'^\d{2,3}\.jpg$', f): + try: (dp / f).unlink() + except: pass + + # 打开模板 + wb = openpyxl.load_workbook(str(template_path)) + ws = wb.active + + # 构建已有产品索引 + existing = {} + for row in range(2, ws.max_row + 1): + n = ws.cell(row, 3).value + if n: + name_key = str(n).strip() + if name_key.startswith("=HYPERLINK"): + m = re.search(r',"([^"]*)"\)', name_key) + if m: name_key = m.group(1) + existing[name_key] = row + print(f"模板已有 {len(existing)} 条数据") + + # 获取全量列表 + print("获取产品列表...") + all_items = fetch_all(client, args.page_size, args.delay) + print(f"金叶网共 {len(all_items)} 条产品") + + new_cnt = upd_cnt = skip_cnt = 0 + next_row = ws.max_row + 1 + fmt_cols = INT_COLS | DEC_COLS | {DATE_COL} + img_tasks = [] # 待下载图片的产品 + + for i, item in enumerate(all_items, 1): + pk, name = item["pk_cigarette"], item["vcgtname"] + + # ── 已存在产品：补缺失字段 ── + if name in existing: + if args.skip_existing: + continue + target_row = existing[name] + detail = fetch_detail(client, pk) + if detail: + fills = 0 + for col_idx, field_path in COL_FIELD.items(): + cur = ws.cell(target_row, col_idx).value + api_val = _g(detail, field_path) + api_str = str(api_val).strip() if api_val is not None else "" + if col_idx in fmt_cols: + val = api_val if api_str else (cur if cur is not None else "") + write_cell(ws, target_row, col_idx, val) + fills += 1 + else: + if api_str and (cur is None or str(cur).strip() == ""): + write_cell(ws, target_row, col_idx, api_val) + fills += 1 + write_cell(ws, target_row, FORMULA_COL, None) + if fills: + upd_cnt += 1 + print(f"[{i}/{len(all_items)}] {name} 补缺 (+{fills}字段)") + time.sleep(args.delay) + if i % 100 == 0: + print(f"[{i}/{len(all_items)}] 扫描中 | 补{upd_cnt}") + if upd_cnt > 0: wb.save(output_path) + continue + + # ── 新产品：完整填充 ── + detail = fetch_detail(client, pk) + if detail is None: + print(f"[{i}/{len(all_items)}] {name} 详情失败，跳过") + skip_cnt += 1; continue + + target_row = next_row; next_row += 1 + existing[name] = target_row + new_cnt += 1 + + fills = 0 + for col_idx, field_path in COL_FIELD.items(): + api_val = _g(detail, field_path) + api_str = str(api_val).strip() if api_val is not None else "" + if col_idx in fmt_cols: + val = api_val if api_str else "" + write_cell(ws, target_row, col_idx, val) + fills += 1 + elif api_str: + write_cell(ws, target_row, col_idx, api_val) + fills += 1 + write_cell(ws, target_row, FORMULA_COL, None) + + # C 列：先写产品名，图片下载后再替换为 HYPERLINK + ws.cell(target_row, 3).value = name + + # 图片：记录任务（或同步下载） + img_urls = detail.get("imgPaths") or [] + if img_urls and not args.no_images: + safe_name = safe_product_dir(name) + folder = images_root / safe_name + folder.mkdir(parents=True, exist_ok=True) + dl_count = 0 + for url in img_urls: + if download_single(url, folder): dl_count += 1 + if dl_count or any(not re.match(r'^\d{2,3}\.jpg$', f) for f in os.listdir(str(folder)) if folder.exists()): + ws.cell(target_row, 3).value = f'=HYPERLINK("卷烟产品详情图/{safe_name}","{name}")' + + print(f"[{i}/{len(all_items)}] {name} 新增 (+{fills}字段) | 新{new_cnt} 跳{skip_cnt}") + + if (new_cnt + skip_cnt) % SAVE_INTERVAL == 0: + wb.save(output_path) + print(f" --- 已保存 ({new_cnt + skip_cnt}/{len(all_items) - len(existing)}) ---") + + time.sleep(args.delay) + + client.close() + wb.save(output_path) + + print(f"\n数据完成: 新增 {new_cnt} | 补缺 {upd_cnt} | 跳过 {skip_cnt} | 共 {ws.max_row - 1} 行") + print(f"Excel: {output_path}") + print(f"图片: {images_root}") + +# ── 仅图片模式 ──────────────────────────────────────── + +def run_images_only(template_path, images_root, cookies, args): + """独立图片下载模式""" + import openpyxl as _xl + wb = _xl.load_workbook(str(template_path)) + ws = wb.active + + products = [] + for row in range(2, ws.max_row + 1): + name = ws.cell(row, 3).value + if not name: continue + name_str = str(name).strip() + if name_str.startswith("=HYPERLINK"): + m = re.search(r',"([^"]*)"\)', name_str) + if m: name_str = m.group(1) + products.append(name_str) + + print(f"Excel 共 {len(products)} 个产品，获取 ID 映射...") + client = make_client(cookies) + name_to_pk = {} + page = 1 + while True: + payload = {"pk_ownerorg":"","pk_brand":"","pk_cigarette":"","wholesaleprice":"", + "leftpricerange":"","rightpricerange":"","moreCondition":"all", + "pageNumber":str(page),"pageSize":"100","keyWord":""} + data = {"_JSONPARA": json.dumps(payload, ensure_ascii=False), "funCode": FUNCODE} + r = client.post(LIST_API, data=data) + items = r.json()["data"].get("items", []) + if not items: break + for it in items: + name_to_pk[it["vcgtname"]] = it["pk_cigarette"] + page += 1; time.sleep(0.1) + client.close() + print(f"ID 映射: {len(name_to_pk)} 个") + + tasks = [(name_to_pk[n], n) for n in products if n in name_to_pk] + print(f"待处理: {len(tasks)} 个产品，{args.img_workers} 线程并发") + + # 清理旧编号图片 + if images_root.exists(): + for d in os.listdir(str(images_root)): + dp = images_root / d + if not dp.is_dir(): continue + for f in os.listdir(str(dp)): + if re.match(r'^\d{2,3}\.jpg$', f): + try: (dp / f).unlink() + except: pass + + total_dl = skipped = 0 + with ThreadPoolExecutor(max_workers=args.img_workers) as executor: + futures = {executor.submit(download_images_for_product, pk, name, images_root, cookies): name for pk, name in tasks} + for i, future in enumerate(as_completed(futures), 1): + name, dl, skip = future.result() + total_dl += dl; skipped += skip + if i % 50 == 0 or dl > 0: + print(f"[{i}/{len(tasks)}] {name}: 下载{dl} 跳过{skip}") + + print(f"\n图片完成: 下载 {total_dl} | 已有跳过 {skipped}") + + # 更新 HYPERLINK + print("更新超链接...") + wb = _xl.load_workbook(str(template_path)) + ws = wb.active + updated = 0 + for row in range(2, ws.max_row + 1): + val = ws.cell(row, 3).value + if val is None: continue + s = str(val).strip() + if s.startswith("=HYPERLINK"): continue + safe_name = safe_product_dir(s) + folder = images_root / safe_name + if folder.is_dir() and os.listdir(str(folder)): + ws.cell(row, 3).value = f'=HYPERLINK("卷烟产品详情图/{safe_name}","{s}")' + updated += 1 + wb.save(str(template_path)) + print(f"超链接更新: {updated} 个") + +if __name__ == "__main__": + main() diff --git a/Cigarette/卷烟产品数据系统_v1.xlsx b/Cigarette/卷烟产品数据系统_v1.xlsx new file mode 100644 index 0000000..5031ea4 Binary files /dev/null and b/Cigarette/卷烟产品数据系统_v1.xlsx differ