688 lines
26 KiB
Python
688 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
hn.jinye.cn 卷烟订货平台数据采集脚本 v4（去星号匹配版）
|
||
- 匹配产品名称时自动忽略星号（*）
|
||
- 条码格式为 '0'，其他数值为 General
|
||
- 增量更新，变更记录在 AH 列
|
||
- 烟气烟碱量、一氧化碳量留空
|
||
"""
|
||
|
||
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
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 常量配置
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
BASE_URL = "https://hn.jinye.cn/wdk"
|
||
API_URL = f"{BASE_URL}?action=ecw.page&method=call_service"
|
||
|
||
DETAIL_SRVNAME = "service.ecw.tbc.product.show.v5"
|
||
DETAIL_SRVMETHOD = "queryProductInfo"
|
||
LIST_SRVNAME = "service.ecw.portlet.catalog.v5"
|
||
LIST_SRVMETHOD = "pageQueryWithUserType"
|
||
|
||
SAVE_INTERVAL = 20
|
||
|
||
# ── Excel 列映射 ──
|
||
COL_MAPPING = {
|
||
1: ("api", "品牌", "brand_name"),
|
||
2: ("api", "产品类型", "product_type_codename"),
|
||
3: ("api", "产品名称", "product_name"),
|
||
4: ("api", "小盒条码", "bar_code"),
|
||
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","条包装支数", ""),
|
||
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","爆珠", ""),
|
||
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}
|
||
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 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"
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# wppm 编码
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
B64_TABLE = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/="
|
||
|
||
def _js_escape(s: str) -> str:
|
||
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]:
|
||
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:
|
||
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
|
||
else:
|
||
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:
|
||
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:
|
||
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:
|
||
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 = os.environ.get("JY_HN_ACCOUNT_ID", "")
|
||
if not account_id:
|
||
print("[警告] 未设置 JY_HN_ACCOUNT_ID 环境变量")
|
||
return {
|
||
"account_id": account_id,
|
||
"cust_uuid": cust_uuid,
|
||
"manageunituuid": manage_unituuid,
|
||
}
|
||
|
||
def _build_api_payload(srvname: str, srvmethod: str, data: dict) -> dict:
|
||
inner = {
|
||
"_SRVNAME": srvname,
|
||
"_SRVMETHOD": srvmethod,
|
||
"_DATA": json.dumps(data, ensure_ascii=False, separators=(",", ":")),
|
||
}
|
||
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:
|
||
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:
|
||
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]:
|
||
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", "Referer": BASE_URL + "/"},
|
||
timeout=20,
|
||
follow_redirects=False,
|
||
trust_env=False,
|
||
)
|
||
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:
|
||
tags = list_item.get("tags", [])
|
||
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", ""),
|
||
"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", ""),
|
||
"spec_length": "", # 固定留空
|
||
"filter_color": api_data.get("filter_color", ""),
|
||
}
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Excel 写入
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def write_cell(ws, row: int, col: int, value):
|
||
cell = ws.cell(row, col)
|
||
if col == FORMULA_COL:
|
||
cell.value = (
|
||
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),"粗支","非常规"))))'
|
||
)
|
||
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 = 'General'
|
||
return
|
||
cell.value = str(value).strip() if value else ""
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 加载已有产品（忽略星号）
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def load_existing_products(ws) -> dict:
|
||
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)
|
||
# 去除所有星号，用于匹配
|
||
name_key = name_key.replace("*", "")
|
||
existing[name_key] = row
|
||
return existing
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 主流程
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def main():
|
||
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}")
|
||
sys.exit(1)
|
||
|
||
output_path = args.output if args.output else str(template_path)
|
||
|
||
cookie_str = os.environ.get("JY_HN_CK", "") or os.environ.get("JY_CK", "")
|
||
if not cookie_str:
|
||
print("错误: 未设置 JY_HN_CK 环境变量")
|
||
sys.exit(1)
|
||
|
||
user_info = _extract_user_info(cookie_str)
|
||
if not user_info.get("account_id"):
|
||
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['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阶段一：列表 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)
|
||
|
||
# 详情采集
|
||
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 = 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("*", "")
|
||
|
||
# 构造数据（焦油、烟碱、一氧化碳均留空）
|
||
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", [])
|
||
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_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 = ""
|
||
|
||
row_data = {
|
||
1: brand,
|
||
2: product_type,
|
||
3: clean_name,
|
||
4: bar_code,
|
||
5: "", # 规格长度留空
|
||
7: "", # 焦油含量留空（不提取）
|
||
8: "", # 烟气烟碱量留空
|
||
13: style,
|
||
14: bar_code2,
|
||
17: retail,
|
||
19: wholesale,
|
||
27: has_tag(tags, "爆珠"),
|
||
28: has_tag(tags, "异型"),
|
||
31: factory,
|
||
32: price_type,
|
||
}
|
||
|
||
# ── 已存在产品 ──
|
||
if clean_name in existing:
|
||
target_row = existing[clean_name]
|
||
fills = 0
|
||
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
|
||
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[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)
|
||
|
||
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)
|
||
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}")
|
||
print(f"{'='*50}")
|
||
|
||
if __name__ == "__main__":
|
||
main() |