platform-assistant/scripts/parse_products.py

#!/usr/bin/env python3
"""
商品列表解析脚本
用法: python3 parse_products.py <文件路径>
输出: JSON [{ "name": "商品名", "spec": "规格", "category": "分类" }, ...]
"""

import sys
import json
import os

def _read_excel(filepath: str):
    import pandas as pd
    is_unnamed_cols = False
    
    # Attempt 1: default header=0
    df = pd.read_excel(filepath, dtype=str)
    raw_cols = list(df.columns)
    unnamed_count = sum(1 for c in raw_cols if c.startswith('Unnamed'))
    total_cols = len(raw_cols)
    
    # If most columns are "Unnamed" or header looks like a title, retry with header=1
    if unnamed_count > total_cols / 2 or total_cols == 1:
        is_unnamed_cols = True
        df = pd.read_excel(filepath, dtype=str, header=1)
    
    # If still mostly unnamed, try without header and use first non-empty row as header
    raw_cols = list(df.columns)
    unnamed_count = sum(1 for c in raw_cols if c.startswith('Unnamed'))
    if unnamed_count > len(raw_cols) / 2:
        is_unnamed_cols = True
        raw = pd.read_excel(filepath, dtype=str, header=None)
        for i in range(min(5, len(raw))):
            row = raw.iloc[i]
            vals = [str(v).strip() for v in row if str(v).strip().lower() != 'nan']
            if len(vals) > 2:
                df = pd.read_excel(filepath, dtype=str, header=i)
                break
    
    return df

def _read_csv(filepath: str):
    import pandas as pd
    df = pd.read_csv(filepath, dtype=str, encoding='utf-8-sig')
    return df

def _match_col(norm_map: dict, keywords: list[str]) -> str:
    for kw in keywords:
        nk = kw.lower().replace(' ', '').replace('　', '')
        if nk in norm_map:
            return norm_map[nk]
    # 模糊匹配：关键词是列名的子串
    for kw in keywords:
        nk = kw.lower().replace(' ', '').replace('　', '')
        for norm, orig in norm_map.items():
            if nk in norm:
                return orig
    return ''

def parse_file(filepath: str) -> list[dict]:
    ext = os.path.splitext(filepath)[1].lower()
    
    if ext in ('.xlsx', '.xls'):
        df = _read_excel(filepath)
    elif ext == '.csv':
        df = _read_csv(filepath)
    else:
        raise ValueError(f'不支持的文件格式: {ext}')
    
    if df.empty:
        return []
    
    # 标准化列名
    raw_cols = list(df.columns)
    norm_map = {}
    for c in raw_cols:
        norm = c.strip().lower().replace(' ', '').replace('　', '')
        norm_map[norm] = c
    
    name_col = _match_col(norm_map, ['商品名称', '商品名', '名称', '产品名称', '品名', 'name', 'productname', '产品名'])
    spec_col = _match_col(norm_map, ['规格', '规格型号', '型号', 'spec', '规格描述', '规格参数'])
    cat_col = _match_col(norm_map, ['分类', '商品分类', '品类', 'category', '类别', '类目'])
    price_col = _match_col(norm_map, ['价格', '单价', '报价', 'price', '售价', '销售价', '报价（元）', '报价(元)', '含税价'])
    unit_col = _match_col(norm_map, ['单位', '计量单位', 'unit', '数量单位'])
    brand_col = _match_col(norm_map, ['品牌', 'brand', '厂家', '制造商', '生产厂家'])
    qty_col = _match_col(norm_map, ['数量', 'quantity', '库存', '库存数量', 'qty'])
    
    if not name_col and raw_cols:
        name_col = raw_cols[0]
    
    results = []
    for _, row in df.iterrows():
        name_raw = row.get(name_col)
        if name_raw is None:
            continue
        name = str(name_raw).strip()
        if not name or name.lower() == 'nan':
            continue
        spec = str(row.get(spec_col, '') or '').strip()
        if spec.lower() == 'nan':
            spec = ''
        category = str(row.get(cat_col, '') or '').strip()
        if category.lower() == 'nan':
            category = ''
        price = row.get(price_col)
        if price is not None:
            try:
                price = float(str(price).replace(',', '').replace('￥', '').replace('¥', '').strip())
            except (ValueError, TypeError):
                price = 0
        else:
            price = 0
        unit = str(row.get(unit_col, '') or '').strip()
        if unit.lower() == 'nan':
            unit = ''
        brand = str(row.get(brand_col, '') or '').strip()
        if brand.lower() == 'nan':
            brand = ''
        qty_raw = row.get(qty_col)
        quantity = 0
        if qty_raw is not None:
            try:
                quantity = int(float(str(qty_raw).replace(',', '').strip()))
            except (ValueError, TypeError):
                quantity = 0
        results.append({
            'name': name,
            'spec': spec,
            'category': category,
            'price': price,
            'unit': unit,
            'brand': brand,
            'quantity': quantity,
        })
    
    return results


if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(json.dumps({'error': '请提供文件路径'}))
        sys.exit(1)
    try:
        data = parse_file(sys.argv[1])
        print(json.dumps(data, ensure_ascii=False))
    except Exception as e:
        print(json.dumps({'error': str(e)}, ensure_ascii=False))
        sys.exit(1)