a8fa159f0f
- 新增 aio-mcp 项目框架 - 新增 .trae/skills/ OCR/SKU 设计工具 - 更新 auto-check 配置和 stress 包 - 删除 fnhelp 文档目录 - 更新 Docker 压力测试脚本
367 lines
13 KiB
Python
367 lines
13 KiB
Python
import json
|
||
import re
|
||
import os
|
||
|
||
# Load OCR results
|
||
with open(r"D:\workspace\membank\售价\ocr_results.json", "r", encoding="utf-8") as f:
|
||
all_results = json.load(f)
|
||
|
||
# Extract structured SKU prices from each image
|
||
# Pattern: SKU name patterns like "蜗牛Nas-xxx-4g+64g", "J1900-4g+32g", etc.
|
||
# Prices appear as decimal numbers typically followed by a discount of 5.00
|
||
|
||
def extract_sku_prices(text_lines):
|
||
"""Extract SKU name and price pairs from OCR text lines"""
|
||
entries = []
|
||
i = 0
|
||
while i < len(text_lines):
|
||
text, conf = text_lines[i]
|
||
text = text.strip()
|
||
|
||
# Look for SKU names containing known patterns
|
||
sku_patterns = [
|
||
r'(?:蜗牛Nas|蜗牛|Nas|cNas)[\s\w\-\.\+\(\)一-鿿]+?\d+g[+一-鿿]*\d*g',
|
||
r'(?:J1900|j1900|J3160|j3160|I3|i3|I5|i5|1037U|1037u|N5105|N5100|N5095|J3355|J3060)[\-\.\+\(\)\s\w一-鿿]*?\d+g[\+一-鿿]*\d*g',
|
||
r'(?:J1900|j1900|J3160|j3160|I3|i3|I5|i5|1037U|1037u|N5105|N5100|N5095|J3355|J3060)[\-\.\+\(\)\s\w一-鿿]*?\d+g',
|
||
r'(?:16G[BW])[\-\s]*(?:windows|蜗牛|群晖)',
|
||
r'(?:DDR|D3|D4)[\-\.\s]*\d+\s*[Gg]',
|
||
]
|
||
|
||
is_sku = False
|
||
for pat in sku_patterns:
|
||
if re.search(pat, text, re.IGNORECASE):
|
||
is_sku = True
|
||
break
|
||
|
||
if is_sku:
|
||
sku_name = text
|
||
# Look for price pattern nearby: two numbers where one is ~5.00 off from the other
|
||
# Pattern: price1 (like 247.50) ... 5.00 ... price2 (like 242.50)
|
||
prices_found = []
|
||
for j in range(i, min(i + 10, len(text_lines))):
|
||
t2 = text_lines[j][0].strip()
|
||
# Match decimal prices
|
||
price_match = re.findall(r'\d{2,4}\.\d{2}', t2)
|
||
for p in price_match:
|
||
prices_found.append(float(p))
|
||
# Also match integer prices
|
||
price_match2 = re.findall(r'(?<!\d\.)(?:\d{2,4})(?:\.\d{2})?(?!\d|\.\d{3,})', t2)
|
||
for p in price_match2:
|
||
try:
|
||
prices_found.append(float(p))
|
||
except:
|
||
pass
|
||
|
||
# Deduplicate and find likely actual price (the larger number, minus 5 = smaller number)
|
||
prices_found = sorted(set(prices_found))
|
||
actual_price = None
|
||
for p in prices_found:
|
||
if p > 10 and (p - 5) in prices_found:
|
||
actual_price = p # The larger one is the original price
|
||
break
|
||
|
||
if actual_price:
|
||
discounted = actual_price - 5
|
||
entries.append({
|
||
'sku': sku_name,
|
||
'price': discounted,
|
||
'original_price': actual_price,
|
||
'confidence': conf
|
||
})
|
||
|
||
i += 1
|
||
|
||
return entries
|
||
|
||
def extract_prices_v2(text_lines):
|
||
"""
|
||
Better approach: parse each image by looking for patterns:
|
||
SKU name -> (price, discount 5.00, discounted_price)
|
||
"""
|
||
results = []
|
||
|
||
# Flatten texts
|
||
all_text = [t[0].strip() for t in text_lines]
|
||
all_with_conf = list(text_lines)
|
||
|
||
# Known CPU/memory patterns
|
||
pattern = re.compile(
|
||
r'.*?(?:蜗牛|Nas|NAS|nas|cNas|黑裙|黑群晖|客服|咨询).*?(?:'
|
||
r'J1900|j1900|J3160|j3160|'
|
||
r'i3|I3|i5|I5|'
|
||
r'1037[Uu]|1037u|'
|
||
r'N5105|N5100|N5095|'
|
||
r'J3355|J3060|'
|
||
r'16G[Bb]'
|
||
r').*',
|
||
re.IGNORECASE
|
||
)
|
||
|
||
# Scan through all texts looking for price patterns
|
||
for i, (text, conf) in enumerate(all_with_conf):
|
||
text = text.strip()
|
||
|
||
# Skip non-SKU lines
|
||
if len(text) < 3:
|
||
continue
|
||
|
||
# Check if this looks like a SKU description
|
||
has_cpu = bool(re.search(r'(J1900|j1900|J3160|j3160|i3|I3|i5|I5|1037[Uu]|N5105|N5100|N5095|J3355|J3060|J4125)', text))
|
||
has_nas = bool(re.search(r'(蜗牛|Nas|NAS|cNas|黑裙)', text))
|
||
has_mem = bool(re.search(r'\d+g[\+一-鿿]*\d+g|\d+g\b|16G[Bb]', text))
|
||
|
||
if has_cpu or (has_nas and has_mem) or ('16G' in text.upper() and ('windows' in text.lower() or '蜗牛' in text or '群晖' in text)):
|
||
# This is a SKU name - look for nearby prices
|
||
sku_name = text
|
||
|
||
# Look at next 8 lines for prices
|
||
prices = []
|
||
for j in range(i+1, min(i+12, len(all_with_conf))):
|
||
t = all_with_conf[j][0].strip()
|
||
# Find decimal numbers
|
||
nums = re.findall(r'(\d{2,4}\.\d{2})', t)
|
||
prices.extend([float(n) for n in nums])
|
||
|
||
# Also look at previous 2 lines
|
||
for j in range(max(0, i-2), i):
|
||
t = all_with_conf[j][0].strip()
|
||
nums = re.findall(r'(\d{2,4}\.\d{2})', t)
|
||
prices.extend([float(n) for n in nums])
|
||
|
||
if prices:
|
||
# Find pairs where diff = 5.00
|
||
prices = sorted(set(prices))
|
||
found_price = None
|
||
for p in prices:
|
||
if p > 20 and (p - 5.00) in [round(x, 2) for x in prices]:
|
||
found_price = p
|
||
break
|
||
|
||
# If no pair found, take the most common price range
|
||
if not found_price:
|
||
valid_prices = [p for p in prices if 100 < p < 500]
|
||
if valid_prices:
|
||
found_price = max(valid_prices)
|
||
|
||
if found_price and found_price > 20:
|
||
discounted = round(found_price - 5.0, 2)
|
||
results.append({
|
||
'sku': sku_name,
|
||
'price': discounted,
|
||
'original_price': round(found_price, 2),
|
||
'confidence': float(conf) if conf else 0
|
||
})
|
||
|
||
# Deduplicate - keep highest confidence for each SKU
|
||
seen = {}
|
||
for r in results:
|
||
key = r['sku']
|
||
if key not in seen or r['confidence'] > seen[key]['confidence']:
|
||
seen[key] = r
|
||
|
||
return sorted(seen.values(), key=lambda x: x['price'])
|
||
|
||
|
||
def extract_prices_v3(text_lines):
|
||
"""
|
||
Most robust: look for every <SKU name> ... <price_a> ... 5.00 ... <price_b> pattern
|
||
where price_a = price_b - 5 (or vice versa)
|
||
"""
|
||
results = []
|
||
all_with_conf = list(text_lines)
|
||
|
||
for i, (text, conf) in enumerate(all_with_conf):
|
||
text = text.strip()
|
||
|
||
# Must contain key SKU indicators
|
||
has_indicator = (
|
||
bool(re.search(r'(J1900|j1900|J3160|j3160|i3|I3|i5|I5|1037[Uu]|N5105|N5100|N5095|J3355|J3060|J4125|5005)', text)) or
|
||
bool(re.search(r'(蜗牛|Nas|NAS|黑裙|黑群晖|客服|咨询)', text)) or
|
||
bool(re.search(r'16G[Bb].*(?:windows|蜗牛|群晖|系统)', text)) or
|
||
bool(re.search(r'(?:预装系统|系统).*(?:双盘位|单盘位)', text))
|
||
)
|
||
|
||
if not has_indicator or len(text) < 4:
|
||
continue
|
||
|
||
# But also filter out pure UI text
|
||
skip_patterns = ['商品价格', '关注券', '折扣', '券后价', '商品ID', '活动', '修改', '查看', '营销', '官方', '拼多多', '商家']
|
||
if any(s in text for s in skip_patterns):
|
||
continue
|
||
|
||
sku_name = text
|
||
|
||
# Search window around this line for prices
|
||
window_texts = []
|
||
for j in range(max(0, i-3), min(i+15, len(all_with_conf))):
|
||
window_texts.append(all_with_conf[j][0].strip())
|
||
|
||
# Extract all decimal prices from the window
|
||
prices = []
|
||
for wt in window_texts:
|
||
nums = re.findall(r'(\d{2,4}\.\d{2})', wt)
|
||
prices.extend([float(n) for n in nums])
|
||
|
||
if not prices:
|
||
continue
|
||
|
||
prices = sorted(set(prices))
|
||
|
||
# Filter noise prices (too small)
|
||
prices = [p for p in prices if p > 15]
|
||
|
||
# Find the discount pattern: original_price and (original_price - 5)
|
||
found_price = None
|
||
price_set = set(round(p, 2) for p in prices)
|
||
|
||
for p in prices:
|
||
if round(p - 5.0, 2) in price_set:
|
||
# p is the higher (original), p-5 is the discounted
|
||
found_price = p
|
||
break
|
||
if round(p + 5.0, 2) in price_set:
|
||
# p is the discounted, p+5 is the original
|
||
found_price = p + 5.0
|
||
break
|
||
|
||
if not found_price:
|
||
# Fallback: take the highest price that looks like a product price
|
||
valid = [p for p in prices if 100 < p < 600]
|
||
if valid:
|
||
found_price = max(valid)
|
||
|
||
if found_price and found_price > 20:
|
||
results.append({
|
||
'sku': sku_name,
|
||
'price': round(found_price - 5.0, 2),
|
||
'original_price': round(found_price, 2),
|
||
'confidence': float(conf) if conf else 0
|
||
})
|
||
|
||
# Deduplicate by SKU name similarity
|
||
def sku_key(s):
|
||
return re.sub(r'[\s\-\.\+]+', '', s.lower())
|
||
|
||
deduped = {}
|
||
for r in results:
|
||
key = sku_key(r['sku'])
|
||
if key not in deduped or r['confidence'] > deduped[key]['confidence']:
|
||
deduped[key] = r
|
||
|
||
return sorted(deduped.values(), key=lambda x: x['price'])
|
||
|
||
|
||
print("=" * 80)
|
||
print("Extracting SKU Price Summary")
|
||
print("=" * 80)
|
||
|
||
all_entries = []
|
||
for img_name, lines in all_results.items():
|
||
print(f"\nProcessing: {img_name}")
|
||
entries = extract_prices_v3(lines)
|
||
print(f" Found {len(entries)} SKUs")
|
||
all_entries.extend(entries)
|
||
|
||
# Global deduplication
|
||
def normalize_sku(name):
|
||
"""Normalize SKU name for deduplication"""
|
||
name = name.strip()
|
||
# Remove leading symbols
|
||
name = re.sub(r'^[>\s\-\.]+', '', name)
|
||
# Standardize separators
|
||
name = name.replace('+', '+').replace('＋', '+')
|
||
# Remove trailing garbage
|
||
name = re.sub(r'[>\s\.\-]+$', '', name)
|
||
return name
|
||
|
||
# Categorize entries
|
||
# Dedup keeping highest confidence
|
||
global_dedup = {}
|
||
for e in all_entries:
|
||
norm = normalize_sku(e['sku'])
|
||
# Further normalize for key
|
||
key = re.sub(r'[\s\-\>\.\(\)]', '', norm.lower())
|
||
# Extract CPU type for grouping
|
||
if key not in global_dedup or e['confidence'] > global_dedup[key]['confidence']:
|
||
global_dedup[key] = e
|
||
|
||
final_entries = sorted(global_dedup.values(), key=lambda x: x['price'])
|
||
|
||
# Group by CPU/platform
|
||
groups = {}
|
||
for e in final_entries:
|
||
name = e['sku']
|
||
# Extract group
|
||
if '16G' in name.upper() or '16g' in name.lower():
|
||
group = '系统U盘 (16GB)'
|
||
elif re.search(r'(i3|I3|i5|I5)', name):
|
||
if '双盘' in name or '雙盤' in name:
|
||
group = 'NAS 双盘位 - i3/i5系列'
|
||
else:
|
||
group = 'NAS - i3/i5系列'
|
||
elif re.search(r'1037[Uu]', name):
|
||
if '单盘' in name or '單盤' in name or 'A4' in name:
|
||
group = 'NAS 单盘位 - 1037U系列'
|
||
elif '双盘' in name or '雙盤' in name:
|
||
group = 'NAS 双盘位 - 1037U系列'
|
||
else:
|
||
group = 'NAS - 1037U系列'
|
||
elif re.search(r'J1900|j1900', name):
|
||
group = 'NAS 双盘位 - J1900系列'
|
||
elif re.search(r'i3|I3', name):
|
||
group = 'NAS - i3系列'
|
||
elif re.search(r'i5|I5', name):
|
||
group = 'NAS - i5系列'
|
||
elif '黑裙' in name or '客服' in name or '咨询' in name:
|
||
group = 'NAS - 黑裙晖系列'
|
||
else:
|
||
group = '其他'
|
||
|
||
if group not in groups:
|
||
groups[group] = []
|
||
groups[group].append(e)
|
||
|
||
# Generate markdown output
|
||
md_lines = []
|
||
md_lines.append("# SKU 售价汇总")
|
||
md_lines.append("")
|
||
md_lines.append(f"> 从拼多多商家后台截图 OCR 识别提取")
|
||
md_lines.append(f"> 提取日期: 2026-06-16")
|
||
md_lines.append(f"> 数据来源: `售价/` 目录下 {len(all_results)} 张截图")
|
||
md_lines.append(f"> 共识别 {len(final_entries)} 个 SKU 售价（已去重）")
|
||
md_lines.append("")
|
||
md_lines.append("---")
|
||
md_lines.append("")
|
||
|
||
for group_name in sorted(groups.keys()):
|
||
entries = groups[group_name]
|
||
md_lines.append(f"## {group_name}")
|
||
md_lines.append("")
|
||
md_lines.append("| SKU 名称 | 券后售价 (¥) | 券前原价 (¥) | OCR 置信度 |")
|
||
md_lines.append("|---------|------------:|------------:|----------:|")
|
||
|
||
for e in entries:
|
||
name = normalize_sku(e['sku'])
|
||
# Clean up OCR artifacts in name
|
||
name = name.replace('關注券5元优惠券', '').strip()
|
||
name = re.sub(r'\s+', ' ', name)
|
||
md_lines.append(f"| {name} | {e['price']:.2f} | {e['original_price']:.2f} | {e['confidence']:.2f} |")
|
||
|
||
md_lines.append("")
|
||
|
||
md_lines.append("---")
|
||
md_lines.append("")
|
||
md_lines.append("## 说明")
|
||
md_lines.append("")
|
||
md_lines.append("- **券后售价** = 券前原价 - 5.00元 (店铺关注券)")
|
||
md_lines.append("- 价格数据通过 RapidOCR 从拼多多商家后台截图自动识别提取")
|
||
md_lines.append("- 同一SKU在不同截图中出现时保留置信度最高的识别结果")
|
||
md_lines.append("- OCR 识别可能存在误差，请以实际后台数据为准")
|
||
|
||
output_path = r"D:\workspace\membank\售价\SKU售价汇总.md"
|
||
with open(output_path, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(md_lines))
|
||
|
||
print(f"\n\nSummary saved to: {output_path}")
|
||
print(f"Total: {len(final_entries)} SKUs, {len(groups)} categories")
|