#!/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 sys import subprocess import threading import time from datetime import datetime import urllib.request from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # 脚本/ → 父目录 DATA_DIR = os.path.join(ROOT, '数据') TODOS_FILE = os.path.join(DATA_DIR, 'todos.json') 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() 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 = 'fwEO41E0Ef0CfulmtQHYgYQy_jBsbl6tXE5x6VYLIoA=' # 宝锐诊断原料销售 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': 'admin', '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 天滑动续期 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 _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 [] 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() 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""" email = email.strip().lower() 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 _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.send_header('Access-Control-Allow-Origin', '*') 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.send_header('Access-Control-Allow-Origin', '*') 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.send_header('Access-Control-Allow-Origin', '*') 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 == '/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') 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.send_header('Access-Control-Allow-Origin', '*') 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/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/permissions': # ── 统一权限管理：读取按人权限（仅 admin/gm 可看全量）── _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 _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 [], }) 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.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) 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'): if not self._auth_user(): self._send_json({'ok': False, 'error': '登录已过期，请重新登录'}, 401) return # ── 客户特殊要求 保存/删除 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') 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 # ── 生成文件标记公共资源 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 # ── 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': 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', {}), 'pwd': user.get('pwd', ''), 'open_id': user.get('open_id', ''), } }) 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', {}), 'pwd': user.get('pwd', ''), 'open_id': user.get('open_id', ''), } }) 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()] 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 停止') # 生成文件过期清理：启动时 + 每 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()