atomk-hermes-skills/skills/cross-border-ecommerce/premiumproducts-db/SKILL.md

name, description, version, category, tags

name	description	version	category	tags
premiumproducts-db	premiumproducts.products MongoDB 数据库操作手册 — 连接、Schema、查询、更新、变参产品集创建、CDP+FlairGS 产品资料补齐流水线。	1.0	cross-border-ecommerce	premiumproducts mongodb product-pool variants

premiumproducts.products 数据库操作手册

连接信息

项目	值
地址	43.134.190.229:27018
用户	root
密码	MongoDB8-2026!
数据库	premiumproducts
集合	products
Python	pymongo.MongoClient('mongodb://root:MongoDB8-2026!@43.134.190.229:27018/')

Schema 核心字段

必填字段

• product_code — 8位产品编码（如 1LTDS5ZY, EE9A114A），唯一索引
• name — 中文产品名称（atomlisting 产品列表显示）
• source_url — 1688/detail 源链接
• images — [{url, is_primary}] 对象数组，非字符串数组
• translations — 嵌套 dict：{zh: {name, description}, en: {name, description}, ru: {name, description}}
• price 或 supplier_price — 数字

推荐字段

• category — 产品类目
• brand — 品牌
• material — 材质
• color / colors — 颜色
• weight / weight_g — 重量(g)
• specs — 规格 dict
• variants — SKU 变体数组 [{name, price, stock}]
• store — 店铺信息 {name, years, return_rate, service_score, shipping_rate, positive_rate}
• moq — 最小起订量
• sku — SKU 字符串
• shipping — 发货信息
• price_range — {min, max}
• packaging — 包装信息

变参产品集字段

• variant_group — 父产品 code（如 EE9A114A）
• variant_of — 所属父产品
• variant_type — 变参类型（如 color）
• variant_value — 变参值（如 橙色）
• supplier_price — 供应商价格
• supplier_stock — 供应商库存

父产品标记

• is_variant_group: true
• variant_codes — 子产品 code 数组
• variant_count — 子产品数量

常见操作

查询

from pymongo import MongoClient
client = MongoClient('mongodb://root:MongoDB8-2026!@43.134.190.229:27018/')
db = client['premiumproducts']

# 按 code 查询
doc = db['products'].find_one({'product_code': '1LTDS5ZY'})

# 按变参组查询
variants = list(db['products'].find({'variant_group': 'EE9A114A'}))

# 查找缺失字段
docs = list(db['products'].find({'name': {'$exists': False}}))

更新

result = db['products'].update_one(
    {'product_code': '1LTDS5ZY'},
    {'$set': {'name': '...', 'price': 42.20, 'images': [...]}}
)

字段改名（两步操作，pymongo 某些版本不支持 $rename）

# Step 1: $set 新字段
db['products'].update_one({'product_code': code}, {'$set': {'supplier_price': old_val}})
# Step 2: $unset 旧字段
db['products'].update_one({'product_code': code}, {'$unset': {'price': ''}})

产品资料补齐流水线

流程

1. MongoDB 查询产品 → 获取 source_url
2. Desktop CDP → /cdp/navigate 1688 详情页
3. /cdp/attach → /cdp/evaluate 提取页面数据
4. 解析：document.title, document.body.innerText, 图片扫描
5. 用 FlairGS VL 分析图片颜色/特征（可选）
6. 组装文档并 update_one 写入 MongoDB

CDP 提取模板

import urllib.request, json, time

K = chr(66)+chr(105)+chr(110)+chr(103)+chr(50)+chr(48)+chr(50)+chr(54)+chr(67)+chr(97)+chr(111)+chr(36)+chr(36)+chr(36)
BASE = 'http://127.0.0.1:9228'
SLOT = 'desktop-mrbgzsdn'
SH = {'Authorization': f'Bearer {K}', 'Content-Type': 'application/json', 'X-Desktop-Id': SLOT}

def ev(tid, expr):
    body = json.dumps({'expression': expr}).encode()
    req = urllib.request.Request(f'{BASE}/cdp/evaluate?targetId={tid}', data=body, method='POST')
    for k, v in SH.items(): req.add_header(k, v)
    resp = json.loads(urllib.request.urlopen(req, timeout=20).read())
    r = resp.get('result', {})
    if isinstance(r, dict) and 'result' in r:
        return r['result'].get('value', r['result'])
    return r

# 1. Navigate (always first)
body = json.dumps({'url': source_url}).encode()
req = urllib.request.Request(f'{BASE}/cdp/navigate', data=body, method='POST')
for k, v in SH.items(): req.add_header(k, v)
urllib.request.urlopen(req, timeout=30)

time.sleep(5)

# 2. Attach
req = urllib.request.Request(f'{BASE}/cdp/attach', data=b'{}', method='POST')
for k, v in SH.items(): req.add_header(k, v)
tid = json.loads(urllib.request.urlopen(req, timeout=60).read())['targetId']

# 3. Extract
title = ev(tid, 'document.title')
body_text = ev(tid, 'document.body.innerText.substring(0,4000)')

图片提取

# CDP 表达式过滤器禁止 Array.from/map/filter/for 循环，只能用逐个索引扫描
total = ev(tid, 'document.querySelectorAll("img").length')
urls = []
for i in range(total):
    src = ev(tid, f'document.querySelectorAll("img")[{i}] ? document.querySelectorAll("img")[{i}].src : ""')
    w = ev(tid, f'document.querySelectorAll("img")[{i}] ? document.querySelectorAll("img")[{i}].width : 0')
    if src and 'alicdn' in str(src) and isinstance(w, (int, float)) and w > 100:
        urls.append({'url': str(src).replace('_.webp', ''), 'is_primary': len(urls) == 0})

FlairGS VL 分析

import base64, io
from PIL import Image

key = '8Ax4TyZc66qrZgXK'
api = 'http://43.160.244.125:17870/api/analyze'

# 下载图片 → 转 RGB → 调 API
img_data = urllib.request.urlopen(img_url).read()
img = Image.open(io.BytesIO(img_data)).convert('RGB')
img.thumbnail((512, 512), Image.LANCZOS)
buf = io.BytesIO()
img.save(buf, format='JPEG', quality=80)
b64 = base64.b64encode(buf.getvalue()).decode()

body = json.dumps({'base64': b64, 'prompt': '用中文描述商品颜色'}).encode()
req = urllib.request.Request(api, data=body, method='POST')
req.add_header('X-API-Key', key)
req.add_header('Content-Type', 'application/json')
desc = json.loads(urllib.request.urlopen(req, timeout=60).read())['description']

数据格式陷阱

1. ⚠️ images 必须是 [{url, is_primary}] 对象数组，不能是字符串数组
2. ⚠️ translations 是嵌套 dict {zh: {name, description}}，不能是字符串
3. ⚠️ name 必填 — atomlisting 产品列表用此字段，缺失显示 "Unnamed"
4. ⚠️ atomlisting 双 API：/remote 读 translations.zh.name，/premium-products 读 name，两个都要设
5. ⚠️ product_code 统一 8 位十六进制，不带前缀
6. ⚠️ pymongo 某些版本 $rename 不可用，用 $set + $unset 两步改名
7. ⚠️ 测试/占位链接的产品（source_url 含 test1/test2/test3）主动删除
8. ⚠️ CDP 表达式过滤器禁止 async/for/forEach/Array.from，索引扫描用 while 递减循环
9. ⚠️ FlairGS 图片必须 .convert('RGB') 后再 JPEG 编码，否则返回空描述

变参产品集创建模式

当产品有多个颜色/SKU 变体时：

1. 每个变体建独立 product_code
2. 分配到对应颜色的图片（用 FlairGS 识别）
3. 统一设置 variant_group, variant_of, variant_type, variant_value
4. 使用 supplier_price 和 supplier_stock
5. 父产品标记 is_variant_group: true, variant_codes: [...]

FlairGS 连通性

• 优先：http://43.160.244.125:17870 (FRP)
• 备用：Desktop CDP → http://192.168.9.105:7870
• API Key: 8Ax4TyZc66qrZgXK
• 端点：POST /api/analyze — 字段名 base64 (非 image_base64)
• 模型：GS3/qwen3-vl:8b，典型延迟 7-8s