kunthawat/sales-trainer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sales Trainer v0.1: corporate sales-training simulator (Flask+Vue, 15 personas, chat simulator, judge, analytics)

Browse Source 
- Auth/roles (no self-reg), admin user provision, JWT
- Analyze: sales kit + initial pain-fit from form/upload
- Persona generator: 15 personas (5/tier) w/ pain variety, negotiation, init mode, channel, latent/revealable, wrong_text special
- Chat simulator: per-mode initiation, one-shot, hidden signals, judge-LLM debrief+coaching
- Trainee loop: win/lose board, weak-areas, user-generated personas
- Admin analytics; EN+TH Vue SPA served by Flask
- Deploy: Dockerfile, docker-compose, README, eng-log + HANDOFF
- Tests (mock LLM): m0/m1/routes/e2e all pass
This commit is contained in:
Macky
2026-08-07 15:31:06 +07:00
commit c3d31c06e2
70 changed files with 6135 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
backend/scripts/mock_llm.py Normal file
View File

@@ -0,0 +1,111 @@
"""Mock LLM for deterministic end-to-end tests (no external API needed).
Substitutes for app.llm.LLMClient. Returns canned JSON for structured calls and
simple replies for chat calls, so the full analyze→persona→chat→debrief flow runs.
"""
from __future__ import annotations
import json
from typing import Any
SAMPLE_SALES_KIT = {
"productName": "CloudPOS",
"category": "POS software",
"valueProps": ["faster checkout", "inventory sync"],
"features": ["tablets", "reports"],
"pricingAnchors": ["1,000 THB/month"],
"targetAudience": {"segment": "SME restaurants", "demographics": "", "useCases": ["front counter"]},
"objectionHandlers": ["free trial", "setup included"],
"initialPainFit": [
{"pain": "slow checkout queues", "fit": "strong", "evidence": "faster checkout"},
{"pain": "lost sales from stockouts", "fit": "partial", "evidence": "inventory sync"},
],
"scenarioFrame": "Cloud POS sold over LINE to Bangkok SME restaurants.",
}
def _sample_persona(idx: int, tier: str) -> dict[str, Any]:
return {
"id": f"persona-{idx:02d}",
"name": f"Persona {idx}",
"tier": tier,
"channel": "line",
"initiation_mode": "customer" if idx % 3 else "seller",
"profession": "restaurant owner",
"age_group": "30s",
"location": "Bangkok",
"product_context": "running a small noodle shop",
"background": "Runs a family noodle shop for 8 years.",
"income": "60k THB/month",
"lifestyle": "works long hours",
"personality": "practical and cautious",
"communication_style": "short, direct, casual",
"budget": "1,500 THB/month max",
"decision_timeline": "within 2 weeks",
"goal": "reduce lunch-rush queues",
"objections": ["too expensive", "hard to learn"],
"pains": [
{"id": "p1", "name": "slow checkout", "fit": "strong",
"description": "Long queues at lunch", "rootCause": "manual order taking",
"resolutionConditions": ["show faster checkout", "offer a trial"]},
{"id": "p2", "name": "stockouts", "fit": "partial",
"description": "Runs out of ingredients", "rootCause": "no inventory tracking",
"resolutionConditions": ["show inventory feature"]},
],
"negotiation_levers": ["price reduction", "free setup"],
"opener": "Hi, I saw your POS ad. Does it work with small shops?",
"special": "wrong_text" if (tier == "C" and idx % 5 == 4) else "",
"difficulty": 2 if tier == "A" else (3 if tier == "B" else 4),
"notes": "sample",
}
def make_personas() -> list[dict[str, Any]]:
out = []
idx = 1
for tier in ["A", "B", "C"]:
for _ in range(5):
out.append(_sample_persona(idx, tier))
idx += 1
return out
class MockLLM:
"""Drop-in for app.llm.LLMClient — reads config the same way."""
persona_count = 0
def __init__(self, **kwargs):
pass
def complete(self, system_prompt: str, user_prompt: str, **kw) -> str:
if "Persona generation prompts" in system_prompt or "persona designer" in system_prompt.lower():
return json.dumps({"personas": make_personas()}, ensure_ascii=False)
if "market-research persona designer" in system_prompt.lower():
return json.dumps({"personas": make_personas()}, ensure_ascii=False)
return "ok"
def complete_json(self, system_prompt: str, user_prompt: str, **kw) -> dict[str, Any]:
sp = system_prompt.lower()
if "ecommerce/b2b analyst" in sp:
return dict(SAMPLE_SALES_KIT)
if "market-research persona designer" in sp:
return {"personas": make_personas()}
if "sales-training simulator" in sp and "PRIVATE" in system_prompt:
return {"persona": _sample_persona(99, "C")}
if "judge" in sp and "sales-training chat" in sp:
return {
"outcome": "won",
"score": 82,
"pain": "slow checkout queues",
"why": "resolved the pain and secured acceptance",
"failurePoints": [],
"coaching": [],
"painProgress": {"slow checkout": 100},
}
return {}
def complete_conversation(self, messages, **kw) -> str:
# persona chat: echo a short in-character reply
return json.dumps({"reply": "I see. Tell me more about the price then."}, ensure_ascii=False)

141
backend/scripts/test_e2e.py Normal file
View File

@@ -0,0 +1,141 @@
"""Full E2E test with a mock LLM: analyze → personas → chat → debrief → board/analytics."""
import os
import sys
import tempfile
import warnings
from pathlib import Path
warnings.filterwarnings("ignore", message="The HMAC key is")
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # for mock_llm
tempdir = tempfile.mkdtemp(prefix="st_e2e_")
os.environ["DATA_DIR"] = tempdir
os.environ["JWT_SECRET"] = "test-secret-key-0123456789abcdef"
from mock_llm import MockLLM # noqa: E402
from app.factory import create_app # noqa: E402
from app.config import Config # noqa: E402
Config.DATA_DIR = Path(tempdir)
Config.LLM_API_KEY = ""
Config.LLM_BASE_URL = ""
def main():
app = create_app()
app.extensions["llm"] = MockLLM()
client = app.test_client()
# admin login
r = client.post("/api/auth/login", json={"email": "admin@salestrainer.local", "password": "admin123"})
AT = r.get_json()["token"]
AH = {"Authorization": f"Bearer {AT}"}
# create group
r = client.post("/api/groups", json={
"product": "Cloud POS for small restaurants", "segment": "SME restaurants",
"channel": "line", "language": "th"}, headers=AH)
assert r.status_code == 201, r.get_json()
gid = r.get_json()["group"]["id"]
# analyze -> sales kit + 15 personas
r = client.post(f"/api/groups/{gid}/analyze", headers=AH)
assert r.status_code == 200, r.get_json()
body = r.get_json()
assert body["sales_kit"]["productName"] == "CloudPOS", body["sales_kit"]
personas = body["personas"]
assert len(personas) == 15, f"expected 15 personas, got {len(personas)}"
tiers = {}
for p in personas:
tiers.setdefault(p["tier"], 0)
tiers[p["tier"]] += 1
assert tiers == {"A": 5, "B": 5, "C": 5}, tiers
# wrong_text special in tier C
assert any(p["tier"] == "C" and p["special"] == "wrong_text" for p in personas), "no wrong_text persona"
print(f"[ok] analyze -> sales kit + 15 personas (tiers {tiers}), wrong_text present")
# create a trainee
client.post("/api/admin/users", json={
"name": "Trainee", "email": "t@x.com", "password": "pass123", "role": "user"}, headers=AH)
r = client.post("/api/auth/login", json={"email": "t@x.com", "password": "pass123"})
UT = r.get_json()["token"]
UH = {"Authorization": f"Bearer {UT}"}
# trainee sees group + personas but revealable-only (no pain/income)
r = client.get(f"/api/groups/{gid}/personas", headers=UH)
assert r.status_code == 200
plist = r.get_json()["personas"]
assert len(plist) == 15
first = plist[1]
assert "pains" not in first and "income" not in first, "latent fields leaked!"
assert "profession" in first and "initiation_mode" in first
print("[ok] trainee sees revealable-only persona fields (latent hidden)")
# pick a customer-initiated persona -> start session (customer opens)
cust = next(p for p in personas if p["initiation_mode"] == "customer")
pid = cust["id"]
r = client.post(f"/api/chat/{gid}/personas/{pid}/chat/start", headers=UH)
assert r.status_code == 200, r.get_json()
session = r.get_json()["session"]
assert session["status"] == "active"
assert len(session["messages"]) >= 1 and session["messages"][0]["role"] == "customer", "customer should open"
print("[ok] customer-initiated session starts with customer opener")
# send messages
r = client.post(f"/api/chat/{gid}/personas/{pid}/chat/send",
json={"text": "Hi, I run a small noodle shop. Tell me about pricing."}, headers=UH)
assert r.status_code == 200, r.get_json()
assert r.get_json()["reply"]
print("[ok] send message -> persona replies")
# finish -> debrief reveals latent + outcome won
r = client.post(f"/api/chat/{gid}/personas/{pid}/chat/finish", headers=UH)
assert r.status_code == 200, r.get_json()
debrief = r.get_json()["debrief"]
assert debrief["outcome"] == "won"
assert "revealed_persona" in debrief and "pains" in debrief["revealed_persona"]
print("[ok] finish -> debrief with latent reveal + outcome")
# ONE-SHOT: cannot start again on same persona
r = client.post(f"/api/chat/{gid}/personas/{pid}/chat/start", headers=UH)
assert r.status_code == 400, r.get_json()
print("[ok] one-shot enforced (cannot re-chat same persona)")
# seller-initiated persona -> session starts WITHOUT opener (task for seller)
sel = next(p for p in personas if p["initiation_mode"] == "seller")
r = client.post(f"/api/chat/{gid}/personas/{sel['id']}/chat/start", headers=UH)
assert r.status_code == 200, r.get_json()
s2 = r.get_json()["session"]
assert "task" in r.get_json() or "initiation_mode" in r.get_json()
assert s2["messages"] == [] , "seller-initiated should not have a customer opener"
print("[ok] seller-initiated session (no customer opener, seller must open)")
# board
r = client.get("/api/me/board", headers=UH)
board = r.get_json()["board"]
assert any(b["persona_id"] == pid and b["my_outcome"] == "won" for b in board)
print("[ok] win/lose board reflects won persona")
# weak-areas (no losses yet -> empty insight but endpoint works)
r = client.get("/api/me/weak-areas", headers=UH)
assert r.status_code == 200
print("[ok] weak-areas endpoint")
# generate own persona (manual, mock)
r = client.post("/api/me/personas/generate", json={"mode": "manual", "spec": {"target": "price-hardball"}}, headers=UH)
assert r.status_code == 201, r.get_json()
print("[ok] user generates own persona (manual)")
# analytics (admin)
r = client.get("/api/analytics", headers=AH)
assert r.status_code == 200
a = r.get_json()
assert a["overall"]["wins"] >= 1
print("[ok] admin analytics aggregates wins")
print("\nALL E2E TESTS PASSED")
if __name__ == "__main__":
main()

107
backend/scripts/test_m0.py Normal file
View File

@@ -0,0 +1,107 @@
"""M0 smoke test: auth + roles via Flask test client."""
import json
import os
import sys
import tempfile
import warnings
from pathlib import Path
warnings.filterwarnings("ignore", message="The HMAC key is")
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
os_env_data = tempfile.mkdtemp(prefix="salestrainer_m0_")
os.environ["DATA_DIR"] = os_env_data
os.environ["JWT_SECRET"] = "test-secret"
from app.factory import create_app # noqa: E402
from app.config import Config # noqa: E402
from pathlib import Path as _Path
# .env with override=True would clobber DATA_DIR, so pin the store root directly.
Config.DATA_DIR = _Path(os_env_data)
def main() -> None:
app = create_app()
client = app.test_client()
# 1. Health
r = client.get("/health")
assert r.status_code == 200 and r.get_json()["status"] == "ok", r.get_json()
print("[ok] health")
# 2. No self-registration: register route must NOT exist (404)
r = client.post("/api/auth/register", json={"email": "a@b.c", "password": "x"})
assert r.status_code == 404, f"register should not exist, got {r.status_code}"
print("[ok] no self-registration (register -> 404)")
# 3. Bootstrap super-admin login
r = client.post(
"/api/auth/login",
json={"email": "admin@salestrainer.local", "password": "admin123"},
)
assert r.status_code == 200, r.get_json()
admin_token = r.get_json()["token"]
print("[ok] super-admin login")
# 4. /me with token
r = client.get("/api/auth/me", headers={"Authorization": f"Bearer {admin_token}"})
assert r.status_code == 200
assert r.get_json()["user"]["role"] == "super_admin"
print("[ok] /me super_admin")
# 5. Create a regular user (admin)
r = client.post(
"/api/admin/users",
json={"name": "Trainee One", "email": "t1@x.com", "password": "pass123", "role": "user"},
headers={"Authorization": f"Bearer {admin_token}"},
)
assert r.status_code == 201, r.get_json()
print("[ok] admin creates user")
# 6. Trainee login + cannot access admin users list (403)
r = client.post("/api/auth/login", json={"email": "t1@x.com", "password": "pass123"})
user_token = r.get_json()["token"]
r = client.get("/api/admin/users", headers={"Authorization": f"Bearer {user_token}"})
assert r.status_code == 403, f"trainee should be denied, got {r.status_code}"
print("[ok] trainee denied admin route (403)")
# 7. No token -> 401
r = client.get("/api/admin/users")
assert r.status_code == 401
print("[ok] no token -> 401")
# 8. Role restriction: admin cannot create another admin (only super_admin)
# create an 'admin' actor first
client.post(
"/api/admin/users",
json={"name": "Admin Two", "email": "a2@x.com", "password": "pass123", "role": "admin"},
headers={"Authorization": f"Bearer {admin_token}"},
)
r = client.post(
"/api/auth/login", json={"email": "a2@x.com", "password": "pass123"}
)
admin2_token = r.get_json()["token"]
r = client.post(
"/api/admin/users",
json={"name": "Bogus Admin", "email": "ba@x.com", "password": "pass123", "role": "super_admin"},
headers={"Authorization": f"Bearer {admin2_token}"},
)
assert r.status_code == 403, f"admin should not promote, got {r.status_code}"
print("[ok] admin cannot grant super_admin (403)")
# 9. Duplicate email rejected
r = client.post(
"/api/admin/users",
json={"name": "Dup", "email": "t1@x.com", "password": "pass123", "role": "user"},
headers={"Authorization": f"Bearer {admin_token}"},
)
assert r.status_code == 400
print("[ok] duplicate email rejected (400)")
print("\nALL M0 TESTS PASSED")
if __name__ == "__main__":
main()

79
backend/scripts/test_m1.py Normal file
View File

@@ -0,0 +1,79 @@
"""Verify the whole backend imports without errors (no LLM calls)."""
import os
import sys
import tempfile
import warnings
from pathlib import Path
warnings.filterwarnings("ignore", message="The HMAC key is")
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
tempdir = tempfile.mkdtemp(prefix="st_import_")
os.environ["DATA_DIR"] = tempdir
os.environ["JWT_SECRET"] = "test-secret-key-0123456789abcdef"
from app.factory import create_app # noqa: E402
from app.config import Config # noqa: E402
Config.DATA_DIR = Path(tempdir)
# Force no LLM for import/structural test (analyze path tested separately)
Config.LLM_API_KEY = ""
Config.LLM_BASE_URL = ""
def main():
app = create_app()
# LLM should be None (no API key in test env)
assert app.extensions["llm"] is None, "expect no LLM in test env"
client = app.test_client()
# login as super-admin
r = client.post("/api/auth/login", json={
"email": "admin@salestrainer.local", "password": "admin123"})
assert r.status_code == 200, r.get_json()
token = r.get_json()["token"]
H = {"Authorization": f"Bearer {token}"}
# create a group via JSON form (no files)
r = client.post("/api/groups", json={
"product": "Cloud POS system for small restaurants",
"segment": "SME restaurants",
"description": "Target Bangkok SME restaurants, 1-3 branches",
"channel": "line",
"language": "th",
}, headers=H)
assert r.status_code == 201, r.get_json()
gid = r.get_json()["group"]["id"]
assert r.get_json()["group"]["status"] == "draft"
print("[ok] group created (draft)")
# analyze should fail cleanly (LLM None)
r = client.post(f"/api/groups/{gid}/analyze", headers=H)
assert r.status_code == 500, r.get_json()
print("[ok] analyze fails cleanly when LLM unset (500)")
# list groups as admin
r = client.get("/api/groups", headers=H)
assert r.status_code == 200 and len(r.get_json()["groups"]) == 1
print("[ok] admin lists 1 group")
# personas empty until analyze
r = client.get(f"/api/groups/{gid}", headers=H)
assert r.get_json()["group"]["personas"] == []
print("[ok] group has no personas before analyze")
# trainee created; can list groups but only 'ready' ones (this one is 'failed' -> hidden)
client.post("/api/admin/users", json={
"name": "Trainee", "email": "t@x.com", "password": "pass123", "role": "user"}, headers=H)
r = client.post("/api/auth/login", json={"email": "t@x.com", "password": "pass123"})
ut = r.get_json()["token"]
UH = {"Authorization": f"Bearer {ut}"}
r = client.get("/api/groups", headers=UH)
assert r.get_json()["groups"] == [], "trainee should not see non-ready groups"
print("[ok] trainee cannot see non-ready groups")
print("\nALL M1/M2-IMPORT TESTS PASSED")
if __name__ == "__main__":
main()

51
backend/scripts/test_routes.py Normal file
View File

@@ -0,0 +1,51 @@
"""Verify full backend imports + all routes registered (no LLM needed)."""
import os
import sys
import tempfile
import warnings
from pathlib import Path
warnings.filterwarnings("ignore", message="The HMAC key is")
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
tempdir = tempfile.mkdtemp(prefix="st_import_")
os.environ["DATA_DIR"] = tempdir
os.environ["JWT_SECRET"] = "test-secret-key-0123456789abcdef"
from app.factory import create_app # noqa: E402
from app.config import Config # noqa: E402
Config.DATA_DIR = Path(tempdir)
Config.LLM_API_KEY = ""
Config.LLM_BASE_URL = ""
def main():
app = create_app()
rules = sorted({str(rule) for rule in app.url_map.iter_rules() if str(rule).startswith("/api")})
expected = [
"/api/auth/login", "/api/auth/me",
"/api/admin/users", "/api/admin/users/<email>",
"/api/groups", "/api/groups/<gid>",
"/api/groups/<gid>/analyze", "/api/groups/<gid>/personas",
"/api/groups/<gid>/personas/<pid>", "/api/groups/<gid>/personas/<pid>",
"/api/groups/<gid>/reanalyze",
"/api/chat/<gid>/personas/<pid>/chat/start",
"/api/chat/<gid>/personas/<pid>/chat/send",
"/api/chat/<gid>/personas/<pid>/chat/finish",
"/api/chat/sessions", "/api/chat/sessions/<sid>",
"/api/me/board", "/api/me/weak-areas", "/api/me/personas",
"/api/me/personas/generate",
"/api/analytics",
]
missing = [e for e in expected if e not in rules]
if missing:
raise SystemExit(f"MISSING ROUTES: {missing}")
print(f"[ok] all {len(expected)} expected routes registered")
for r in sorted(rules):
print(" ", r)
print("ALL ROUTE REGISTRATION TESTS PASSED")
if __name__ == "__main__":
main()

4
backend/scripts/verify_env.py Normal file
View File

@@ -0,0 +1,4 @@
import flask, jwt, dotenv, openai, fitz, pydantic, werkzeug
print("flask", flask.__version__)
print("openai", openai.__version__)
print("all imports ok")