diff --git a/app/engine/config.py b/app/engine/config.py index f80059f..2d5bc44 100644 --- a/app/engine/config.py +++ b/app/engine/config.py @@ -1,6 +1,15 @@ -from pydantic_settings import BaseSettings +"""进程级环境配置(.env)——Phase 1 的"平台默认"层。 + +Task 9 后业务参数(邮件/推送/AI 凭据)优先读租户 config(engine 服务的 +resolve_*_settings 空槽回退到这里);本文件只剩部署级开关与平台默认密钥。 +Task 9 死代码清理: DATABASE_URL/ensure_dirs/DATA_DIR 已无消费方(每租户 +独立 SQLite 由 registry.make_tenant_engine 构造)——删除;ADMIN_PASSWORD +保留(Phase 2 超管初始密码引导用)。 +""" from pathlib import Path +from pydantic_settings import BaseSettings + class Settings(BaseSettings): APP_HOST: str = "0.0.0.0" @@ -14,8 +23,6 @@ class Settings(BaseSettings): ZHIPU_BASE_URL: str = "" AI_MODEL: str = "glm-4-flash" - DATABASE_URL: str = "sqlite+aiosqlite:///./data/hesupipe_ai_kefu.db" - SMTP_HOST: str = "" SMTP_PORT: int = 465 SMTP_USER: str = "" @@ -41,11 +48,6 @@ class Settings(BaseSettings): # app/engine/config.py 比 jc 的 app/config.py 深一层,多退一级才到项目根 BASE_DIR = Path(__file__).resolve().parent.parent.parent -DATA_DIR = BASE_DIR / "data" STATIC_DIR = BASE_DIR / "static" settings = Settings() - - -def ensure_dirs(): - DATA_DIR.mkdir(exist_ok=True) diff --git a/app/engine/routes/chat_rest.py b/app/engine/routes/chat_rest.py index 4aafced..2b797cd 100644 --- a/app/engine/routes/chat_rest.py +++ b/app/engine/routes/chat_rest.py @@ -8,7 +8,8 @@ Red lines (see spec): - WS path (`app/api/routes/chat.py`) shares `ConversationManager` with these routes; both paths stay drift-free. - `context` seeding is conditional + whitelist (handled in - `ConversationManager.create_conversation`). + `ConversationManager.create_conversation`; Task 9 起白名单 = + 租户 config.engine.context_seed_keys——jiancai product_*/auto vehicle_*). - The message route consumes `stream_process_message`; lead email is triggered inside `ConversationManager` when contact info is detected (contact-driven-lead-email plan). @@ -20,6 +21,7 @@ from sqlalchemy import select from app.engine.models import ConversationSource, ConversationStatus, Message from app.engine.routes.runtime import get_runtime from app.engine.services.conversation import ConversationManager +from app.engine.services.rate_limit import DEFAULT_MAX from app.engine.services.rate_limit import check as rate_check router = APIRouter(prefix="/api/chat", tags=["chat-rest"]) @@ -111,15 +113,19 @@ async def chat_message(req: MessageRequest, request: Request): `ConversationManager` when contact info is detected (contact-driven-lead-email plan) — this route carries no email logic. - A light in-memory rate limit (per ip + X-Device-Id, 30/h) guards GLM - spend; only this route is throttled. + A light in-memory rate limit (per tenant + ip + X-Device-Id, rolling 1h) + guards GLM spend; only this route is throttled. Task 9: 桶键加租户维度 + (两租户同出口 IP 不互挤配额),上限读租户 config.engine.rate_limit_per_hour + (默认 30,与出厂一致)。 """ + runtime = get_runtime(request) ip = request.client.host if request.client else "" dev = request.headers.get("X-Device-Id", "") - if not rate_check(ip, dev): + engine_cfg = runtime.config.data.get("engine") or {} + limit = engine_cfg.get("rate_limit_per_hour") or DEFAULT_MAX + if not rate_check(runtime.config.slug, ip, dev, max_requests=limit): raise HTTPException(status_code=429, detail="rate limit exceeded") - runtime = get_runtime(request) async with runtime.session_factory() as db: manager = ConversationManager(db, runtime=runtime) conv = await manager.get_by_session(req.session_id) diff --git a/app/engine/services/ai_service.py b/app/engine/services/ai_service.py index 161b255..b76f80d 100644 --- a/app/engine/services/ai_service.py +++ b/app/engine/services/ai_service.py @@ -1,12 +1,14 @@ import json import logging from collections.abc import AsyncGenerator +from dataclasses import dataclass from pathlib import Path from app.engine.providers.base import AIProvider from app.engine.providers.glm import GLMProvider from app.engine.config import settings from app.engine.services.boundary_filter import DEFAULT_WALL +from app.engine.wall_schema import CompiledWall logger = logging.getLogger(__name__) @@ -45,9 +47,13 @@ def _format_deflection_pool(deflections) -> str: ) +@dataclass(frozen=True) class PromptProvider: """租户提示词四件套持有者(Task 6, R1 话术单一事实源)。 + frozen(Task 9 收口, Task 6 审查遗留): 实例跨租户运行时/模块级 + DEFAULT_PROVIDER 共享, 不可变防共享单例被毒化。 + - from_dir(dir): 读 system/company_facts/extractor/evaluator 四个 md, 存实例字段——提示词属于租户(seed_profiles//prompts/),不再有 代码级 app/engine/prompts 副本; @@ -57,13 +63,10 @@ class PromptProvider: - build_extractor/build_evaluator: $existing_info/$messages/$profile/ $recent_messages 替换逻辑自 AIService 原样搬入(行为等价)。 """ - - def __init__(self, system: str, company_facts: str, - extractor: str, evaluator: str): - self.system = system - self.company_facts = company_facts - self.extractor = extractor - self.evaluator = evaluator + system: str + company_facts: str + extractor: str + evaluator: str @classmethod def from_dir(cls, prompts_dir) -> "PromptProvider": @@ -100,25 +103,54 @@ DEFAULT_PROVIDER = PromptProvider.from_dir( Path(__file__).resolve().parents[3] / "seed_profiles" / "jiancai" / "prompts") -def create_provider() -> AIProvider: - if settings.AI_PROVIDER == "glm": +# model 的出厂默认值——视同"未设置"哨兵: 租户未定制(merged 值 == 出厂默认) +# 时回退 env AI_MODEL,部署级模型选择不因接线失效。显式配了其他值则用租户的。 +_FACTORY_DEFAULT_MODEL = "glm-4-flash" + + +def resolve_ai_settings(config_data: dict | None) -> dict: + """租户 config.ai → provider 实例化参数(Task 9 接线; env 回退)。 + + api_key/base_url 空槽(None/空串)→ env 智谱凭据(平台默认集团 key, + spec §5.5 "用平台默认"状态下租户侧不存密钥);model 空或等于出厂默认 + → env AI_MODEL。 + """ + ai = ((config_data or {}).get("ai")) or {} + model = ai.get("model") + if not model or model == _FACTORY_DEFAULT_MODEL: + model = settings.AI_MODEL or _FACTORY_DEFAULT_MODEL + return { + "provider": ai.get("provider") or settings.AI_PROVIDER, + "api_key": ai.get("api_key") or settings.ZHIPU_API_KEY, + "base_url": ai.get("base_url") or settings.ZHIPU_BASE_URL, + "model": model, + } + + +def create_provider(config_data: dict | None = None) -> AIProvider: + """provider 工厂(Task 9: 按租户配置实例化 glm;None 走 env 平台默认)。""" + cfg = resolve_ai_settings(config_data) + if cfg["provider"] == "glm": return GLMProvider( - api_key=settings.ZHIPU_API_KEY, - model=settings.AI_MODEL, - base_url=settings.ZHIPU_BASE_URL, + api_key=cfg["api_key"], + model=cfg["model"], + base_url=cfg["base_url"], ) - raise ValueError(f"Unknown AI provider: {settings.AI_PROVIDER}") + raise ValueError(f"Unknown AI provider: {cfg['provider']}") class AIService: def __init__(self, provider: AIProvider | None = None, prompts: PromptProvider | None = None, - wall_deflections: dict | None = None): - self.provider = provider or create_provider() + wall: CompiledWall | None = None, + config_data: dict | None = None): + self.provider = provider or create_provider(config_data) # Task 6: 提示词经 PromptProvider 注入(租户四件套);默认 = jiancai # 出厂 + 出厂墙话术池(与 DEFAULT_PROVIDER 同源,单一事实源 wall.json) self.prompts = prompts or DEFAULT_PROVIDER - self.wall_deflections = (wall_deflections if wall_deflections is not None + # Task 9 收口: wall 参数定型为 CompiledWall(话术池只取 .deflections, + # 与 prompts 同为租户运行时产物);缺省 = jiancai 出厂墙 + self.wall_deflections = (wall.deflections if wall is not None else DEFAULT_WALL.deflections) # 对话基座: 行为 prompt 在前(红线显著性),事实卡居中(参考资料), # 当轮 context 提醒最后拼接(在 chat/stream_chat 内,保持最新鲜) diff --git a/app/engine/services/cc_completion.py b/app/engine/services/cc_completion.py index db70829..484c189 100644 --- a/app/engine/services/cc_completion.py +++ b/app/engine/services/cc_completion.py @@ -171,7 +171,8 @@ def _consume_round(state: dict) -> dict: def transit(state: dict | None, reply: dict | None, - new_canon: str | None = None) -> dict | None: + new_canon: str | None = None, + extras: dict[str, str] | None = None) -> dict | None: """消费一轮 cc_reply 推进状态机(纯函数,不变异入参;spec §4 答语树)。 参数: @@ -179,6 +180,8 @@ def transit(state: dict | None, reply: dict | None, reply —— None(无回应/extractor 漏填兜底)或 {"affirm": bool, "country": str|None}(extractor 单轮分类,消费后由调用方剥离); new_canon —— 本轮最新号码的纯数字形态;与 state["canon"] 不同 → 换号。 + extras —— misc.name_to_region_extras(Task 9:租户国名映射扩表, + 报国重算的查表经 name_to_region 合并消费)。 返回: None —— RESET(换号或无态):调用方对号码重走 classify 重建状态 @@ -186,7 +189,7 @@ def transit(state: dict | None, reply: dict | None, dict —— 新状态: - affirm=true → confirmed=true(source=ip_confirm;报国重算后的确认 保持 customer_stated,号源如实标注)→ 落定; - - affirm=false + country → NAME_TO_REGION 转 ISO 重算 e164 + - affirm=false + country → NAME_TO_REGION(⊕extras) 转 ISO 重算 e164 (source=customer_stated)→ 再回显,attempts+1; - affirm=false 无 country / 无回应 / 国名不识别 → attempts+1, 达上限(常规 2 轮)落定 confirmed=false、source=fallback。 @@ -210,7 +213,7 @@ def transit(state: dict | None, reply: dict | None, else "ip_confirm") return new if country: - iso = name_to_region(country) + iso = name_to_region(country, extras=extras) recalc = _complete_with_region(canon or "", iso) if iso else None if recalc: new["e164"] = recalc["e164"] @@ -288,21 +291,28 @@ NAME_TO_REGION: dict[str, str] = { "西班牙": "ES", "Spain": "ES", } -# 大小写容错影子表(中文 casefold 为恒等,不受影响) -_NAME_LOOKUP = {k.casefold(): v for k, v in NAME_TO_REGION.items()} - -def name_to_region(name: str | None) -> str | None: +def name_to_region(name: str | None, + extras: dict[str, str] | None = None) -> str | None: """国名(extractor 回传原样字符串)→ ISO;未命中返回 None。 容错:去首尾空白、大小写不敏感、弯撇号(U+2019)归一为 ASCII。 + Task 9: extras = misc.name_to_region_extras(租户扩表),可加键也可 + 覆盖基础表同名键(租户纠偏)——查表序 extras 优先。 """ if not name: return None key = name.strip().replace("’", "'") - if key in NAME_TO_REGION: - return NAME_TO_REGION[key] - return _NAME_LOOKUP.get(key.casefold()) + for table in (extras, NAME_TO_REGION): + if table and key in table: + return table[key] + folded = key.casefold() + for table in (extras, NAME_TO_REGION): + if table: + hit = {k.casefold(): v for k, v in table.items()}.get(folded) + if hit: + return hit + return None # ---- Task 6 出站渲染(spec §5):首封补全号的展示派生(纯函数,邮件/推送共用) ---- diff --git a/app/engine/services/conversation.py b/app/engine/services/conversation.py index 1e8c9ae..d364e8e 100644 --- a/app/engine/services/conversation.py +++ b/app/engine/services/conversation.py @@ -18,8 +18,7 @@ from app.engine.services import ( ) from app.engine.services.ai_service import AIService from app.engine.services.boundary_filter import ( - DEFAULT_WALL, NO_CONTACT_ALREADY, NO_CONTACT_APP, NO_CONTACT_WEB, - asks_contact, deflection_for, no_contact_value, pivot_for, + DEFAULT_WALL, asks_contact, deflection_for, no_contact_value, pivot_for, split_sentences, violates_boundary, ) from app.engine.services.contact_pipeline import ( @@ -32,6 +31,34 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) + +# ---- Task 9: contact 行为开关 / context 种子白名单(config → 引擎消费口) ---- + +def resolve_contact_flags(config_data: dict | None) -> dict: + """租户 config.contact → 联系方式行为开关。 + + cc_completion: 号码区号补全(判定树/确认状态机/CC 强制墙总闸); + email_light_ask: 邮箱轻补(缺失渠道一次性追问)。False = 该行为整体 + 跳过(引擎级门控: 不建状态/不注入/不写计数键)。 + (template_languages 暂无运行期消费方, 不在此解析——Phase 2 后台批次 + 再接。) + """ + contact = ((config_data or {}).get("contact")) or {} + return { + "cc_completion": bool(contact.get("cc_completion", True)), + "email_light_ask": bool(contact.get("email_light_ask", True)), + } + + +def allowed_seed_keys(config_data: dict | None) -> set[str]: + """context 种子白名单 = config.engine.context_seed_keys 的键集合。 + + Task 9 前白名单(product_name/product_id)硬编码在本模块——两租户差异 + (auto 是 vehicle_*)由此配置化;create_conversation 按映射值写 raw_info。 + """ + engine = ((config_data or {}).get("engine")) or {} + return {str(k) for k in (engine.get("context_seed_keys") or {})} + # 职能边界触发词(Task 13.2, 评审#1/#4 修订): 命中即注入红线提醒。精修说明: # - 裸 "how long"/"多久" 不收(规格参数类问题常用),英文要求 take/arrive/deliver/ship 邻接, # 中文只收 到货/多久能到/交货 等明确交付语境 @@ -166,7 +193,7 @@ def _cc_confirm_note(wa_state: dict | None, lang: str) -> str: def _contact_completion_note( raw_info: dict | None, confirmed: dict | None, suspect: dict | None, - lang: str = "en", + lang: str = "en", email_ask: bool = True, ) -> tuple[str, list[str]]: """邮箱轻补提醒(spec §4.2/§4.3b,2026-10-08 判定树设计收编)。 纯函数,返回(注入文案,计数键)。 @@ -177,7 +204,11 @@ def _contact_completion_note( 轻补会让两个"逐字使用这句"同轮竞争(旧 §4.3b 场景①教训);落定即解锁。 带 + 号客户无 _cc 状态 → cc_settled(None)=True,旧轻补行为不回归。 suspect 仅为与调用处画像判定同签名,规则只看 confirmed(spec §4.2)。 + email_ask=False(Task 9: contact.email_light_ask 开关)→ 恒空,不注入 + 也不产计数键。 """ + if not email_ask: + return "", [] info = raw_info or {} numbers = [str(v).strip() for k, v in (confirmed or {}).items() if k in ("whatsapp", "phone") and v] @@ -333,17 +364,23 @@ class ConversationManager: self.db = db # Task 6: runtime 必填(Task 5b 质量审查关门建议)——提示词/墙/会话 # 工厂全部租户化,无租户上下文的构造不再合法;默认 AI 服务用该租户 - # 的 PromptProvider + compiled_wall.deflections 组装($DEFLECTION_POOL - # 单一事实源注入)。email/push 等 settings→config 业务接线是 Task 9。 + # 的 PromptProvider + compiled_wall 组装($DEFLECTION_POOL 单一事实源 + # 注入)。Task 9: AI 凭据/config 全部经 runtime.config.data 接线。 self.runtime = runtime self.ai = ai_service or AIService( prompts=runtime.prompts, - wall_deflections=runtime.compiled_wall.deflections) + wall=runtime.compiled_wall, + config_data=runtime.config.data) + # Task 9: contact 行为开关(cc 补全/邮箱轻补)与国名映射扩表—— + # manager 生命周期内配置不变(reload 换 runtime 即换 manager 装配线) + self._contact_flags = resolve_contact_flags(runtime.config.data) + self._region_extras = (((runtime.config.data.get("misc")) or {}) + .get("name_to_region_extras") or None) @property def session_factory(self): """推送后台任务写结果用的工厂(写回派发租户的库);runtime 未懒建 - engine 时 None,lead_push_service 回退模块级兜底。""" + engine 时 None(仅测试直连场景——生产装配线总在 ensure_ready 之后)。""" return self.runtime.session_factory async def create_conversation( @@ -365,14 +402,17 @@ class ConversationManager: await self.db.refresh(conv) # Conditional whitelist seed of app context into raw_info (red line): - # only when `context` is provided, and only product_name/product_id - # mapped to app_* keys. WS path passes no context -> raw_info stays {}. + # only when `context` is provided, and only keys in the tenant's + # engine.context_seed_keys map (jiancai: product_*; auto: vehicle_* — + # Task 9 前白名单硬编码 product_name/product_id,现读租户配置). + # WS path passes no context -> raw_info stays {}. raw_info: dict = {} if context: - if context.get("product_name"): - raw_info["app_product_name"] = context["product_name"] - if context.get("product_id"): - raw_info["app_product_id"] = context["product_id"] + engine_cfg = self.runtime.config.data.get("engine") or {} + seed_map = engine_cfg.get("context_seed_keys") or {} + for ctx_key, raw_key in seed_map.items(): + if context.get(ctx_key): + raw_info[raw_key] = context[ctx_key] profile = CustomerProfile(conversation_id=conv.id, raw_info=raw_info) self.db.add(profile) await self.db.commit() @@ -446,13 +486,17 @@ class ConversationManager: # _build_context_reminder 之前)——即时扫描并入裸号后同步跑判定树 # (纯计算毫秒级)建/重置/迁移 _cc,当轮即可被墙门控/轮末消费/出站 # 渲染读到;落库与下方一次性计数写回同事务。 + # Task 9: contact.cc_completion=False → 判定树整体跳过(不建不迁 + # _cc;存量 _cc 冻结为只读出站渲染,不再推进)。 _ip_iso = ip_service.resolve_region_code( (conv.client_info or {}).get("ip")) - _cc_changed = _cc_pre_turn(effective_raw, _confirmed_now, _ip_iso) + _cc_changed = (_cc_pre_turn(effective_raw, _confirmed_now, _ip_iso) + if self._contact_flags["cc_completion"] else False) _wa_cc = (effective_raw.get("_cc") or {}).get("whatsapp") context_reminder = self._build_context_reminder( - effective_raw, content, history, source=conv.source.value) + effective_raw, content, history, source=conv.source.value, + contact_flags=self._contact_flags) if CONTACT_ASK_TRIGGER_RE.search(content or ""): logger.info("Contact-ask reminder fired (source=%s)", conv.source.value) @@ -464,7 +508,8 @@ class ConversationManager: # 变更任一存在"(评审 Step 3b)。 _, completion_marks = _contact_completion_note( effective_raw, _confirmed_now, _suspect_now, - "zh" if _CJK_RE.search(content or "") else "en") + "zh" if _CJK_RE.search(content or "") else "en", + email_ask=self._contact_flags["email_light_ask"]) if completion_marks or _cc_changed: raw = dict(profile.raw_info or {}) for key in completion_marks: @@ -506,7 +551,10 @@ class ConversationManager: # 检查,优先级不变)。Task 4(spec §4/§4b): 回显轮(等待态有 e164)裁决 # 锚 = 注入 e164 的 digits 等值在场,替换句 = ECHO 模板;no_ip 轮 # (无可回显号)沿用 ASK 模板与 "country code" 问句短语锚。 - cc_wall = cc_completion.cc_due(_wa_cc) + # Task 9: cc_completion=False → CC 强制墙同样关闭(开关管整个补全 + # 行为族: 判定树/确认流/墙);存量等待态不触发缓冲整段裁决 + cc_wall = (self._contact_flags["cc_completion"] + and cc_completion.cc_due(_wa_cc)) # Task 4 白名单传参(spec §4): 零号码墙逐段放行系统注入的补全号—— # 仅回显轮(等待态有 e164)传,其余轮 None 保持原行为(全拦) cc_allow_number = ((_wa_cc or {}).get("e164") if cc_wall else None) @@ -587,9 +635,12 @@ class ConversationManager: # 价格墙路径(violation=="boundary")不变 closing = _cc_wall_template(_wa_cc, _cc_lang_of(content)) else: - closing = (NO_CONTACT_ALREADY if has_contact - else (NO_CONTACT_APP if conv.source.value == "app" - else NO_CONTACT_WEB)) + # Task 9 收口: 视图直取 DEFAULT_WALL.no_contact(单一 + # 事实源),不再经 boundary_filter 的模块级常量转引 + closing = (DEFAULT_WALL.no_contact["already"] if has_contact + else (DEFAULT_WALL.no_contact["app"] + if conv.source.value == "app" + else DEFAULT_WALL.no_contact["web"])) else: logger.warning( "Boundary filter intercepted sentence (conv=%s): %r | tail=%r", @@ -648,10 +699,11 @@ class ConversationManager: if not violated and dropped_ask and not emitted.strip(): logger.warning("Re-ask wall dropped entire reply (conv=%s), " "falling back to closing", conv.id) - for i in range(0, len(NO_CONTACT_ALREADY), 40): + fallback = DEFAULT_WALL.no_contact["already"] + for i in range(0, len(fallback), 40): yield {"type": "stream_chunk", - "content": NO_CONTACT_ALREADY[i:i + 40]} - emitted = NO_CONTACT_ALREADY + "content": fallback[i:i + 40]} + emitted = fallback yield {"type": "stream_end"} ai_reply = emitted # 落库与流出统一为已流出文本(下方持久化) @@ -797,8 +849,11 @@ class ConversationManager: # 询盘推送(设计 8.1): 主事务提交后才派生后台任务——payload 内存 # 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞。 - # session_factory 随任务传递: 写回本租户的库(Task 5b) - lead_push_service.fire_pending_pushes(pending_pushes, self.session_factory) + # session_factory 随任务传递: 写回本租户的库(Task 5b);config_data + # 同随任务传递(Task 9: 重试矩阵参数走租户配置) + lead_push_service.fire_pending_pushes( + pending_pushes, self.session_factory, + config_data=self.runtime.config.data) yield { "type": "evaluated", @@ -880,7 +935,11 @@ class ConversationManager: - canon 变化(换号/号消失)→ RESET 删组状态。 - 补建 pass: 对最终联系方式重跑判定树(_cc_pre_turn)——本轮才经 extractor 入管道的裸号/换号后的新号当场建状态,不下轮补课。 + - Task 9: contact.cc_completion=False → 整钩子跳过(轮前判定树 + 已不建态,此处不消费不推进;国名映射查表带租户 extras)。 """ + if not self._contact_flags["cc_completion"]: + return original_cc = (profile.raw_info or {}).get("_cc") ip_iso = ip_service.resolve_region_code( (conv.client_info or {}).get("ip")) @@ -914,7 +973,8 @@ class ConversationManager: if (reply is None and not first_ask and cc_completion.cc_due(state)): signal = _country_signal(raw.get("目标国家"), pre_turn_country) - if signal and cc_completion.name_to_region(signal): + if signal and cc_completion.name_to_region( + signal, extras=self._region_extras): reply = {"affirm": False, "country": signal} if state is not None: @@ -923,7 +983,8 @@ class ConversationManager: elif first_ask: cc_map["whatsapp"] = {**state, "asked": True} # 首问轮免计 elif reply is not None or cc_completion.cc_due(state): - cc_map["whatsapp"] = cc_completion.transit(state, reply) + cc_map["whatsapp"] = cc_completion.transit( + state, reply, extras=self._region_extras) if cc_map: raw["_cc"] = cc_map else: @@ -956,6 +1017,7 @@ class ConversationManager: if not ev.email_sent: # S3: 任一值已见于24h内其他会话 → 轻提醒;否则全量。 # Task 6 双保险: 比对候选集含 _cc.e164 的 canon(见 _channel_canons) + # Task 9: 邮件/推送参数走租户配置(config_data 透传给 resolve_*) seen = await self._recently_emailed_canons() if any(cands & seen for cands in _channel_canons(confirmed, cc_map).values()): @@ -963,7 +1025,8 @@ class ConversationManager: pending_pushes, ev, conv, profile, confirmed, "returning", evaluation, evaluated) sent = await email_service.send_returning_lead_email( - confirmed, cc_info=cc_map) + confirmed, cc_info=cc_map, + config_data=self.runtime.config.data) else: await self._collect_push( pending_pushes, ev, conv, profile, confirmed, @@ -976,6 +1039,7 @@ class ConversationManager: conversation_messages=messages, intent_level=evaluation.get("intent_level", "") if evaluated else "", cc_info=cc_map, + config_data=self.runtime.config.data, ) if sent: ev.email_sent = True @@ -993,7 +1057,8 @@ class ConversationManager: pending_pushes, ev, conv, profile, confirmed, "correction", evaluation, evaluated) if await email_service.send_correction_email( - confirmed, count=count, cc_info=cc_map): + confirmed, count=count, cc_info=cc_map, + config_data=self.runtime.config.data): ev.correction_count = count ev.contact_info = _panel_contact_info(confirmed, cc_map) self.db.add(ev) @@ -1009,11 +1074,13 @@ class ConversationManager: ) -> None: """询盘推送派发(设计 §8): 触发条件=进入发信分支,与 SMTP 成败解耦。 - dispatch 内部吞异常(失败只记日志),这里无需再兜。 + dispatch 内部吞异常(失败只记日志),这里无需再兜。config_data = + 租户配置(Task 9: 开关/前缀走 config, 空槽回退 env)。 """ log = await lead_push_service.dispatch_lead_push( self.db, ev, conv, profile, confirmed, push_type, evaluation, evaluated, + config_data=self.runtime.config.data, ) if log is not None and pending is not None: pending.append(log) @@ -1070,13 +1137,17 @@ class ConversationManager: @staticmethod def _build_context_reminder(raw_info: dict, latest_msg: str, history: list[dict] | None = None, - source: str = "") -> str: + source: str = "", + contact_flags: dict | None = None) -> str: """Build a context reminder from extracted profile info. Task 4 词表数据化: 六类轮中提醒话术(boundary/contact_ask/ language_follow/lead_given/cc_guide/email_light)的数据源 = DEFAULT_WALL.reminders(seed_profiles/jiancai/wall.json); {shown}/{no_contact_web}/{no_contact_already} 占位由本函数 - .format() 回填,值与数据化前逐字等价。""" + .format() 回填,值与数据化前逐字等价。 + contact_flags(Task 9): contact 行为开关——cc_completion=False 跳过 + cc 确认指引,email_light_ask=False 跳过邮箱轻补;None = 全开 + (直调本函数的旧测试语义)。""" # 职能边界提醒(评审#7: 前置到早退之前,首条消息画像为空也要生效) boundary_note = "" if BOUNDARY_TRIGGER_RE.search(latest_msg or ""): @@ -1107,7 +1178,7 @@ class ConversationManager: contact_note = DEFAULT_WALL.reminders["contact_ask"]["app"] else: contact_note = DEFAULT_WALL.reminders["contact_ask"]["web"].format( - no_contact_web=NO_CONTACT_WEB) + no_contact_web=DEFAULT_WALL.no_contact["web"]) # 语言跟随提醒(Task 11): 与 boundary/contact 提醒同挂载位—— # 首条消息画像为空也要生效(中文客户开场即漂英文是实测重灾区) @@ -1126,6 +1197,11 @@ class ConversationManager: if val: parts.append(f"{key}: {val}") + # Task 9: contact 行为开关注入本函数(staticmethod 保持旧直调签名, + # None = 全开);cc 确认指引与邮箱轻补按开关短路 + flags = (contact_flags if contact_flags is not None + else resolve_contact_flags(None)) + # Task 3(spec §4.2/§4.3b): 邮箱轻补注入(cc 分支已删,判定树接管号码 # 区号补全——确认提醒由 cc_due 驱动,Task 4 接)。计数键(_marks)此处 # 弃用: 本函数是 staticmethod 收内存副本,写回由 stream 调用方以同参 @@ -1133,28 +1209,31 @@ class ConversationManager: # confirmed 块之前: 未落定轮邮箱轻补不注入(cc_settled 串行)。 completion_note, _marks = _contact_completion_note( raw_info, confirmed, suspect, - lang="zh" if _CJK_RE.search(latest_msg or "") else "en") + lang="zh" if _CJK_RE.search(latest_msg or "") else "en", + email_ask=flags["email_light_ask"]) # Task 4(spec §4 回显确认轮): cc 等待态注入确认句指引——有 e164 用 # 回显模板(逐字 intl),no_ip 用 ASK 追问句;与邮箱轻补 cc_settled # 串行互斥,两个"逐字使用这句"永不同轮。无计数键: 轮次上限由状态机 # attempts 管理(轮末 transit 递增),提醒每轮等待态都在场。 # P3: 语言路由扩 ES(_cc_lang_of),西语轮注入西语模板。 - cc_note = _cc_confirm_note( + cc_note = (_cc_confirm_note( (raw_info.get("_cc") or {}).get("whatsapp"), _cc_lang_of(latest_msg)) + if flags["cc_completion"] else "") # 联系方式专项(D6) 与 职能边界(D6联动) if confirmed: shown = ", ".join( f"{k}: {' '.join(str(v).split())[:60]}" for k, v in confirmed.items()) - # 逐字话术直接引用 NO_CONTACT_ALREADY 常量(Task 5 已 import)—— - # 与零号码墙替换句同源,三处"经理主动联系"话术不漂移 + # 逐字话术直接引用 DEFAULT_WALL.no_contact["already"](Task 9 收口: + # 视图直取单一事实源)——与零号码墙替换句同源,三处"经理主动联系" + # 话术不漂移 # Task 3 防自锁改写(spec 4.1-6): 原"绝对禁止再索要任何联系方式!" # 会抑制缺失渠道轻补——只禁已提供过的渠道,缺失渠道放行一次 parts.append(DEFAULT_WALL.reminders["lead_given"]["forbidden"].format( shown=shown)) parts.append(DEFAULT_WALL.reminders["lead_given"]["respond_script"].format( - no_contact_already=NO_CONTACT_ALREADY)) + no_contact_already=DEFAULT_WALL.no_contact["already"])) if boundary_note: parts.append(boundary_note) # Task 3 防自锁改写: 尾部补缺失渠道豁免(与上方改写同因) diff --git a/app/engine/services/email_service.py b/app/engine/services/email_service.py index 917f08a..2ba11a8 100644 --- a/app/engine/services/email_service.py +++ b/app/engine/services/email_service.py @@ -16,6 +16,33 @@ TAG_LABELS = { } CONTACT_LABELS = {"phone": "手机", "whatsapp": "WhatsApp", "email": "邮箱", "wechat": "微信"} +# 平台默认发件人显示名(Task 9 前的硬编码值, 租户未配 sender_name 时沿用) +_DEFAULT_SENDER_NAME = "HESUPIPE" + + +def resolve_email_settings(config_data: dict | None) -> dict: + """租户 config.email → 发信参数(Task 9 接线; env settings 回退)。 + + 回退规则(语义钉在 test_config_merge: 显式 None = 清空/未设置): + - smtp 缺省或显式 None → 整段用 env SMTP_*; + - smtp 给了 dict → 段内逐键回退(缺键/空串 → env 对应值, 租户可只改 + 密码/只换 host); + - sender_name 空 → 平台默认(_DEFAULT_SENDER_NAME); + - notify_emails 空列表 → env NOTIFY_EMAIL 单元素。 + """ + email = ((config_data or {}).get("email")) or {} + smtp = email.get("smtp") or {} + notify = [e for e in (email.get("notify_emails") or []) if e] + return { + "host": smtp.get("host") or settings.SMTP_HOST, + "port": smtp.get("port") or settings.SMTP_PORT, + "user": smtp.get("user") or settings.SMTP_USER, + "password": smtp.get("password") or settings.SMTP_PASS, + "sender_name": email.get("sender_name") or _DEFAULT_SENDER_NAME, + "notify_emails": notify or ( + [settings.NOTIFY_EMAIL] if settings.NOTIFY_EMAIL else []), + } + def _esc(v) -> str: return html.escape(str(v)) if v else "" @@ -57,17 +84,17 @@ def _cc_views(contacts: dict, cc_info: dict | None) -> tuple[dict, dict]: return body, subj -async def _send(msg: MIMEMultipart) -> bool: - if not all([settings.SMTP_HOST, settings.SMTP_USER, settings.NOTIFY_EMAIL]): +async def _send(msg: MIMEMultipart, cfg: dict) -> bool: + if not all([cfg["host"], cfg["user"], cfg["notify_emails"]]): logger.warning("SMTP not configured, skipping email") return False try: await aiosmtplib.send( - msg, hostname=settings.SMTP_HOST, port=settings.SMTP_PORT, - username=settings.SMTP_USER, password=settings.SMTP_PASS, + msg, hostname=cfg["host"], port=cfg["port"], + username=cfg["user"], password=cfg["password"], use_tls=True, ) - logger.info("Lead email sent to %s", settings.NOTIFY_EMAIL) + logger.info("Lead email sent to %s", cfg["notify_emails"]) return True except Exception: logger.exception("Failed to send lead email") @@ -81,9 +108,12 @@ async def send_lead_email( conversation_messages: list[dict], intent_level: str = "", cc_info: dict | None = None, + config_data: dict | None = None, ) -> bool: """cc_info = raw_info["_cc"](Task 6, spec §5):正文联系方式行渲染 - 补全号+标注、主题行号码用 e164 无标注;缺省(无 _cc/显式 + 号)行为不变。""" + 补全号+标注、主题行号码用 e164 无标注;缺省(无 _cc/显式 + 号)行为不变。 + config_data = 租户配置(Task 9):邮件参数经 resolve_email_settings 解析, + None 走 env settings(平台默认)。""" tag = TAG_LABELS.get(intent_level, "未评估") contacts, subj_contacts = _cc_views(contact_info, cc_info) country = (customer_profile or {}).get("目标国家", "") @@ -97,9 +127,10 @@ async def send_lead_email( subject = f"[新询盘·{tag}] {_flat(country)}" if contact_parts: subject += " | " + ", ".join(contact_parts) + cfg = resolve_email_settings(config_data) msg["Subject"] = subject - msg["From"] = f"【HESUPIPE】 <{settings.SMTP_USER}>" - msg["To"] = settings.NOTIFY_EMAIL + msg["From"] = f"【{cfg['sender_name']}】 <{cfg['user']}>" + msg["To"] = ", ".join(cfg["notify_emails"]) info_rows = "".join( f"{_esc(k)}" @@ -125,39 +156,43 @@ async def send_lead_email( """ msg.attach(MIMEText(html_body, "html", "utf-8")) - return await _send(msg) + return await _send(msg, cfg) async def send_returning_lead_email( contact_info: dict, cc_info: dict | None = None, + config_data: dict | None = None, ) -> bool: """D5: 24h 内同联系方式再访,只发一行轻提醒。主题行号码走 e164(Task 6)。""" _body, subj_contacts = _cc_views(contact_info, cc_info) contact_parts = [ f"{CONTACT_LABELS.get(k, k)}: {_flat(v)}" for k, v in subj_contacts.items() ] + cfg = resolve_email_settings(config_data) msg = MIMEMultipart("alternative") msg["Subject"] = f"[线索回访] {' | '.join(contact_parts) or '已知客户'}再次来访" - msg["From"] = f"【HESUPIPE】 <{settings.SMTP_USER}>" - msg["To"] = settings.NOTIFY_EMAIL + msg["From"] = f"【{cfg['sender_name']}】 <{cfg['user']}>" + msg["To"] = ", ".join(cfg["notify_emails"]) msg.attach(MIMEText( f"

该联系方式 24h 内已发过完整询盘,客户又开启了新对话," f"详情请查后台。

", "html", "utf-8")) - return await _send(msg) + return await _send(msg, cfg) async def send_correction_email( contact_info: dict, count: int = 1, cc_info: dict | None = None, + config_data: dict | None = None, ) -> bool: """D4: 联系方式变化(改号/新增/无提示换值)后的轻量更正邮件,每会话上限3封。 正文联系方式行渲染补全号+标注(Task 6, spec §5 更正路径)。""" body_contacts, _subj = _cc_views(contact_info, cc_info) + cfg = resolve_email_settings(config_data) msg = MIMEMultipart("alternative") msg["Subject"] = f"[询盘·更正{count}] 客户联系方式已更新(以本封为准)" - msg["From"] = f"【HESUPIPE】 <{settings.SMTP_USER}>" - msg["To"] = settings.NOTIFY_EMAIL + msg["From"] = f"【{cfg['sender_name']}】 <{cfg['user']}>" + msg["To"] = ", ".join(cfg["notify_emails"]) msg.attach(MIMEText( f"

这是本会话第 {count} 次更正(上限 3 次),请以本封为准:

" f"{_contact_rows(body_contacts)}
" f"", "html", "utf-8")) - return await _send(msg) + return await _send(msg, cfg) diff --git a/app/engine/services/lead_push_service.py b/app/engine/services/lead_push_service.py index afe9591..9d1dbab 100644 --- a/app/engine/services/lead_push_service.py +++ b/app/engine/services/lead_push_service.py @@ -8,7 +8,12 @@ 后台任务红线(设计 8.1): 不读库只写结果,payload 内存传递; 结果写入用独立 session——session_factory 由派发方传入(Task 5b 起无全局 -单例,写回派发它的那个租户的库)。 +单例,写回派发它的那个租户的库;Task 9 收口: 转必填,漏传在调用点即 +TypeError,不再拖到任务深处)。 + +Task 9 配置接线: PUSH 组(开关/URL/密钥/前缀/超时/重试)经 +resolve_push_settings 从租户 config 解析,空槽回退 env settings—— +各入口收 config_data(租户配置 dict),None = 平台默认(env)。 """ import asyncio import logging @@ -28,11 +33,26 @@ from app.engine.services.ip_service import resolve_country logger = logging.getLogger(__name__) -# Task 5b 装配切换: app.main 全局单例已删,写库入口改为显式传 session_factory -# (租户正确: 后台任务写结果必须落回派发租户的库)。模块级 async_session 仅作 -# 无工厂上下文时的兜底(测试直接驱动 _run_push/sweep_stale_pending 时注入), -# 生产装配不设值——装配线(main.lifespan/ConversationManager)总是显式传。 -async_session = None + +def resolve_push_settings(config_data: dict | None) -> dict: + """租户 config.push → 推送参数(Task 9 接线; env settings 回退)。 + + 出厂默认的空槽(False/空串)视同"未设置"→ 回退 env PUSH_*——部署级 + .env 开关是平台默认集的一部分,租户未配置时继承,不静默关停。 + 已知边界: env 开 + 租户显式 false 无法区分于未配置(布尔空槽只有 + False),Phase 2 admin 落地时以"平台清 env、配置显式写 false"解决。 + """ + p = ((config_data or {}).get("push")) or {} + return { + "enabled": bool(p.get("enabled") or settings.PUSH_ENABLED), + "url": p.get("url") or settings.PUSH_API_URL, + "api_key": p.get("api_key") or settings.PUSH_API_KEY, + "prefix": p.get("prefix") or settings.PUSH_EXTERNAL_KEY_PREFIX, + "timeout": p.get("timeout") or settings.PUSH_TIMEOUT_SECONDS, + "max_attempts": p.get("max_attempts") or settings.PUSH_MAX_ATTEMPTS, + "retry_backoff_base": (p.get("retry_backoff_base") + or settings.PUSH_RETRY_BACKOFF_BASE), + } # 对方契约(docs/AI_INQUIRY_PUSH_API.md): phone/whatsapp/email 至少一个有效 _TRIO = ("phone", "whatsapp", "email") @@ -70,6 +90,7 @@ async def _build_payload( db: AsyncSession, ev: Evaluation, conv: Conversation, profile: CustomerProfile, confirmed: dict, push_type: str, evaluation: dict, evaluated: bool, + config_data: dict | None = None, ) -> dict: """组装契约报文(设计 7.2 映射表)。仅读当前事务,不产生副作用。""" # 7.1 event 规则: 仅更正可能 update,且必须同会话此前推送成功过 @@ -125,7 +146,7 @@ async def _build_payload( return { "event": event, - "external_key": f"{settings.PUSH_EXTERNAL_KEY_PREFIX}-{conv.id}", + "external_key": f"{resolve_push_settings(config_data)['prefix']}-{conv.id}", "occurred_at": _fmt_cst(datetime.now(timezone.utc)), "contact": { "name": None, # 我方 contact_info 快照无称呼字段,契约可空 @@ -153,6 +174,7 @@ async def dispatch_lead_push( db: AsyncSession, ev: Evaluation, conv: Conversation, profile: CustomerProfile, confirmed: dict, push_type: str, evaluation: dict, evaluated: bool, + config_data: dict | None = None, ) -> LeadPushLog | None: """触发点唯一入口: 返回待派发的 pending 行(加入当前事务,随主事务提交)。 @@ -160,8 +182,10 @@ async def dispatch_lead_push( 触发条件 = 调用方进入发信分支,与 SMTP 成败解耦。 DB 层失败用 SAVEPOINT 隔离: 只回滚推送写入,不毒化调用方事务 (否则 flush 失败 → PendingRollbackError → email_sent 落不了库 → 重复发邮件)。 + config_data = 租户配置(Task 9): 开关/前缀经 resolve_push_settings, + None 走 env settings(平台默认)。 """ - if not settings.PUSH_ENABLED: + if not resolve_push_settings(config_data)["enabled"]: return None try: async with db.begin_nested(): @@ -174,7 +198,8 @@ async def dispatch_lead_push( )) return None payload = await _build_payload( - db, ev, conv, profile, confirmed, push_type, evaluation, evaluated) + db, ev, conv, profile, confirmed, push_type, evaluation, + evaluated, config_data) log = LeadPushLog( evaluation_id=ev.id, conversation_id=conv.id, push_type=push_type, status="pending", attempts=0, @@ -189,19 +214,22 @@ async def dispatch_lead_push( return None -async def _do_post(payload: dict) -> tuple[httpx.Response | None, str | None]: +async def _do_post(payload: dict, + config_data: dict | None = None) -> tuple[httpx.Response | None, str | None]: """实际 HTTP 调用(独立函数便于测试 monkeypatch)。 返回 (resp, transport_error): 传输层异常不抛出,转成错误串由重试矩阵消化。 + URL/密钥/超时经 resolve_push_settings(Task 9: 租户配置, None 回退 env)。 """ + cfg = resolve_push_settings(config_data) try: async with httpx.AsyncClient( - timeout=settings.PUSH_TIMEOUT_SECONDS, + timeout=cfg["timeout"], follow_redirects=True) as client: # 307/308 不落终态失败 resp = await client.post( - settings.PUSH_API_URL, + cfg["url"], json={"messages": [payload]}, # 契约 §2: 批量结构,我方单条 - headers={"X-Api-Key": settings.PUSH_API_KEY}, + headers={"X-Api-Key": cfg["api_key"]}, ) return resp, None except httpx.HTTPError as exc: @@ -226,26 +254,27 @@ async def _update_log(session_factory, push_log_id: str, **fields) -> None: async def _run_push(push_log_id: str, payload: dict, - session_factory=None) -> None: + session_factory, config_data: dict | None = None) -> None: """后台任务体: 分类重试 + 200 逐条结果解析(设计 6.3/§10)。 红线: 全套 try/except,异常只记日志绝不逃逸(fire-and-forget)。 - session_factory 缺省回退模块级 async_session(测试注入缝)。 + session_factory 必填(Task 9 收口): 派发租户的工厂,写回同一租户库。 + 重试矩阵参数(上限/退避)经 resolve_push_settings(租户配置, None 回退 env)。 """ - factory = session_factory or async_session + cfg = resolve_push_settings(config_data) try: attempts = 0 while True: attempts += 1 - resp, transport_error = await _do_post(payload) - if _retryable(resp) and attempts < settings.PUSH_MAX_ATTEMPTS: + resp, transport_error = await _do_post(payload, config_data) + if _retryable(resp) and attempts < cfg["max_attempts"]: await _update_log( - factory, push_log_id, attempts=attempts, + session_factory, push_log_id, attempts=attempts, error_message=transport_error or (f"http {resp.status_code}" if resp is not None else "unknown"), ) await _sleep( - settings.PUSH_RETRY_BACKOFF_BASE * 2 ** (attempts - 1)) + cfg["retry_backoff_base"] * 2 ** (attempts - 1)) continue break @@ -255,7 +284,7 @@ async def _run_push(push_log_id: str, payload: dict, except ValueError: # 非 JSON 的 200(如网关重定向到 HTML 登录页)—— # 无法确认对方已接受,诚实落 failed(幂等键保证补推安全) - await _update_log(factory, push_log_id, attempts=attempts, + await _update_log(session_factory, push_log_id, attempts=attempts, status="failed", response_status=200, error_message="non-JSON 200 body") @@ -268,8 +297,8 @@ async def _run_push(push_log_id: str, payload: dict, ] if failed_mine: await _update_log( - factory, push_log_id, attempts=attempts, status="failed", - response_status=200, + session_factory, push_log_id, attempts=attempts, + status="failed", response_status=200, error_message=str(failed_mine[0].get("reason") or "rejected"), ) else: @@ -277,8 +306,8 @@ async def _run_push(push_log_id: str, payload: dict, outcome = ("duplicated" if (data.get("duplicated") or 0) > 0 else "accepted") await _update_log( - factory, push_log_id, attempts=attempts, status="success", - response_status=200, outcome=outcome, + session_factory, push_log_id, attempts=attempts, + status="success", response_status=200, outcome=outcome, ) return @@ -288,7 +317,7 @@ async def _run_push(push_log_id: str, payload: dict, "密钥/报文问题需人工介入,不重试", push_log_id, resp.status_code) await _update_log( - factory, push_log_id, attempts=attempts, status="failed", + session_factory, push_log_id, attempts=attempts, status="failed", response_status=resp.status_code if resp is not None else None, error_message=transport_error or (f"http {resp.status_code}" if resp is not None else "unknown"), @@ -301,34 +330,33 @@ async def _run_push(push_log_id: str, payload: dict, _tasks: set[asyncio.Task] = set() -def fire_pending_pushes(pushes: list[LeadPushLog], session_factory=None) -> None: +def fire_pending_pushes(pushes: list[LeadPushLog], session_factory, + config_data: dict | None = None) -> None: """主事务 commit 之后调用(设计 8.1): payload 内存传递,任务不读库。 - 同步函数、纳秒级返回——这就是"不阻塞"的派发点。session_factory = - 派发租户的工厂,随任务传递(后台写结果落回同一租户库);缺省回退 - 模块级 async_session(测试注入缝)。 + 同步函数、纳秒级返回——这就是"不阻塞"的派发点。session_factory + 必填(Task 9 收口): 派发租户的工厂,随任务传递(后台写结果落回同一 + 租户库);config_data 同随任务传递(重试参数走租户配置)。 """ - factory = session_factory or async_session for log in pushes: if log.status != "pending": continue task = asyncio.create_task( - _run_push(log.id, log.request_payload, factory)) + _run_push(log.id, log.request_payload, session_factory, + config_data=config_data)) _tasks.add(task) task.add_done_callback(_tasks.discard) -async def sweep_stale_pending(stale_minutes: int = 30, session_factory=None) -> int: +async def sweep_stale_pending(session_factory, stale_minutes: int = 30) -> int: """启动清扫(设计 8.2): 崩溃遗留的 pending → failed。 - session_factory: 目标租户的工厂(lifespan 逐租户调用);缺省回退 - 模块级 async_session(测试注入缝)。 + session_factory 必填(Task 9 收口): 目标租户的工厂(lifespan 逐租户调用)。 已知限制(设计 §14): 推送成功但结果写回失败也会被标 failed—— 幂等键保证将来补推安全(对方返回 duplicated),概率极低,接受。 """ - factory = session_factory or async_session cutoff = datetime.now(timezone.utc) - timedelta(minutes=stale_minutes) - async with factory() as session: + async with session_factory() as session: result = await session.execute( update(LeadPushLog).where( LeadPushLog.status == "pending", diff --git a/app/engine/services/rate_limit.py b/app/engine/services/rate_limit.py index 9da7289..5f68eea 100644 --- a/app/engine/services/rate_limit.py +++ b/app/engine/services/rate_limit.py @@ -1,32 +1,36 @@ -"""Minimal in-memory rate limiter (per (ip, device_id), rolling 1h window). +"""Minimal in-memory rate limiter (per (tenant, ip, device_id), rolling 1h window). Kept aligned with the reference implementation's rate_limit.py shape. Only the chat message route uses this; session/history endpoints are cheap and not throttled. State is process-local and resets on restart, which is acceptable for a single-worker ai-kefu deploy. +Task 9: 桶键加租户维度——多租户同 IP(共享出口 NAT/公司网络)不互相挤占 +配额;limit 经 config.engine.rate_limit_per_hour 逐租户可调(消费点 +chat_rest, 默认 DEFAULT_MAX)。 Caveats: - Single-process, in-memory. Under `uvicorn --workers >1` each worker keeps its own `_buckets`, so the effective cap is `max_requests * workers`. -- Keyed on (ip, device_id); a client rotating the `X-Device-Id` header per +- Keyed on (tenant, ip, device_id); a client rotating the `X-Device-Id` header per request bypasses it. This is a light safeguard against accidental GLM-cost burn, NOT abuse-proof. """ import time from collections import defaultdict -_buckets: dict[tuple[str, str], list[float]] = defaultdict(list) +_buckets: dict[tuple[str, str, str], list[float]] = defaultdict(list) WINDOW = 3600 DEFAULT_MAX = 30 def check( + tenant: str | None, ip: str | None, device_id: str | None, *, max_requests: int = DEFAULT_MAX, ) -> bool: """Return True if the request is under the cap (and record it), else False.""" - key = (ip or "", device_id or "") + key = (tenant or "", ip or "", device_id or "") now = time.time() _buckets[key] = [t for t in _buckets[key] if t > now - WINDOW] if len(_buckets[key]) >= max_requests: diff --git a/tests/conftest.py b/tests/conftest.py index bdd7ef6..5a5e8c8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -93,8 +93,8 @@ def build_jiancai_runtime(): 只走加载期(config/prompts/compiled_wall 就位),从不 ensure_ready—— engine/session_factory 保持 None: 直连测试自管 async_db(NullPool 跨 - 循环纪律),manager.session_factory 为 None 时 lead_push_service 回退 - 模块级兜底,与旧 runtime=None 行为一致。无 engine 即无句柄,无需清理; + 循环纪律;Task 9 起 lead_push_service 无模块级 session 兜底,session_factory + 由调用方/测试显式传)。无 engine 即无句柄,无需清理; data_dir 仅占位(只拼 db_path 字符串,无任何 IO——从不 ensure_ready, 不会创建 tenants/ 目录或库文件),模块级缓存单例跨测试复用。 """ diff --git a/tests/test_email_rendering.py b/tests/test_email_rendering.py index 76dfe91..7e330f4 100644 --- a/tests/test_email_rendering.py +++ b/tests/test_email_rendering.py @@ -7,7 +7,7 @@ from app.engine.services import email_service def _capture_send(monkeypatch): captured = {} - async def fake_send(msg): + async def fake_send(msg, cfg=None): captured["msg"] = msg return True diff --git a/tests/test_lead_email_trigger.py b/tests/test_lead_email_trigger.py index f733c78..102cd98 100644 --- a/tests/test_lead_email_trigger.py +++ b/tests/test_lead_email_trigger.py @@ -60,15 +60,16 @@ def _setup(async_db, monkeypatch, source=ConversationSource.widget, ai=None): sent = [] async def fake_lead(customer_profile, contact_info, reasoning, - conversation_messages, intent_level="", cc_info=None): + conversation_messages, intent_level="", cc_info=None, + config_data=None): sent.append(("lead", dict(contact_info), intent_level)) return True - async def fake_nudge(contact_info, cc_info=None): + async def fake_nudge(contact_info, cc_info=None, config_data=None): sent.append(("nudge", dict(contact_info), "")) return True - async def fake_fix(contact_info, count=1, cc_info=None): + async def fake_fix(contact_info, count=1, cc_info=None, config_data=None): sent.append(("fix", dict(contact_info), "")) return True diff --git a/tests/test_lead_push_service.py b/tests/test_lead_push_service.py index e6e9ce1..2972540 100644 --- a/tests/test_lead_push_service.py +++ b/tests/test_lead_push_service.py @@ -335,19 +335,24 @@ def _mk_push_log(async_db, status="pending"): def _run_push_case(async_db, monkeypatch, do_post, log_id): - """注入 _do_post/_sleep/async_session 后驱动 _run_push,返回退避序列。""" + """注入 _do_post/_sleep 后驱动 _run_push(Task 9: session_factory 直传, + 不再借模块级 async_session——已删除),返回退避序列。""" from app.engine.services import lead_push_service as svc sleeps = [] async def fake_sleep(s): sleeps.append(s) - monkeypatch.setattr(svc, "_do_post", do_post) + # Task 9: _do_post 多了 config_data 参数——适配层吸收,各用例 fake 保持单参 + async def adapted_do_post(payload, config_data=None): + return await do_post(payload) + + monkeypatch.setattr(svc, "_do_post", adapted_do_post) monkeypatch.setattr(svc, "_sleep", fake_sleep) - monkeypatch.setattr(svc, "async_session", async_db) async def run(): - await svc._run_push(log_id, {"external_key": "ai-kefu-x", "event": "new"}) + await svc._run_push( + log_id, {"external_key": "ai-kefu-x", "event": "new"}, async_db) asyncio.run(run()) return sleeps @@ -482,7 +487,7 @@ def test_fire_pending_creates_tasks(monkeypatch): from app.engine.services import lead_push_service as svc pushed = [] - async def fake_run(log_id, payload, session_factory=None): + async def fake_run(log_id, payload, session_factory=None, config_data=None): pushed.append((log_id, payload)) monkeypatch.setattr(svc, "_run_push", fake_run) @@ -494,7 +499,7 @@ def test_fire_pending_creates_tasks(monkeypatch): ] async def run(): - svc.fire_pending_pushes(logs) + svc.fire_pending_pushes(logs, None) await asyncio.sleep(0) # 同 loop 内让 create_task 跑一轮 asyncio.run(run()) assert pushed == [("id-pending", {"event": "new"})] @@ -503,7 +508,6 @@ def test_fire_pending_creates_tasks(monkeypatch): def test_sweep_stale_pending(async_db, monkeypatch): from datetime import datetime, timezone from app.engine.services import lead_push_service as svc - monkeypatch.setattr(svc, "async_session", async_db) _seed_push_ctx(async_db) async def run(): @@ -524,7 +528,7 @@ def test_sweep_stale_pending(async_db, monkeypatch): updated_at=datetime(2020, 1, 1, tzinfo=timezone.utc)) db.add_all([stale, fresh, done]) await db.commit() - n = await svc.sweep_stale_pending(stale_minutes=30) + n = await svc.sweep_stale_pending(async_db, stale_minutes=30) assert n == 1 async with async_db() as db: logs = list((await db.execute(select(LeadPushLog))).scalars()) diff --git a/tests/test_lead_push_wiring.py b/tests/test_lead_push_wiring.py index 5620e58..6053975 100644 --- a/tests/test_lead_push_wiring.py +++ b/tests/test_lead_push_wiring.py @@ -33,15 +33,16 @@ def _setup(async_db, monkeypatch, ai, lead_succeeds=True): sent = [] async def fake_lead(customer_profile, contact_info, reasoning, - conversation_messages, intent_level="", cc_info=None): + conversation_messages, intent_level="", cc_info=None, + config_data=None): sent.append(("lead", dict(contact_info))) return lead_succeeds - async def fake_nudge(contact_info, cc_info=None): + async def fake_nudge(contact_info, cc_info=None, config_data=None): sent.append(("returning", dict(contact_info))) return lead_succeeds - async def fake_fix(contact_info, count=1, cc_info=None): + async def fake_fix(contact_info, count=1, cc_info=None, config_data=None): sent.append(("correction", dict(contact_info))) return lead_succeeds @@ -51,7 +52,8 @@ def _setup(async_db, monkeypatch, ai, lead_succeeds=True): fired = [] monkeypatch.setattr(lead_push_service, "fire_pending_pushes", - lambda logs, session_factory=None: fired.extend(logs)) + lambda logs, session_factory=None, config_data=None: + fired.extend(logs)) monkeypatch.setattr(lead_push_service.settings, "PUSH_ENABLED", True) async def run(): @@ -160,14 +162,16 @@ def test_push_disabled_writes_nothing(async_db, monkeypatch): sent = [] async def fake_lead(customer_profile, contact_info, reasoning, - conversation_messages, intent_level="", cc_info=None): + conversation_messages, intent_level="", cc_info=None, + config_data=None): sent.append("lead") return True monkeypatch.setattr(email_service, "send_lead_email", fake_lead) fired = [] monkeypatch.setattr(lead_push_service, "fire_pending_pushes", - lambda logs, session_factory=None: fired.extend(logs)) + lambda logs, session_factory=None, config_data=None: + fired.extend(logs)) async def run(): async with async_db() as db: @@ -186,7 +190,7 @@ def test_fire_end_to_end_single_loop(async_db, monkeypatch): """真 fire + 假 _run_push: 同一事件循环内 stream 完成后任务确实被派发执行。""" pushed = [] - async def fake_run(log_id, payload, session_factory=None): + async def fake_run(log_id, payload, session_factory=None, config_data=None): pushed.append(log_id) monkeypatch.setattr(lead_push_service, "_run_push", fake_run) @@ -195,7 +199,8 @@ def test_fire_end_to_end_single_loop(async_db, monkeypatch): sent = [] async def fake_lead(customer_profile, contact_info, reasoning, - conversation_messages, intent_level="", cc_info=None): + conversation_messages, intent_level="", cc_info=None, + config_data=None): sent.append("lead") return True @@ -234,11 +239,12 @@ def test_fire_happens_only_after_commit(async_db, monkeypatch): fired = [] async def fake_lead(customer_profile, contact_info, reasoning, - conversation_messages, intent_level="", cc_info=None): + conversation_messages, intent_level="", cc_info=None, + config_data=None): sent.append("lead") return True - async def fake_run(log_id, payload, session_factory=None): + async def fake_run(log_id, payload, session_factory=None, config_data=None): async with async_db() as db: try: row = (await db.execute( @@ -253,9 +259,9 @@ def test_fire_happens_only_after_commit(async_db, monkeypatch): real_fire = lead_push_service.fire_pending_pushes - def recording_fire(logs, session_factory=None): + def recording_fire(logs, session_factory=None, config_data=None): fired.append(True) - real_fire(logs, session_factory) + real_fire(logs, session_factory, config_data) monkeypatch.setattr(email_service, "send_lead_email", fake_lead) monkeypatch.setattr(lead_push_service, "_run_push", fake_run) diff --git a/tests/test_rate_limit.py b/tests/test_rate_limit.py index e2adfe5..8013472 100644 --- a/tests/test_rate_limit.py +++ b/tests/test_rate_limit.py @@ -4,18 +4,18 @@ from app.engine.services.rate_limit import check, _reset_for_test def test_allows_under_cap(): _reset_for_test() for _ in range(30): - assert check("1.2.3.4", "dev1") is True + assert check("t1", "1.2.3.4", "dev1") is True def test_blocks_over_cap(): _reset_for_test() for _ in range(30): - assert check("1.2.3.4", "dev1") is True - assert check("1.2.3.4", "dev1") is False + assert check("t1", "1.2.3.4", "dev1") is True + assert check("t1", "1.2.3.4", "dev1") is False def test_isolated_buckets(): _reset_for_test() for _ in range(30): - check("1.2.3.4", "dev1") - assert check("5.6.7.8", "dev2") is True # different ip+device unaffected + check("t1", "1.2.3.4", "dev1") + assert check("t1", "5.6.7.8", "dev2") is True # different ip+device unaffected diff --git a/tests/test_tenant_config_wiring.py b/tests/test_tenant_config_wiring.py new file mode 100644 index 0000000..998ffa8 --- /dev/null +++ b/tests/test_tenant_config_wiring.py @@ -0,0 +1,305 @@ +"""Task 9: 服务层配置接线——DEFAULT_CONFIG 九组 → 真实消费方的缺口测试。 + +既有测试 monkeypatch 的都是 settings(环境层): 配置组没人读时它们照样绿, +两租户联调只会静默错配。本文件从 TenantConfig 出发逐组逼出消费点: + +- email: resolve_email_settings 租户段优先, 空槽回退 env settings; + send_* 消费 resolved 产物(From/To 走租户发件名/通知列表) +- push: resolve_push_settings 全字段(租户段优先, 空槽回退 env) +- engine.rate_limit_per_hour + rate_limit 桶键含租户维度 +- ai: resolve_ai_settings 覆盖生效; api_key/base_url/model 空槽回退 env +- contact: cc_completion/email_light_ask 行为开关在引擎级真实门控 +- misc.name_to_region_extras ⊕ cc_completion.NAME_TO_REGION +- engine.context_seed_keys 种子白名单(create_conversation 消费) +""" +import asyncio +from dataclasses import replace + +from sqlalchemy import select + +from app.engine.config import settings +from app.engine.services import email_service, lead_push_service +from app.engine.services.ai_service import resolve_ai_settings +from app.engine.services.cc_completion import name_to_region +from app.engine.services.conversation import ( + ConversationManager, allowed_seed_keys, resolve_contact_flags, +) +from app.engine.services.rate_limit import _reset_for_test, check +from app.platform.tenant_config import TenantConfig +from tests.conftest import build_jiancai_runtime + + +def _cfg(snapshot): + return TenantConfig.from_snapshot("t1", "T", snapshot) + + +# ---- email --------------------------------------------------------------- + +def test_email_reads_tenant_smtp(monkeypatch): + """租户段优先于全局 settings: smtp/sender_name/notify_emails 全量覆盖。""" + monkeypatch.setattr(settings, "SMTP_HOST", "env-host") + monkeypatch.setattr(settings, "SMTP_USER", "env@x.com") + r = email_service.resolve_email_settings(_cfg({"email": { + "smtp": {"host": "t.smtp", "port": 587, "user": "t@x.com", + "password": "t-pass"}, + "sender_name": "Auto Co", + "notify_emails": ["a@t.com", "b@t.com"], + }}).data) + assert r == { + "host": "t.smtp", "port": 587, "user": "t@x.com", + "password": "t-pass", "sender_name": "Auto Co", + "notify_emails": ["a@t.com", "b@t.com"], + } + + +def test_email_falls_back_to_env_settings(monkeypatch): + """smtp=None(缺省/显式清空)→ env SMTP_*; 发件名空 → 平台默认 + (旧硬编码【HESUPIPE】); notify_emails 空列表 → env NOTIFY_EMAIL。""" + for k, v in (("SMTP_HOST", "env-host"), ("SMTP_PORT", 465), + ("SMTP_USER", "env@x.com"), ("SMTP_PASS", "env-pass"), + ("NOTIFY_EMAIL", "notify@env.com")): + monkeypatch.setattr(settings, k, v) + r = email_service.resolve_email_settings(_cfg({}).data) + assert r == { + "host": "env-host", "port": 465, "user": "env@x.com", + "password": "env-pass", "sender_name": "HESUPIPE", + "notify_emails": ["notify@env.com"], + } + + +def test_email_partial_smtp_override(monkeypatch): + """smtp 段内逐键回退: 只给 host, 其余(user/password/port)沿用 env。""" + monkeypatch.setattr(settings, "SMTP_HOST", "env-host") + monkeypatch.setattr(settings, "SMTP_PORT", 465) + monkeypatch.setattr(settings, "SMTP_USER", "env@x.com") + monkeypatch.setattr(settings, "SMTP_PASS", "env-pass") + r = email_service.resolve_email_settings( + _cfg({"email": {"smtp": {"host": "t.smtp"}}}).data) + assert (r["host"], r["port"], r["user"], r["password"]) == ( + "t.smtp", 465, "env@x.com", "env-pass") + + +def test_send_lead_email_uses_tenant_sender_and_recipients(monkeypatch): + """send_* 消费接线: From = 租户发件名, To = 租户通知列表。""" + captured = {} + + async def fake_send(msg, cfg=None): + captured["msg"] = msg + return True + + monkeypatch.setattr(email_service, "_send", fake_send) + cfg = _cfg({"email": { + "smtp": {"host": "h", "user": "t@x.com"}, + "sender_name": "ACME", + "notify_emails": ["x@t.com", "y@t.com"], + }}) + ok = asyncio.run(email_service.send_lead_email( + customer_profile={}, contact_info={"email": "c@x.com"}, + reasoning="", conversation_messages=[], + config_data=cfg.data, + )) + assert ok is True + assert str(captured["msg"]["From"]) == "【ACME】 " + assert str(captured["msg"]["To"]) == "x@t.com, y@t.com" + + +# ---- push ---------------------------------------------------------------- + +def _patch_push_env(monkeypatch): + monkeypatch.setattr(settings, "PUSH_ENABLED", True) + monkeypatch.setattr(settings, "PUSH_API_URL", "env-url") + monkeypatch.setattr(settings, "PUSH_API_KEY", "env-key") + monkeypatch.setattr(settings, "PUSH_EXTERNAL_KEY_PREFIX", "env-pfx") + monkeypatch.setattr(settings, "PUSH_TIMEOUT_SECONDS", 10.0) + monkeypatch.setattr(settings, "PUSH_MAX_ATTEMPTS", 3) + monkeypatch.setattr(settings, "PUSH_RETRY_BACKOFF_BASE", 1.0) + + +def test_push_reads_tenant_section(monkeypatch): + """PUSH 组全部字段: 租户段给了就用租户的。""" + _patch_push_env(monkeypatch) + r = lead_push_service.resolve_push_settings(_cfg({"push": { + "enabled": True, "url": "https://t.example/push", "api_key": "t-key", + "prefix": "t-pfx", "timeout": 5, "max_attempts": 4, + "retry_backoff_base": 0.5, + }}).data) + assert r == { + "enabled": True, "url": "https://t.example/push", "api_key": "t-key", + "prefix": "t-pfx", "timeout": 5, "max_attempts": 4, + "retry_backoff_base": 0.5, + } + + +def test_push_empty_section_falls_back_to_env(monkeypatch): + """空槽(默认 False/空串)视同未设置 → 回退 env PUSH_*。""" + _patch_push_env(monkeypatch) + r = lead_push_service.resolve_push_settings(_cfg({}).data) + assert r == { + "enabled": True, "url": "env-url", "api_key": "env-key", + "prefix": "env-pfx", "timeout": 10.0, "max_attempts": 3, + "retry_backoff_base": 1.0, + } + + +# ---- rate limit ---------------------------------------------------------- + +def test_rate_limit_key_includes_tenant(): + """桶键 = (tenant, ip, device): 同 ip+device 在两租户不共享预算; + limit 读 config.engine.rate_limit_per_hour。""" + _reset_for_test() + for _ in range(3): + assert check("tenant-a", "1.2.3.4", "dev1", max_requests=3) is True + assert check("tenant-a", "1.2.3.4", "dev1", max_requests=3) is False + assert check("tenant-b", "1.2.3.4", "dev1", max_requests=3) is True + + limit = _cfg({"engine": {"rate_limit_per_hour": 2}})\ + .data["engine"]["rate_limit_per_hour"] + assert check("t", "9.9.9.9", "d", max_requests=limit) is True + assert check("t", "9.9.9.9", "d", max_requests=limit) is True + assert check("t", "9.9.9.9", "d", max_requests=limit) is False + + +# ---- AI ------------------------------------------------------------------ + +def test_ai_settings_override_and_fallback(monkeypatch): + monkeypatch.setattr(settings, "ZHIPU_API_KEY", "env-key") + monkeypatch.setattr(settings, "ZHIPU_BASE_URL", "env-url") + monkeypatch.setattr(settings, "AI_MODEL", "glm-4-plus") + assert resolve_ai_settings(_cfg({"ai": { + "api_key": "t-key", "base_url": "t-url", "model": "glm-4-air", + }}).data) == { + "provider": "glm", "api_key": "t-key", "base_url": "t-url", + "model": "glm-4-air", + } + # 未覆盖回退 env: api_key/base_url 空槽(None)→ env; + # model 出厂默认值(glm-4-flash)视同未设置 → env AI_MODEL + r = resolve_ai_settings(_cfg({}).data) + assert r == {"provider": "glm", "api_key": "env-key", + "base_url": "env-url", "model": "glm-4-plus"} + + +# ---- contact 行为开关 ------------------------------------------------------ + +def test_contact_toggles_resolve(): + assert resolve_contact_flags(_cfg({}).data) == { + "cc_completion": True, "email_light_ask": True} + assert resolve_contact_flags(_cfg({"contact": { + "cc_completion": False, "email_light_ask": False}}).data) == { + "cc_completion": False, "email_light_ask": False} + + +def _runtime_with(snapshot): + """jiancai 运行时 + 覆盖配置段(共享加载期产物, 无 engine 无句柄)。""" + cfg = TenantConfig.from_snapshot("jiancai", "T", snapshot) + return replace(build_jiancai_runtime(), config=cfg) + + +class _PassThroughAI: + """确定性桩: 不拦截不抽取, 只回固定句。""" + + async def stream_chat(self, messages, context=""): + for token in ("ok ", "reply"): + yield token + + async def extract_info(self, messages, existing_info): + return {} + + async def evaluate(self, profile, recent_messages): + return {"intent_level": "pending", "reasoning": "", + "contact_info": {}, "info_sufficient": False} + + +def _send_and_get_raw(async_db, runtime, text): + async def run(): + async with async_db() as db: + mgr = ConversationManager(db, runtime=runtime, + ai_service=_PassThroughAI()) + conv = await mgr.create_conversation(source=None) + async for _ in mgr.stream_process_message(conv.session_id, text): + pass + return conv.id + cid = asyncio.run(run()) + + async def fetch(): + from app.engine.models import CustomerProfile + async with async_db() as db: + profile = (await db.execute(select(CustomerProfile).where( + CustomerProfile.conversation_id == cid))).scalar_one() + return dict(profile.raw_info or {}) + return asyncio.run(fetch()) + + +def test_cc_completion_toggle_gates_engine(async_db): + """cc_completion=False: 轮前判定树不建 _cc(裸号无状态机痕迹); + True(默认): 照建。""" + raw_off = _send_and_get_raw( + async_db, _runtime_with({"contact": {"cc_completion": False}}), + "my whatsapp is 0501234567") + assert "_cc" not in raw_off + assert raw_off.get("联系方式", {}).get("whatsapp") == "0501234567" + + raw_on = _send_and_get_raw(async_db, build_jiancai_runtime(), + "my whatsapp is 0501234567") + assert raw_on.get("_cc", {}).get("whatsapp", {}).get("canon") == "0501234567" + + +def test_email_light_ask_toggle_gates_engine(async_db): + """email_light_ask=False: 一次性计数键 _email_asked 不写; + True(默认): 写。""" + raw_off = _send_and_get_raw( + async_db, _runtime_with({"contact": {"email_light_ask": False}}), + "my whatsapp is +966501234567") + assert "_email_asked" not in raw_off + + raw_on = _send_and_get_raw(async_db, build_jiancai_runtime(), + "my whatsapp is +966501234567") + assert raw_on.get("_email_asked") is True + + +# ---- 映射表 / 种子键 ------------------------------------------------------- + +def test_name_to_region_extras_merged(): + """misc.name_to_region_extras ⊕ NAME_TO_REGION: extras 可扩表也可覆盖。""" + extras = {"德意志": "DE", "沙特": "QA"} + assert name_to_region("德意志") is None # 基础表无此键 + assert name_to_region("德意志", extras=extras) == "DE" + assert name_to_region("沙特", extras=extras) == "QA" # extras 覆盖基础表 + assert name_to_region("中国", extras=extras) == "CN" # 基础表照常可用 + assert name_to_region("SAUDI ARABIA", extras=extras) == "SA" + + +def test_chat_rest_seed_keys_from_config(): + """种子键白名单 = config.engine.context_seed_keys 的键集合。""" + cfg = _cfg({"engine": {"context_seed_keys": { + "vehicle_name": "app_vehicle_name", "vehicle_id": "app_vehicle_id"}}}) + assert allowed_seed_keys(cfg.data) == {"vehicle_name", "vehicle_id"} + # jiancai 出厂 = 旧硬编码白名单的配置化形态 + assert allowed_seed_keys(build_jiancai_runtime().config.data) == { + "product_name", "product_id"} + + +def test_seed_keys_drive_create_conversation(async_db): + """create_conversation 按映射种子化: 白名单外的 context 键被丢弃。""" + + async def run(): + async with async_db() as db: + mgr = ConversationManager( + db, runtime=_runtime_with({"engine": {"context_seed_keys": { + "vehicle_name": "app_vehicle_name", + "vehicle_id": "app_vehicle_id"}}})) + conv = await mgr.create_conversation(context={ + "vehicle_name": "Toyota Hilux", "vehicle_id": "v9", + "product_name": "should_be_ignored", + }) + return conv.id + cid = asyncio.run(run()) + + from app.engine.models import CustomerProfile + + async def fetch(): + async with async_db() as db: + profile = (await db.execute(select(CustomerProfile).where( + CustomerProfile.conversation_id == cid))).scalar_one() + return profile.raw_info + assert asyncio.run(fetch()) == { + "app_vehicle_name": "Toyota Hilux", "app_vehicle_id": "v9"}