#!/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()