mem_product_cli.py•7.5 kB
#!/usr/bin/env python3
"""
ProductMemory命令行接口
提供mem product命令来管理商品记忆
"""
import sys
import json
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from product_memory import create_product_memory
def print_usage():
"""打印使用说明"""
print("""
ProductMemory命令行工具
用法:
python mem_product_cli.py <command> [arguments]
命令:
add <sku> <name> <price> <category> <brand> [selling_points...]
添加商品
示例: python mem_product_cli.py add SKU-001 "华为Mate60" 6999 "智能手机" "华为" "5G" "长续航"
search <query>
搜索商品
示例: python mem_product_cli.py search "华为"
示例: python mem_product_cli.py search "SKU-001"
get <sku>
根据SKU获取商品详情
示例: python mem_product_cli.py get SKU-001
list
列出所有商品
示例: python mem_product_cli.py list
stats
显示统计信息
示例: python mem_product_cli.py stats
help
显示此帮助信息
""")
def cmd_add(args):
"""添加商品命令"""
if len(args) < 5:
print("❌ 参数不足。用法: add <sku> <name> <price> <category> <brand> [selling_points...]")
return False
sku = args[0]
name = args[1]
try:
price = float(args[2])
except ValueError:
print("❌ 价格必须是数字")
return False
category = args[3]
brand = args[4]
selling_points = args[5:] if len(args) > 5 else []
pm = create_product_memory(verbose=True)
attributes = {
"name": name,
"price": price,
"category": category,
"brand": brand
}
success = pm.add_product(sku, attributes, selling_points)
if success:
print(f"✅ 商品 {sku} 添加成功")
else:
print(f"❌ 商品 {sku} 添加失败")
pm.close()
return success
def cmd_search(args):
"""搜索商品命令"""
if len(args) < 1:
print("❌ 参数不足。用法: search <query>")
return False
query = " ".join(args)
pm = create_product_memory(verbose=False)
results = pm.search_product(query, limit=10)
if results:
print(f"🔍 搜索'{query}' 找到 {len(results)} 个结果:")
print()
for i, result in enumerate(results, 1):
print(f"{i}. SKU: {result['sku']}")
print(f" 名称: {result['attributes'].get('name', '未知')}")
print(f" 品牌: {result['attributes'].get('brand', '未知')}")
print(f" 价格: ¥{result['attributes'].get('price', 0)}")
print(f" 分类: {result['attributes'].get('category', '未知')}")
print(f" 卖点: {', '.join(result['selling_points'])}")
if result['relationships']:
print(f" 关系: {len(result['relationships'])} 个")
print()
else:
print(f"🔍 搜索'{query}' 未找到结果")
pm.close()
return True
def cmd_get(args):
"""获取商品详情命令"""
if len(args) < 1:
print("❌ 参数不足。用法: get <sku>")
return False
sku = args[0]
pm = create_product_memory(verbose=False)
product = pm.get_product_by_sku(sku)
if product:
print(f"📦 商品详情: {sku}")
print(f"名称: {product['attributes'].get('name', '未知')}")
print(f"品牌: {product['attributes'].get('brand', '未知')}")
print(f"价格: ¥{product['attributes'].get('price', 0)}")
print(f"分类: {product['attributes'].get('category', '未知')}")
print(f"创建时间: {product.get('created_at', '未知')}")
print(f"更新时间: {product.get('updated_at', '未知')}")
print(f"\n🎯 卖点 ({len(product['selling_points'])} 个):")
for i, point in enumerate(product['selling_points'], 1):
print(f" {i}. {point}")
if product['relationships']:
print(f"\n🔗 关系 ({len(product['relationships'])} 个):")
for i, rel in enumerate(product['relationships'], 1):
print(f" {i}. {rel.get('type', '未知')} -> {rel.get('target', '未知')}")
print(f"\n🏷️ 标签:")
tags = product.get('tags', [])
if tags:
print(f" {', '.join(tags)}")
else:
print(" 无标签")
else:
print(f"❌ 未找到商品: {sku}")
pm.close()
return product is not None
def cmd_list(args):
"""列出所有商品命令"""
pm = create_product_memory(verbose=False)
products = pm.list_all_products()
if products:
print(f"📋 所有商品 ({len(products)} 个):")
print()
# 按品牌分组显示
brands = {}
for product in products:
brand = product['attributes'].get('brand', '未知品牌')
if brand not in brands:
brands[brand] = []
brands[brand].append(product)
for brand, brand_products in brands.items():
print(f"🏢 {brand} ({len(brand_products)} 个商品):")
for product in brand_products:
name = product['attributes'].get('name', '未知')
price = product['attributes'].get('price', 0)
category = product['attributes'].get('category', '未知')
print(f" • {product['sku']}: {name} - ¥{price} ({category})")
print()
else:
print("📋 暂无商品")
pm.close()
return True
def cmd_stats(args):
"""显示统计信息命令"""
pm = create_product_memory(verbose=False)
stats = pm.get_stats()
print("📊 ProductMemory统计信息:")
print(f"总商品数: {stats.get('total_products', 0)}")
print(f"记忆状态: {stats.get('memory_status', '未知')}")
print(f"记忆数量: {stats.get('memory_count', 0)}")
brands = stats.get('brands', [])
categories = stats.get('categories', [])
print(f"\n🏢 品牌 ({len(brands)} 个):")
if brands:
for brand in sorted(brands):
print(f" • {brand}")
else:
print(" 无品牌")
print(f"\n📂 分类 ({len(categories)} 个):")
if categories:
for category in sorted(categories):
print(f" • {category}")
else:
print(" 无分类")
print(f"\n⏰ 统计时间: {stats.get('timestamp', '未知')}")
pm.close()
return True
def main():
"""主函数"""
if len(sys.argv) < 2:
print_usage()
return False
command = sys.argv[1].lower()
args = sys.argv[2:]
commands = {
'add': cmd_add,
'search': cmd_search,
'get': cmd_get,
'list': cmd_list,
'stats': cmd_stats,
'help': lambda _: print_usage() or True
}
if command not in commands:
print(f"❌ 未知命令: {command}")
print_usage()
return False
try:
return commands[command](args)
except KeyboardInterrupt:
print("\n\n⏹️ 用户中断操作")
return False
except Exception as e:
print(f"❌ 执行命令时出错: {e}")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)