#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 销售工作台 · 本地数据服务（仅依赖 Python 标准库） 用法： 1. 把本文件放在同时包含「工具」和「数据」两个文件夹的目录里 2. python3 销售工作台服务.py （默认端口 8000） python3 销售工作台服务.py 8080 （指定端口） 功能： - 静态文件服务（替代原先的静态服务器，原有页面/JSON 路径不变） - GET /api/todos 读取共享待办（存于 数据/todos.json） - POST /api/todos 写入共享待办，action 支持： save {user, todos} 整体覆盖某用户的待办列表 close {item} 追加一条团队已关闭记录（按 id 去重，最多保留 200 条） unclose {id} 按 id 移除已关闭记录（重新打开/删除时调用） - 所有 .json 响应自动附加 Cache-Control: no-store，避免浏览器缓存旧数据 """ import json import os import re import sys import subprocess import hashlib import threading import time from datetime import datetime, timedelta import urllib.request from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # 脚本/ → 父目录 if os.path.dirname(os.path.abspath(__file__)) not in sys.path: sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import activity_v2 def _expense_api(): """加载报销助手服务模块（多用户上传/台账/打印包）""" _r = os.path.join(ROOT, '报销助手', '脚本') if _r not in sys.path: sys.path.insert(0, _r) import importlib return importlib.import_module('expense_api') DATA_DIR = os.path.join(ROOT, '数据') TODOS_FILE = os.path.join(DATA_DIR, 'todos.json') ROLLUP_FILE = os.path.join(DATA_DIR, 'customer_rollup_v2.json') ROLLUP_LINKS = os.path.join(DATA_DIR, 'activity_opp_links.json') OPP_STATUS_FILE = os.path.join(DATA_DIR, 'opportunity_status.json') OPP_STATUSES = ['沟通需求确认', '产品试用中', '订单签订中', '实现销售'] _ROLLUP_LOCK = threading.Lock() def _rollup_cutoff(): return (datetime.now() - timedelta(days=183)).strftime('%Y-%m-%d') def _rollup_load(): with open(ROLLUP_FILE, encoding='utf-8') as f: return json.load(f) def _rollup_links_load(): try: with open(ROLLUP_LINKS, encoding='utf-8') as f: d = json.load(f) return d.get('links') or {} except Exception: return {} def _rollup_links_save(links): tmp = ROLLUP_LINKS + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump({'updated': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'links': links}, f, ensure_ascii=False, indent=2) os.replace(tmp, ROLLUP_LINKS) def _opp_status_load(): try: with open(OPP_STATUS_FILE, encoding='utf-8') as f: d = json.load(f) return d.get('items') or {} except Exception: return {} def _opp_status_save(items): tmp = OPP_STATUS_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump({'updated': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'items': items}, f, ensure_ascii=False, indent=2) os.replace(tmp, OPP_STATUS_FILE) def _rollup_apply_status(data, statuses=None): statuses = statuses if statuses is not None else _opp_status_load() for p in data.get('projects') or []: key = (p.get('cust') or '') + '||' + (p.get('opp') or '') it = statuses.get(key) or {} p['system_status'] = it.get('status') or '' p['status_by'] = it.get('by') or '' p['status_at'] = it.get('at') or '' p['admin_stage'] = it.get('stage') or '' p['admin_node_state'] = it.get('node_state') or '' p['admin_node_start'] = it.get('node_start') or '' p['admin_node_end'] = it.get('node_end') or '' p['admin_note'] = it.get('note') or '' p['admin_updated_at'] = it.get('updated_at') or '' return data def _rollup_apply_links(data, links=None): links = links if links is not None else _rollup_links_load() pmap = {} for p in data.get('projects') or []: pmap[(p.get('cust'), p.get('opp'))] = p for a in data.get('activities') or []: lk = links.get(a.get('id')) if not lk: continue opp = (lk.get('opp') or '').strip() if not opp: continue a['opp_names'] = [opp] a['opp_basis'] = '手动关联' p = pmap.get((a.get('customer'), opp)) if p: a['project_scene'] = p.get('scene') or a.get('project_scene') a['scene_basis'] = '按手动关联商机场景' return data FEISHU_DATA_FILE = os.path.join(DATA_DIR, 'customer_feishu_data.json') LOCK = threading.Lock() MAX_CLOSED = 200 # MCP 配置 MCP_TOKEN = 'm-abfb29e8-3104-434f-9944-8d0bb592f8cd' MCP_API = 'https://project.feishu.cn/mcp_server/v1' MCP_PK = '6593cd71471290e3cc6be6e6' ACTIVITY_TYPE_KEY = '65ae1e5d44338dbe7c39a29a' # MCP 数据缓存（TTL 秒） MCP_CACHE = {} MCP_CACHE_TTL = { 'customers': 1800, # 30分钟 'orders': 300, # 5分钟 'complaints': 900, # 15分钟 'activities': 300, # 5分钟 'opportunities': 300 # 5分钟 } MCP_CACHE_LOCK = threading.Lock() # ───────── 测试报告定向收集（2026-09-15 新增） ───────── TEST_REPORT_DIR = os.path.join(DATA_DIR, 'test_reports') TEST_REPORT_SCHEMA = os.path.join(DATA_DIR, 'test_report_schema.json') TEST_REPORT_OPT_TTL = 60 # 选择项缓存（秒） def build_test_report_options(user=None): """订单（联销售员）+ 货号 精简选择项。销售只看自己名下订单，admin/gm 全量。 销售若有权限但名下无订单，回退全量（避免空白页）。""" now = time.time() c = MCP_CACHE.get('__test_report_options__') if c and now - c[0] < TEST_REPORT_OPT_TTL: data = c[1] else: orders, seen = [], set() try: with open(os.path.join(DATA_DIR, 'mcp_orders_input.json'), encoding='utf-8') as f: raw = json.load(f) except Exception: raw = [] try: with open(os.path.join(DATA_DIR, 'order_sp_map.json'), encoding='utf-8') as f: spmap = (json.load(f) or {}).get('map') or {} except Exception: spmap = {} for o in (raw if isinstance(raw, list) else []): no = str(o.get('单据编号') or '').strip() if not no or no in seen: continue seen.add(no) orders.append({ 'no': no, 'customer': str(o.get('销售订单关联客户') or o.get('购货单位#') or '').strip(), 'date': str(o.get('创建时间') or '')[:10], 'doc_customer': str(o.get('购货单位#') or '').strip(), 'creator': str(o.get('创建者') or '').strip(), 'dept': str(o.get('销售部门') or '').strip(), 'sp': spmap.get(no) or str(o.get('创建者') or '').strip(), }) orders.sort(key=lambda x: x['date'], reverse=True) skus = [] try: with open(os.path.join(DATA_DIR, 'product_master.json'), encoding='utf-8') as f: pm = json.load(f) for s in (pm.get('skus') or []): skus.append({'sku': s.get('sku', ''), 'name': s.get('name', ''), 'spec': s.get('spec', ''), 'line': s.get('line', ''), 'cat': s.get('cat', '')}) except Exception: pass data = {'orders': orders, 'skus': skus} MCP_CACHE['__test_report_options__'] = (now, data) role = (user or {}).get('role') or '' name = (user or {}).get('name') or '' if role in ('admin', 'gm', 'director', 'readonly') or not name: return data if role == 'manager': mem = _dept_members((user or {}).get('dept') or '') mine = [o for o in data['orders'] if o.get('sp') in mem] if mem else [] return {'orders': mine, 'skus': data['skus']} mine = [o for o in data['orders'] if o.get('sp') == name] return {'orders': mine, 'skus': data['skus']} def save_test_report(payload, user=None): """落盘一份测试报告（含原始值、图片、AI 整理的 MD）＋ 维护汇总索引。""" os.makedirs(TEST_REPORT_DIR, exist_ok=True) rid = str(payload.get('id') or '').strip() or ('TR' + datetime.now().strftime('%Y%m%d%H%M%S')) # 原始填写值：优先取扁平的 values；否则从 sections 展开（字段标签 → 值），便于详情页按标签展示 values = payload.get('values') if not isinstance(values, dict) or not values: values = {} for sec in (payload.get('sections') or []): for f in (sec.get('fields') or []): lbl = str(f.get('label') or '').strip() val = str(f.get('value') or '').strip() if lbl and val: values[lbl] = val rec = { 'id': rid, 'created': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'submitted_by': (user or {}).get('name') or payload.get('sales') or '', 'product': payload.get('product') or '', 'product_type': payload.get('product_type') or '', 'order_no': payload.get('order_no') or '', 'customer': payload.get('customer') or '', 'sku': payload.get('sku') or '', 'sku_name': payload.get('sku_name') or '', 'status': payload.get('status') or 'submitted', 'values': values, 'sections': payload.get('sections') or [], 'images': payload.get('images') or [], 'md': payload.get('md') or '', } with LOCK: with open(os.path.join(TEST_REPORT_DIR, rid + '.json'), 'w', encoding='utf-8') as f: json.dump(rec, f, ensure_ascii=False, indent=1) idx_f = os.path.join(TEST_REPORT_DIR, '_index.json') try: with open(idx_f, encoding='utf-8') as f: idx = json.load(f) if not isinstance(idx, list): idx = [] except Exception: idx = [] idx = [x for x in idx if x.get('id') != rid] idx.append({k: rec.get(k) for k in ('id', 'created', 'submitted_by', 'product', 'product_type', 'order_no', 'customer', 'sku', 'sku_name', 'status')}) idx.sort(key=lambda x: x.get('created') or '', reverse=True) with open(idx_f, 'w', encoding='utf-8') as f: json.dump(idx, f, ensure_ascii=False, indent=1) return rid def list_test_reports(limit=200): idx_f = os.path.join(TEST_REPORT_DIR, '_index.json') try: with open(idx_f, encoding='utf-8') as f: idx = json.load(f) return idx[:limit] if isinstance(idx, list) else [] except Exception: return [] def structure_test_report(payload): """把销售填写的测试反馈交给 GLM 整理成结构化 Markdown（只整理、不编造）。""" _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat head = [ '产品方向：%s' % (payload.get('product') or '（未选择）'), '产品类型：%s' % (payload.get('product_type') or '未提供'), '订单号：%s' % (payload.get('order_no') or '未提供'), '客户：%s' % (payload.get('customer') or '未提供'), '货号：%s %s' % (payload.get('sku') or '未提供', payload.get('sku_name') or ''), '提交人：%s' % (payload.get('sales') or ''), '测试时间：%s' % (payload.get('test_date') or ''), '', '【表单填写内容】', ] body = [] for sec in (payload.get('sections') or []): body.append('## %s' % (sec.get('title') or '')) for f in (sec.get('fields') or []): v = str(f.get('value') or '').strip() if not v: continue u = f.get('unit') or '' body.append('- %s：%s%s' % (f.get('label') or '', v, (' ' + u) if u and u not in v else '')) body.append('') imgs = payload.get('images') or [] if imgs: body.append('【上传图片】共 %d 张：%s' % ( len(imgs), '、'.join([str(i.get('name') or i.get('url') or '') for i in imgs]))) sysp = ( '你是宝锐生物（诊断原料销售）的销售管理助理。请把销售提交的「客户测试反馈」整理成一份结构化 Markdown 报告。\n' '硬性要求：\n' '1) 只使用用户提供的信息，**绝不编造**任何数据、客户名、结论；缺失项写「未提供」。\n' '2) 保留所有数值与单位原样。\n' '3) 输出结构：一级标题（产品方向 · 客户）→ 一、基本信息（表格）→ 二、测试关键数据（表格）→ ' '三、性能评分（表格，含平均分）→ 四、问题与采购意向 → 五、开放反馈 → 六、结论与建议。\n' '4) 「结论与建议」必须基于填写内容推导：指出 1-3 条可执行建议（如需要补充的测试、需要澄清的数据），' '不确定的明确写「需进一步确认」，不要给泛泛而谈的话。\n' '5) 直接输出 Markdown 正文，不要任何解释性前言。' ) content, err = ai_chat._call_glm(sysp, [{'role': 'user', 'content': '\n'.join(head + body)}]) return content, err def mcp_pull_all(mql_query, max_pages=50): """拉取全量 MCP 数据，自动翻页，展开 field_list 为扁平 JSON""" all_items = [] session_id = None for page in range(1, max_pages + 1): args = {'project_key': MCP_PK, 'mql': mql_query} if session_id: args['session_id'] = session_id args['group_pagination_list'] = [{'page_num': page, 'group_id': 'default'}] result = mcp_call('search_by_mql', args, timeout=60) if not result: break # 从 data dict 中提取 items data_dict = result.get('data', {}) if not data_dict: break for gid, items in data_dict.items(): for item in items: flat = {} for f in (item.get('moql_field_list') or item.get('field_list') or []): k = f.get('key', '') v = f.get('value', None) # 展开 key_label_value 格式 if isinstance(v, list): # 数组类型（multi-select, multi-user, workitem_related_multi_select） flat[k] = [] for elem in v: if isinstance(elem, dict) and 'key_label_value' in elem: flat[k].append({'key': elem['key_label_value'].get('key',''), 'label': elem['key_label_value'].get('label','')}) elif isinstance(elem, dict) and len(elem) == 1: flat[k].append(list(elem.values())[0]) else: flat[k].append(elem) elif isinstance(v, dict) and 'key_label_value' in v: flat[k] = v['key_label_value'].get('label', v['key_label_value'].get('key','')) elif isinstance(v, dict) and len(v) == 1: flat[k] = list(v.values())[0] else: flat[k] = v all_items.append(flat) if len(items) < 50: return all_items # 最后一页 if not session_id: session_id = result.get('session_id', '') # 检查总数 group_infos = result.get('list', []) for gi in group_infos: total = gi.get('count', 0) if len(all_items) >= total: return all_items return all_items def get_mcp_cache(cache_key): """获取缓存，过期返回 None""" with MCP_CACHE_LOCK: entry = MCP_CACHE.get(cache_key) if entry and time.time() - entry['time'] < MCP_CACHE_TTL.get(cache_key, 300): return entry['data'] return None def set_mcp_cache(cache_key, data): """写入缓存 + 写入 JSON 文件作为降级备份""" with MCP_CACHE_LOCK: MCP_CACHE[cache_key] = {'time': time.time(), 'data': data} # 异步写 JSON 备份 backup_path = os.path.join(DATA_DIR, f'mcp_{cache_key}.json') try: with open(backup_path + '.tmp', 'w', encoding='utf-8') as f: json.dump({'updated_at': time.strftime('%Y-%m-%d %H:%M:%S'), 'customers' if cache_key == 'customers' else 'items': data}, f, ensure_ascii=False) os.replace(backup_path + '.tmp', backup_path) except Exception: pass def refresh_mcp_cache(cache_key, mql_query): """强制刷新缓存""" try: data = mcp_pull_all(mql_query) sys.stderr.write(f'[MCP] refresh {cache_key}: got {len(data)} items\n') sys.stderr.flush() set_mcp_cache(cache_key, data) return len(data) except Exception as e: sys.stderr.write(f'[MCP] refresh {cache_key} ERROR: {e}\n') sys.stderr.flush() # 尝试从备份文件恢复 backup_path = os.path.join(DATA_DIR, f'mcp_{cache_key}.json') if os.path.exists(backup_path): try: with open(backup_path, encoding='utf-8') as f: old = json.load(f) return len(old.get('data', [])) except Exception: pass raise e def mcp_call(method, args, timeout=20): """调用飞书 MCP API""" req = urllib.request.Request( MCP_API, data=json.dumps({ 'jsonrpc': '2.0', 'id': 1, 'method': 'tools/call', 'params': {'name': method, 'arguments': args} }).encode('utf-8'), headers={'X-Mcp-Token': MCP_TOKEN, 'Content-Type': 'application/json'}, method='POST' ) try: with urllib.request.urlopen(req, timeout=timeout) as r: d = json.loads(r.read()) if 'error' in d: return {'ok': False, 'error': d['error']} for c in d['result']['content']: t = c.get('text', '') if 'log_id' in t: continue return json.loads(t) except Exception as e: return {'ok': False, 'error': str(e)} return {'ok': False, 'error': 'no content'} # IMA 凭证 IMA_CLIENT_ID = os.environ.get('IMA_OPENAPI_CLIENTID') or open(os.path.expanduser('~/.config/ima/client_id')).read().strip() IMA_API_KEY = os.environ.get('IMA_OPENAPI_APIKEY') or open(os.path.expanduser('~/.config/ima/api_key')).read().strip() IMA_KB_DIAG = 'ima_kb_IktEcZCS' # 宝锐诊断原料销售 def ima_api(path, body): """调用 IMA OpenAPI""" req = urllib.request.Request( f'https://ima.qq.com/{path}', data=json.dumps(body).encode('utf-8'), headers={ 'ima-openapi-clientid': IMA_CLIENT_ID, 'ima-openapi-apikey': IMA_API_KEY, 'Content-Type': 'application/json' }, method='POST' ) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode('utf-8')) # ── 飞书通知 + IMA 回写审核队列 ── FEISHU_APP_ID = 'cli_aaee6a2f90b89bc1' FEISHU_APP_SECRET = 'arht5fd8wyppZysFvOjUcb2bV0UMVVYF' ADMIN_OPEN_ID = 'ou_901b790c9afdda4a8f14554962b32664' # 刘新元 IMA_REVIEW_FILE = os.path.join(DATA_DIR, 'ima_pending_review.json') def feishu_notify(open_id, text): """发送飞书文本消息，成功返回 True""" try: body = json.dumps({'app_id': FEISHU_APP_ID, 'app_secret': FEISHU_APP_SECRET}).encode() req = urllib.request.Request( 'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal', data=body, headers={'Content-Type': 'application/json'}, method='POST') r = json.loads(urllib.request.urlopen(req, timeout=10).read()) token = r.get('tenant_access_token', '') if not token: return False content = json.dumps({'text': text}, ensure_ascii=False) msg = json.dumps({'receive_id': open_id, 'msg_type': 'text', 'content': content}).encode('utf-8') req = urllib.request.Request( 'https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id', data=msg, headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + token}, method='POST') r = json.loads(urllib.request.urlopen(req, timeout=15).read()) return r.get('code') == 0 except Exception: return False def _load_ima_review(): try: if os.path.exists(IMA_REVIEW_FILE): with open(IMA_REVIEW_FILE, encoding='utf-8') as f: d = json.load(f) if isinstance(d, list): return d except Exception: pass return [] def _save_ima_review(data): try: os.makedirs(DATA_DIR, exist_ok=True) tmp = IMA_REVIEW_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=1) os.replace(tmp, IMA_REVIEW_FILE) return True except Exception: return False def read_store(): if not os.path.exists(TODOS_FILE): return {'users': {}, 'closed': [], 'updated_at': None} try: with open(TODOS_FILE, encoding='utf-8') as f: d = json.load(f) d.setdefault('users', {}) d.setdefault('closed', []) return d except Exception: return {'users': {}, 'closed': [], 'updated_at': None} def write_store(d): os.makedirs(DATA_DIR, exist_ok=True) d['updated_at'] = time.strftime('%Y-%m-%d %H:%M:%S') tmp = TODOS_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(d, f, ensure_ascii=False, indent=2) os.replace(tmp, TODOS_FILE) # 默认用户表（邮箱前缀 → 完整用户信息） DEFAULT_USERS = { 'admin': {'name': '超级管理员', 'role': 'admin', 'dept': '诊断原料销售部'}, 'liuxinyuan@biori.com': {'name': '刘新元', 'role': 'admin', 'dept': '诊断原料销售部'}, 'liuxin@biori.com': {'name': '刘欣', 'role': 'manager', 'dept': '销售拓展部'}, 'liuziyan@biori.com': {'name': '刘子研', 'role': 'sales', 'dept': '诊断原料大客户销售部'}, 'zhaoyunhao@biori.com': {'name': '赵云浩', 'role': 'sales', 'dept': '诊断原料大客户销售部'}, 'wuyun@biori.com': {'name': '吴云', 'role': 'manager', 'dept': '生命科学-浙江'}, 'hanyuanhuai@biori.com': {'name': '韩远怀', 'role': 'manager', 'dept': '生命科学-广东'}, 'songxuewei@biori.com': {'name': '宋学伟', 'role': 'sales', 'dept': '诊断原料销售拓展部'}, } # ============ 数据质检 ============ def run_quality_check(filepath, ext): """质检上传文件：检测类型、列名、数据完整性""" result = {'pass': False, 'type': 'unknown', 'columns': [], 'issues': [], 'summary': ''} if ext not in ('.xlsx', '.xls', '.csv'): result['summary'] = '不支持的文件格式，仅支持 .xlsx/.xls/.csv 数据文件' return result try: if ext == '.csv': import csv with open(filepath, 'r', encoding='utf-8-sig') as f: reader = csv.reader(f) headers = [h.strip() for h in next(reader)] else: import openpyxl wb = openpyxl.load_workbook(filepath, data_only=True) ws = wb.active first_row = next(ws.iter_rows(min_row=1, max_row=1, values_only=True)) headers = [str(c).strip() if c else '' for c in first_row] wb.close() result['columns'] = headers has_order = any('单据编号' in h for h in headers) has_tax = any('价税合计' in h or '税合计' in h for h in headers) has_customer = any('购货单位' in h or '客户' in h for h in headers) has_activity = any('活动' in h for h in headers) has_creator = any('创建者' in h or '创建人' in h for h in headers) if has_order and has_tax: result['type'] = 'erp_order' result['summary'] = f'识别为 ERP 销售订单导出，{len(headers)} 列' elif has_activity or has_creator: result['type'] = 'feishu_activity' result['summary'] = f'识别为 飞书活动导出，{len(headers)} 列' else: result['summary'] = f'未识别，{len(headers)} 列' result['issues'].append('无法匹配处理管线') return result if result['type'] == 'erp_order': for c in ['单据编号', '价税合计']: if not any(c in h for h in headers): result['issues'].append(f'缺少关键列: {c}') if not result['issues']: result['pass'] = True result['summary'] += '，质检通过' except Exception as e: result['issues'].append(str(e)) result['summary'] = f'质检失败: {e}' return result def process_uploaded_file(filepath, ftype, saved_name): """处理上传文件：智能识别月份 → 路由到正确数据文件 → 合并部署""" import re result = {'ok': False, 'summary': '', 'updated': []} try: ext = os.path.splitext(saved_name)[1].lower() if ext == '.csv': import csv with open(filepath, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) headers = reader.fieldnames rows_list = list(reader) else: import openpyxl wb = openpyxl.load_workbook(filepath, data_only=True) ws = wb.active headers = [str(c).strip() if c else '' for c in next(ws.iter_rows(min_row=1, max_row=1, values_only=True))] rows_list = [] for row in ws.iter_rows(min_row=2, values_only=True): if not row: continue rows_list.append({headers[i]: row[i] for i in range(min(len(headers), len(row)))}) wb.close() if ftype == 'erp_order': from collections import defaultdict, Counter cols = headers bn_col = amt_col = sp_col = None for c in cols: if '单据编号' in c: bn_col = c if '价税合计' in c or '税合计' in c: amt_col = c if '销售员' in c or '下单人' in c: sp_col = c # ── 智能月份检测 ── month_counter = Counter() for row in rows_list[:100]: bn = str(row.get(bn_col, '') or '').strip() if bn_col else '' if bn.startswith('ZHBR') and len(bn) >= 10: ym = '20' + bn[4:6] + '-' + bn[6:8] month_counter[ym] += 1 m = re.search(r'(2026\d{2})', saved_name) file_ym = m.group(1)[:4] + '-' + m.group(1)[4:6] if m else None detected_month = month_counter.most_common(1)[0][0] if month_counter else (file_ym or datetime.now().strftime('%Y-%m')) current_month = datetime.now().strftime('%Y-%m') # ── 数据汇总 ── by_sp = defaultdict(float) total = rows = 0 for row in rows_list: bn = str(row.get(bn_col, '') or '').strip() if bn_col else '' if not bn or bn == 'None' or '合计' in bn: continue amt = 0 try: amt = float(row.get(amt_col, 0) or 0) if amt_col else 0 except: continue total += amt sp = str(row.get(sp_col, '') or '').strip() if sp_col else '' if sp and amt > 0: by_sp[sp] += amt rows += 1 now_str = datetime.now().strftime('%Y-%m-%d %H:%M') people = [{'name': k, 'amount': round(v,2)} for k, v in sorted(by_sp.items(), key=lambda x: -x[1])] # ── 按月份路由 ── if detected_month == current_month: kpi_data = {'month': detected_month, 'updated': now_str, 'source': f'上传: {saved_name}', 'total': round(total,2), 'orders': rows, 'people': [{'name': k, 'current': round(v,2), 'ytd': round(v,2)} for k,v in sorted(by_sp.items(), key=lambda x: -x[1])]} with open(os.path.join(DATA_DIR, 'kpi_progress.json'), 'w', encoding='utf-8') as f: json.dump(kpi_data, f, ensure_ascii=False, indent=2) result['updated'].append('kpi_progress.json') result['summary'] = f'{detected_month} KPI: {rows}条, {total/1e4:.1f}万, {len(people)}人' else: # 历史月 → three_year_monthly + kpi_dashboard month_num = int(detected_month.split('-')[1]) ty_path = os.path.join(DATA_DIR, 'three_year_monthly.json') with open(ty_path, 'r', encoding='utf-8') as f: ty = json.load(f) ty['2026'][str(month_num)] = round(total, 2) with open(ty_path, 'w', encoding='utf-8') as f: json.dump(ty, f, ensure_ascii=False, indent=2) result['updated'].append('three_year_monthly.json') kd_path = os.path.join(DATA_DIR, 'kpi_dashboard.json') with open(kd_path, 'r', encoding='utf-8') as f: kd = json.load(f) kd['total_h2_actual'] = round(kd.get('total_h2_actual', 0) + total, 2) kd['total_h1_h2'] = round(kd.get('total_h1', 0) + kd['total_h2_actual'], 2) kd['updated_at'] = now_str with open(kd_path, 'w', encoding='utf-8') as f: json.dump(kd, f, ensure_ascii=False, indent=2) result['updated'].append('kpi_dashboard.json') result['summary'] = f'{detected_month} 校准: {rows}条, {total/1e4:.1f}万, {len(people)}人' subprocess.run(['python3', os.path.join(ROOT, '脚本', 'merge_workbench_data.py')], capture_output=True, timeout=30) result['updated'].append('workbench_data.json → CloudBase') subprocess.run(['bash', os.path.join(ROOT, '脚本', 'cb_sync.sh')], capture_output=True, timeout=60) result['ok'] = True else: result['summary'] = f'类型 {ftype} 暂不支持自动处理' except Exception as e: result['summary'] = f'处理失败: {e}' return result # ============ Token 鉴权（登录发 token，API 验 token，身份从会话表取，不信任前端传参） ============ import secrets as _secrets AUTH_SESSIONS = {} # {token: {name, role, dept, line, email, expire_at}} AUTH_LOCK = threading.Lock() AUTH_FILE = os.path.join(DATA_DIR, 'auth_sessions.json') TOKEN_TTL = 7 * 86400 # 7 天滑动续期 # ══════════ 只读角色（readonly，2026-10-15 新增，供只读代理/审计账号使用）══════════ # 语义：可读不可写。数据范围沿用 sales 逻辑（= 本人名下客户，由 customer_sp_map 决定）， # 即「最小权限」；如需更宽范围，请改用 manager/gm 并按需配置。 # 写端点：POST 一律 403；白名单默认空。若确有纯查询型 POST 需放行，显式加入并注明理由。 READONLY_ROLE = 'readonly' READONLY_ALLOW_POST = set() # ── CORS（2026-10-06）：精确来源白名单 ──────────────────────────────────────── # 为什么不是 '*'：/api/sandept 等接口靠 Authorization: Bearer 凭据鉴权， # 「ACAO=* + 凭据」既是反模式、也会被未来开启 credentials 的调用方拒绝。 # 这里只回显白名单来源（CloudBase 静态托管域 / Tunnel 域 / Tailscale 域 / 本机）， # 无 Origin 的同源或命令行请求不注入 ACAO。Function Call 与工作台其它 Tab 不受影响。 CORS_ORIGIN_HOSTS = ('api.biorisales.site',) def cors_allow_origin(origin): """返回允许回显的 ACAO 值；不在白名单 / 非法来源 → None（不注入头）。""" o = (origin or '').strip() if not o: return None try: from urllib.parse import urlparse p = urlparse(o) except Exception: return None if p.scheme not in ('http', 'https'): return None host = (p.hostname or '').lower() if not host: return None if host in ('127.0.0.1', 'localhost') or host in CORS_ORIGIN_HOSTS: return o if host.endswith('.tcloudbaseapp.com') or host.endswith('.ts.net'): return o return None USER_ROLES_FILE = os.path.join(DATA_DIR, 'user_roles.json') # CloudBase 同步（tcb symlink 在 Node v24 下静默失效，必须 node 直调 cli.js） TCB_CLI = "/Users/liuxinyuan/.npm-global/lib/node_modules/@cloudbase/cli/dist/standalone/cli.js" CB_ENV = "bier-sales-d0gatbvlx288724e9" def _dept_members(dept): """本部门在职成员姓名集合（部门名做归一：诊断原料销售拓展部 ≡ 销售拓展部）""" try: sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import ai_chat return set(ai_chat._dept_personnel(ai_chat._norm_dept(dept))) except Exception: return set() def _test_report_visible_people(role, name, user=None): """测试报告的可见提交人集合；None = 不限（admin/gm/director）""" role = role or '' if role in ('admin', 'gm', 'director', 'readonly') or not name: return None if role == 'manager': mem = _dept_members((user or {}).get('dept') or '') return mem | {name} if mem else {name} return {name} def _load_user_roles(): """加载本地用户表（从 CloudBase user_roles 同步的快照），失败返回 []""" try: with open(USER_ROLES_FILE, encoding='utf-8') as f: d = json.load(f) if isinstance(d, list): return d except Exception: pass return [] # ══════════════════ 结构化活动录入（项目跟进 / 采购跟进）══════════════════ ACT_DIR = os.path.join(DATA_DIR, 'activities') ACT_ATTACH_DIR = os.path.join(DATA_DIR, '活动附件') ACT_ORDERS_INDEX = os.path.join(DATA_DIR, 'orders_2026_index.json') ACT_CB_COLLECTION = 'activities' ACT_LOCK = threading.Lock() _ACT_CACHE = {'orders': None, 'mtime': 0} def _act_orders(): """2026 订单索引（懒加载；文件变更自动失效）""" try: mt = os.path.getmtime(ACT_ORDERS_INDEX) except Exception: return [] if _ACT_CACHE['orders'] is None or _ACT_CACHE['mtime'] != mt: try: with open(ACT_ORDERS_INDEX, encoding='utf-8') as f: _ACT_CACHE['orders'] = (json.load(f) or {}).get('orders') or [] _ACT_CACHE['mtime'] = mt except Exception: _ACT_CACHE['orders'] = [] return _ACT_CACHE['orders'] def _act_visible_sales(user): """可查看的销售员姓名集合；None = 不限（admin/gm）""" role = (user or {}).get('role') or 'sales' name = (user or {}).get('name') or '' if role in ('admin', 'gm', 'readonly'): return None if role == 'manager': mem = _dept_members((user or {}).get('dept') or '') return (mem | {name}) if mem else {name} return {name} def _act_can_see(user, rec): vis = _act_visible_sales(user) if vis is None: return True return (rec.get('submitted_by') in vis) or (rec.get('sales') in vis) def _mp_parse(raw, boundary): """极简 multipart 解析 -> (files[(filename, bytes)], fields{name: str})""" files, fields = [], {} try: parts = raw.split(('--' + boundary).encode()) except Exception: return files, fields for part in parts[1:]: if part.startswith(b'--'): continue if part.startswith(b'\r\n'): part = part[2:] if part.endswith(b'\r\n'): part = part[:-2] h = part.find(b'\r\n\r\n') if h < 0: continue head = part[:h].decode('utf-8', errors='replace') body = part[h + 4:] mf = re.search(r'filename="([^"]*)"', head) mn = re.search(r'name="([^"]*)"', head) if mf and mf.group(1): files.append((mf.group(1), body)) elif mn: fields[mn.group(1)] = body.decode('utf-8', errors='replace').strip() return files, fields def _tcb_storage_upload(local_path, cloud_path): """经 CLI 上传到 CloudBase 存储桶 -> (ok, msg)""" try: r = subprocess.run(['node', TCB_CLI, 'storage', 'upload', local_path, cloud_path, '-e', CB_ENV], capture_output=True, text=True, timeout=300) ok = (r.returncode == 0) and ('成功' in (r.stdout or '')) return ok, ((r.stdout or '') + (r.stderr or ''))[-300:] except Exception as e: return False, str(e) def _tcb_storage_url(cloud_path): """签发 CloudBase 存储临时 URL（约 1 小时）""" try: r = subprocess.run(['node', TCB_CLI, 'storage', 'url', cloud_path, '-e', CB_ENV, '--json'], capture_output=True, text=True, timeout=90) m = re.search(r'\{[\s\S]*\}', r.stdout or '') if m: return ((json.loads(m.group(0)).get('data') or {}).get('url') or '') except Exception: pass return '' def _act_block_lines(b, typ): """单个商机区块的描述行""" L = [] if typ == '客户档案完善': dec = b.get('decision') or [] if dec: L.append('【决策链】') for i2, x in enumerate(dec, 1): seg = [x.get('name') or ''] if x.get('title'): seg.append('（%s）' % x['title']) extra = [] if x.get('role'): extra.append('角色 %s' % x['role']) if x.get('power'): extra.append('影响力 %s' % x['power']) if x.get('phone'): extra.append('联系方式 %s' % x['phone']) if x.get('note'): extra.append('沟通偏好 %s' % x['note']) L.append('%d. %s%s' % (i2, ''.join(seg), ('｜' + '｜'.join(extra)) if extra else '')) if b.get('biz_desc'): L.append('【主营业务】\n%s' % b['biz_desc']) revs = b.get('revenues') or [] if revs: L.append('【近三年营收】') for x in revs: src = ('（来源：%s）' % x['source']) if x.get('source') else '' L.append('- %s：%s 万元%s' % (x.get('year') or '', x.get('amt') or '', src)) sups = b.get('suppliers') or [] if sups: L.append('【TOP%d 供应商】' % min(len(sups), 3) if len(sups) <= 3 else '【供应商】') for x in sups: seg2 = [x.get('name') or ''] if x.get('project'): seg2.append('主供 %s' % x['project']) share = [] if x.get('amt'): share.append('份额 %s 万元' % x['amt']) if x.get('pct'): share.append('占比 %s%%' % x['pct']) L.append('- %s%s' % ('｜'.join(seg2), ('（%s）' % '，'.join(share)) if share else '')) if b.get('self_share') or b.get('main_rival'): tail = [] if b.get('self_share'): tail.append('我司当前份额 %s%%' % b['self_share']) if b.get('main_rival'): tail.append('主要竞对 / 卡位 %s' % b['main_rival']) L.append('【竞争态势】%s' % '｜'.join(tail)) elif typ == '采购跟进': if b.get('visit_goal'): L.append('【拜访目标】%s' % b['visit_goal']) if b.get('visit_content'): L.append('【拜访内容】\n%s' % b['visit_content']) else: if b.get('need'): L.append('【客户需求】\n%s' % b['need']) if b.get('params'): L.append('【技术参数 / 测试数据】') for x in b['params']: L.append('- %s：%s' % (x.get('sku') or '', x.get('desc') or '')) t = b.get('test_order') or {} if t.get('no'): L.append('【测试单结果跟进】订单 %s%s' % (t['no'], ('（%s）' % t['customer']) if t.get('customer') else '')) for x in (t.get('items') or []): att = x.get('attachments') or [] L.append('- %s%s：%s%s' % (x.get('sku') or '', (' ' + x['name']) if x.get('name') else '', x.get('result') or '', ('｜附件 %d 个：%s' % (len(att), '、'.join(a.get('name') or '' for a in att))) if att else '')) if b.get('issue'): L.append('【问题现象】%s' % b['issue']) if b.get('feedback'): L.append('【客户反馈】%s' % b['feedback']) comp = b.get('competitors') or [] if comp: L.append('【竞对信息】') for c in comp: L.append('- %s%s：%s%s' % (c.get('brand') or '', ('（' + c['sku'] + '）') if c.get('sku') else '', c.get('perf') or '', ('｜客户评价：' + c['feedback']) if c.get('feedback') else '')) prics = [x for x in (b.get('prices') or []) if (x.get('brand') or '').strip() or (x.get('sku') or '').strip() or (x.get('price') or '').strip()] if prics: L.append('【价格信息】') for x in prics: seg = [] if x.get('brand'): seg.append('品牌 %s' % x['brand']) if x.get('sku'): seg.append('货号 %s' % x['sku']) if x.get('spec'): seg.append('规格 %s' % x['spec']) if x.get('qty'): seg.append('采购数量 %s' % x['qty']) if x.get('price'): seg.append('单价 %s' % x['price']) src = x.get('source') or '' if src == '其他' and (x.get('source_other') or '').strip(): src = '其他（%s）' % x['source_other'].strip() if src: seg.append('来源 %s' % src) L.append('- ' + '｜'.join(seg)) if b.get('plan'): L.append('【下一步计划】') for i, pp in enumerate(b['plan'], 1): L.append('%d. %s（%s / %s）' % (i, pp.get('item') or '', pp.get('owner') or '未定', pp.get('due') or '未定')) return L def _act_feishu_text(rec): """生成可粘贴到飞书的结构化描述（唯一来源：本地记录与前端共用） 新结构：按关联商机分块；每块一套跟进内容（含各自的下一步计划）""" L = [] if rec.get('type'): L.append('【活动类型】%s%s' % (rec['type'], ('｜' + rec['subtype']) if rec.get('subtype') else '')) for k, lbl in (('customer', '客户'), ('contact', '联系人'), ('date', '活动日期')): if rec.get(k): L.append('【%s】%s' % (lbl, rec[k])) opps = [o for o in (rec.get('opportunities') or []) if (o.get('name') or '').strip()] if opps: L.append('【关联商机】%s' % '、'.join((o.get('name') or '').strip() for o in opps)) blocks = rec.get('blocks') or [] if blocks: multi = (len(blocks) > 1) for n, b in enumerate(blocks, 1): name = (b.get('opp') or '').strip() if multi or name: L.append('') L.append('━━ %s ━━' % (('商机：' + name) if name else '未关联商机')) L.extend(_act_block_lines(b, rec.get('type') or '项目跟进')) return '\n'.join([x for x in L if x]) # ── 兜底：老结构（平铺字段），保持历史记录渲染不变 ── if rec.get('type') == '采购跟进': if rec.get('visit_goal'): L.append('【拜访目标】%s' % rec['visit_goal']) if rec.get('visit_content'): L.append('【拜访内容】\n%s' % rec['visit_content']) else: if rec.get('need'): L.append('【客户需求】\n%s' % rec['need']) if rec.get('params'): L.append('【技术参数 / 测试数据】') for x in rec['params']: L.append('- %s：%s' % (x.get('sku') or '', x.get('desc') or '')) t = rec.get('test_order') or {} if t.get('no'): L.append('【测试单结果跟进】订单 %s%s' % (t['no'], ('（%s）' % t['customer']) if t.get('customer') else '')) for x in (t.get('items') or []): att = x.get('attachments') or [] L.append('- %s%s：%s%s' % (x.get('sku') or '', (' ' + x['name']) if x.get('name') else '', x.get('result') or '', ('｜附件 %d 个：%s' % (len(att), '、'.join(a.get('name') or '' for a in att))) if att else '')) if rec.get('issue'): L.append('【问题现象】%s' % rec['issue']) if rec.get('feedback'): L.append('【客户反馈】%s' % rec['feedback']) comp = rec.get('competitors') or [] if comp: L.append('【竞对信息】') for c in comp: L.append('- %s%s：%s%s' % (c.get('brand') or '', ('（' + c['sku'] + '）') if c.get('sku') else '', c.get('perf') or '', ('｜客户评价：' + c['feedback']) if c.get('feedback') else '')) prics_old = [x for x in (rec.get('prices') or []) if (x.get('brand') or '').strip() or (x.get('sku') or '').strip() or (x.get('price') or '').strip()] if prics_old: L.append('【价格信息】') for x in prics_old: seg = [] if x.get('brand'): seg.append('品牌 %s' % x['brand']) if x.get('sku'): seg.append('货号 %s' % x['sku']) if x.get('spec'): seg.append('规格 %s' % x['spec']) if x.get('qty'): seg.append('采购数量 %s' % x['qty']) if x.get('price'): seg.append('单价 %s' % x['price']) src = x.get('source') or '' if src == '其他' and (x.get('source_other') or '').strip(): src = '其他（%s）' % x['source_other'].strip() if src: seg.append('来源 %s' % src) L.append('- ' + '｜'.join(seg)) plan = rec.get('plan') or [] if plan: L.append('【下一步计划】') for i, pp in enumerate(plan, 1): L.append('%d. %s（%s / %s）' % (i, pp.get('item') or '', pp.get('owner') or '未定', pp.get('due') or '未定')) return '\n'.join([x for x in L if x]) # ══════════ P1：活动 × 待办打通 + 拜访视图 ══════════ ORDERS_INDEX_FILE = os.path.join(DATA_DIR, 'orders_2026_index.json') _ORDERS_CACHE = {'data': None, 'mtime': None} def _orders_index(): """订单索引（2.8MB，进程内缓存，按 mtime 失效）。""" try: mt = os.path.getmtime(ORDERS_INDEX_FILE) except OSError: return {} if _ORDERS_CACHE['data'] is None or _ORDERS_CACHE['mtime'] != mt: try: with open(ORDERS_INDEX_FILE, encoding='utf-8') as f: raw = json.load(f) _ORDERS_CACHE['data'] = {str(o.get('no')): o for o in (raw.get('orders') or [])} _ORDERS_CACHE['mtime'] = mt except Exception: return {} return _ORDERS_CACHE['data'] def _act_classify_orders(nos): """订单号列表 → [{no,kind,customer,date,in_index}]。 kind 由服务端按索引 is_test 判定，**不让前端自报类型**（前端自报可造假/可填错）。 """ out = [] for _no in (nos or []): _no = str(_no).strip() if not _no: continue h = _orders_index().get(_no) out.append({'no': _no, 'kind': ('test' if (h or {}).get('is_test') else 'ship'), 'customer': (h or {}).get('customer') or '', 'date': (h or {}).get('date') or '', 'in_index': bool(h)}) return out ACT_TYPES = ['线下拜访', '线上跟进', '陌拜', '协同拜访'] # 飞书「活动类型」field_76654e 的选项，顺序=展示顺序 CH_OF_TYPE = {'offline': '线下拜访', 'online': '线上跟进', 'cold': '陌拜', 'joint': '协同拜访'} DEPT_SPLIT = {'生命科学-浙江': ('生命科学销售部', '浙江'), '生命科学-广东': ('生命科学销售部', '广东')} def _dept_overrides(): """归属补充映射（数据/dept_overrides.json）。 给不在 user_roles 的人（离职/历史账号）指定部门与区域，**仅供统计归属**—— 刻意与 user_roles 分离，避免为只做归属统计的人造出登录账号。用户表优先。 """ try: with open(os.path.join(DATA_DIR, 'dept_overrides.json'), encoding='utf-8') as f: d = json.load(f) return {k: v for k, v in d.items() if not k.startswith('_') and isinstance(v, dict)} except Exception: return {} def _sales_dept_map(): """姓名 → (部门, 区域)。部门两级归一：生命科学-浙江/广东 → 生命科学销售部 + 区域。 数据源：数据/user_roles.json（服务端快照）。不在表里的（离职/代管/订单销售员）归「未匹配部门」。 """ m = {} try: for u in _load_user_roles(): nm = (u.get('name') or '').strip() if not nm: continue raw = (u.get('dept') or '').strip() dept, region = DEPT_SPLIT.get(raw, (raw or '未匹配部门', '')) m[nm] = (dept, region) except Exception: pass for _nm, _v in _dept_overrides().items(): if _nm in m or not isinstance(_v, dict): continue # 用户表优先；本表只填空缺 _raw = (_v.get('dept') or '').strip() if not _raw: continue _dept, _region = DEPT_SPLIT.get(_raw, (_raw, '')) m[_nm] = (_dept, (_v.get('region') or _region or '').strip()) return m TD_SCOPE = { 'internal_project': 'internal', 'internal_complaint': 'internal', 'internal_test_plan': 'internal', 'internal_other': 'internal', 'external_visit': 'external', 'external_test_req': 'external', 'external_test_result': 'external', 'external_project': 'external', 'external_appointment': 'external', } def _todo_fp(seed, uid): """指纹只锚 (seed, 稳定行 uid) —— 绝不锚会变的文本，否则改措辞会变出新条目。""" return hashlib.sha1(('%s#%s' % (seed, uid)).encode('utf-8')).hexdigest()[:16] def _cbt_save(user, todos): """CloudBase todos 整体覆盖写（与前端 pushTodos 同形态；失败不阻断本地）。""" try: cmd = json.dumps({'update': 'todos', 'updates': [{'q': {'user': user}, 'u': {'$set': {'todos': todos, 'updated': datetime.now().strftime('%Y-%m-%dT%H:%M:%S.000Z')}}}]}, ensure_ascii=False) subprocess.run(['node', TCB_CLI, 'db', 'nosql', 'execute', '--command', json.dumps([{'TableName': 'todos', 'CommandType': 'UPDATE', 'Command': cmd}], ensure_ascii=False), '--json', '-e', CB_ENV], capture_output=True, text=True, timeout=120) return True except Exception: return False FEISHU_ACT_FILE = os.path.join(DATA_DIR, 'activities_feishu.json') _FS_ACT = {'data': None, 'mtime': None, 'idx': {}} def _feishu_act_rows(): """飞书历史活动（只读基础数据，由 脚本/build_feishu_activities.py 生成）。按 mtime 缓存。""" try: mt = os.path.getmtime(FEISHU_ACT_FILE) except OSError: return [] if _FS_ACT['data'] is None or _FS_ACT['mtime'] != mt: try: with open(FEISHU_ACT_FILE, encoding='utf-8') as f: d = json.load(f) rows = d.get('items') or [] _FS_ACT['data'] = rows _FS_ACT['idx'] = {r.get('id'): r for r in rows} _FS_ACT['mtime'] = mt except Exception: return [] return _FS_ACT['data'] def _feishu_act_one(fid): _feishu_act_rows() return _FS_ACT['idx'].get(fid) def _act_sync_todos(rec, confirmed, draft_id): """把「销售提交前确认过的待办清单」落库。 规则（用户 2026-10-07 定版）： · 生成 ≠ 落库：只落 confirmed=True 的条目； · 幂等：fp = sha1((draft_id|活动id) + '#' + 稳定行 uid)，重复提交只更新不新增； · 绝不覆盖手工条：只处理 source.kind == 'activity' 的记录； · 绝不抹掉人的动作：命中已有条目时只更新 text/type/scope/due/owner/tech_support， 永不动 done / closed / closedBy / conclusion。 """ seed = (draft_id or '').strip() or rec['id'] store = read_store() users = store.setdefault('users', {}) created = updated = removed = skipped = 0 touched = set() live = {} for c in (confirmed or []): uid = str(c.get('uid') or '').strip() text = str(c.get('text') or c.get('item') or '').strip() if not uid or not text or not c.get('confirmed'): skipped += 1 continue owner = str(c.get('owner') or rec.get('submitted_by') or '').strip() if not owner: skipped += 1 continue tp = c.get('type') if c.get('type') in TD_SCOPE else 'external_project' fp = _todo_fp(seed, uid) live[fp] = True lst = users.setdefault(owner, []) hit = None for t in lst: if t.get('fp') == fp: hit = t break src = {'kind': 'activity', 'activity_id': rec['id'], 'draft_id': (draft_id or ''), 'customer': rec.get('customer', ''), 'channel': rec.get('channel', '')} if hit is not None: if (hit.get('source') or {}).get('kind') == 'activity': hit['text'] = text hit['type'] = tp hit['cat'] = tp hit['scope'] = TD_SCOPE[tp] hit['due'] = str(c.get('due') or '').strip() hit['owner'] = owner hit['tech_support'] = bool(c.get('tech_support')) hit['source'] = src hit['updated'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') updated += 1 touched.add(owner) else: skipped += 1 else: lst.insert(0, { 'id': datetime.now().strftime('%y%m%d%H%M%S') + str(int(fp[:6], 16) % 1000).zfill(3), 'text': text, 'done': False, 'type': tp, 'cat': tp, 'scope': TD_SCOPE[tp], 'owner': owner, 'tech_support': bool(c.get('tech_support')), 'client': rec.get('customer', ''), 'due': str(c.get('due') or '').strip(), 'note': '', 'created': datetime.now().strftime('%Y-%m-%d'), 'fp': fp, 'source': src, }) created += 1 touched.add(owner) # 活动派生条中，本轮清单已不含的 → 标 source_removed（不物理删除，留痕给人决定） for owner, lst in list(users.items()): if not isinstance(lst, list): continue for t in lst: s = t.get('source') or {} if s.get('kind') != 'activity': continue same = (s.get('activity_id') == rec['id']) or (draft_id and s.get('draft_id') == draft_id) if same and t.get('fp') and t['fp'] not in live and not t.get('source_removed'): t['source_removed'] = True removed += 1 touched.add(owner) if touched: store['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') write_store(store) for o in touched: _cbt_save(o, users.get(o) or []) return {'created': created, 'updated': updated, 'removed': removed, 'skipped': skipped, 'owners': sorted(touched)} def _act_save(rec): os.makedirs(ACT_DIR, exist_ok=True) with ACT_LOCK: with open(os.path.join(ACT_DIR, rec['id'] + '.json'), 'w', encoding='utf-8') as f: json.dump(rec, f, ensure_ascii=False, indent=1) idx_f = os.path.join(ACT_DIR, '_index.json') try: with open(idx_f, encoding='utf-8') as f: idx = json.load(f) if not isinstance(idx, list): idx = [] except Exception: idx = [] idx = [x for x in idx if x.get('id') != rec['id']] idx.append({k: rec.get(k) for k in ('id', 'created', 'submitted_by', 'sales', 'type', 'subtype', 'customer', 'contact', 'date', 'channel', 'order_no', 'sku_count', 'att_count', 'opp_count', 'block_count')}) idx.sort(key=lambda x: x.get('created') or '', reverse=True) with open(idx_f, 'w', encoding='utf-8') as f: json.dump(idx, f, ensure_ascii=False, indent=1) def _act_cb_insert(rec): """写入 CloudBase activities 集合（失败不阻断本地落盘）""" try: cmd_obj = {'TableName': ACT_CB_COLLECTION, 'CommandType': 'ADD', 'Command': json.dumps({'insert': ACT_CB_COLLECTION, 'documents': [rec]}, ensure_ascii=False)} cmd = json.dumps([cmd_obj], ensure_ascii=False) r = subprocess.run(['node', TCB_CLI, 'db', 'nosql', 'execute', '--command', cmd, '--json', '-e', CB_ENV], capture_output=True, text=True, timeout=120) out = ((r.stdout or '') + (r.stderr or '')).strip() ok = (r.returncode == 0) and ('ok' in out.lower()) and ('false' not in out.lower()[:200]) and ('error' not in out.lower()[:200]) return ok, out[-300:] except Exception as e: return False, str(e) def _act_bind_attach_dir(dir_id, rec): """把附件目录绑定到活动（供 /attachment_url 鉴权；目录 id 可能是上传时的临时 id）""" if not dir_id: return try: d = os.path.join(ACT_ATTACH_DIR, os.path.basename(dir_id)) os.makedirs(d, exist_ok=True) with open(os.path.join(d, '_owner.json'), 'w', encoding='utf-8') as f: json.dump({'activity': rec.get('id'), 'submitted_by': rec.get('submitted_by'), 'sales': rec.get('sales')}, f, ensure_ascii=False) except Exception: pass def _act_attach_owner(dir_id): """附件目录的归属（先查活动记录，再查 _owner.json 绑定）""" did = os.path.basename(dir_id or '') if not did: return None try: with open(os.path.join(ACT_DIR, did + '.json'), encoding='utf-8') as f: return json.load(f) except Exception: pass try: with open(os.path.join(ACT_ATTACH_DIR, did, '_owner.json'), encoding='utf-8') as f: o = json.load(f) # 绑定里只有姓名 → 转成可判定结构，再从活动记录补 sales rec = {'submitted_by': o.get('submitted_by'), 'sales': o.get('sales')} try: with open(os.path.join(ACT_DIR, (o.get('activity') or '') + '.json'), encoding='utf-8') as f: a = json.load(f) rec = {'submitted_by': a.get('submitted_by'), 'sales': a.get('sales')} except Exception: pass return rec except Exception: return None def _act_load_index(): try: with open(os.path.join(ACT_DIR, '_index.json'), encoding='utf-8') as f: idx = json.load(f) return idx if isinstance(idx, list) else [] except Exception: return [] def _price_scope_of(user): """该用户「价格查询」口径（all/own/none）：role 默认 + 权限面板覆盖（供登录响应下发给前端）""" try: _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import price_api return price_api.scope_for(user or {})[0] except Exception: _r = (user or {}).get('role') or 'sales' return 'all' if _r in ('admin', 'gm', 'readonly') else 'own' def _feishu_identity(code): """飞书免登：用授权 code 换用户身份（直连飞书 API，不依赖 CloudBase）。 返回 (user_dict, None) 或 (None, error_str)。""" import urllib.request FEISHU_APP_ID = 'cli_aaee6a2f90b89bc1' FEISHU_APP_SECRET = 'arht5fd8wyppZysFvOjUcb2bV0UMVVYF' try: # Step1: app_access_token req = urllib.request.Request( 'https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal', data=json.dumps({'app_id': FEISHU_APP_ID, 'app_secret': FEISHU_APP_SECRET}).encode('utf-8'), headers={'Content-Type': 'application/json'}, method='POST') with urllib.request.urlopen(req, timeout=10) as r: tk = json.loads(r.read()) app_token = tk.get('app_access_token') if not app_token: return None, 'no access token' # Step2: code 换身份（OIDC） req2 = urllib.request.Request( 'https://open.feishu.cn/open-apis/authen/v1/oidc/access_token', data=json.dumps({'grant_type': 'authorization_code', 'code': code}).encode('utf-8'), headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + app_token}, method='POST') with urllib.request.urlopen(req2, timeout=10) as r: auth = json.loads(r.read()) if auth.get('code') != 0: return None, 'auth failed: %s' % str(auth.get('msg', '')) data = auth.get('data') or {} name = (data.get('name') or '').strip() open_id = (data.get('open_id') or '').strip() if not name and not open_id: return None, 'no identity' # Step3: 匹配本地用户表（open_id 精确优先，name 兜底） users = _load_user_roles() user = None if open_id: for u in users: if u.get('open_id') == open_id: user = u break if not user and name: for u in users: if u.get('name') == name: user = u break if not user: return None, '账号未绑定：%s' % (name or open_id) return user, None except Exception as e: return None, str(e) def _sync_tabs_to_cloudbase(name, tabs): """把某用户的 tabs 字段同步到 CloudBase user_roles 集合（node 直调 cli.js）。 返回 (ok: bool, err: str)。用 _id 定位文档，$set 更新 tabs。""" try: roles = _load_user_roles() doc_id = None for u in roles: if u.get('name') == name: doc_id = u.get('_id') break if not doc_id: return False, '本地用户表无该用户 _id，无法同步 CloudBase' cmd_obj = { "TableName": "user_roles", "CommandType": "UPDATE", "Command": json.dumps({ "update": "user_roles", "updates": [{"q": {"_id": {"$oid": doc_id}}, "u": {"$set": {"tabs": tabs}}}], }), } cmd = json.dumps([cmd_obj], ensure_ascii=False) r = subprocess.run(["node", TCB_CLI, "db", "nosql", "execute", "--command", cmd, "--json", "-e", CB_ENV], capture_output=True, text=True, timeout=60) raw = r.stdout or '' start = raw.find("{") if start < 0: return False, 'CloudBase 同步失败：' + (r.stderr or raw)[:300] d = json.loads(raw[start:]) if "error" in d: return False, 'CloudBase 同步出错：' + json.dumps(d.get("error"), ensure_ascii=False)[:300] # 校验是否真的匹配并修改到文档（$oid 匹配失败会返回 n=0，命令仍不报 error） try: res = ((d.get("data", {}).get("results") or [[]])[0] or [{}])[0] n = res.get("n") if res.get("n") is not None else res.get("nMatched") if isinstance(n, dict): n = n.get("$numberInt") or n.get("$numberLong") if n is not None and int(n) == 0: return False, 'CloudBase 未匹配到该用户文档（_id 可能已变更，请重新同步本地用户表）' except Exception: pass return True, '' except Exception as e: return False, 'CloudBase 同步异常：' + str(e) def _load_sessions(): """加载会话表，剔除过期项""" try: if os.path.exists(AUTH_FILE): with open(AUTH_FILE, encoding='utf-8') as f: d = json.load(f) now = time.time() return {k: v for k, v in d.items() if v.get('expire_at', 0) > now} except Exception: pass return {} def _save_sessions(): try: os.makedirs(DATA_DIR, exist_ok=True) tmp = AUTH_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(AUTH_SESSIONS, f, ensure_ascii=False) os.replace(tmp, AUTH_FILE) except Exception: pass AUTH_SESSIONS = _load_sessions() # ══════════ 三部门逐月拆分看板：按登录态过滤的数据接口（方案A · 服务端过滤）══════════ _SANDEPT_FILE = os.path.join(DATA_DIR, '三部门逐月_2026.json') _SANDEPT_CACHE = {'mtime': None, 'data': None, 'lock': threading.Lock()} def _load_sandept_data(): """读取并缓存看板数据（按文件 mtime 失效）。返回 dict 或 None。""" try: mt = os.path.getmtime(_SANDEPT_FILE) except OSError: return None with _SANDEPT_CACHE['lock']: if _SANDEPT_CACHE['mtime'] == mt and _SANDEPT_CACHE['data'] is not None: return _SANDEPT_CACHE['data'] try: with open(_SANDEPT_FILE, encoding='utf-8') as f: data = json.load(f) except Exception: return None with _SANDEPT_CACHE['lock']: _SANDEPT_CACHE['mtime'] = mt _SANDEPT_CACHE['data'] = data return data # ══════════ 目标基础数据（唯一权威源 · 管理台）：仅 admin/gm 可读 ══════════ _TARGETS_FILE = os.path.join(DATA_DIR, '目标基础数据_2026.json') _TARGETS_CACHE = {'mtime': None, 'data': None, 'lock': threading.Lock()} def _load_targets_data(): """读取并缓存权威目标源（按文件 mtime 失效）。返回 dict 或 None。""" try: mt = os.path.getmtime(_TARGETS_FILE) except OSError: return None with _TARGETS_CACHE['lock']: if _TARGETS_CACHE['mtime'] == mt and _TARGETS_CACHE['data'] is not None: return _TARGETS_CACHE['data'] try: with open(_TARGETS_FILE, encoding='utf-8') as f: data = json.load(f) except Exception: return None with _TARGETS_CACHE['lock']: _TARGETS_CACHE['mtime'] = mt _TARGETS_CACHE['data'] = data return data def _targets_payload(user): """仅 admin/gm 可读（服务端硬门槛）；返回 (payload, err)。 数据源自唯一权威源 数据/目标基础数据_2026.json；不做写接口（改动走页面导出→落盘→重跑管道）。""" role = (user or {}).get('role') or 'sales' if role not in ('admin', 'gm', 'readonly'): return None, 'forbidden' d = _load_targets_data() if d is None: return None, '数据文件缺失：数据/目标基础数据_2026.json' try: import target_source as _ts ok, rep = _ts.check_closure(d) except Exception as e: ok, rep = False, {'errors': ['closure 计算异常: %s' % e]} return { 'ok': True, 'scope': 'admin', 'user': {'name': (user or {}).get('name') or '', 'role': role}, 'closure_ok': ok, 'closure': rep, 'data': d, }, None def _sandept_group_owners(all_customers): """（仅内部用于判定 partial）全量数据里 key(=grp||cust) → 归属方集合。 sales 值 '' 代表「无归属」。返回结构只在本函数/归集函数内部使用， 绝不出现在响应里 —— 只允许输出布尔与计数。""" owners = {} for c in all_customers or []: if not c: continue key = c.get('grp') or c.get('cust') if key in (None, ''): continue bucket = owners.setdefault(str(key), set()) sv = c.get('sales') vals = sv if isinstance(sv, list) else [sv] for v in vals: bucket.add('' if v in (None, '') else str(v)) return owners def _agg_own_group_rows(own_rows, owners, name): """用「本人名下」的 customers 行现场归集集团口径（key = grp || cust，与全局一致）。 · 金额 = 本人名下这些行的 amt / amt_2025 之和（不含任何他人金额） · 每行附加 partial(布尔) 与 other_owner_count(整数)：由全量 owners 判定， 但只输出布尔/计数，不含他人姓名、不含他人客户名。""" agg, order = {}, [] for c in own_rows or []: if not c: continue grp = c.get('grp') key = grp or c.get('cust') if key in (None, ''): key = '（未标注）' key = str(key) ak = (c.get('month'), c.get('dept'), key) o = agg.get(ak) if o is None: o = agg[ak] = {'month': c.get('month'), 'dept': c.get('dept'), 'cust': key, 'grp': (grp or None), 'amt': None, 'amt_2025': None} order.append(ak) a = c.get('amt') if isinstance(a, (int, float)): o['amt'] = (o['amt'] or 0) + a b = c.get('amt_2025') if isinstance(b, (int, float)): o['amt_2025'] = (o['amt_2025'] or 0) + b rows = [] for ak in order: o = agg[ak] bucket = owners.get(o['cust']) or set() own = name or '' others = set(x for x in bucket if x and x != own) has_unassigned = ('' in bucket) cnt = len(others) + (1 if has_unassigned else 0) # 无归属记作 1 个「其他归属方」(只计数，不具名) o['partial'] = bool(cnt > 0) o['other_owner_count'] = int(cnt) rows.append(o) return rows def _sandept_payload(user): """按角色过滤后返回看板数据 —— 服务端过滤是唯一硬保证： · admin / gm → 完整 customers + customers_grp（行为不变） · 其他角色(含 manager) → 仅 sales == 本人姓名 的客户行；customers_grp 由其本人 名下的 customers 行现场归集（集团口径可见，但只含本人金额；跨人集团标 partial） 聚合字段（months / depts / total / salespersons / meta）不含客户名，全量返回。""" data = _load_sandept_data() if data is None: return None, '数据文件缺失' role = (user or {}).get('role') or 'sales' name = (user or {}).get('name') or '' priv = role in ('admin', 'gm', 'readonly') all_c = data.get('customers') or [] if priv: customers = all_c customers_grp = data.get('customers_grp') or [] else: def _own_match(c): if not name: return False sv = c.get('sales') if isinstance(sv, list): return name in sv return sv == name customers = [c for c in all_c if c and _own_match(c)] customers_grp = _agg_own_group_rows(customers, _sandept_group_owners(all_c), name) # 销售员聚合（含姓名+金额）同样按本人过滤：否则普通销售可在「按销售员」下钻里 # 看到同事的月度金额与客户数 —— 与「不泄露同事」口径冲突 salespersons = [r for r in (data.get('salespersons') or []) if r and r.get('sp') == name] return { 'ok': True, 'user': {'name': name, 'role': role, 'dept': (user or {}).get('dept') or ''}, 'scope': 'all' if priv else 'own', 'data': { 'meta': data.get('meta') or {}, 'months': data.get('months') or [], 'depts': data.get('depts') or {}, 'total': data.get('total') or {}, 'q4': data.get('q4'), 'salespersons': (data.get('salespersons') or []) if priv else salespersons, 'customers': customers, 'customers_grp': customers_grp, }, }, None class Handler(SimpleHTTPRequestHandler): def _auth_user(self): """从 Authorization: Bearer 验证身份，返回 user dict 或 None（含滑动续期）""" auth = self.headers.get('Authorization', '') or '' token = auth[7:].strip() if auth.startswith('Bearer ') else '' if not token: return None with AUTH_LOCK: sess = AUTH_SESSIONS.get(token) if not sess: return None if sess.get('expire_at', 0) < time.time(): with AUTH_LOCK: AUTH_SESSIONS.pop(token, None) return None sess['expire_at'] = time.time() + TOKEN_TTL # 滑动续期 return {'name': sess.get('name', ''), 'role': sess.get('role', 'sales'), 'dept': sess.get('dept', ''), 'line': sess.get('line', 'da'), 'email': sess.get('email', '')} def _find_user(self, email, password): """验证用户：仅当本地用户表里没有该邮箱时，才允许默认表兜底（密码 biori2026）。 加这道闸是因为：本地表里已改过密码的用户，若不拦，旧密码 biori2026 仍能登录， 并拿到 DEFAULT_USERS 里写死的角色（曾导致刘欣可用公共密码拿到 admin）。""" email = email.strip().lower() if any((u.get('email') or '').lower() == email for u in _load_user_roles()): return None user = DEFAULT_USERS.get(email) if not user: return None valid = password == 'biori2026' if not valid: return None return dict(user) def log_message(self, fmt, *args): sys.stderr.write('%s - %s\n' % (self.address_string(), fmt % args)) def end_headers(self): path = getattr(self, 'path', '').split('?')[0] if hasattr(self, 'path') else '' if path.endswith('.json') or path.startswith('/api/') or path.endswith('.html'): self.send_header('Cache-Control', 'no-store, no-cache, must-revalidate') super().end_headers() def _cors_headers(self): """注入精确来源 CORS 头（白名单外不注入；绝不与凭据共用 '*'）。""" o = cors_allow_origin(self.headers.get('Origin')) if o: self.send_header('Access-Control-Allow-Origin', o) self.send_header('Access-Control-Allow-Credentials', 'true') self.send_header('Vary', 'Origin') def _send_json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self._cors_headers() self.end_headers() self.wfile.write(body) def _stream_sse(self, generator): """SSE 流式响应：逐事件写 data: {json}\n\n，供 /api/chat 流式透传""" self.send_response(200) self.send_header('Content-Type', 'text/event-stream; charset=utf-8') self._cors_headers() self.send_header('X-Accel-Buffering', 'no') self.end_headers() try: for event in generator: line = 'data: ' + json.dumps(event, ensure_ascii=False) + '\n\n' self.wfile.write(line.encode('utf-8')) self.wfile.flush() self.wfile.write(b'data: [DONE]\n\n') self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass def do_OPTIONS(self): """处理 CORS 预检请求""" self.send_response(204) self._cors_headers() self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization') self.end_headers() def do_GET(self): path = self.path.split('?')[0] # 统一鉴权兜底：所有 /api/ 端点（除登录/换 token/飞书免登）都需有效 token if path.startswith('/api/') and path not in ('/api/login', '/api/auth/token', '/api/feishu/login'): if not self._auth_user(): self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return # ══════════ 销售下单（三合一 + 审批规则引擎）══════════ if path.startswith('/api/order/'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return from urllib.parse import parse_qs _oqs = {k: v[0] for k, v in parse_qs(self.path.split('?', 1)[1]).items()} if '?' in self.path else {} try: import order_api order_api.setup(visible_sales=_act_visible_sales, notify=feishu_notify) _res = order_api.handle_get(path, _oqs, _au) except Exception as _e: import traceback traceback.print_exc() _res = {'ok': False, 'error': '下单接口异常: %s' % _e} self._send_json(_res if _res is not None else {'ok': False, 'error': '未知下单接口'}) return # ══════════ 顺丰物流：按需刷新（订单管理页「🔄 刷新物流」/ 进页自动刷）══════════ if path in ('/api/shipment/refresh', '/api/shipment/summary'): from urllib.parse import parse_qs _sqs = {k: v[0] for k, v in parse_qs(self.path.split('?', 1)[1]).items()} if '?' in self.path else {} try: import shipment_api if path == '/api/shipment/summary': _res = shipment_api.summary() else: _res = shipment_api.refresh(force=(_sqs.get('force') == '1')) except Exception as _e: import traceback traceback.print_exc() _res = {'ok': False, 'error': '物流刷新异常: %s' % _e} self._send_json(_res if _res is not None else {'ok': False, 'error': '未知物流接口'}) return # ══════════ 拜访与活动统计（当月线下拜访次数 / 线上跟进次数 + 近 12 月趋势）══════════ # scope=me|team；team 仅 admin/gm/manager 可用，其余强制 me（服务端兜底，不信前端） if path == '/api/visit/stats': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return from urllib.parse import parse_qs _qs = {k: v[0] for k, v in parse_qs(self.path.split('?', 1)[1]).items()} if '?' in self.path else {} month = ((_qs.get('month') or '').strip() or datetime.now().strftime('%Y-%m'))[:7] if not re.match(r'^\d{4}-\d{2}$', month): self._send_json({'ok': False, 'error': 'month 需为 YYYY-MM'}, 400) return _role = _au.get('role') _allow_team = _role in ('admin', 'gm', 'manager', 'readonly') scope = (_qs.get('scope') or '').strip().lower() if scope not in ('me', 'team'): scope = 'team' if _allow_team else 'me' if scope == 'team' and not _allow_team: scope = 'me' vis = _act_visible_sales(_au) _wts = _act_load_index() _fsh = _feishu_act_rows() rows = [x for x in (_wts + _fsh) if vis is None or (x.get('submitted_by') in vis) or (x.get('sales') in vis)] if scope == 'me': _me = _au.get('name') or '' rows = [x for x in rows if x.get('submitted_by') == _me or x.get('sales') == _me] monthly = [x for x in rows if (x.get('date') or '').startswith(month)] def _atype(x): """活动形式 = 飞书「活动类型」原值（field_76654e）；工作台自记的走 channel 映射。 不用文本推断：飞书该字段 100% 有值，推断出来的「未标」是假缺失。 """ t = (x.get('atype') or '').strip() if t.startswith('陌拜'): return '陌拜' if t in ACT_TYPES: return t return CH_OF_TYPE.get((x.get('channel') or '').strip(), '未填') def _bump(d, lab): d[lab] = d.get(lab, 0) + 1 gtypes = {} offline = online = noclass = 0 cust = {} for x in monthly: _L = _atype(x) if _L == '线下拜访': offline += 1 elif _L == '线上跟进': online += 1 elif _L == '未填': noclass += 1 _bump(gtypes, _L) c = (x.get('customer') or '').strip() or '(未填客户)' d = cust.setdefault(c, {'customer': c, 'types': {}, 'acts': []}) _bump(d['types'], _L) d['acts'].append({'id': x.get('id', ''), 'date': x.get('date', ''), 'atype': _L, 'type': x.get('type', ''), 'sales': x.get('sales', ''), 'src': x.get('src', '')}) for d in cust.values(): d['total'] = sum(d['types'].values()) d['acts'].sort(key=lambda a: a.get('date') or '', reverse=True) # ── 部门 → 销售 → 客户 三级下钻管道 ── _dmap = _sales_dept_map() _g = {} for x in monthly: _L = _atype(x) _own = (x.get('submitted_by') or '').strip() or (x.get('sales') or '').strip() or '(未知)' _dept, _reg = _dmap.get(_own, ('未匹配部门', '')) gd = _g.setdefault(_dept, {'dept': _dept, 'types': {}, 'sales': {}}) _bump(gd['types'], _L) gs = gd['sales'].setdefault(_own, {'name': _own, 'dept': _dept, 'region': _reg, 'types': {}, 'customers': {}}) _bump(gs['types'], _L) _c = (x.get('customer') or '').strip() or '(未填客户)' gc = gs['customers'].setdefault(_c, {'customer': _c, 'types': {}, 'acts': []}) _bump(gc['types'], _L) gc['acts'].append({'id': x.get('id', ''), 'date': x.get('date', ''), 'atype': _L, 'type': x.get('type', ''), 'sales': x.get('sales', ''), 'src': x.get('src', '')}) groups = [] for gd in _g.values(): gd['total'] = sum(gd['types'].values()) _sl = [] for gs in gd['sales'].values(): gs['total'] = sum(gs['types'].values()) _cl = list(gs['customers'].values()) for gc in _cl: gc['total'] = sum(gc['types'].values()) gc['acts'].sort(key=lambda a: a.get('date') or '', reverse=True) _cl.sort(key=lambda d: -d['total']) gs['customers'] = _cl gs['customer_count'] = len(_cl) _sl.append(gs) _sl.sort(key=lambda d: -d['total']) gd['sales'] = _sl gd['sales_count'] = len(_sl) groups.append(gd) groups.sort(key=lambda d: -d['total']) # 近 12 个月趋势（同一权限与 scope 口径，同样按活动类型分桶） allm = sorted({(x.get('date') or '')[:7] for x in rows if x.get('date')}) trend = [] for m in allm[-12:]: mm = [x for x in rows if (x.get('date') or '').startswith(m)] _tt = {} for x in mm: _bump(_tt, _atype(x)) trend.append({'month': m, 'types': _tt, 'total': len(mm)}) self._send_json({'ok': True, 'month': month, 'scope': scope, 'can_team': _allow_team, 'type_order': ACT_TYPES, 'types': gtypes, 'offline_count': offline, 'online_count': online, 'unclassified': noclass, 'total': len(monthly), 'customer_count': len(cust), 'months': allm[-12:], 'trend': trend, 'departments': groups, 'dept_count': len(groups), 'src_breakdown': { 'workbench': len([x for x in monthly if x.get('src') != 'feishu']), 'feishu': len([x for x in monthly if x.get('src') == 'feishu'])}, 'customers': sorted(cust.values(), key=lambda d: -d['total'])}) return # ══════════ 销售管理后台（仅刘新元）══════════ if path.startswith('/api/admin/'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: import admin_api from urllib.parse import parse_qs _q = {k: v[0] for k, v in parse_qs(self.path.split('?', 1)[1]).items()} if '?' in self.path else {} _res = admin_api.handle_get(path, _q, _au) except Exception as _e: _res = {'ok': False, 'error': '管理后台接口异常: %s' % _e} self._send_json(_res, 200 if _res.get('ok') else (403 if '仅刘新元' in (_res.get('error') or '') else 400)) return # ══════════ 客户拉通大表真实数据（仅刘新元）══════════ if path == '/api/customer-rollup/data': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return if (_au.get('name') or '') != '刘新元': self._send_json({'ok': False, 'error': '该数据仅刘新元可见'}, 403) return _fp = os.path.join(DATA_DIR, 'customer_rollup_v2.json') try: # 默认只回传首屏 500 条活动，避免 9MB 大包拖慢浏览器；项目全量。 from urllib.parse import parse_qs as _pq _q = {k: v[0] for k, v in _pq(self.path.split('?', 1)[1]).items()} if '?' in self.path else {} _limit = max(1, min(int(_q.get('limit', '500')), 5000)) _offset = max(0, int(_q.get('offset', '0'))) with open(_fp, encoding='utf-8') as _f: _d = json.load(_f) _acts = _d.get('activities') or [] _total = len(_acts) _d['activities'] = _acts[_offset:_offset + _limit] _d = _rollup_apply_links(_d) _d = _rollup_apply_status(_d) _m = _d.setdefault('meta', {}) _m['served_at'] = time.strftime('%Y-%m-%d %H:%M:%S') _m['activity_total'] = _total _m['offset'] = _offset _m['limit'] = _limit self._send_json({'ok': True, 'data': _d}) except Exception as _e: self._send_json({'ok': False, 'error': '真实数据读取失败: %s' % _e}, 500) return # ══════════ 飞书活动→商机 手动关联（销售本人 / 刘新元）══════════ if path == '/api/customer-rollup/unlinked': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: from urllib.parse import parse_qs as _pq _q = {k: v[0] for k, v in _pq(self.path.split('?', 1)[1]).items()} if '?' in self.path else {} _limit = max(1, min(int(_q.get('limit', '100')), 500)) _offset = max(0, int(_q.get('offset', '0'))) _d = _rollup_apply_links(_rollup_load()) _cut = _rollup_cutoff() _name = _au.get('name') or '' _items, _projects = [], _d.get('projects') or [] _bycust = {} for _p in _projects: _bycust.setdefault(_p.get('cust'), []).append(_p.get('opp')) _cust_q = (_q.get('customer') or '').strip() for _a in _d.get('activities') or []: if (_a.get('date') or '') < _cut: continue if _a.get('opp_names'): continue if _cust_q and _cust_q not in (_a.get('customer') or ''): continue if _name != '刘新元' and _a.get('sales') != _name and _a.get('submitted_by') != _name: continue _items.append({ 'id': _a.get('id'), 'date': _a.get('date'), 'customer': _a.get('customer'), 'sales': _a.get('sales') or _a.get('submitted_by') or '', 'activity_type': _a.get('activity_type'), 'content_type': _a.get('content_type'), 'project_scene': _a.get('project_scene'), 'desc': (_a.get('desc') or '')[:280], 'opps': _bycust.get(_a.get('customer')) or [] }) _total = len(_items) self._send_json({'ok': True, 'cutoff': _cut, 'total': _total, 'items': _items[_offset:_offset + _limit], 'offset': _offset, 'limit': _limit}) except Exception as _e: self._send_json({'ok': False, 'error': '待关联活动读取失败: %s' % _e}, 500) return # ══════════ 活动录入：订单 / SKU / 列表 / 详情 / 附件地址 ══════════ if path.startswith('/api/activity/') and path != '/api/activity/options': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return from urllib.parse import parse_qs _qs = {k: v[0] for k, v in parse_qs(self.path.split('?', 1)[1]).items()} if '?' in self.path else {} if path == '/api/activity/orders': # 订单下拉：按登录人权限过滤（销售只看到自己名下订单） q = (_qs.get('q') or '').strip().upper() lim = max(1, min(500, int(_qs.get('limit') or 200))) vis = _act_visible_sales(_au) out = [] for o in _act_orders(): if vis is not None and (o.get('sales') not in vis): continue if q and q not in ((o.get('no') or '') + ' ' + (o.get('customer') or '')).upper(): continue out.append({'no': o['no'], 'customer': o.get('customer', ''), 'sales': o.get('sales', ''), 'date': o.get('date', ''), 'is_test': o.get('is_test', False), 'sku_count': len(o.get('skus') or [])}) if len(out) >= lim: break self._send_json({'ok': True, 'total': len(out), 'orders': out}) return if path == '/api/activity/order_skus': no = (_qs.get('no') or '').strip() vis = _act_visible_sales(_au) for o in _act_orders(): if o.get('no') == no: if vis is not None and o.get('sales') not in vis: self._send_json({'ok': False, 'error': '无权查看该订单'}, 403) return self._send_json({'ok': True, 'order': { 'no': o.get('no'), 'customer': o.get('customer', ''), 'sales': o.get('sales', ''), 'date': o.get('date', ''), 'is_test': o.get('is_test', False), 'skus': o.get('skus') or []}}) return self._send_json({'ok': False, 'error': '未找到该订单'}, 404) return if path == '/api/activity/list': vis = _act_visible_sales(_au) rows = [x for x in _act_load_index() if vis is None or (x.get('submitted_by') in vis) or (x.get('sales') in vis)] self._send_json({'ok': True, 'total': len(rows), 'activities': rows[:500]}) return if path == '/api/activity/detail': rid = os.path.basename((_qs.get('id') or '').strip()) if rid.startswith('FS'): # 飞书历史活动（只读基础数据，不在本地文件里） _fr = _feishu_act_one(rid) if not _fr: self._send_json({'ok': False, 'error': '未找到该飞书活动'}, 404) return _ow = (_fr.get('submitted_by') or '').strip() _vis = _act_visible_sales(_au) if _vis is not None and _ow not in _vis: self._send_json({'ok': False, 'error': '无权查看该活动'}, 403) return _at = (_fr.get('atype') or '').strip() or CH_OF_TYPE.get((_fr.get('channel') or '').strip(), '未填') self._send_json({'ok': True, 'activity': { 'id': rid, 'src': 'feishu', 'date': _fr.get('date', ''), 'customer': _fr.get('customer', ''), 'sales': _ow, 'submitted_by': _ow, 'channel': _fr.get('channel', ''), 'atype': _at, 'channel_ev': [], 'dept_raw': _fr.get('dept_raw', ''), 'type': _fr.get('type', ''), 'order_ref_nos': [], 'feishu_text': ('【来源】飞书历史活动（基础数据 · 只读）\n【日期】%s\n【客户】%s\n' '【归属】%s\n【活动名称】%s\n【活动类型】%s\n【飞书跟进部门】%s\n\n【活动描述】\n%s' % (_fr.get('date', ''), _fr.get('customer') or '—', _ow, _fr.get('type') or '—', _at, _fr.get('dept_raw') or '—', _fr.get('text') or ''))}}) return try: with open(os.path.join(ACT_DIR, rid + '.json'), encoding='utf-8') as f: rec = json.load(f) except Exception: self._send_json({'ok': False, 'error': '未找到该活动'}, 404) return if not _act_can_see(_au, rec): self._send_json({'ok': False, 'error': '无权查看该活动'}, 403) return self._send_json({'ok': True, 'activity': rec}) return if path == '/api/activity/attachment_url': cp = (_qs.get('path') or '').strip() if not cp.startswith('activity-attachments/'): self._send_json({'ok': False, 'error': '非法附件路径'}, 400) return aid = cp.split('/')[1] if len(cp.split('/')) > 1 else '' allowed = _au.get('role') in ('admin', 'gm', 'readonly') if not allowed and aid: _rec = _act_attach_owner(aid) allowed = bool(_rec) and _act_can_see(_au, _rec) if not allowed: self._send_json({'ok': False, 'error': '无权访问该附件'}, 403) return u = _tcb_storage_url(cp) self._send_json({'ok': bool(u), 'url': u, 'expires': 3600, 'error': '' if u else '签发临时地址失败'}) return self._send_json({'ok': False, 'error': '未知活动接口'}, 404) return # ── 用户表（服务端；客户端不再直连 CloudBase user_roles）── if path in ('/api/users/list', '/api/users/roster'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return if path == '/api/users/roster': rows = [{'name': u.get('name', ''), 'email': u.get('email', ''), 'dept': u.get('dept', ''), 'role': u.get('role', '')} for u in _load_user_roles() if u.get('status', '在职') != '离职'] self._send_json({'ok': True, 'users': rows}) return if _au.get('role') not in ('admin', 'gm'): self._send_json({'ok': False, 'error': '仅管理员可访问'}, 403) return rows = [] for u in _load_user_roles(): v = {k: val for k, val in u.items() if k != 'pwd'} v['has_pwd'] = bool(u.get('pwd')) rows.append(v) self._send_json({'ok': True, 'users': rows}) return # ── 报销助手（多用户：状态/台账/下载）── if path in ('/api/expense/status', '/api/expense/list', '/api/expense/pack', '/api/expense/page', '/api/expense/mail'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return _E = _expense_api() if path == '/api/expense/status': return _E.handle_status(self, _au) if path == '/api/expense/list': return _E.handle_list(self, _au) if path == '/api/expense/mail': return _E.handle_mail_get(self, _au) return _E.handle_download(self, _au, 'pack' if path.endswith('pack') else 'page') if path == '/api/history': # 读取对话历史（权限：admin/gm 可读任何人，否则仅自己） from urllib.parse import parse_qs qs = {k: v[0] for k, v in parse_qs(self.path.split('?')[1]).items()} if '?' in self.path else {} name = qs.get('name', '') # Token 鉴权：身份从 token 解析（name 仍 query 传，用于 admin 查指定人历史） _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return vrole = _au.get('role', '') viewer = _au.get('name', '') if not name: self._send_json({'ok': False, 'error': 'name required'}, 400) return if vrole not in ('admin', 'gm', 'readonly') and name != viewer: self._send_json({'ok': False, 'error': '无权查看他人历史'}, 403) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat self._send_json({'ok': True, 'name': name, 'history': ai_chat._get_history(name)}) elif path == '/api/token_usage': # AI Token 用量统计（权限：admin/gm 看全部，否则仅自己） from urllib.parse import parse_qs qs = {k: v[0] for k, v in parse_qs(self.path.split('?')[1]).items()} if '?' in self.path else {} # Token 鉴权：身份从 token 解析 _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return vrole = _au.get('role', '') viewer = _au.get('name', '') try: data = {} tf = os.path.join(DATA_DIR, 'ai_token_usage.json') if os.path.exists(tf): with open(tf, encoding='utf-8') as fh: _d = json.load(fh) if isinstance(_d, dict): data = _d result = [] for n, recs in data.items(): if vrole not in ('admin', 'gm') and n != viewer: continue if not isinstance(recs, list) or not recs: continue total = sum(int(r.get('total') or 0) for r in recs) prompt = sum(int(r.get('prompt') or 0) for r in recs) completion = sum(int(r.get('completion') or 0) for r in recs) cnt = sum(int(r.get('est_count') or 1) for r in recs) est_cnt = sum(int(r.get('est_count') or 0) for r in recs if r.get('estimated')) real_dates = sorted(set(r.get('date', '') for r in recs if r.get('date') and r.get('date') != '历史')) result.append({ 'name': n, 'count': cnt, 'est_count': est_cnt, 'total_tokens': total, 'prompt_tokens': prompt, 'completion_tokens': completion, 'first_date': real_dates[0] if real_dates else '', 'last_date': real_dates[-1] if real_dates else '', 'recent': recs[-15:], }) result.sort(key=lambda x: -x['total_tokens']) self._send_json({'ok': True, 'usage': result}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/files/download': # 下载 AI 生成文件（鉴权：admin/gm 全见，否则仅自己） from urllib.parse import parse_qs, quote qs = {k: v[0] for k, v in parse_qs(self.path.split('?')[1]).items()} if '?' in self.path else {} fid = qs.get('fid', '') inline = qs.get('inline', '') == '1' # iframe 预览用 inline，下载用 attachment # Token 鉴权：身份从 token 解析，不信任 query 传参 _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return viewer = _au.get('name', '') vrole = _au.get('role', 'sales') if not fid: self._send_json({'ok': False, 'error': 'fid required'}, 400) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat fp, err = ai_chat._get_file_path(fid, viewer, vrole) if fp is None: self._send_json({'ok': False, 'error': err}, 403 if '无权' in err else 404) return try: ext = os.path.splitext(fp)[1].lower() ctype = { '.html': 'text/html; charset=utf-8', '.htm': 'text/html; charset=utf-8', '.md': 'text/markdown; charset=utf-8', '.txt': 'text/plain; charset=utf-8', '.csv': 'text/csv; charset=utf-8', '.json': 'application/json; charset=utf-8', '.pdf': 'application/pdf', }.get(ext, 'application/octet-stream') with open(fp, 'rb') as fh: body = fh.read() self.send_response(200) self.send_header('Content-Type', ctype) self.send_header('Content-Length', str(len(body))) self.send_header('Content-Disposition', "%s; filename*=UTF-8''%s" % ('inline' if inline else 'attachment', quote(os.path.basename(fp)))) self._cors_headers() self.end_headers() self.wfile.write(body) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/files/list': # 列出 AI 生成文件（权限：admin/gm 全见，否则仅自己） from urllib.parse import parse_qs qs = {k: v[0] for k, v in parse_qs(self.path.split('?')[1]).items()} if '?' in self.path else {} # Token 鉴权：身份从 token 解析，不信任 query 传参 _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return viewer = _au.get('name', '') vrole = _au.get('role', 'sales') _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat try: self._send_json({'ok': True, 'files': ai_chat._list_files(viewer, vrole)}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/opportunity/list': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return from urllib.parse import parse_qs qs = parse_qs(self.path.split('?', 1)[1]) if '?' in self.path else {} st = (qs.get('status') or [''])[0] self._send_json({'ok': True, 'items': activity_v2.list_for(_au, st)}) elif path == '/api/activity/options': # 活动录入下拉选项（客户/商机/联系人/货号，权限过滤） _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat try: self._send_json({'ok': True, 'options': ai_chat._activity_options(_au)}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/feishu/customer_data': # 读取飞书缓存数据 qs = {} if '?' in self.path: from urllib.parse import parse_qs qs = {k: v[0] for k, v in parse_qs(self.path.split('?')[1]).items()} cid = qs.get('customer_id', '') try: with open(FEISHU_DATA_FILE, encoding='utf-8') as f: all_data = json.load(f) customer_data = all_data.get(str(cid)) if customer_data: self._send_json({'ok': True, 'data': customer_data}) else: self._send_json({'ok': False, 'error': 'no data for this customer, please ask Hermes to generate it'}, 404) except FileNotFoundError: self._send_json({'ok': False, 'error': 'no cache file yet'}, 404) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/todos': with LOCK: self._send_json(read_store()) elif path == '/api/visit-requests': # 拜访请求队列状态（pending + processed） qdir = os.path.join(ROOT, '数据', 'visit_requests') processed_dir = os.path.join(qdir, 'processed') pending, processed = [], [] for fn in sorted(os.listdir(qdir)): if not fn.endswith('.json'): continue fp = os.path.join(qdir, fn) if os.path.isdir(fp): continue try: with open(fp, encoding='utf-8') as f: req = json.load(f) req['_file'] = fn pending.append(req) except Exception: pass if os.path.isdir(processed_dir): for fn in sorted(os.listdir(processed_dir)): if not fn.endswith('.json'): continue try: with open(os.path.join(processed_dir, fn), encoding='utf-8') as f: req = json.load(f) req['_file'] = fn processed.append(req) except Exception: pass self._send_json({'ok': True, 'pending': pending, 'processed': processed}) elif path == '/api/uploads': # 返回上传日志 log_file = os.path.join(DATA_DIR, '上传', 'upload_log.json') if os.path.exists(log_file): try: with open(log_file, 'r', encoding='utf-8') as lf: log = json.load(lf) self._send_json({'ok': True, 'uploads': log}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) else: self._send_json({'ok': True, 'uploads': []}) elif path == '/api/special-requirements': # ── 客户特殊要求（权限随客户归属：sales 仅见自己客户）── _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return from urllib.parse import parse_qs as _pqs _qs = {k: v[0] for k, v in _pqs(self.path.split('?')[1]).items()} if '?' in self.path else {} _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat try: _res = ai_chat.special_requirements_for(_au, (_qs.get('customer') or '').strip() or None) _res['ok'] = True self._send_json(_res) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/price/query': # ── 价格查询：ERP 真实成交价（权限服务端强制：all/own/none）── _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import price_api price_api.handle_get(self, _au) return elif path == '/api/permissions': # ── 统一权限管理：读取按人权限（admin/gm 全量；readonly 只读代理可查看）── _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return if _au.get('role') not in ('admin', 'gm', 'readonly'): self._send_json({'ok': False, 'error': '仅管理员可访问'}, 403) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat try: users = [u for u in _load_user_roles() if u.get('status', '在职') != '离职'] perms = ai_chat._load_user_permissions() out_users = [] for u in users: nm = u.get('name', '') p = perms.get(nm, {}) out_users.append({ 'name': nm, 'role': u.get('role', 'sales'), 'dept': u.get('dept', ''), 'line': u.get('line', 'da'), 'email': u.get('email', ''), 'ds_limit': p.get('ds_limit', ai_chat.DEEPSEEK_DAILY_LIMIT), 'tabs': dict(u.get('tabs') or {}), 'extra_customers': p.get('extra_customers') or [], 'blocked_customers': p.get('blocked_customers') or [], 'price_scope': p.get('price_scope') or ('all' if u.get('role') in ('admin', 'gm', 'readonly') else 'own'), 'price_scope_default': 'all' if u.get('role') in ('admin', 'gm', 'readonly') else 'own', 'price_alias': p.get('price_alias') or [], }) custs, groups = ai_chat._load_customer_data() cust_names = sorted(set((c.get('name') or '') for c in custs if (c.get('name') or ''))) group_names = sorted(set((g.get('name') or '') for g in groups if (g.get('name') or ''))) self._send_json({'ok': True, 'users': out_users, 'customers': cust_names, 'groups': group_names, 'default_ds_limit': ai_chat.DEEPSEEK_DAILY_LIMIT}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/test-report/schema': # 测试报告收集 · 表单字段定义（由 xlsx 模板抽取，20 个产品方向） try: with open(TEST_REPORT_SCHEMA, encoding='utf-8') as f: self._send_json({'ok': True, 'schema': json.load(f)}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/test-report/options': # 订单号 / 客户 / 货号 选择项（服务端联表，销售只看自己名下订单） _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: self._send_json({'ok': True, 'options': build_test_report_options(_au)}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) elif path == '/api/test-reports': # 已提交测试报告列表；?id=TRxxx 返回单份完整记录（含原始值/图片/MD） _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return from urllib.parse import parse_qs _qs = {k: v[0] for k, v in parse_qs(self.path.split('?')[1]).items()} if '?' in self.path else {} _rid = (_qs.get('id') or '').strip() _role = (_au or {}).get('role') or '' _name = (_au or {}).get('name') or '' if _rid: try: with open(os.path.join(TEST_REPORT_DIR, os.path.basename(_rid) + '.json'), encoding='utf-8') as f: _rec = json.load(f) except Exception: self._send_json({'ok': False, 'error': '记录不存在'}, 404) return if _test_report_visible_people(_role, _name, _au) is not None and _name \ and _rec.get('submitted_by') not in _test_report_visible_people(_role, _name, _au): self._send_json({'ok': False, 'error': '无权查看他人报告'}, 403) return self._send_json({'ok': True, 'item': _rec}) return _rows = list_test_reports() _vis_p = _test_report_visible_people(_role, _name, _au) if _vis_p is not None and _name: _rows = [r for r in _rows if r.get('submitted_by') in _vis_p] self._send_json({'ok': True, 'items': _rows}) elif path.startswith('/api/mcp/'): # ── MCP 实时数据 API ── from urllib.parse import parse_qs qs = {k: v[0] for k, v in parse_qs(self.path.split('?')[1]).items()} if '?' in self.path else {} force = qs.get('refresh') == '1' MQL_MAP = { 'customers': 'SELECT `name`,`work_item_id`,`field_17186c`,`field_6415cf`,`field_5ed7ab`,`field_2d5b6a`,`field_e62869`,`field_c3224b`,`field_c8e80d`,`field_a1ff52`,`field_79aa0b`,`owner`,`start_time`,`business`,`current_status_operator` FROM `6593cd71471290e3cc6be6e6`.`65ae1e403c87b152f3365ca6`', 'orders': 'SELECT `name`,`work_item_id`,`field_e1001d`,`field_a6e501`,`field_5ed7ab`,`field_2d6367`,`field_e0cce5`,`field_2e095b`,`field_4a1b47`,`field_29d947`,`field_3a3ccb`,`field_8c4b7b`,`field_5b2b3b`,`field_dbd4ab`,`field_a98458`,`field_8d9153`,`owner`,`start_time`,`business`,`current_status_operator`,`work_item_status` FROM `6593cd71471290e3cc6be6e6`.`662612aea6bb7089fea044ef`', 'complaints': 'SELECT `name`,`work_item_id`,`start_time`,`work_item_status`,`field_9e2144`,`field_6fb810`,`field_3ce9fe`,`field_d107b0`,`field_502a85`,`field_ac5caf`,`owner`,`business`,`current_status_operator`,`finish_time`,`finish_status` FROM `658288abfb8bd616b17025f1`.`6669433056a98249604376de`', 'activities': 'SELECT `name`,`work_item_id`,`field_b54da4`,`field_b99055`,`field_76654e`,`field_a2b3f6`,`field_5f20fc`,`field_5ed7ab`,`field_d6ee93`,`owner`,`start_time`,`business`,`current_status_operator`,`work_item_status` FROM `6593cd71471290e3cc6be6e6`.`65ae1e5d44338dbe7c39a29a`', 'opportunities': 'SELECT `name`,`work_item_id`,`owner`,`start_time`,`business`,`current_status_operator`,`work_item_status`,`description` FROM `6593cd71471290e3cc6be6e6`.`story`', } cache_key = path.replace('/api/mcp/', '') if cache_key == 'dashboard': self._send_json({'ok': False, 'error': 'use POST /api/mcp/refresh or GET with ?refresh=1 to compute'}, 400) return if cache_key in MQL_MAP: try: if force: count = refresh_mcp_cache(cache_key, MQL_MAP[cache_key]) data = get_mcp_cache(cache_key) else: data = get_mcp_cache(cache_key) if data is None: count = refresh_mcp_cache(cache_key, MQL_MAP[cache_key]) data = get_mcp_cache(cache_key) self._send_json({'ok': True, 'type': cache_key, 'total': len(data), 'items': data}) except Exception as e: self._send_json({'ok': False, 'error': str(e), 'type': cache_key}, 500) else: self._send_json({'ok': False, 'error': f'unknown type: {cache_key}'}, 400) elif path == '/api/sandept': # 三部门逐月拆分看板数据：服务端按登录态过滤（方案A · 客户端不再持有全量数据） _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return _payload, _err = _sandept_payload(_au) if _payload is None: self._send_json({'ok': False, 'error': _err or '数据不可用'}, 500) return self._send_json(_payload) # end_headers 已对 /api/ 注入 Cache-Control: no-store return elif path == '/api/targets': # 目标基础数据（唯一权威源）：仅 admin/gm 可读；其他角色 403（服务端硬门槛，非前端隐藏） _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return _tp, _terr = _targets_payload(_au) if _tp is None: if _terr == 'forbidden': self._send_json({'ok': False, 'error': '仅 admin/gm 可访问目标基础数据'}, 403) else: self._send_json({'ok': False, 'error': _terr or '数据不可用'}, 500) return self._send_json(_tp) return else: super().do_GET() def do_POST(self): path = self.path.split('?')[0] # 统一鉴权兜底：所有 /api/ 端点（除登录/换 token/飞书免登）都需有效 token if path.startswith('/api/') and path not in ('/api/login', '/api/auth/token', '/api/feishu/login'): _ru = self._auth_user() if not _ru: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return # 只读角色：拒绝一切写端点（白名单外全 403） if (_ru.get('role') or '') == READONLY_ROLE and path not in READONLY_ALLOW_POST: self._send_json({'ok': False, 'error': '只读账号无写权限（readonly）'}, 403) return # ══════════ 销售管理后台写入（仅刘新元）══════════ if path.startswith('/api/admin/'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: _len = int(self.headers.get('Content-Length') or 0) _body = json.loads((self.rfile.read(_len) if _len else b'{}').decode('utf-8') or '{}') import admin_api _res = admin_api.handle_post(path, _body, _au) except Exception as _e: _res = {'ok': False, 'error': '管理后台写入异常: %s' % _e} self._send_json(_res, 200 if _res.get('ok') else (403 if '仅刘新元' in (_res.get('error') or '') else 400)) return # ══════════ 商机系统状态初始化/更新（刘新元 / 商机销售）══════════ if path == '/api/customer-rollup/status': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: _len = int(self.headers.get('Content-Length') or 0) _body = json.loads((self.rfile.read(_len) if _len else b'{}').decode('utf-8') or '{}') _cust = (_body.get('cust') or '').strip() _opp = (_body.get('opp') or '').strip() _status = (_body.get('status') or '').strip() if _status and _status not in OPP_STATUSES: self._send_json({'ok': False, 'error': '非法状态'}, 400) return _d = _rollup_load() _p = next((x for x in (_d.get('projects') or []) if (x.get('cust') or '') == _cust and (x.get('opp') or '') == _opp), None) if not _p: self._send_json({'ok': False, 'error': '未找到该商机'}, 404) return _name = _au.get('name') or '' if _name != '刘新元' and _p.get('sales') != _name: self._send_json({'ok': False, 'error': '只能初始化本人商机'}, 403) return _key = _cust + '||' + _opp with _ROLLUP_LOCK: _items = _opp_status_load() if _status: _items[_key] = {'status': _status, 'by': _name, 'at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')} else: _items.pop(_key, None) _opp_status_save(_items) self._send_json({'ok': True, 'cust': _cust, 'opp': _opp, 'status': _status, 'by': _name}) except Exception as _e: self._send_json({'ok': False, 'error': '状态保存失败: %s' % _e}, 500) return # ══════════ 飞书活动→商机 手动关联保存（销售本人 / 刘新元）══════════ if path == '/api/customer-rollup/link': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: _len = int(self.headers.get('Content-Length') or 0) _body = json.loads((self.rfile.read(_len) if _len else b'{}').decode('utf-8') or '{}') _aid = (_body.get('id') or '').strip() _opp = (_body.get('opp') or '').strip() _d = _rollup_load() _a = next((x for x in (_d.get('activities') or []) if x.get('id') == _aid), None) if not _a: self._send_json({'ok': False, 'error': '未找到该活动'}, 404) return _cut = _rollup_cutoff() if (_a.get('date') or '') < _cut: self._send_json({'ok': False, 'error': '只处理近 6 个月活动'}, 400) return _name = _au.get('name') or '' if _name != '刘新元' and _a.get('sales') != _name and _a.get('submitted_by') != _name: self._send_json({'ok': False, 'error': '只能关联本人活动'}, 403) return if _opp: _valid = {p.get('opp') for p in (_d.get('projects') or []) if p.get('cust') == _a.get('customer')} if _opp not in _valid: self._send_json({'ok': False, 'error': '该商机不属于该客户'}, 400) return with _ROLLUP_LOCK: _links = _rollup_links_load() if _opp: _links[_aid] = {'opp': _opp, 'by': _name, 'at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')} else: _links.pop(_aid, None) _rollup_links_save(_links) self._send_json({'ok': True, 'id': _aid, 'opp': _opp, 'by': _name}) except Exception as _e: self._send_json({'ok': False, 'error': '关联保存失败: %s' % _e}, 500) return # ══════════ 销售下单：提交 / 审批 / 授权 ══════════ if path.startswith('/api/order/'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: _len = int(self.headers.get('Content-Length') or 0) _raw = self.rfile.read(_len) if _len else b'{}' try: _body = json.loads(_raw.decode('utf-8') or '{}') except Exception: _body = {} import order_api order_api.setup(visible_sales=_act_visible_sales, notify=feishu_notify) _res = order_api.handle_post(path, _body, _au) except Exception as _e: import traceback traceback.print_exc() _res = {'ok': False, 'error': '下单接口异常: %s' % _e} self._send_json(_res if _res is not None else {'ok': False, 'error': '未知下单接口'}) return # ══════════ 活动录入：提交 / 附件上传 ══════════ if path == '/api/activity/submit_v2': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length).decode('utf-8') or '{}') except Exception as e: self._send_json({'ok': False, 'error': '参数解析失败: %s' % e}, 400) return rec, err = activity_v2.build_activity(payload, _au) if err: self._send_json({'ok': False, 'error': err}, 400) return # 客户可见性：不能把活动/商机写到无权客户名下 try: _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat _opts = ai_chat._activity_options(_au) if rec.get('customer') not in (_opts.get('customers') or []): self._send_json({'ok': False, 'error': '该客户不在你的可见范围'}, 403) return except Exception: pass rec['feishu_text'] = _act_feishu_text(rec) _act_save(rec) try: cb_ok, cb_err = _act_cb_insert(rec) except Exception as _e: cb_ok, cb_err = False, str(_e) appr = activity_v2.create_from_payload(payload, _au, rec['id']) self._send_json({'ok': True, 'id': rec['id'], 'feishu_text': rec['feishu_text'], 'approval': appr, 'cb': cb_ok, 'cb_err': ('' if cb_ok else cb_err)}) return if path == '/api/opportunity/review': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length).decode('utf-8') or '{}') except Exception as e: self._send_json({'ok': False, 'error': '参数解析失败: %s' % e}, 400) return ok, err, item = activity_v2.review(_au, (payload.get('id') or '').strip(), (payload.get('action') or '').strip(), (payload.get('reason') or '').strip()) if not ok: self._send_json({'ok': False, 'error': err}, 403 if '无权' in err or '仅' in err else 400) return self._send_json({'ok': True, 'item': item}) return if path in ('/api/activity/submit', '/api/activity/attachment'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return if path == '/api/activity/attachment': ct = self.headers.get('Content-Type', '') or '' if 'multipart/form-data' not in ct or 'boundary=' not in ct: self._send_json({'ok': False, 'error': '需要 multipart/form-data'}, 400) return boundary = ct.split('boundary=')[1].strip() if boundary.startswith('"') and boundary.endswith('"'): boundary = boundary[1:-1] length = int(self.headers.get('Content-Length') or 0) files, fields = _mp_parse(self.rfile.read(length), boundary) if not files: self._send_json({'ok': False, 'error': '未收到文件'}, 400) return aid = re.sub(r'[^\w\-]', '', (fields.get('activityId') or '').strip()) or ('TMP' + datetime.now().strftime('%Y%m%d%H%M%S')) sku = (fields.get('sku') or '').strip() d = os.path.join(ACT_ATTACH_DIR, aid) os.makedirs(d, exist_ok=True) out = [] for fname, body in files: if len(body) > 30 * 1024 * 1024: out.append({'name': fname, 'error': '超过 30MB 上限'}) continue safe = re.sub(r'[^\w\u4e00-\u9fa5.\-]+', '_', fname)[-90:] or 'file' local = os.path.join(d, datetime.now().strftime('%H%M%S_') + safe) with open(local, 'wb') as f: f.write(body) cloud = 'activity-attachments/%s/%s' % (aid, os.path.basename(local)) ok, err = _tcb_storage_upload(local, cloud) out.append({'name': fname, 'saved': os.path.basename(local), 'size': len(body), 'sku': sku, 'cloud_ok': ok, 'cloud_path': (cloud if ok else ''), 'err': ('' if ok else err), 'local': os.path.relpath(local, ROOT)}) self._send_json({'ok': True, 'activityId': aid, 'files': out}) return # ── 提交活动 ── length = int(self.headers.get('Content-Length') or 0) try: payload = json.loads(self.rfile.read(length).decode('utf-8') or '{}') except Exception as e: self._send_json({'ok': False, 'error': '参数解析失败: %s' % e}, 400) return typ = (payload.get('type') or '').strip() if typ not in ('项目跟进', '采购跟进', '客户档案完善'): self._send_json({'ok': False, 'error': '活动类型必须是「项目跟进」/「采购跟进」/「客户档案完善」'}, 400) return # ── P1：活动形式必填（下拉，只认线上/线下）── ch = (payload.get('channel') or '').strip() if ch not in ('offline', 'online'): self._send_json({'ok': False, 'error': '请选择活动形式：线下拜访 / 线上跟进（P1 起必填）'}, 400) return # ── P1：关联订单（发货单 / 测试单，类型服务端按 is_test 判定）── order_refs = _act_classify_orders(payload.get('order_refs')) _bad_refs = [o['no'] for o in order_refs if not o.get('in_index')] order_refs = [o for o in order_refs if o.get('in_index')] # ── 按商机分块采集（兼容旧前端：无 blocks 时把平铺字段包成单块）── raw_blocks = payload.get('blocks') or [] if not raw_blocks: raw_blocks = [{'need': payload.get('need'), 'params': payload.get('params'), 'test_order': payload.get('test_order'), 'issue': payload.get('issue'), 'feedback': payload.get('feedback'), 'competitors': payload.get('competitors'), 'prices': payload.get('prices'), 'price_on': payload.get('price_on'), 'decision': payload.get('decision'), 'biz_desc': payload.get('biz_desc'), 'revenues': payload.get('revenues'), 'suppliers': payload.get('suppliers'), 'self_share': payload.get('self_share'), 'main_rival': payload.get('main_rival'), 'visit_goal': payload.get('visit_goal'), 'visit_content': payload.get('visit_content'), 'plan': payload.get('plan')}] def _clean_block(b): bb = { 'opp': (b.get('opp') or '').strip(), 'opp_status': (b.get('opp_status') or '').strip(), 'opp_url': (b.get('opp_url') or '').strip(), 'need': (b.get('need') or '').strip(), 'params': [x for x in (b.get('params') or []) if (x.get('sku') or '').strip() or (x.get('desc') or '').strip()], 'issue': (b.get('issue') or '').strip(), 'feedback': (b.get('feedback') or '').strip(), 'competitors': [c for c in (b.get('competitors') or []) if (c.get('brand') or '').strip()], 'prices': [{'brand': (x.get('brand') or '').strip(), 'sku': (x.get('sku') or '').strip(), 'spec': (x.get('spec') or '').strip(), 'qty': (x.get('qty') or '').strip(), 'price': (x.get('price') or '').strip(), 'source': (x.get('source') or '').strip(), 'source_other': (x.get('source_other') or '').strip()} for x in (b.get('prices') or []) if (x.get('brand') or '').strip() or (x.get('sku') or '').strip() or (x.get('price') or '').strip()], 'price_on': bool(b.get('price_on')), 'decision': [{'name': (x.get('name') or '').strip(), 'title': (x.get('title') or '').strip(), 'role': (x.get('role') or '').strip(), 'power': (x.get('power') or '').strip(), 'phone': (x.get('phone') or '').strip(), 'note': (x.get('note') or '').strip(), 'influence_tier': (x.get('influence_tier') or '').strip(), 'reports_to': (x.get('reports_to') or '').strip(), 'attitude': (x.get('attitude') or '').strip(), 'comm_pref': (x.get('comm_pref') or '').strip(), 'focus_point': (x.get('focus_point') or '').strip()} for x in (b.get('decision') or []) if (x.get('name') or '').strip() or (x.get('title') or '').strip()], 'biz_desc': (b.get('biz_desc') or '').strip(), 'revenues': [{'year': (x.get('year') or '').strip(), 'amt': (x.get('amt') or '').strip(), 'source': (x.get('source') or '').strip()} for x in (b.get('revenues') or []) if (x.get('year') or '').strip() or (x.get('amt') or '').strip()], 'suppliers': [{'name': (x.get('name') or '').strip(), 'project': (x.get('project') or '').strip(), 'amt': (x.get('amt') or '').strip(), 'pct': (x.get('pct') or '').strip()} for x in (b.get('suppliers') or []) if (x.get('name') or '').strip() or (x.get('project') or '').strip() or (x.get('amt') or '').strip()], 'self_share': (b.get('self_share') or '').strip(), 'main_rival': (b.get('main_rival') or '').strip(), 'visit_goal': (b.get('visit_goal') or '').strip(), 'visit_content': (b.get('visit_content') or '').strip(), # ── 客户画像 P0（v0.2 新增，刘新元 2026-09-29 定版）── 'founded_year': (b.get('founded_year') or '').strip(), 'registered_capital_wan': (b.get('registered_capital_wan') or '').strip(), 'financing_status': (b.get('financing_status') or '').strip(), 'headcount_scale': (b.get('headcount_scale') or '').strip(), 'corp_nature': (b.get('corp_nature') or '').strip(), 'biz_type': (b.get('biz_type') or '').strip(), 'license_status': (b.get('license_status') or '').strip(), 'license_count': (b.get('license_count') or '').strip(), 'registering_products': (b.get('registering_products') or '').strip(), 'risk_notes': (b.get('risk_notes') or '').strip(), 'risk_note_source': (b.get('risk_note_source') or '').strip(), 'price_priority': (b.get('price_priority') or '').strip(), 'require_backup_vendor': (b.get('require_backup_vendor') or '').strip(), 'vendor_decider': (b.get('vendor_decider') or '').strip(), 'negotiation_rounds': (b.get('negotiation_rounds') or '').strip(), 'quote_buffer': (b.get('quote_buffer') or '').strip(), 'phase_split_reg_opt': (b.get('phase_split_reg_opt') or '').strip(), 'pain_labor_shortage': (b.get('pain_labor_shortage') or '').strip(), 'pain_import_cost': (b.get('pain_import_cost') or '').strip(), 'pain_notes': (b.get('pain_notes') or '').strip(), 'segment': (b.get('segment') or '').strip(), 'product_line': (b.get('product_line') or '').strip(), 'annual_volume_t': (b.get('annual_volume_t') or '').strip(), 'volume_unit': (b.get('volume_unit') or '').strip(), 'reaction_system': (b.get('reaction_system') or '').strip(), 'incumbent_brand': (b.get('incumbent_brand') or '').strip(), 'test_ref': (b.get('test_ref') or '').strip(), 'opportunity': (b.get('opportunity') or '').strip(), 'challenge': (b.get('challenge') or '').strip(), 'support_needed': (b.get('support_needed') or '').strip(), 'breakthrough': (b.get('breakthrough') or '').strip(), 'strategy_focus': [x for x in (b.get('strategy_focus') or []) if (x or '').strip()], 'plan': [x for x in (b.get('plan') or []) if (x.get('item') or '').strip()], } t = b.get('test_order') or {} items = [x for x in (t.get('items') or []) if (x.get('sku') or '').strip()] bb['test_order'] = ({'no': (t.get('no') or '').strip(), 'customer': (t.get('customer') or '').strip(), 'sales': (t.get('sales') or '').strip(), 'items': items} if (t.get('no') or '').strip() else {}) bb['order_no'] = bb['test_order'].get('no', '') return bb def _block_empty(bb): # v0.2：必须覆盖 P0 新字段，否则「只填新卡片」的块会被判空丢弃 return not (bb['need'] or bb['params'] or bb['test_order'] or bb['issue'] or bb['feedback'] or bb['competitors'] or bb['prices'] or bb['decision'] or bb['biz_desc'] or bb['revenues'] or bb['suppliers'] or bb['self_share'] or bb['main_rival'] or bb['visit_goal'] or bb['visit_content'] or bb['plan'] or bb['founded_year'] or bb['registered_capital_wan'] or bb['financing_status'] or bb['headcount_scale'] or bb['corp_nature'] or bb['biz_type'] or bb['license_status'] or bb['license_count'] or bb['registering_products'] or bb['risk_notes'] or bb['risk_note_source'] or bb['price_priority'] or bb['require_backup_vendor'] or bb['vendor_decider'] or bb['negotiation_rounds'] or bb['quote_buffer'] or bb['phase_split_reg_opt'] or bb['pain_labor_shortage'] or bb['pain_import_cost'] or bb['pain_notes'] or bb['segment'] or bb['product_line'] or bb['annual_volume_t'] or bb['volume_unit'] or bb['reaction_system'] or bb['incumbent_brand'] or bb['test_ref'] or bb['opportunity'] or bb['challenge'] or bb['support_needed'] or bb['breakthrough'] or bb['strategy_focus']) blocks = [_clean_block(b) for b in raw_blocks] blocks = [b for b in blocks if not _block_empty(b)] att_all = [a for b in blocks for x in ((b.get('test_order') or {}).get('items') or []) for a in (x.get('attachments') or [])] order_nos = [b['order_no'] for b in blocks if b.get('order_no')] opportunities = [{'name': (o.get('name') or '').strip(), 'status': (o.get('status') or '').strip(), 'url': (o.get('url') or '').strip()} for o in (payload.get('opportunities') or []) if (o.get('name') or '').strip()] b0 = blocks[0] if blocks else {} plan_flat = [dict(p, opp=b['opp']) for b in blocks for p in b['plan']] rec = { 'id': 'ACT' + datetime.now().strftime('%Y%m%d%H%M%S'), 'created': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'submitted_by': _au.get('name') or '', 'sales': (payload.get('sales') or '').strip() or (_au.get('name') or ''), 'type': typ, 'subtype': (payload.get('subtype') or '').strip(), 'customer': (payload.get('customer') or '').strip(), 'contact': (payload.get('contact') or '').strip(), 'date': (payload.get('date') or '').strip() or datetime.now().strftime('%Y-%m-%d'), 'channel': ch, 'order_refs': order_refs, 'order_ref_nos': [o['no'] for o in order_refs], 'opportunities': opportunities, 'opp_names': [o['name'] for o in opportunities], 'opp_count': len(opportunities), 'blocks': blocks, 'block_count': len(blocks), 'order_no': order_nos[0] if order_nos else '', 'sku_count': sum(len(b['params']) for b in blocks) + sum(len((b['test_order'] or {}).get('items') or []) for b in blocks), 'att_count': len(att_all), 'attachments': att_all, # ── 向后兼容：顶层保留首块字段，老页面/报表仍可读 ── 'need': b0.get('need', ''), 'params': b0.get('params', []), 'test_order': b0.get('test_order', {}), 'issue': b0.get('issue', ''), 'feedback': b0.get('feedback', ''), 'competitors': b0.get('competitors', []), 'prices': b0.get('prices', []), 'price_count': sum(len(b.get('prices') or []) for b in blocks), 'decision': b0.get('decision', []), 'biz_desc': b0.get('biz_desc', ''), 'revenues': b0.get('revenues', []), 'suppliers': b0.get('suppliers', []), 'self_share': b0.get('self_share', ''), 'main_rival': b0.get('main_rival', ''), 'decision_count': sum(len(b.get('decision') or []) for b in blocks), 'visit_goal': b0.get('visit_goal', ''), 'visit_content': b0.get('visit_content', ''), 'plan': plan_flat, } rec['feishu_text'] = _act_feishu_text(rec) errs = [] if not rec['customer'] and not order_nos: errs.append('客户 或 订单编号 至少填一项') if not blocks: errs.append('至少填写一套跟进内容（关联商机后每套内容分别填写）') if blocks and not any(b['plan'] for b in blocks): errs.append('下一步计划至少 1 条') if typ == '客户档案完善': for b in blocks: if not (b['decision'] or b['biz_desc'] or b['revenues'] or b['suppliers'] or b['self_share'] or b['main_rival']): errs.append('档案完善内容至少填一项：决策链 / 主营业务 / 近三年营收 / TOP3供应商') elif typ == '项目跟进': for b in blocks: if not (b['need'] or b['params'] or b['issue'] or b['test_order']): errs.append('商机「%s」缺「客户需求 / 技术参数 / 问题现象」' % (b['opp'] or '未关联商机')) else: for b in blocks: if not (b['visit_goal'] or b['visit_content']): errs.append('商机「%s」缺「拜访目标 / 拜访内容」' % (b['opp'] or '未关联商机')) for b in blocks: if b.get('price_on') and not b.get('prices'): errs.append('商机「%s」已勾选「价格信息收集」，请至少填一条（品牌/货号/单价）' % (b['opp'] or '未关联商机')) if errs: self._send_json({'ok': False, 'error': '；'.join(errs)}, 400) return # 附件目录 → 活动归属绑定（提交时目录 id 还是临时的，必须绑） for _a in att_all: _cp = (_a.get('cloud_path') or '') or '' _lp = (_a.get('local') or '') or '' _d = (_cp.split('/')[1] if _cp.startswith('activity-attachments/') and len(_cp.split('/')) > 1 else (_lp.split('/')[1] if _lp.startswith('数据/活动附件/') and len(_lp.split('/')) > 1 else '')) _act_bind_attach_dir(_d, rec) _act_save(rec) cb_ok, cb_err = _act_cb_insert(rec) # ── P1：活动确认清单 → 待办（生成≠落库，只落已确认；幂等；不碰手工条）── try: todo_res = _act_sync_todos(rec, payload.get('todos') or [], (payload.get('draft_id') or '').strip()) except Exception as _te: import traceback traceback.print_exc() todo_res = {'error': str(_te)} self._send_json({'ok': True, 'id': rec['id'], 'feishu_text': rec['feishu_text'], 'channel': rec['channel'], 'order_refs': rec['order_refs'], 'unmatched_orders': _bad_refs, 'cb': cb_ok, 'cb_err': ('' if cb_ok else cb_err), 'todos': todo_res}) return # ── 用户表维护（仅 admin/gm；写本地快照，登录以此为准）── if path in ('/api/users/save', '/api/users/reset_pwd'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return if _au.get('role') not in ('admin', 'gm'): self._send_json({'ok': False, 'error': '仅管理员可维护用户'}, 403) return try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': '请求格式错误'}, 400) return roles = _load_user_roles() if path == '/api/users/reset_pwd': email = (payload.get('email') or '').strip().lower() hit = False for u in roles: if (u.get('email') or '').lower() == email: u['pwd'] = payload.get('pwd') or 'biori2026' hit = True break if not hit: self._send_json({'ok': False, 'error': '未找到该用户'}, 404) return else: email = (payload.get('email') or '').strip().lower() target = None for u in roles: if payload.get('_id') and u.get('_id') == payload['_id']: target = u break if email and (u.get('email') or '').lower() == email: target = u break if target is None: if not email: self._send_json({'ok': False, 'error': '缺少邮箱'}, 400) return target = {'_id': 'local' + str(int(time.time() * 1000))} roles.append(target) for k in ('name', 'email', 'dept', 'role', 'line', 'open_id', 'status'): if k in payload: target[k] = payload[k] if isinstance(payload.get('tabs'), dict): target['tabs'] = {k: (1 if v else 0) for k, v in payload['tabs'].items()} if payload.get('pwd'): target['pwd'] = payload['pwd'] elif not target.get('pwd'): target['pwd'] = 'biori2026' tmp = USER_ROLES_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(roles, f, ensure_ascii=False, indent=1) os.replace(tmp, USER_ROLES_FILE) self._send_json({'ok': True, 'users_count': len(roles)}) return # ── 报销助手（多用户：上传/重跑）── if path in ('/api/expense/upload', '/api/expense/requeue', '/api/expense/mail', '/api/expense/mail/fetch'): _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return if path == '/api/expense/upload': return _expense_api().handle_upload(self, _au) if path == '/api/expense/mail': return _expense_api().handle_mail_save(self, _au) if path == '/api/expense/mail/fetch': return _expense_api().handle_mail_fetch(self, _au) uid = _expense_api().EU.uid_of(_au.get('email', ''), _au.get('name', '')) _expense_api().enqueue(uid, []) return self._send_json({'ok': True, 'msg': '已重新排队'}) # ── 客户特殊要求 保存/删除 API ── if path == '/api/special-requirements': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat try: _act = (payload.get('action') or 'save').strip() if _act == 'delete': _ok, _msg = ai_chat.delete_special_requirement(_au, (payload.get('id') or '').strip()) self._send_json({'ok': _ok, 'msg': _msg}, 200 if _ok else 403) else: _ok, _msg, _item = ai_chat.save_special_requirement(_au, payload) self._send_json({'ok': _ok, 'msg': _msg, 'item': _item}, 200 if _ok else 403) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── 对话历史保存 API ── if path == '/api/history': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return name = (payload.get('name') or '').strip() history = payload.get('history') or [] vrole = payload.get('vrole') or '' viewer = payload.get('viewer') or '' if not name: self._send_json({'ok': False, 'error': 'name required'}, 400) return if vrole not in ('admin', 'gm', 'readonly') and name != viewer: self._send_json({'ok': False, 'error': '无权操作他人历史'}, 403) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat ai_chat._set_history(name, history) self._send_json({'ok': True, 'name': name}) return # ── AI 生成文档保存 API ── if path == '/api/save_doc': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return filename = (payload.get('filename') or '').strip() content = payload.get('content') or '' ftype = (payload.get('type') or 'md') if not filename or not content: self._send_json({'ok': False, 'error': 'filename and content required'}, 400) return filename = os.path.basename(filename) if not filename.endswith(('.html', '.md', '.txt')): filename = filename + ('.html' if ftype == 'html' else '.md') target_dir = os.path.join(ROOT, '客户管理') os.makedirs(target_dir, exist_ok=True) filepath = os.path.join(target_dir, filename) try: with open(filepath, 'w', encoding='utf-8') as f: f.write(content) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return self._send_json({'ok': True, 'path': '客户管理/' + filename, 'file': filename}) return # ── 测试报告定向收集：保存 / AI 结构化 ── if path == '/api/test-report/save': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: rid = save_test_report(payload, _au) self._send_json({'ok': True, 'id': rid, 'path': '数据/test_reports/%s.json' % rid}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return if path == '/api/test-report/structure': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: md, err = structure_test_report(payload) if err or not md: self._send_json({'ok': False, 'error': err or 'AI 未返回内容'}, 502) else: self._send_json({'ok': True, 'md': md, 'by': (_au or {}).get('name') or ''}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── 生成文件标记公共资源 API ── if path == '/api/files/mark_public': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') # Token 鉴权：身份从 token 解析，role 不信任前端传参 _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return vrole = _au.get('role', 'sales') fid = payload.get('fid') or '' is_public = bool(payload.get('is_public')) if vrole not in ('admin', 'gm'): self._send_json({'ok': False, 'error': '仅管理员可标记公共资源'}, 403) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat ok = ai_chat._mark_public(fid, is_public) self._send_json({'ok': ok, 'error': '' if ok else '未找到该文件'}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) # ── AI 智能问答 API ── if path == '/api/chat': try: length = int(self.headers.get('Content-Length') or 0) if length > 30 * 1024 * 1024: # 30MB 上限（20MB 文件 base64 后约 27MB） self._send_json({'ok': False, 'error': '请求体过大（超过 30MB），请压缩文件后重试'}, 413) return payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return # Token 鉴权：身份从 token 解析，覆盖前端传的 user（防伪造 name/role） _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return payload['user'] = _au # 动态导入 ai_chat 模块（同目录） try: _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat if payload.get('stream'): self._stream_sse(ai_chat.handle_stream(payload)) else: code, resp = ai_chat.handle(payload) self._send_json(resp, code) except Exception as e: sys.stderr.write('[chat] ERROR: %s\n' % str(e)) self._send_json({'ok': False, 'error': str(e)}, 500) return # ── AI 助手图表 → Excel 导出（真 .xlsx；token 鉴权见 do_POST 顶部兜底）── if path == '/api/report/xlsx': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return try: length = int(self.headers.get('Content-Length') or 0) body = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return headers = body.get('headers') or [] rows = body.get('rows') or [] title = (body.get('title') or '数据汇总').strip() if not headers or not isinstance(rows, list): self._send_json({'ok': False, 'error': '缺表头或数据'}, 400) return try: _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) from urllib.parse import quote import ai_chart data = ai_chart.build_xlsx(title, headers, rows) fname = quote((body.get('filename') or ai_chart.export_filename(title)).strip()) self.send_response(200) self.send_header('Content-Type', 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet') self.send_header('Content-Length', str(len(data))) self.send_header('Content-Disposition', "attachment; filename*=UTF-8''%s" % fname) self._cors_headers() self.end_headers() self.wfile.write(data) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── Token 换取 API（登录后换 token，后续 API 带 token 鉴权）── if path == '/api/auth/token': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': '请求格式错误'}, 400) return email = (payload.get('email') or '').strip().lower() pwd = payload.get('password', '') open_id = (payload.get('open_id') or '').strip() users = _load_user_roles() user = None # 方式1：邮箱+密码（本地用户表验证，role 从表查，不信任前端传的 role） if email: for u in users: if u.get('email', '').lower() == email and u.get('pwd', '') == pwd: user = u break if not user: # 默认表仅作兜底：本地用户表里已有该邮箱时，绝不允许它覆盖密码/角色 u2 = DEFAULT_USERS.get(email) if u2 and pwd == 'biori2026' and not any( (x.get('email') or '').lower() == email for x in users): user = dict(u2, email=email) # 方式2：飞书 open_id（本地用户表匹配） elif open_id: for u in users: if u.get('open_id', '') == open_id: user = u break if not user: self._send_json({'ok': False, 'error': '身份验证失败'}, 401) return token = _secrets.token_urlsafe(32) with AUTH_LOCK: AUTH_SESSIONS[token] = { 'name': user.get('name', ''), 'role': user.get('role', 'sales'), 'dept': user.get('dept', ''), 'line': user.get('line', 'da'), 'email': user.get('email', email), 'expire_at': time.time() + TOKEN_TTL, } _save_sessions() self._send_json({ 'ok': True, 'token': token, 'user': { 'name': user.get('name', ''), 'role': user.get('role', 'sales'), 'dept': user.get('dept', ''), 'line': user.get('line', 'da'), 'email': user.get('email', email), 'tabs': user.get('tabs', {}), 'open_id': user.get('open_id', ''), 'price_scope': _price_scope_of(user), } }) return # ── 飞书免登（后端换身份，不依赖 CloudBase）── if path == '/api/feishu/login': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': '请求格式错误'}, 400) return code = (payload.get('code') or '').strip() if not code: self._send_json({'ok': False, 'error': 'missing code'}, 400) return user, err = _feishu_identity(code) if not user: self._send_json({'ok': False, 'error': err or '免登失败'}, 401) return token = _secrets.token_urlsafe(32) with AUTH_LOCK: AUTH_SESSIONS[token] = { 'name': user.get('name', ''), 'role': user.get('role', 'sales'), 'dept': user.get('dept', ''), 'line': user.get('line', 'da'), 'email': user.get('email', ''), 'expire_at': time.time() + TOKEN_TTL, } _save_sessions() self._send_json({ 'ok': True, 'token': token, 'user': { 'name': user.get('name', ''), 'role': user.get('role', 'sales'), 'dept': user.get('dept', ''), 'line': user.get('line', 'da'), 'email': user.get('email', ''), 'tabs': user.get('tabs', {}), 'open_id': user.get('open_id', ''), 'price_scope': _price_scope_of(user), } }) return # ── 竞对情报沉淀 API ── if path == '/api/competitor/intel': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': '请求格式错误'}, 400) return _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat ok, msg = ai_chat._add_competitor_intel(payload, _au) self._send_json({'ok': ok, 'msg': msg}, 200 if ok else 400) return # ── 登录 API（无需 CloudBase 前端 SDK）── if path == '/api/login': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': '请求格式错误'}, 400) return email = (payload.get('email') or '').strip() pwd = payload.get('password', '') if not email: self._send_json({'ok': False, 'error': '请输入邮箱'}, 400) return user = self._find_user(email, pwd) if not user: self._send_json({'ok': False, 'error': '邮箱未注册或密码错误'}, 401) return self._send_json({ 'ok': True, 'user': { 'name': user.get('name', email.split('@')[0]), 'role': user.get('role', 'sales'), 'dept': user.get('dept', ''), 'email': email, 'tabs': user.get('tabs', {}) } }) return # ── IMA 知识库 API ── if path == '/api/ima': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return action = payload.get('action', '') try: if action == 'search': # 搜索知识库 query = payload.get('query', '') kb_id = payload.get('kb_id', IMA_KB_DIAG) result = ima_api('openapi/wiki/v1/search_knowledge', { 'knowledge_base_id': kb_id, 'query': query, 'cursor': '' }) self._send_json({'ok': True, 'data': result}) elif action == 'create_note': # 创建笔记并添加到知识库 title = payload.get('title', '未命名经验') content = payload.get('content', '') # Step 1: 创建笔记 note = ima_api('openapi/note/v1/import_doc', { 'title': title, 'content': content, 'content_format': 1 # markdown }) doc_id = note.get('data',{}).get('note_id') or note.get('note_id') or note.get('data',{}).get('doc_id') or note.get('doc_id') if not doc_id: self._send_json({'ok': False, 'error': '创建笔记失败', 'detail': note}, 500) return # Step 2: 添加到知识库 kb = ima_api('openapi/wiki/v1/add_knowledge', { 'knowledge_base_id': IMA_KB_DIAG, 'media_type': 11, # 笔记 'note_info': {'content_id': doc_id}, 'title': title }) self._send_json({'ok': True, 'doc_id': doc_id, 'note': note, 'kb': kb}) else: self._send_json({'ok': False, 'error': f'unknown action: {action}'}, 400) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── IMA 回写审核 API ── if path == '/api/ima_review': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return action = payload.get('action', '') try: if action == 'submit': title = (payload.get('title') or '').strip() content = (payload.get('content') or '').strip() submitter = (payload.get('submitter') or '').strip() if not title or not content: self._send_json({'ok': False, 'error': 'title and content required'}, 400) return review = _load_ima_review() item = { 'id': 'r' + datetime.now().strftime('%Y%m%d%H%M%S%f'), 'title': title, 'content': content, 'submitter': submitter, 'submit_time': datetime.now().strftime('%Y-%m-%d %H:%M'), 'status': 'pending' } review.append(item) _save_ima_review(review) feishu_notify(ADMIN_OPEN_ID, '📖 新的经验库回写待审核：\n标题：%s\n提交人：%s\n请在「数据管理」Tab 审核' % (title, submitter)) self._send_json({'ok': True, 'id': item['id']}) elif action == 'list': self._send_json({'ok': True, 'items': _load_ima_review()}) elif action == 'approve': pid = payload.get('id', '') review = _load_ima_review() target = next((it for it in review if it.get('id') == pid), None) if not target: self._send_json({'ok': False, 'error': 'not found'}, 404) return note = ima_api('openapi/note/v1/import_doc', { 'title': target['title'], 'content': target['content'], 'content_format': 1 }) doc_id = note.get('data', {}).get('note_id') or note.get('note_id') or note.get('data', {}).get('doc_id') or note.get('doc_id') if not doc_id: self._send_json({'ok': False, 'error': 'IMA 创建笔记失败', 'detail': note}, 500) return kb = ima_api('openapi/wiki/v1/add_knowledge', { 'knowledge_base_id': IMA_KB_DIAG, 'media_type': 11, 'note_info': {'content_id': doc_id}, 'title': target['title'] }) target['status'] = 'approved' target['reviewer'] = payload.get('reviewer', '') target['review_time'] = datetime.now().strftime('%Y-%m-%d %H:%M') _save_ima_review(review) self._send_json({'ok': True, 'doc_id': doc_id, 'kb': kb}) elif action == 'reject': pid = payload.get('id', '') reason = (payload.get('reason') or '').strip() review = _load_ima_review() target = next((it for it in review if it.get('id') == pid), None) if not target: self._send_json({'ok': False, 'error': 'not found'}, 404) return target['status'] = 'rejected' target['reviewer'] = payload.get('reviewer', '') target['reason'] = reason target['review_time'] = datetime.now().strftime('%Y-%m-%d %H:%M') _save_ima_review(review) self._send_json({'ok': True, 'status': 'rejected'}) else: self._send_json({'ok': False, 'error': 'unknown action'}, 400) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── 归档多轮对话为经验 API ── if path == '/api/archive_experience': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return try: _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat messages = payload.get('messages') or [] user = payload.get('user') or {} if not user.get('name'): self._send_json({'ok': False, 'error': '未登录'}, 401) return title, content = ai_chat.archive_experience(messages, user) if not title: self._send_json({'ok': False, 'error': content}, 500) return self._send_json({'ok': True, 'title': title, 'content': content}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── 飞书免登 API ── if path == '/api/feishu/auth': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return code = payload.get('code', '') if not code: self._send_json({'ok': False, 'error': 'missing code'}, 400) return # 用 app_access_token 换用户身份 try: # Step 1: 获取 app_access_token req = urllib.request.Request( 'https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal', data=json.dumps({'app_id': 'cli_aaee6a2f90b89bc1', 'app_secret': 'arht5fd8wyppZysFvOjUcb2bV0UMVVYF'}).encode(), headers={'Content-Type': 'application/json'}, method='POST' ) with urllib.request.urlopen(req, timeout=10) as r: token_res = json.loads(r.read()) access_token = token_res.get('app_access_token', '') if not access_token: self._send_json({'ok': False, 'error': 'no access token'}, 500) return # Step 2: 用 code 换用户身份 req2 = urllib.request.Request( 'https://open.feishu.cn/open-apis/authen/v1/oidc/access_token', data=json.dumps({'grant_type': 'authorization_code', 'code': code}).encode(), headers={'Content-Type': 'application/json', 'Authorization': f'Bearer {access_token}'}, method='POST' ) with urllib.request.urlopen(req2, timeout=10) as r2: auth_res = json.loads(r2.read()) if auth_res.get('code') != 0: self._send_json({'ok': False, 'error': 'auth failed', 'detail': str(auth_res)[:200]}, 500) return name = (auth_res.get('data') or {}).get('name', '') self._send_json({'ok': True, 'name': name}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── 活动创建 API ── if path == '/api/create_activity': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return name = payload.get('name', '拜访纪要') customer_id = payload.get('customer_id') activity_type = payload.get('activity_type', '线下拜访') description = payload.get('description', '') start_date = payload.get('start_date', time.strftime('%Y-%m-%d')) department = payload.get('department') # 活动类型映射 type_map = {'线下拜访': 'v4zlj721e', '线上跟进': '5vezmmm1y', '陌拜': 'wg15dd75b', '协同拜访': 'wsxnxsd60'} fields = [ {'field_key': 'name', 'field_value': name}, {'field_key': 'field_76654e', 'field_value': type_map.get(activity_type, 'v4zlj721e')}, {'field_key': 'field_a2b3f6', 'field_value': int(time.mktime(time.strptime(start_date, '%Y-%m-%d'))) * 1000 if start_date else int(time.time() * 1000)}, ] if description: fields.append({'field_key': 'field_b99055', 'field_value': description}) if customer_id: fields.append({'field_key': 'field_5f20fc', 'field_value': str(customer_id)}) if department: fields.append({'field_key': 'field_5ed7ab', 'field_value': department}) result = mcp_call('create_workitem', { 'project_key': MCP_PK, 'work_item_type': ACTIVITY_TYPE_KEY, 'fields': fields, 'template_id': '700373' }) if result.get('ok') == False: self._send_json({'ok': False, 'error': result.get('error', 'MCP call failed')}, 500) return # 提取工作项 ID wid = result.get('id') or result.get('work_item_id') or result.get('data', {}).get('id') feishu_url = f'https://project.feishu.cn/baoyu/workitem/{wid}' if wid else '' self._send_json({'ok': True, 'work_item_id': wid, 'feishu_url': feishu_url, 'mcp_result': result}) return # ── MCP 数据刷新 API ── if path == '/api/mcp/refresh': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return rtype = payload.get('type', 'all') MQL_MAP = { 'customers': 'SELECT `name`,`work_item_id`,`field_17186c`,`field_6415cf`,`field_5ed7ab`,`field_2d5b6a`,`field_e62869`,`field_c3224b`,`field_c8e80d`,`field_a1ff52`,`field_79aa0b`,`owner`,`start_time`,`business`,`current_status_operator` FROM `6593cd71471290e3cc6be6e6`.`65ae1e403c87b152f3365ca6`', 'orders': 'SELECT `name`,`work_item_id`,`field_e1001d`,`field_a6e501`,`field_5ed7ab`,`field_2d6367`,`field_e0cce5`,`field_2e095b`,`field_4a1b47`,`field_29d947`,`field_3a3ccb`,`field_8c4b7b`,`field_5b2b3b`,`field_dbd4ab`,`field_a98458`,`field_8d9153`,`owner`,`start_time`,`business`,`current_status_operator`,`work_item_status` FROM `6593cd71471290e3cc6be6e6`.`662612aea6bb7089fea044ef`', 'complaints': 'SELECT `name`,`work_item_id`,`start_time`,`work_item_status`,`field_9e2144`,`field_6fb810`,`field_3ce9fe`,`field_d107b0`,`field_502a85`,`field_ac5caf`,`owner`,`business`,`current_status_operator`,`finish_time`,`finish_status` FROM `658288abfb8bd616b17025f1`.`6669433056a98249604376de`', 'activities': 'SELECT `name`,`work_item_id`,`field_b54da4`,`field_b99055`,`field_76654e`,`field_a2b3f6`,`field_5f20fc`,`field_5ed7ab`,`field_d6ee93`,`owner`,`start_time`,`business`,`current_status_operator`,`work_item_status` FROM `6593cd71471290e3cc6be6e6`.`65ae1e5d44338dbe7c39a29a`', 'opportunities': 'SELECT `name`,`work_item_id`,`owner`,`start_time`,`business`,`current_status_operator`,`work_item_status`,`description` FROM `6593cd71471290e3cc6be6e6`.`story`', } if rtype == 'all': results = {} for key, mql in MQL_MAP.items(): try: results[key] = refresh_mcp_cache(key, mql) except Exception as e: results[key] = f'error: {e}' self._send_json({'ok': True, 'results': results}) elif rtype in MQL_MAP: try: count = refresh_mcp_cache(rtype, MQL_MAP[rtype]) self._send_json({'ok': True, 'type': rtype, 'count': count}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) else: self._send_json({'ok': False, 'error': f'unknown type: {rtype}, valid: all, {", ".join(MQL_MAP.keys())}'}, 400) return # ── 拜访卡生成（精简版：写请求队列）── if path == '/api/visit-card': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return customer = (payload.get('customer') or '').strip() if not customer: self._send_json({'ok': False, 'error': 'customer required'}, 400) return # 保存请求队列 try: queue_dir = os.path.join(ROOT, '数据', 'visit_requests') os.makedirs(queue_dir, exist_ok=True) ts = datetime.now().strftime('%Y%m%d_%H%M%S') cn = customer[:10].replace('/', '_') queue_file = os.path.join(queue_dir, f'{ts}_{cn}.json') with open(queue_file, 'w') as qf: json.dump({ 'customer': customer, 'salesperson': payload.get('salesperson', ''), 'dept': payload.get('dept', ''), 'visitDate': payload.get('visitDate', ''), 'note': payload.get('note', ''), 'requirements': payload.get('requirements', ''), 'types': payload.get('types', []), 'requested_at': datetime.now().isoformat() }, qf, ensure_ascii=False) except Exception as e: print(f'[visit-card] Queue error: {e}') self._send_json({ 'ok': True, 'queued': True, 'customer': customer, 'generated_at': datetime.now().isoformat() }) return # ── 文件上传 API ── if path == '/api/upload': try: content_type = self.headers.get('Content-Type', '') if 'multipart/form-data' not in content_type: self._send_json({'ok': False, 'error': '需要 multipart/form-data'}, 400) return # 解析 multipart boundary = content_type.split('boundary=')[1].strip() if boundary.startswith('"') and boundary.endswith('"'): boundary = boundary[1:-1] length = int(self.headers.get('Content-Length') or 0) raw = self.rfile.read(length) # 简单 multipart 解析 boundary_bytes = ('--' + boundary).encode('utf-8') end_boundary = ('--' + boundary + '--').encode('utf-8') parts = raw.split(boundary_bytes) uploaded_files = [] for part in parts[1:]: if part.startswith(b'--'): continue if part.startswith(b'\r\n'): part = part[2:] if part.endswith(b'\r\n'): part = part[:-2] header_end = part.find(b'\r\n\r\n') if header_end < 0: continue headers_raw = part[:header_end].decode('utf-8', errors='replace') body = part[header_end + 4:] # 提取 filename import re as _re fn_match = _re.search(r'filename="([^"]*)"', headers_raw) if not fn_match: continue filename = fn_match.group(1) # 保存文件 upload_dir = os.path.join(DATA_DIR, '上传') os.makedirs(upload_dir, exist_ok=True) ts = datetime.now().strftime('%Y%m%d_%H%M%S') safe_name = f'{ts}_{filename}' filepath = os.path.join(upload_dir, safe_name) with open(filepath, 'wb') as f: f.write(body) # 自动分类 ext = os.path.splitext(filename)[1].lower() ftype_map = {'.xlsx':'ERP/飞书导出', '.xls':'ERP/飞书导出', '.csv':'ERP/飞书导出', '.docx':'文档', '.doc':'文档', '.pdf':'PDF', '.png':'图片', '.jpg':'图片'} ftype = ftype_map.get(ext, '其他') uploaded_files.append({ 'original': filename, 'saved': safe_name, 'size': len(body), 'type': ftype, 'path': f'数据/上传/{safe_name}' }) # 记录上传日志 log_file = os.path.join(DATA_DIR, '上传', 'upload_log.json') log = [] if os.path.exists(log_file): try: with open(log_file, 'r', encoding='utf-8') as lf: log = json.load(lf) except: log = [] for uf in uploaded_files: log.insert(0, {'time': datetime.now().isoformat(), **uf}) log = log[:100] # 保留最近 100 条 with open(log_file, 'w', encoding='utf-8') as lf: json.dump(log, lf, ensure_ascii=False, indent=2) self._send_json({'ok': True, 'files': uploaded_files}) except Exception as e: self._send_json({'ok': False, 'error': str(e)}, 500) return # ── 文件处理 API（质检 + 更新 + 归档）── if path == '/api/upload/process': try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return saved_name = payload.get('file', '') if not saved_name: self._send_json({'ok': False, 'error': 'missing file param'}, 400) return upload_dir = os.path.join(DATA_DIR, '上传') filepath = os.path.join(upload_dir, saved_name) if not os.path.exists(filepath): self._send_json({'ok': False, 'error': f'文件不存在: {saved_name}'}, 404) return ext = os.path.splitext(saved_name)[1].lower() result = {'ok': True, 'file': saved_name, 'steps': []} # Step 1: 质检 qc = run_quality_check(filepath, ext) result['qc'] = qc result['steps'].append({'step': '质检', 'ok': qc.get('pass', False), 'summary': qc.get('summary', '')}) if not qc.get('pass'): self._send_json(result) return # Step 2: 数据处理 ftype = qc.get('type', 'unknown') process_result = process_uploaded_file(filepath, ftype, saved_name) result['process'] = process_result result['steps'].append({'step': '数据更新', 'ok': process_result.get('ok', False), 'summary': process_result.get('summary', '')}) # Step 3: 归档（仅处理成功时归档） if process_result.get('ok'): archive_dir = os.path.join(DATA_DIR, '上传', '_已处理归档') os.makedirs(archive_dir, exist_ok=True) ts = datetime.now().strftime('%Y%m%d_%H%M%S') archive_name = f'{ts}_{saved_name}' import shutil shutil.move(filepath, os.path.join(archive_dir, archive_name)) result['archive'] = archive_name result['steps'].append({'step': '归档', 'ok': True, 'summary': f'已归档至 数据/上传/_已处理归档/{archive_name}'}) else: result['steps'].append({'step': '归档', 'ok': False, 'summary': '处理未成功，文件保留在上传目录'}) self._send_json(result) return # ── 统一权限管理：更新按人权限（仅 admin/gm）── if path == '/api/permissions': _au = self._auth_user() if not _au: self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return if _au.get('role') not in ('admin', 'gm'): self._send_json({'ok': False, 'error': '仅管理员可访问'}, 403) return try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return name = (payload.get('name') or '').strip() if not name: self._send_json({'ok': False, 'error': '缺少用户名'}, 400) return _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat # 1) ds_limit / extra_customers / blocked_customers → 独立覆盖文件 perms = ai_chat._load_user_permissions() p = perms.get(name, {}) if 'ds_limit' in payload: try: p['ds_limit'] = int(payload['ds_limit']) except Exception: self._send_json({'ok': False, 'error': 'ds_limit 必须是整数（-1=不限）'}, 400) return if 'extra_customers' in payload and isinstance(payload['extra_customers'], list): p['extra_customers'] = [str(x).strip() for x in payload['extra_customers'] if str(x).strip()] if 'blocked_customers' in payload and isinstance(payload['blocked_customers'], list): p['blocked_customers'] = [str(x).strip() for x in payload['blocked_customers'] if str(x).strip()] # 价格查询三态：all=全量明细 / own=仅自己客户 / none=不可见 if 'price_scope' in payload: _ps = str(payload.get('price_scope') or '').strip().lower() if _ps not in ('all', 'own', 'none'): self._send_json({'ok': False, 'error': 'price_scope 只能是 all / own / none'}, 400) return p['price_scope'] = _ps if 'price_alias' in payload and isinstance(payload['price_alias'], list): p['price_alias'] = [str(x).strip() for x in payload['price_alias'] if str(x).strip()] perms[name] = p ai_chat._save_user_permissions(perms) # 2) tabs → 改本地 user_roles.json + 同步 CloudBase user_roles 集合 tabs_synced = False cb_synced = False cb_err = '' if 'tabs' in payload and isinstance(payload['tabs'], dict): merged = {k: (1 if v else 0) for k, v in payload['tabs'].items()} try: roles = _load_user_roles() for u in roles: if u.get('name') == name: merged = dict(u.get('tabs') or {}) for k, v in payload['tabs'].items(): merged[k] = 1 if v else 0 u['tabs'] = merged break tmp = USER_ROLES_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(roles, f, ensure_ascii=False, indent=1) os.replace(tmp, USER_ROLES_FILE) tabs_synced = True except Exception as e: self._send_json({'ok': False, 'error': 'tabs 本地保存失败：' + str(e)}, 500) return # 同步 CloudBase（飞书免登直接读 CloudBase，必须同步才生效） cb_synced, cb_err = _sync_tabs_to_cloudbase(name, merged) self._send_json({'ok': True, 'name': name, 'tabs_synced': tabs_synced, 'cb_synced': cb_synced, 'cb_err': cb_err}) return # ── 原有 todos API ── if path != '/api/todos': self._send_json({'ok': False, 'error': 'not found'}, 404) return try: length = int(self.headers.get('Content-Length') or 0) payload = json.loads(self.rfile.read(length) or b'{}') except Exception: self._send_json({'ok': False, 'error': 'bad json'}, 400) return action = payload.get('action') with LOCK: store = read_store() if action == 'save': user = str(payload.get('user') or '') todos = payload.get('todos') if not user or not isinstance(todos, list): self._send_json({'ok': False, 'error': 'bad payload'}, 400) return store['users'][user] = todos elif action == 'close': item = payload.get('item') if not isinstance(item, dict): self._send_json({'ok': False, 'error': 'bad payload'}, 400) return store['closed'] = [x for x in store['closed'] if x.get('id') != item.get('id')] store['closed'].insert(0, item) store['closed'] = store['closed'][:MAX_CLOSED] elif action == 'unclose': tid = payload.get('id') store['closed'] = [x for x in store['closed'] if x.get('id') != tid] else: self._send_json({'ok': False, 'error': 'unknown action'}, 400) return write_store(store) self._send_json({'ok': True}) def _visit_request_loop(): """后台线程：每 30 秒扫描 visit_requests 队列，自动生成拜访卡/市场调研卡。""" import threading as _th qdir = os.path.join(ROOT, '数据', 'visit_requests') processed_dir = os.path.join(qdir, 'processed') while True: try: os.makedirs(processed_dir, exist_ok=True) _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat for fn in sorted(os.listdir(qdir)): if not fn.endswith('.json'): continue fp = os.path.join(qdir, fn) if os.path.isdir(fp): continue try: with open(fp, encoding='utf-8') as f: req = json.load(f) except Exception: continue customer = (req.get('customer') or '').strip() if not customer: continue user = {'name': req.get('salesperson', ''), 'role': 'sales', 'dept': req.get('dept', ''), 'line': 'da'} files, err = ai_chat._process_visit_request(req, user) req['processed_at'] = datetime.now().isoformat() req['files'] = files req['error'] = err try: with open(fp, 'w', encoding='utf-8') as f: json.dump(req, f, ensure_ascii=False, indent=1) os.replace(fp, os.path.join(processed_dir, fn)) print('[visit-request] 已处理 %s -> %s' % (customer, err or ('%d 个文件' % len(files)))) # 刷新文件清单 + 部署到 CDN（前端读的是 CDN manifest，只更新本地会导致前端不同步） try: os.system('bash ~/.hermes/scripts/regenerate_manifest.sh >/dev/null 2>&1') os.system('tcb hosting deploy "%s/数据/file_manifest.json" "数据/file_manifest.json" -e bier-sales-d0gatbvlx288724e9 >/dev/null 2>&1' % ROOT) except Exception: pass except Exception as e: print('[visit-request] 处理 %s 失败: %s' % (customer, e)) except Exception as e: print('[visit-request] loop error: %s' % e) _th.Event().wait(30) if __name__ == '__main__': port = int(sys.argv[1]) if len(sys.argv) > 1 else 8000 print('销售工作台服务已启动：http://0.0.0.0:%d （根目录：%s）' % (port, ROOT)) print('按 Ctrl+C 停止') # 报销助手后台工人（串行处理上传的票据） try: _expense_api().start_worker() except Exception as _e: print('[expense] 工人启动失败: %s' % _e) # 生成文件过期清理：启动时 + 每 24 小时清理一次（30 天，公共资源豁免） import threading as _th def _cleanup_loop(): while True: try: _p = os.path.dirname(os.path.abspath(__file__)) if _p not in sys.path: sys.path.insert(0, _p) import ai_chat n = ai_chat._cleanup_expired(30) if n: print('已清理 %d 个过期生成文件' % n) except Exception: pass _th.Event().wait(86400) _th.Thread(target=_cleanup_loop, daemon=True).start() _th.Thread(target=_visit_request_loop, daemon=True).start() ThreadingHTTPServer(('0.0.0.0', port), Handler).serve_forever()