#!/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 import secrets import hmac import html as html_lib from datetime import datetime, timedelta import urllib.request from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler from http.cookies import SimpleCookie from urllib.parse import unquote 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 天滑动续期 # ══════════ 临时外部演示页：48 小时、3 个账号、服务端会话 ══════════ DEMO_STATE_DIR = os.path.expanduser('~/.hermes/profiles/zone-b/state') DEMO_PAGE_FILE = os.path.join(DEMO_STATE_DIR, 'demo_customer_rollup_20261105.html') DEMO_STATE_FILE = os.path.join(DEMO_STATE_DIR, 'demo_access_sessions.json') DEMO_ACCOUNTS = ('Biori2027-1', 'Biori2027-2', 'Biori2027-3') DEMO_CODE = 'Biori2027' DEMO_TTL = 48 * 3600 DEMO_COOKIE = 'bior_demo_session' _DEMO_LOCK = threading.Lock() _DEMO_FAILS = {} # ip -> [count, blocked_until] # ══════════ 只读角色（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() def _demo_state_load(): """Load the 48h demo state. The expiry is absolute, not sliding.""" os.makedirs(DEMO_STATE_DIR, mode=0o700, exist_ok=True) now = int(time.time()) try: with open(DEMO_STATE_FILE, encoding='utf-8') as f: st = json.load(f) if not isinstance(st, dict): raise ValueError('bad state') except Exception: st = {'created_at': now, 'expires_at': now + DEMO_TTL, 'sessions': {}} if not st.get('expires_at'): st['expires_at'] = now + DEMO_TTL st.setdefault('sessions', {}) return st def _demo_state_save(st): os.makedirs(DEMO_STATE_DIR, mode=0o700, exist_ok=True) tmp = DEMO_STATE_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(st, f, ensure_ascii=False, indent=2) os.replace(tmp, DEMO_STATE_FILE) try: os.chmod(DEMO_STATE_FILE, 0o600) except Exception: pass def _demo_hash(token): return hashlib.sha256((token or '').encode('utf-8')).hexdigest() def _demo_ua_hash(ua): return hashlib.sha256((ua or '').encode('utf-8')).hexdigest() def _demo_cookie_token(cookie_header): try: c = SimpleCookie() c.load(cookie_header or '') m = c.get(DEMO_COOKIE) return m.value if m else '' except Exception: return '' # ══════════ 三部门逐月拆分看板：按登录态过滤的数据接口（方案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', '')} # ---------- 临时外部演示页（48h / 3 账号） ---------- def _demo_is_https(self): host = (self.headers.get('Host') or '').lower() proto = (self.headers.get('X-Forwarded-Proto') or '').lower() return proto == 'https' or host.startswith('api.biorisales.site') or host.endswith(':8443') def _demo_set_cookie(self, token='', clear=False): if clear: val = DEMO_COOKIE + '=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0' else: val = DEMO_COOKIE + '=' + token + '; Path=/; HttpOnly; SameSite=Lax; Max-Age=' + str(DEMO_TTL) if self._demo_is_https(): val += '; Secure' self.send_header('Set-Cookie', val) def _demo_session(self): token = _demo_cookie_token(self.headers.get('Cookie', '')) if not token: return None st = _demo_state_load() if int(time.time()) > int(st.get('expires_at') or 0): return None sess = (st.get('sessions') or {}).get(_demo_hash(token)) if not sess: return None if sess.get('ua') != _demo_ua_hash(self.headers.get('User-Agent', '')): return None return {'account': sess.get('account', ''), 'token': token} def _demo_json(self, obj, code=200, token=None, clear=False): 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('Cache-Control', 'no-store, no-cache, must-revalidate') self._cors_headers() if token is not None or clear: self._demo_set_cookie(token or '', clear=clear) self.end_headers() self.wfile.write(body) def _demo_login_page(self, message='', status=200): msg = html_lib.escape(message or '') body = """ 客户拉通大表 · 演示登录
BIORI BIOTECH · 销售工作台
客户拉通大表 · 演示
本页面为临时演示环境，有效期 48 小时，仅限 3 个账号访问。账号和验证码由管理员单独发送。
__MSG__
演示环境为只读；请勿转发链接或截图外传。访问有效期至 2026-11-07 09:34（北京时间）。
""" body = body.replace('__MSG__', msg) raw = body.encode('utf-8') self.send_response(status) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(raw))) self.send_header('Cache-Control', 'no-store, no-cache, must-revalidate') self.send_header('X-Frame-Options', 'DENY') self.send_header('Referrer-Policy', 'no-referrer') self._cors_headers() self.end_headers() self.wfile.write(raw) def _demo_login(self): ip = self.client_address[0] if self.client_address else '' now = time.time() fail = _DEMO_FAILS.get(ip) if fail and fail[1] > now: self._demo_json({'ok': False, 'error': '尝试次数过多，请 10 分钟后再试'}, 429) return try: n = int(self.headers.get('Content-Length', '0') or 0) data = json.loads(self.rfile.read(n).decode('utf-8') or '{}') except Exception: self._demo_json({'ok': False, 'error': '请求格式错误'}, 400) return account = str(data.get('account') or '').strip() code = str(data.get('code') or '') st = _demo_state_load() if now > int(st.get('expires_at') or 0): self._demo_json({'ok': False, 'error': '演示链接已过期'}, 410) return if account not in DEMO_ACCOUNTS or not hmac.compare_digest(code, DEMO_CODE): cnt = (fail[0] if fail else 0) + 1 _DEMO_FAILS[ip] = [cnt, (now + 600) if cnt >= 10 else 0] self._demo_json({'ok': False, 'error': '账号或验证码不正确'}, 401) return _DEMO_FAILS.pop(ip, None) with _DEMO_LOCK: st = _demo_state_load() sessions = st.setdefault('sessions', {}) # one active session per account; up to 3 accounts for k in [k for k, v in sessions.items() if v.get('account') == account]: sessions.pop(k, None) if len(sessions) >= 3: self._demo_json({'ok': False, 'error': '当前演示账号已满，请联系管理员'}, 403) return token = secrets.token_urlsafe(32) sessions[_demo_hash(token)] = { 'account': account, 'created_at': int(now), 'ua': _demo_ua_hash(self.headers.get('User-Agent', '')), } _demo_state_save(st) self._demo_json({'ok': True, 'account': account, 'expires_at': int(st.get('expires_at') or 0)}, token=token) def _demo_logout(self): token = _demo_cookie_token(self.headers.get('Cookie', '')) if token: with _DEMO_LOCK: st = _demo_state_load() (st.get('sessions') or {}).pop(_demo_hash(token), None) _demo_state_save(st) self._demo_json({'ok': True}, clear=True) def _demo_serve_page(self): if 'logout=1' in (self.path or ''): self._demo_logout() return st = _demo_state_load() if int(time.time()) > int(st.get('expires_at') or 0): self._demo_login_page('演示链接已过期（48 小时有效期已结束）。', 410) return if not self._demo_session(): self._demo_login_page('') return try: with open(DEMO_PAGE_FILE, 'rb') as f: raw = f.read() except Exception as e: self._demo_json({'ok': False, 'error': '演示页面文件缺失: %s' % e}, 500) return self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(raw))) self.send_header('Content-Disposition', 'inline') self.send_header('Cache-Control', 'no-store, no-cache, must-revalidate') self.send_header('X-Frame-Options', 'DENY') self.send_header('Referrer-Policy', 'no-referrer') self.end_headers() self.wfile.write(raw) 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] # 临时外部演示页（48h / 3 账号）：登录页与受保护正文 if path in ('/demo/customer-rollup', '/demo/customer-rollup/'): self._demo_serve_page() return # 原公开页面转到门禁页，避免绕过登录直接拿数据 if unquote(path) == '/工具/客户拉通大表_demo_v11_活动录入V2同步_20261101_V01.html': self.send_response(302) self.send_header('Location', '/demo/customer-rollup') self.send_header('Cache-Control', 'no-store') self.end_headers() return # 统一鉴权兜底：所有 /api/ 端点（除登录/换 token/飞书免登）都需有效 token _demo_rollup = path == '/api/customer-rollup/data' and bool(self._demo_session()) if path.startswith('/api/') and path not in ('/api/login', '/api/auth/token', '/api/feishu/login') and not _demo_rollup: 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': _demo = self._demo_session() if _demo: # 演示账号只读；数据仍走同一个快照与脱敏/状态逻辑 _au = {'name': '演示-' + _demo.get('account', ''), 'role': 'demo', 'dept': '', 'line': 'da'} else: _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] if path == '/api/demo/login': self._demo_login() return if path == '/api/demo/logout': self._demo_logout() return # 统一鉴权兜底：所有 /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()