From 0ad7e7a850ac2bede22112dde7abb199e9b3af50 Mon Sep 17 00:00:00 2001 From: timxx <3337598077@qq.com> Date: Sat, 1 Aug 2026 20:14:58 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=20Cigarette/hn=5Fjinye=5Fscr?= =?UTF-8?q?aper.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cigarette/hn_jinye_scraper.py | 797 ++++++++++++---------------------- 1 file changed, 281 insertions(+), 516 deletions(-) diff --git a/Cigarette/hn_jinye_scraper.py b/Cigarette/hn_jinye_scraper.py index 41b9664..92e6e14 100644 --- a/Cigarette/hn_jinye_scraper.py +++ b/Cigarette/hn_jinye_scraper.py @@ -1,25 +1,10 @@ #!/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浏览器（本地测试用） +hn.jinye.cn 卷烟订货平台数据采集脚本 v4（去星号匹配版） +- 匹配产品名称时自动忽略星号（*） +- 条码格式为 '0'，其他数值为 General +- 增量更新，变更记录在 AH 列 +- 烟气烟碱量、一氧化碳量留空 """ import os @@ -36,59 +21,39 @@ 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" +LIST_SRVNAME = "service.ecw.portlet.catalog.v5" +LIST_SRVMETHOD = "pageQueryWithUserType" -TOTAL_PAGES = 24 -ITEMS_PER_PAGE = 16 -TOTAL_PRODUCTS = 378 -SAVE_INTERVAL = 20 # 每 N 条保存一次 -BROWSER_TIMEOUT = 30000 # 30 秒 +SAVE_INTERVAL = 20 -# ── 列表页字段提取（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"=不可用 +# ── Excel 列映射 ── 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 未提供 + 5: ("none","规格长度", ""), # 留空 + 6: ("none","烟支周长", ""), + 7: ("api", "焦油含量", ""), + 8: ("none","烟气烟碱量", ""), # 不提取 + 9: ("none","烟气一氧化碳量", ""), 10: ("formula","烟支分类", ""), 11: ("none","每盒数量", ""), 12: ("none","小盒价格", ""), 13: ("api", "包装类型", "product_style_codename"), 14: ("api", "条装条码", "bar_code2"), 15: ("none","条装盒数", ""), - 16: ("none","条包装支数", ""), # API 未提供 + 16: ("none","条包装支数", ""), 17: ("api", "官方指导价", "retail_price"), 18: ("none","条装售价", ""), 19: ("api", "官方进价", "whole_sale_price"), @@ -99,87 +64,48 @@ COL_MAPPING = { 24: ("none","最高出货价", ""), 25: ("none","出货价更新时间", ""), 26: ("none","热度", ""), - 27: ("list","爆珠", "tags"), - 28: ("list","异型", "tags"), - 29: ("none","主体颜色", ""), # API 未提供 - 30: ("none","批复日期", ""), # API 未提供 + 27: ("list","爆珠", ""), + 28: ("list","异型", ""), + 29: ("none","主体颜色", ""), + 30: ("none","批复日期", ""), 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 # 烟支分类公式 +INT_COLS = {4, 14} # 条码格式 '0' +DEC_COLS = {6, 7, 8, 9, 16, 17, 19} +DATE_COL = 30 +FORMULA_COL = 10 +NOTE_COL = 34 # AH列 # ═══════════════════════════════════════════════════════════ # 工具函数 # ═══════════════════════════════════════════════════════════ 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 请求参数封装） +# wppm 编码 # ═══════════════════════════════════════════════════════════ -# ── Base64.encode2 自定义字符表 ── B64_TABLE = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=" - def _js_escape(s: str) -> str: - """模拟 JavaScript escape() 函数""" result = [] for c in s: o = ord(c) @@ -191,9 +117,7 @@ def _js_escape(s: str) -> str: 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) @@ -208,9 +132,7 @@ def _ucs2_utf8(s: str) -> list[int]: 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) @@ -229,25 +151,18 @@ def _b64_encode2(s: str) -> str: b2 |= (tmp & 0xc0) >> 6 b3 = tmp & 0x3f else: - b3 = 64 # padding + b3 = 64 else: - b2 = b3 = 64 # padding + b2 = b3 = 64 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: @@ -258,45 +173,26 @@ def _decode_jwt_payload(jwt_token: str) -> dict: 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=") - + print("[警告] 未设置 JY_HN_ACCOUNT_ID 环境变量") return { "account_id": account_id, - "personuuid": cust_uuid, + "cust_uuid": 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）""" +def _build_api_payload(srvname: str, srvmethod: str, data: dict) -> dict: inner = { "_SRVNAME": srvname, "_SRVMETHOD": srvmethod, @@ -304,22 +200,83 @@ def _build_api_payload(srvname: str, srvmethod: str, data: dict, user_info: dict } return {"wppm": _encode_wppm(inner)} +# ═══════════════════════════════════════════════════════════ +# 列表 API 采集 +# ═══════════════════════════════════════════════════════════ + +def fetch_product_list_via_api(cookie_str: str, user_uuid: str, account_id: str) -> list[dict]: + import time as _time + ajax_ts = str(int(_time.time() * 1000)) + url = f"{API_URL}&ajaxparamtime={ajax_ts}" + inner = { + "_SRVNAME": LIST_SRVNAME, + "_SRVMETHOD": LIST_SRVMETHOD, + "_DATA": json.dumps({ + "user_type": "CUSTOMER", + "user_uuid": user_uuid, + "op_acc": account_id, + "search_fields": {}, + "order_fields": "" + }), + "_RSTYPE": "grid", + "_RSFIELD": "resultset", + "_RSPARAM": json.dumps({"pagequery": "1"}), + "page": 1, + "rows": 500, + } + cookies = {} + for item in cookie_str.split("; "): + 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", + "Referer": BASE_URL + "/", + }, + timeout=30, + follow_redirects=False, + trust_env=False, + ) + items = [] + try: + r = client.post(url, data={"wppm": _encode_wppm(inner)}, + headers={"Content-Type": "application/x-www-form-urlencoded"}) + if r.status_code == 200: + result = r.json() + rows = result.get("rows", []) + if rows: + for row in rows: + product_uuid = row.get("product_uuid", "") + product_name = row.get("product_name", "") + items.append({ + "uuid": product_uuid, + "product_name": product_name, + "factory": "", + "wholesale": "", + "retail": "", + "tags": [], + }) + print(f"[列表API] 成功获取 {len(items)} 条产品") + else: + print(f"[列表API] 无数据: {result.get('desc', '')}") + else: + print(f"[列表API] HTTP {r.status_code}") + except Exception as e: + print(f"[列表API] 请求失败: {e}") + finally: + client.close() + return items # ═══════════════════════════════════════════════════════════ # 详情 API 采集 # ═══════════════════════════════════════════════════════════ -def fetch_detail_via_api( - client: httpx.Client, product_uuid: str, user_info: dict -) -> dict | None: - """ - 通过 call_service API 获取单个产品的完整详情。 - 返回 resultset 字典，失败返回 None。 - """ +def fetch_detail_via_api(client: httpx.Client, product_uuid: str, user_info: dict) -> dict | 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, @@ -329,17 +286,12 @@ def fetch_detail_via_api( "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, - ) + 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") @@ -350,40 +302,21 @@ def fetch_detail_via_api( _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 +def fetch_details_concurrent(cookie_str: str, list_items: list[dict], user_info: dict, + workers: int, delay: float) -> list[dict]: 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, - }, + headers={"User-Agent": "Mozilla/5.0", "Referer": BASE_URL + "/"}, timeout=20, - follow_redirects=False, # 不跟随重定向（200 才是成功） + follow_redirects=False, + trust_env=False, ) - detailed = [None] * len(list_items) total = len(list_items) @@ -391,7 +324,6 @@ def fetch_details_concurrent( 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) @@ -400,12 +332,8 @@ def fetch_details_concurrent( 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) - } + 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() @@ -413,92 +341,47 @@ def fetch_details_concurrent( 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", ""), + "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, + "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(ISNUMBER(FIND("雪茄",B{row})),"雪茄",' + 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),"粗支","非常规")))' + 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") @@ -509,8 +392,6 @@ def write_cell(ws, row: int, col: int, value): cell.value = str(value).strip() cell.number_format = 'yyyy-mm-dd' return - - # 整数列（条码） if col in INT_COLS: s = str(value).strip() if s: @@ -520,8 +401,6 @@ def write_cell(ws, row: int, col: int, value): cell.value = s cell.number_format = '0' return - - # 小数列（价格、理化指标） if col in DEC_COLS: s = str(value).strip() if s: @@ -529,15 +408,15 @@ def write_cell(ws, row: int, col: int, value): cell.value = float(s) except ValueError: cell.value = s - cell.number_format = '#,##0.##' + cell.number_format = 'General' 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 @@ -547,349 +426,250 @@ def load_existing_products(ws) -> dict: m = re.search(r',"([^"]*)"\)', name_key) if m: name_key = m.group(1) + # 去除所有星号，用于匹配 + name_key = name_key.replace("*", "") 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）") + parser = argparse.ArgumentParser() + parser.add_argument("--template", default="") + parser.add_argument("--output", default="") + parser.add_argument("--delay", type=float, default=0.3) + parser.add_argument("--quick", action="store_true") + parser.add_argument("--api-workers", type=int, default=10) 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 指定路径") + print(f"错误: 模板不存在 {args.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 字符串") + print("错误: 未设置 JY_HN_CK 环境变量") 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 采集") + print("错误: 未设置 JY_HN_ACCOUNT_ID 环境变量") + sys.exit(1) + if not user_info.get("cust_uuid"): + print("错误: 无法从 JWT 提取 cust_uuid，Cookie 可能过期") 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')}") - # ── 加载模板 ── + print(f"[user_info] account_id={user_info['account_id']}, cust_uuid={user_info['cust_uuid']}") + 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("\n阶段一：列表 API 采集") + list_items = fetch_product_list_via_api( + cookie_str, + user_info["cust_uuid"], + user_info["account_id"] + ) + if not list_items: + print("错误: 未获取到任何产品") + sys.exit(1) - # 阶段一：列表页采集 - 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}") + # 详情采集 + if args.quick: + detailed_items = list_items + else: + print("\n阶段二：详情API并发采集") + detailed_items = fetch_details_concurrent( + cookie_str, list_items, user_info, args.api_workers, args.delay + ) + # 写入 Excel + print("\n写入 Excel...") existing = load_existing_products(ws) - next_row = ws.max_row + 1 + + # 从第一个空行开始 + next_row = 2 + while next_row <= ws.max_row and ws.cell(next_row, 3).value is not None: + next_row += 1 + new_cnt = upd_cnt = skip_cnt = 0 + # 字段映射（不含焦油含量） + field_names = { + 1: "品牌", 2: "产品类型", 4: "小盒条码", + 13: "包装类型", 14: "条装条码", 17: "官方指导价", + 19: "官方进价", 31: "所属工业", 32: "卷烟价类" + } + # 检查的列（不含焦油含量） + check_cols = [1, 2, 13, 17, 19, 31, 32] + # 数值列（不含焦油含量） + numeric_cols = {17, 19} + for item in detailed_items: name = item.get("product_name", "") if not name: skip_cnt += 1 continue + clean_name = name.replace("*", "") - # 获取数据（优先详情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", "") + style_raw = item.get("product_style_codename", "") + style = "" if style_raw == "其他" else style_raw 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, "异型") + price_type_raw = item.get("price_type_codename", "") + if price_type_raw: + if price_type_raw == "无价类": + price_type = "" + else: + m = re.match(r'([一二三四五]类)烟', price_type_raw) + if m: + price_type = m.group(1) + else: + price_type = price_type_raw + else: + price_type = "" - # 数据字典（Excel 列号 → 值） row_data = { 1: brand, 2: product_type, - 3: name, + 3: clean_name, 4: bar_code, - 5: spec_length, - 7: tar, - 8: nicotine, + 5: "", # 规格长度留空 + 7: "", # 焦油含量留空（不提取） + 8: "", # 烟气烟碱量留空 13: style, 14: bar_code2, 17: retail, 19: wholesale, - 27: is_pearl, - 28: is_abnormal, + 27: has_tag(tags, "爆珠"), + 28: has_tag(tags, "异型"), 31: factory, 32: price_type, } - # ── 已存在产品：补缺 ── - fmt_cols = INT_COLS | DEC_COLS | {DATE_COL} - if name in existing: - target_row = existing[name] + # ── 已存在产品 ── + if clean_name in existing: + target_row = existing[clean_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) + change_notes = [] + + # 条码处理 + for barcode_col in [4, 14]: + old_val = ws.cell(target_row, barcode_col).value + new_val = row_data.get(barcode_col, "") + if old_val is not None and str(old_val).strip() != "": + try: + num_val = int(float(str(old_val).strip())) + ws.cell(target_row, barcode_col).value = num_val + ws.cell(target_row, barcode_col).number_format = '0' + except ValueError: + pass + elif new_val and str(new_val).strip() != "": + try: + num_val = int(float(str(new_val).strip())) + ws.cell(target_row, barcode_col).value = num_val + ws.cell(target_row, barcode_col).number_format = '0' + fills += 1 + except ValueError: + ws.cell(target_row, barcode_col).value = str(new_val).strip() + fills += 1 + + if old_val is not None and new_val: + old_str = str(old_val).strip() + new_str = str(new_val).strip() + try: + old_num = int(float(old_str)) + new_num = int(float(new_str)) + if old_num != new_num: + change_notes.append(f"{field_names[barcode_col]}: {old_str} -> {new_str}") + except ValueError: + if old_str != new_str: + change_notes.append(f"{field_names[barcode_col]}: {old_str} -> {new_str}") + + # 其他字段（不含焦油） + for col_idx in check_cols: + old_val = ws.cell(target_row, col_idx).value + new_val = row_data.get(col_idx, "") + if not new_val or str(new_val).strip() == "": + continue + old_is_empty = (old_val is None or str(old_val).strip() == "") + if old_is_empty: + write_cell(ws, target_row, col_idx, new_val) fills += 1 - elif val and (cur is None or str(cur).strip() == ""): - write_cell(ws, target_row, col_idx, val) - fills += 1 - # 公式 + change_notes.append(f"{field_names[col_idx]}: 空 -> {new_val}") + else: + old_str = str(old_val).strip() + new_str = str(new_val).strip() + if col_idx in numeric_cols: + try: + old_num = float(old_str) + new_num = float(new_str) + if abs(old_num - new_num) > 1e-9: + change_notes.append(f"{field_names[col_idx]}: {old_str} -> {new_str}") + except ValueError: + if old_str != new_str: + change_notes.append(f"{field_names[col_idx]}: {old_str} -> {new_str}") + else: + if old_str != new_str: + change_notes.append(f"{field_names[col_idx]}: {old_str} -> {new_str}") + + if change_notes: + note_cell = ws.cell(target_row, NOTE_COL) + old_note = note_cell.value or "" + new_note = "; ".join(change_notes) + if old_note: + note_cell.value = f"{old_note}; {new_note}" + else: + note_cell.value = new_note + write_cell(ws, target_row, FORMULA_COL, None) + if fills: upd_cnt += 1 + print(f"补缺 [{clean_name}] +{fills}字段") + if change_notes: + print(f"变更记录 [{clean_name}] {len(change_notes)}项") continue # ── 新产品 ── target_row = next_row next_row += 1 - existing[name] = target_row + existing[clean_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}") + for bc in [4, 14]: + cell = ws.cell(target_row, bc) + if cell.value is not None: + try: + cell.value = int(float(str(cell.value).strip())) + except ValueError: + pass + cell.number_format = '0' + + print(f"新增 [{clean_name}]") if new_cnt % SAVE_INTERVAL == 0: wb.save(output_path) @@ -897,27 +677,12 @@ def main(): 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() + main() \ No newline at end of file