434 lines
17 KiB
Python
434 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
卷烟平台数据采集脚本 v4（通用格式 + 强制禁用代理）
|
||
- 数字列格式设为 'General'（G/通用格式）
|
||
- 强制忽略系统代理（HTTP_PROXY/HTTPS_PROXY）
|
||
- 修复调货价（第20列）转数字
|
||
- 从第一个空白行开始写入
|
||
"""
|
||
|
||
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 条装条码 → 整数
|
||
DEC_COLS = {6, 7, 8, 9, 16, 17, 19, 20} # 数值列 → 通用格式
|
||
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:
|
||
# 强制禁用代理，避免走 SOCKS/HTTP 代理
|
||
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,
|
||
proxies=None # ← 关键修改：忽略系统代理
|
||
)
|
||
|
||
# ── 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 = 'General' # G/通用格式
|
||
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
|
||
fmt_cols = INT_COLS | DEC_COLS | {DATE_COL}
|
||
|
||
# 从第一个空行开始写入
|
||
next_row = 2
|
||
while next_row <= ws.max_row and ws.cell(next_row, 3).value is not None:
|
||
next_row += 1
|
||
|
||
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)
|
||
|
||
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}")
|
||
|
||
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() |