BOM：R2 物料清单与淘宝 / 立创比价（表格 + 抓价脚本与数据）

Co-Authored-By: Claude Opus 5.5 <[email protected]>

BOM/_src/visit.py

@@ -0,0 +1,91 @@
+import asyncio, json, os, re, math, time
+import cdp
+exec(open('build.py', encoding='utf-8').read().split('def relevant')[0])  # CORE, SPECIAL, norm, cap_tokens ...
+PLAN = json.load(open('cost_plan.json', encoding='utf-8'))
+Qd = {x['sid']: x for x in json.load(open('queries.json', encoding='utf-8'))}
+OUT = 'cost_detail.json'
+out = json.load(open(OUT, encoding='utf-8')) if os.path.exists(OUT) else {}
+
+SKUJS = r'''(()=>{const r=(window.__ICE_APP_CONTEXT__||{}).loaderData?.home?.data?.res; if(!r) return null;
+const props=(r.skuBase?.props||[]).map(p=>({name:p.name,values:(p.values||[]).map(v=>v.name)}));
+return JSON.stringify({title:r.item?.title,props})})()'''
+CLICK = r'''((txt)=>{const el=[...document.querySelectorAll("span,div")].filter(e=>e.children.length<=2&&e.innerText&&e.innerText.trim()===txt);if(!el.length)return false;el[el.length-1].click();return true})'''
+
+def tokens(p):
+    sid, q = p['sid'], Qd[p['sid']]['q']
+    if '贴片电阻' in q:
+        v = q.split()[1].upper(); m = re.match(r'([\d.]+)([KMR])', v); num, u = m.group(1), m.group(2)
+        f = {v, num + u, num + u + 'Ω'}
+        if '.' in num: a, b = num.split('.'); f.add(f'{a}{u}{b}')
+        return [f], [q.split()[0]]
+    if '贴片电容' in q:
+        return [cap_tokens(p['desc'])], [re.search(r'(\d+(\.\d)?V)', p['desc']).group(1).upper()]
+    c = CORE.get(sid, '')
+    return [{c}] if c else [], []
+
+def score_sku(name, p):
+    must, nice = tokens(p)
+    U = name.upper()
+    s = 0
+    for grp in must:
+        if any(re.search(r'(?<![\d.])' + re.escape(t.upper()) + r'(?![\d])', U) or (len(t) >= 5 and norm(t) in norm(name)) for t in grp if t):
+            s += 5
+    for t in nice:
+        if re.search(r'(?<![\d.])' + re.escape(t.upper()), U): s += 2
+    return s
+
+def parse_price(txt, title):
+    i = txt.find('￥', max(0, txt.find(title[:12])) if title else 0)
+    if i < 0: i = txt.find('￥')
+    if i < 0: return None
+    seg = txt[i + 1:i + 60].split('\n'); val = ''
+    for s in seg:
+        s = s.strip()
+        if not s: continue
+        if re.fullmatch(r'[\d.]+|\.\d+|-|~', s): val += s
+        elif val: break
+        else: break
+    return val or None
+
+async def main():
+    ws = await cdp.open_ws()
+    for n, p in enumerate(PLAN):
+        if p['sid'] in out and out[p['sid']].get('price'): continue
+        rec = None
+        for c in p['cands'][:2]:
+            url = ('https://detail.tmall.com/item.htm?id=' if c['tmall'] else 'https://item.taobao.com/item.htm?id=') + c['id']
+            await cdp.send(ws, 'Page.navigate', {'url': url})
+            info = None
+            for k in range(20):
+                await asyncio.sleep(1)
+                v = await cdp.ev(ws, SKUJS)
+                if v: info = json.loads(v); break
+            if not info:
+                t = await cdp.ev(ws, 'location.href+" "+document.title')
+                print('  no page', c['id'], t, flush=True)
+                if t and ('login' in t or '验证' in t or 'punish' in t):
+                    print('CAPTCHA/LOGIN — waiting', flush=True)
+                    for w in range(120):
+                        await asyncio.sleep(5)
+                        if await cdp.ev(ws, SKUJS): break
+                await asyncio.sleep(30); continue
+            chosen = []
+            for pr in info['props']:
+                vals = pr['values']
+                if len(vals) <= 1:
+                    chosen.append(vals[0] if vals else None); continue
+                best = max(vals, key=lambda x: (score_sku(x, p), -vals.index(x)))
+                chosen.append(best)
+                await cdp.ev(ws, '(' + CLICK + ')(' + json.dumps(best) + ')')
+                await asyncio.sleep(1.2)
+            txt = await cdp.ev(ws, 'document.body.innerText') or ''
+            price = parse_price(txt, info['title'] or '')
+            ship = re.search(r'快递[:：]\s*([^\n]+)', txt)
+            rec = dict(url=url, shop=c['shop'], title=info['title'], props=info['props'], chosen=chosen,
+                       sku_score=[score_sku(x, p) for x in chosen if x], price=price, ship=ship.group(1) if ship else None)
+            break
+        out[p['sid']] = rec or {'price': None}
+        json.dump(out, open(OUT, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
+        print(n, p['mid'], rec and (rec['price'], rec['chosen'], rec['ship']), flush=True)
+        await asyncio.sleep(30)
+asyncio.run(main())