#!/usr/bin/env python3 """客户档案数据采集 — 拉取全维度数据生成JSON""" import subprocess, json, sys, os from datetime import datetime BASE = os.path.dirname(os.path.abspath(__file__)) TK = 'm-abfb29e8-3104-434f-9944-8d0bb592f8cd' SALES_PK = '6593cd71471290e3cc6be6e6' CLIENT_PK = '658bb60520ea78a2125f1b99' AFTERSALES_PK = '658288abfb8bd616b17025f1' def mcp(method, args, timeout=30): 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=timeout) 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 try: return json.loads(t) except: return None return None def fetch_profile(cust_name): result = {'customer': cust_name, 'fetched_at': datetime.now().isoformat()[:19]} # Step 1: Get customer work_item_id + decision chain MQL = f'SELECT work_item_id, `决策链URL链接`, `客户级别` FROM `销售管理`.`客户` WHERE `客户名称` = "{cust_name}" LIMIT 1' r = mcp('search_by_mql', {'project_key': SALES_PK, 'mql': MQL}) wid = None if r: for items in r.get('data',{}).values(): for item in items: for f in item.get('moql_field_list',[]): vv = list(f.get('value',{}).values()) if f.get('value') else [''] v = str(vv[0]) if vv else '' if f['name'] == '工作项id': wid = v if f['name'] == '决策链URL链接' and v and v != 'None': result['decision_url'] = v if f['name'] == '客户级别': result['level'] = v if isinstance(r, dict): pass # it's a nested dict if not wid: result['error'] = '未找到客户' return result result['work_item_id'] = wid # Step 2: Pricing from local JSON safe_name = cust_name.replace('/', '_')[:30] pricing_file = os.path.join(BASE, f'数据/pricing/{safe_name}.json') if os.path.exists(pricing_file): with open(pricing_file) as f: pdata = json.load(f) result['pricing'] = { 'total_amount': pdata.get('total_amount', 0), 'total_orders': pdata.get('total_orders', 0), 'top_products': [{ 'sku': p['sku'][:25], 'name': p['name'][:40], 'latest_price': p['latest_price'], 'total_amt': p['total_amt'], 'history_price': p.get('history_price',''), 'is_gift': p.get('is_gift',False) } for p in sorted(pdata.get('products',[]), key=lambda x: -x['total_amt'])[:8]] } else: result['pricing'] = {'total_amount': 0, 'total_orders': 0, 'top_products': []} # Step 3: Contacts via relation result['contacts'] = [] # Try list_related_workitem r = mcp('list_related_workitem', { 'project_key': SALES_PK, 'work_item_id': wid, 'work_item_type': '65b8a6cacde27fd415d769ca' # 联系人 type_key }) if r: for item in r.get('list', r.get('data', [])): wa = item.get('work_item_attribute', {}) result['contacts'].append({ 'name': wa.get('work_item_name', ''), 'id': wa.get('work_item_id', ''), }) # Step 4: Activities (销售管理.活动) result['activities'] = [] act_type = '65ae1e5d44338dbe7c39a29a' # Try to find activities related to this customer # MQL approach: search for customer name in activity content MQL = f'SELECT `活动内容`, `创建时间`, `创建者`, work_item_id FROM `销售管理`.`活动` ORDER BY `创建时间` DESC LIMIT 100' r = mcp('search_by_mql', {'project_key': SALES_PK, 'mql': MQL}) if r: for items in r.get('data',{}).values(): for item in items: fields = {} for f in item.get('moql_field_list',[]): vv = list(f.get('value',{}).values()) if f.get('value') else [''] fields[f['name']] = str(vv[0])[:100] if vv else '' content = fields.get('活动内容','') # Check if related to this customer (simple substring match) if cust_name[:4] in content or cust_name[:6] in content: result['activities'].append({ 'content': content[:80], 'time': fields.get('创建时间','')[:19], 'creator': fields.get('创建者',''), 'id': fields.get('工作项id','') }) result['activities_note'] = f'匹配{len(result["activities"])}条(模糊匹配)' # Step 5: Projects (客户项目管理) result['projects'] = [] r = mcp('list_related_workitem', { 'project_key': CLIENT_PK, 'work_item_id': wid, 'work_item_type': '658bc2e62f7a609dbd1bf0b8' # 项目 }) if r: for item in r.get('list', r.get('data', [])): wa = item.get('work_item_attribute', {}) result['projects'].append({ 'name': wa.get('work_item_name', ''), 'status': wa.get('work_item_status', {}).get('name', ''), 'id': wa.get('work_item_id', ''), }) if not result['projects']: # Try with CLIENT_PK relation lookup r2 = mcp('list_workitem_relations', {'project_key': CLIENT_PK, 'work_item_id': wid}) if r2: result['projects_note'] = f'关联类型: {len(r2.get("list",[]))}种' # Step 6: Complaints (售后管理) result['complaints'] = [] comp_type = '6669433056a98249604376de' # 客户反馈单 r = mcp('list_related_workitem', { 'project_key': AFTERSALES_PK, 'work_item_id': wid, 'work_item_type': comp_type }) if r: for item in r.get('list', r.get('data', [])): wa = item.get('work_item_attribute', {}) result['complaints'].append({ 'name': wa.get('work_item_name', ''), 'status': wa.get('work_item_status', {}).get('name', ''), 'id': wa.get('work_item_id', ''), }) return result if __name__ == '__main__': cust = sys.argv[1] if len(sys.argv) > 1 else '圣湘生物科技股份有限公司' print(f"采集: {cust}...") profile = fetch_profile(cust) safe = cust.replace('/', '_')[:20] out = os.path.join(BASE, f'数据/profiles/{safe}.json') os.makedirs(os.path.dirname(out), exist_ok=True) with open(out, 'w') as f: json.dump(profile, f, ensure_ascii=False, indent=2) summary = {k: (str(v)[:60] if not isinstance(v, list) else f'[{len(v)}条]') for k, v in profile.items()} print(json.dumps(summary, ensure_ascii=False, indent=2)) print(f"\n✅ 已保存: {out}")