#!/usr/bin/env python3 """拉取 H1 订单存档（按季度分批） 用法: python3 pull_h1_archive.py --q1 # 拉取 Q1 (1-3月) python3 pull_h1_archive.py --q2 # 拉取 Q2 (4-6月) python3 pull_h1_archive.py --all # 全部 输出: h1_archive/q1_orders.json, q2_orders.json, h1_summary.json """ import subprocess, json, os, sys, time TK = "m-abfb29e8-3104-434f-9944-8d0bb592f8cd" PK = "6593cd71471290e3cc6be6e6" BASE = os.path.dirname(os.path.abspath(__file__)) ARCHIVE = os.path.join(BASE, "h1_archive") COLUMNS = 'name, start_time, field_96a245, field_e1001d, field_51d592, work_item_id, owner, field_3f38c1, field_0e6822, field_290c00, field_5ed7ab, field_4a1b47' QUARTERS = { 'm01': ('2026-01-01', '2026-01-31', '1月'), 'm02': ('2026-02-01', '2026-02-28', '2月'), 'm03': ('2026-03-01', '2026-03-31', '3月'), 'm04': ('2026-04-01', '2026-04-30', '4月'), 'm05': ('2026-05-01', '2026-05-31', '5月'), 'm06': ('2026-06-01', '2026-06-30', '6月'), } def mcp(method, args): r = subprocess.run(['curl','-s','-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=60) try: d = json.loads(r.stdout) if 'error' in d: return None for c in d['result']['content']: t = c.get('text','') if 'log_id' in t: continue return json.loads(t) except: return None def parse_items(result): items = [] if not result: return items for gid, gitems in result.get('data', {}).items(): for item in gitems: fields = {} for f in (item.get('moql_field_list') or []): k = f['key'] v = f.get('value') if isinstance(v, list) and len(v) > 0: v = v[0] if v is None: fields[k] = '' elif 'string_value' in v: fields[k] = v['string_value'] elif 'double_value' in v: fields[k] = v['double_value'] elif 'long_value' in v: fields[k] = v['long_value'] elif 'key_label_value' in v: fields[k] = v['key_label_value'] elif 'user_value' in v: fields[k] = v['user_value'] else: fields[k] = '' items.append(fields) return items def pull_quarter(qkey): start, end, label = QUARTERS[qkey] print(f'\n{"="*50}') print(f'拉取 {label}: {start} ~ {end}') print(f'{"="*50}') all_items = [] seen_wid = set() page = 0 while page < 200: offset = page * 50 mql = f'SELECT {COLUMNS} FROM `销售管理`.`销售订单` WHERE start_time >= "{start}" AND start_time <= "{end}" ORDER BY start_time DESC LIMIT 50 OFFSET {offset}' result = mcp("search_by_mql", {"project_key": PK, "mql": mql}) if not result: break items = parse_items(result) if not items: break new = [] for i in items: wid = str(i.get('work_item_id', '')) if wid and wid not in seen_wid: seen_wid.add(wid) new.append(i) all_items.extend(new) page += 1 if len(items) < 50: break if page % 5 == 0: print(f" page {page}: +{len(new)} (total {len(all_items)})") print(f" 共 {len(all_items)} 条（按 work_item_id 去重）") # H1 存档用客户→销售员映射（不逐一查角色，省时间） cust_file = os.path.join(BASE, 'mcp_customers.json') sales_map = {} if os.path.exists(cust_file): custs = json.load(open(cust_file, encoding='utf-8')).get('customers', []) for c in custs: owner = c.get('owner', '') name = c.get('name', '') if isinstance(owner, dict) and owner.get('email') and '@biori.com' in owner.get('email', ''): sales_map[name] = owner print(f" 销售员映射: {len(sales_map)} 客户") # 转为输出格式 output = [] seen_oid = set() for o in all_items: oid = o.get('name', '') if oid in seen_oid: continue seen_oid.add(oid) customer = o.get('field_e1001d', '') if isinstance(customer, dict): customer = customer.get('label', '') role = sales_map.get(customer, {}) owner_name = role.get('name_cn', role.get('name_en', '')) owner_email = role.get('email', '') output.append({ 'orderId': oid, 'customer': customer, 'dept': (lambda v: (v.get('label','') if isinstance(v, dict) else str(v)) if v else '')(o.get('field_5ed7ab','')), 'createDate': o.get('start_time', ''), 'trackingNo': o.get('field_96a245', ''), 'amount': o.get('field_51d592', 0), 'taxTotal': o.get('field_51d592', 0), 'contractNo': o.get('field_4a1b47', ''), 'creator': owner_name, 'creatorEmail': owner_email, 'recipient': o.get('field_0e6822', ''), 'phone': o.get('field_3f38c1', ''), 'address': o.get('field_290c00', ''), 'workItemId': str(o.get('work_item_id', '')), }) out_path = os.path.join(ARCHIVE, f'{qkey}_orders.json') with open(out_path, 'w', encoding='utf-8') as f: json.dump({ 'period': f'{start} ~ {end}', 'pulled_at': time.strftime('%Y-%m-%d %H:%M:%S'), 'total': len(output), 'orders': output }, f, ensure_ascii=False, indent=2) print(f" ✅ 写入 {out_path} ({len(output)} 条)") return output def main(): os.makedirs(ARCHIVE, exist_ok=True) quarters = [] if '--all' in sys.argv: quarters = ['m01', 'm02', 'm03', 'm04', 'm05', 'm06'] elif '--h1' in sys.argv: quarters = ['m01', 'm02', 'm03', 'm04', 'm05', 'm06'] else: for a in sys.argv: if a.startswith('--m') and len(a) >= 5: quarters.append(a[2:]) # "--m01" → "m01" if not quarters: print("用法: --m01 | --m02 | ... | --m06 | --h1 | --all") sys.exit(1) all_orders = [] for q in quarters: orders = pull_quarter(q) all_orders.extend(orders) # 生成汇总 summary = { 'generated_at': time.strftime('%Y-%m-%d %H:%M:%S'), 'total_orders': len(all_orders), 'total_tax': sum(float(o.get('taxTotal', 0) or 0) for o in all_orders), 'by_month': {}, 'by_dept': {}, } for o in all_orders: m = o.get('createDate', '')[:7] d = o.get('dept', '未知') summary['by_month'][m] = summary['by_month'].get(m, 0) + 1 summary['by_dept'][d] = summary['by_dept'].get(d, 0) + 1 with open(os.path.join(ARCHIVE, 'h1_summary.json'), 'w', encoding='utf-8') as f: json.dump(summary, f, ensure_ascii=False, indent=2) print(f'\n{"="*50}') print(f'H1 存档完成: {len(all_orders)} 条, 税合计 {summary["total_tax"]/10000:.1f}万') print(f'{"="*50}') if __name__ == '__main__': main()