#!/usr/bin/env python3 """MCP → 销售单JSON 完整管道 拉取销售管理全部销售订单，写入 daily_sales_mcp.json - 全量聚合：按日期/产品/销售员汇总 - 近90天明细：含客户+货号+金额完整字段 """ import subprocess, json, time, os, re from datetime import datetime, timedelta from collections import defaultdict TK = "m-abfb29e8-3104-434f-9944-8d0bb592f8cd" SALES_PK = "6593cd71471290e3cc6be6e6" BASE = "/Users/liuxinyuan/Desktop/Hermes输出-工作类/数据" TODAY = datetime.now() RECENT_DAYS = 90 RECENT_CUTOFF = (TODAY - timedelta(days=RECENT_DAYS)).strftime('%Y-%m-%d') # Relations REL_CUSTOMER = "6593cd71471290e3cc6be6e6:work_item_relation:relation_1713785263022" def mcp(method, args, timeout=30): try: r = subprocess.run(['curl','-s','--resolve','project.feishu.cn:443:120.233.177.47','-X','POST','https://project.feishu.cn/mcp_server/v1', '-H',f'X-Mcp-Token: {TK}','-H','Content-Type: application/json', '-d', json.dumps({"jsonrpc":"2.0","method":"tools/call","params":{"name":method,"arguments":args},"id":1})], capture_output=True, text=True, timeout=timeout) d = json.loads(r.stdout) result = None for c in d['result']['content']: t = c['text'] if 'log_id' in t: continue if t.startswith('{'): result = json.loads(t) return result except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, Exception): return None def parse_items(result): items = [] for gid, gitems in result.get('data', {}).items(): for item in gitems: fields = {} for f in item.get('moql_field_list', []): k = f['name'] vdict = f.get('value', {}) vals = list(vdict.values()) if vdict else [''] v = vals[0] if vals else '' if isinstance(v, dict): v = list(v.values())[0] if v else '' fields[k] = v if v else '' # Normalize if k == '工作项id': fields['work_item_id'] = str(v) elif k == '单据编号': fields['order_no'] = v elif k == '货号#': fields['product_code'] = v elif k == '金额': fields['amount'] = float(v) if v else 0 elif k == '单价': fields['unit_price'] = float(v) if v else 0 elif k == '销售数量': fields['qty'] = float(v) if v else 0 elif k == '创建时间': fields['date'] = v elif k == '创建者': fields['creator'] = v items.append(fields) return items # ═══════════════════════════ # Step 1: Pull all orders (aggregates only) # ═══════════════════════════ print("=== 拉取销售订单(分页) ===") daily_amount = defaultdict(float) # date → total amount daily_count = defaultdict(int) # date → count product_amount = defaultdict(float) # product_code → amount product_count = defaultdict(int) # product_code → count all_orders = [] recent_orders = [] page = 0 while True: offset = page * 50 mql = f"SELECT `单据编号`, `创建时间`, `创建者`, `金额`, `货号#`, `单价`, `销售数量`, work_item_id FROM `销售管理`.`销售订单` ORDER BY `创建时间` DESC LIMIT 50 OFFSET {offset}" # Retry up to 3 times on None result result = None for attempt in range(3): result = mcp("search_by_mql", {"project_key": SALES_PK, "mql": mql}) if result is not None: break print(f" 第{page+1}页 API返回None，重试 {attempt+1}/3...") time.sleep(2) if result is None: print(f" 第{page+1}页: 3次重试后仍返回None，跳过此页继续") page += 1 time.sleep(1) continue items = parse_items(result) if not items: print(f" 第{page+1}页: 0条，停止") break for item in items: date = item.get('date', '')[:10] if item.get('date') else '' amt = item.get('amount', 0) code = item.get('product_code', '未知') if date: daily_amount[date] += amt daily_count[date] += 1 product_amount[code] += amt product_count[code] += 1 all_orders.append(item) if date >= RECENT_CUTOFF: recent_orders.append(item) page += 1 print(f" 第{page}页: {len(items)}条 | 累计{len(all_orders)}条 | 近{RECENT_DAYS}天{len(recent_orders)}条") if len(items) < 50: break if page >= 150: # 150 pages = 7500 orders print(f" 达到上限150页，停止（近90天+聚合已完成）") break time.sleep(0.3) print(f"\n✅ 总计: {len(all_orders)}条订单") # ═══════════════════════════ # Step 2: Enrich recent orders with customer name # ═══════════════════════════ print(f"\n=== 补全近{RECENT_DAYS}天客户名 ===") customer_cache = {} # work_item_id → customer name for i, order in enumerate(recent_orders): wid = order.get('work_item_id', '') if not wid: if i < 3: print(f" ⚠️ order {order.get('order_no','?')} has no work_item_id - fields: {list(order.keys())}") continue # Check cache first if wid in customer_cache: continue related = None for attempt in range(3): related = mcp("list_related_workitem", { "project_key": SALES_PK, "work_item_id": wid, "relation_id": REL_CUSTOMER }, timeout=30) if related is not None: break time.sleep(1) if related: items = related.get('list', []) if items: customer_cache[wid] = items[0].get('name', '未知') else: customer_cache[wid] = '未关联' else: customer_cache[wid] = '查询失败' if (i+1) % 10 == 0: print(f" {i+1}/{len(recent_orders)}... 缓存{len(customer_cache)}条") # Enrich for order in recent_orders: wid = order.get('work_item_id', '') order['customer'] = customer_cache.get(wid, '未查询') print(f"✅ 客户补全完成") # ═══════════════════════════ # Step 3: Build JSON output # ═══════════════════════════ print("\n=== 生成JSON ===") # Recent detail per day daily_detail = defaultdict(list) for order in recent_orders: d = order.get('date', '') if d: daily_detail[d].append({ 'order_no': order.get('order_no', ''), 'customer': order.get('customer', ''), 'product_code': order.get('product_code', ''), 'amount': order.get('amount', 0), 'unit_price': order.get('unit_price', 0), 'qty': order.get('qty', 0), 'creator': order.get('creator', ''), 'date': d, 'url': f"https://project.feishu.cn/xsguanli/xsdd/detail/{order.get('work_item_id','')}" }) # Product aggregation product_agg = [{'code': k, 'amount': v, 'count': product_count[k]} for k, v in sorted(product_amount.items(), key=lambda x: -x[1])] # Daily aggregation days_sorted = sorted(daily_amount.keys(), reverse=True) output = { 'days': days_sorted, 'daily': dict(daily_detail), 'daily_summary': {d: {'amount': daily_amount.get(d, 0), 'count': daily_count.get(d, 0)} for d in days_sorted}, 'products': product_agg[:50], 'total_amount': sum(daily_amount.values()), 'total_orders': len(all_orders), 'recent_orders': len(recent_orders), 'generated': TODAY.strftime('%Y-%m-%d %H:%M'), 'source': '飞书项目MCP · 销售订单(xsdd)' } outpath = f'{BASE}/daily_sales_mcp.json' with open(outpath, 'w', encoding='utf-8') as f: json.dump(output, f, ensure_ascii=False, default=str) size = os.path.getsize(outpath) print(f'✅ 保存: {outpath} ({size//1024}KB)') print(f'\n📊 统计:') print(f' 总订单: {output["total_orders"]}条') print(f' 近{RECENT_DAYS}天明细: {output["recent_orders"]}条') print(f' 覆盖天数: {len(days_sorted)}') print(f' 总金额: ¥{output["total_amount"]:,.0f}') print(f' 覆盖: {days_sorted[-1] if days_sorted else "N/A"} → {days_sorted[0] if days_sorted else "N/A"}') print(f' 今日: {daily_count.get(TODAY.strftime("%Y-%m-%d"),0)}条 ¥{daily_amount.get(TODAY.strftime("%Y-%m-%d"),0):,.0f}') # Top products print(f'\n🏆 畅销货号 Top5:') for p in product_agg[:5]: print(f' {p["code"]}: {p["count"]}笔 ¥{p["amount"]:,.0f}')