diff --git a/dealix/api/main.py b/dealix/api/main.py index 31d26ab2..5dbcba49 100644 --- a/dealix/api/main.py +++ b/dealix/api/main.py @@ -21,6 +21,7 @@ from api.routers import ( automation, autonomous, business, + cards, command_center, connector_router, customer_ops, @@ -179,6 +180,7 @@ def create_app() -> FastAPI: app.include_router(revenue_launch.router) app.include_router(public_launch.router) app.include_router(business.router) + app.include_router(cards.router) app.include_router(personal_operator.router) app.include_router(public.router) app.include_router(admin.router) @@ -197,6 +199,7 @@ def create_app() -> FastAPI: "personal_operator_launch_report": "/api/v1/personal-operator/launch-report", "business_pricing": "/api/v1/business/pricing", "innovation_command_feed_demo": "/api/v1/innovation/command-feed/demo", + "role_cards_feed": "/api/v1/cards/feed?role=ceo", } return app diff --git a/dealix/api/routers/cards.py b/dealix/api/routers/cards.py new file mode 100644 index 00000000..34d45d87 --- /dev/null +++ b/dealix/api/routers/cards.py @@ -0,0 +1,112 @@ +"""Role-based Revenue Command Cards API — feeds + decisions (draft/approval-first).""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException, Query +from pydantic import BaseModel, Field + +from auto_client_acquisition.revenue_company_os.card_factory import ( + build_role_command_feed, + build_whatsapp_daily_brief_lines, +) +from auto_client_acquisition.revenue_company_os.cards import ( + UserRole, + is_known_role, + normalize_role_param, +) + +router = APIRouter(prefix="/api/v1/cards", tags=["revenue-command-cards"]) + +# Demo-only: in-process decision log (not durable storage). +_decision_log: dict[str, dict[str, Any]] = {} + + +class CardDecisionBody(BaseModel): + """Human decision on a card — never triggers live channel send.""" + + action: str | None = Field(None, description="approve | edit | skip") + button_action: str | None = Field(None, description="machine key from pressed button") + note: str | None = Field(None, max_length=2000) + + +def _find_card(card_id: str) -> dict[str, Any] | None: + for role in UserRole: + for c in build_role_command_feed(role.value)["cards"]: + if str(c.get("card_id")) == card_id: + return c + return None + + +def _allowed_roles() -> list[str]: + return [e.value for e in UserRole] + + +@router.get("/feed") +async def get_card_feed(role: str = Query("ceo", description="ceo | sales_manager | growth_manager | ...")) -> dict[str, Any]: + nr = normalize_role_param(role) + if not is_known_role(nr): + raise HTTPException( + status_code=400, + detail={"error": "unknown_role", "allowed": _allowed_roles()}, + ) + return build_role_command_feed(nr) + + +@router.get("/whatsapp/daily-brief") +async def get_whatsapp_daily_brief( + role: str = Query("ceo", description="Role for brief lines (still approval-first)"), +) -> dict[str, Any]: + """Compact lines for WhatsApp-style surfaces — no auto-send.""" + nr = normalize_role_param(role) + if not is_known_role(nr): + raise HTTPException( + status_code=400, + detail={"error": "unknown_role", "allowed": _allowed_roles()}, + ) + return { + "role": nr, + "lines_ar": build_whatsapp_daily_brief_lines(nr), + "demo": True, + "no_auto_send": True, + } + + +@router.post("/{card_id}/decision") +async def post_card_decision(card_id: str, body: CardDecisionBody) -> dict[str, Any]: + card = _find_card(card_id) + if not card: + raise HTTPException(status_code=404, detail={"error": "unknown_card_id"}) + + base_proof = list(card.get("proof_impact") or []) + extra: list[str] = ["decision_recorded"] + if body.button_action: + extra.append(f"button:{body.button_action}") + if body.action: + extra.append(f"action:{body.action}") + proof_events = base_proof + extra + + record = { + "card_id": card_id, + "role": card.get("role"), + "action": body.action, + "button_action": body.button_action, + "note": body.note, + "proof_events": proof_events, + "execution_mode": "draft_only", + "draft_export_ar": ( + "مسودة تنفيذية: راجع المحتوى ثم نفّذ يدوياً عبر قناتك المعتمدة. " + "لا يُرسل Dealix نيابةً عنك في وضع Paid Beta الحالي." + ), + } + _decision_log[card_id] = record + + return { + "card_id": card_id, + "status": "logged", + "proof_events": proof_events, + "execution_mode": record["execution_mode"], + "draft_export_ar": record["draft_export_ar"], + "demo": True, + } diff --git a/dealix/auto_client_acquisition/revenue_company_os/__init__.py b/dealix/auto_client_acquisition/revenue_company_os/__init__.py index 1d522653..17c65608 100644 --- a/dealix/auto_client_acquisition/revenue_company_os/__init__.py +++ b/dealix/auto_client_acquisition/revenue_company_os/__init__.py @@ -1,6 +1,25 @@ """Revenue Company OS — events to cards, RWU, merged command feed.""" +from auto_client_acquisition.revenue_company_os.card_factory import ( + build_role_command_feed, + build_whatsapp_daily_brief_lines, +) +from auto_client_acquisition.revenue_company_os.cards import ( + MAX_CARD_BUTTONS, + UserRole, + normalize_card, + normalize_role_param, +) from auto_client_acquisition.revenue_company_os.command_feed_engine import build_company_os_command_feed from auto_client_acquisition.revenue_company_os.event_to_card import event_to_card -__all__ = ["build_company_os_command_feed", "event_to_card"] +__all__ = [ + "MAX_CARD_BUTTONS", + "UserRole", + "build_company_os_command_feed", + "build_role_command_feed", + "build_whatsapp_daily_brief_lines", + "event_to_card", + "normalize_card", + "normalize_role_param", +] diff --git a/dealix/auto_client_acquisition/revenue_company_os/card_factory.py b/dealix/auto_client_acquisition/revenue_company_os/card_factory.py new file mode 100644 index 00000000..abeea04b --- /dev/null +++ b/dealix/auto_client_acquisition/revenue_company_os/card_factory.py @@ -0,0 +1,347 @@ +"""Deterministic role-scoped command cards for Revenue OS (demo / in-process).""" + +from __future__ import annotations + +from typing import Any + +from auto_client_acquisition.revenue_company_os.cards import ( + MAX_CARDS_VISIBLE, + CardType, + UserRole, + normalize_card, +) + + +def _btn(label_ar: str, action: str) -> dict[str, str]: + return {"label_ar": label_ar, "action": action} + + +def _card( + *, + card_id: str, + role: str, + ctype: str, + title_ar: str, + why_now_ar: str, + recommended_action_ar: str, + risk_level: str, + buttons: list[dict[str, str]], + action_mode: str, + proof_impact: list[str], + context: dict[str, Any] | None = None, + status: str = "pending", +) -> dict[str, Any]: + return normalize_card( + { + "card_id": card_id, + "tenant_id": "demo_tenant", + "role": role, + "type": ctype, + "title_ar": title_ar, + "why_now_ar": why_now_ar, + "context": context or {}, + "recommended_action_ar": recommended_action_ar, + "risk_level": risk_level, + "buttons": buttons, + "action_mode": action_mode, + "proof_impact": proof_impact, + "status": status, + } + ) + + +def _ceo_cards() -> list[dict[str, Any]]: + r = UserRole.CEO.value + return [ + _card( + card_id=f"{r}_brief_1", + role=r, + ctype=CardType.CEO_BRIEF.value, + title_ar="👑 أهم ٣ قرارات اليوم", + why_now_ar="تغيّر نشاط الوكالات أعلى من شركات SaaS في السجل الأخير.", + recommended_action_ar="ركّز لمسات اليوم على الوكالات؛ أرسل Pilot ٤٩٩ لعميلين جاهزين؛ لا تستخدم واتساب بارد.", + risk_level="medium", + buttons=[ + _btn("اعتمد الخطة", "approve_daily_plan"), + _btn("اعرض التفاصيل", "expand_brief"), + _btn("أرسل للمدير", "delegate_sales_manager"), + ], + action_mode="approval_required", + proof_impact=["executive_brief_created", "decisions_recommended"], + ), + _card( + card_id=f"{r}_partner_1", + role=r, + ctype=CardType.PARTNER.value, + title_ar="🤝 شريك محتمل — وكالة Riyadh SMB", + why_now_ar="لديها عملاء SMB وتحتاج Proof متكرر لعملائها.", + recommended_action_ar="رسالة شراكة + Agency Partner Pilot على عميل واحد.", + risk_level="low", + buttons=[ + _btn("جهّز رسالة", "draft_partner_message"), + _btn("احجز اجتماع", "draft_meeting_invite"), + _btn("تخطي", "skip"), + ], + action_mode="draft_only", + proof_impact=["partner_suggested", "partner_scorecard_created"], + ), + _card( + card_id=f"{r}_risk_1", + role=r, + ctype=CardType.RISK.value, + title_ar="⛔ قناة عالية المخاطر", + why_now_ar="اقتراح حملة واتساب لقائمة باردة ظهر في المسودات.", + recommended_action_ar="استخدم إيميل أو LinkedIn يدوي مع opt-in؛ لا إرسال واتساب بارد.", + risk_level="high", + buttons=[ + _btn("اعرض السياسة", "show_policy"), + _btn("مسودة بديلة", "draft_safe_alternative"), + _btn("إغلاق", "dismiss"), + ], + action_mode="blocked", + proof_impact=["risks_blocked"], + context={"channel": "whatsapp", "policy": "no_cold_whatsapp"}, + ), + _card( + card_id=f"{r}_proof_1", + role=r, + ctype=CardType.PROOF.value, + title_ar="📊 Proof Pack — جاهز للمراجعة", + why_now_ar="اكتمل أسبوع تشغيل؛ جمع لقطات وموافقة عميل.", + recommended_action_ar="أرسل الملخص + عرض الترقية إلى Growth Starter.", + risk_level="low", + buttons=[ + _btn("اعرض التقرير", "open_proof"), + _btn("أرسل للعميل", "draft_client_email"), + _btn("حضّر عرض الترقية", "draft_upgrade_offer"), + ], + action_mode="approval_required", + proof_impact=["proof_generated", "upgrade_recommended"], + ), + ] + + +def _sales_cards() -> list[dict[str, Any]]: + r = UserRole.SALES_MANAGER.value + return [ + _card( + card_id=f"{r}_deal_1", + role=r, + ctype=CardType.DEAL_FOLLOWUP.value, + title_ar="📌 صفقة تحتاج متابعة", + why_now_ar="اكتمل الديمو قبل ٣ أيام ولا يوجد follow-up مسجّل.", + recommended_action_ar="إرسال follow-up يختصر القيمة ويقترح Pilot ٤٩٩.", + risk_level="medium", + buttons=[ + _btn("جهّز Follow-up", "draft_followup_email"), + _btn("حوّل للـ CEO", "escalate_ceo"), + _btn("تخطي", "skip"), + ], + action_mode="approval_required", + proof_impact=["deal_risk_detected", "followup_created", "approval_requested"], + ), + _card( + card_id=f"{r}_neg_1", + role=r, + ctype=CardType.NEGOTIATION.value, + title_ar="💬 اعتراض: السعر مرتفع", + why_now_ar="العميل لم يرَ Proof كافياً بعد؛ الخصم المباشر يضعف القيمة.", + recommended_action_ar="اقترح Pilot ٤٩٩ بدل خصم على الاشتراك الشهري.", + risk_level="medium", + buttons=[ + _btn("استخدم الرد", "use_objection_reply"), + _btn("عدّل النبرة", "edit_tone"), + _btn("جهّز عرض بديل", "draft_counter_offer"), + ], + action_mode="draft_only", + proof_impact=["objection_handled", "draft_created"], + ), + _card( + card_id=f"{r}_close_1", + role=r, + ctype=CardType.CLOSE.value, + title_ar="💰 فرصة إغلاق", + why_now_ar="العميل طلب تفاصيل بعد الديمو.", + recommended_action_ar="أرسل نص الدفع + نموذج intake (يدوي) الآن.", + risk_level="low", + buttons=[ + _btn("أرسل نص الدفع", "draft_payment_message"), + _btn("جهّز intake", "draft_intake_form"), + _btn("تخطي", "skip"), + ], + action_mode="approval_required", + proof_impact=["payment_link_drafted", "approval_requested"], + ), + ] + + +def _growth_cards() -> list[dict[str, Any]]: + r = UserRole.GROWTH_MANAGER.value + return [ + _card( + card_id=f"{r}_plan_1", + role=r, + ctype=CardType.GROWTH_PLAN.value, + title_ar="📣 خطة النمو اليوم", + why_now_ar="ردود الوكالات أعلى من شركات SaaS في آخر أسبوع.", + recommended_action_ar="١٠ رسائل لوكالات B2B + ٥ متابعات + منشور LinkedIn واحد.", + risk_level="low", + buttons=[ + _btn("اعتمد الخطة", "approve_daily_plan"), + _btn("غيّر الشريحة", "change_segment"), + _btn("اعرض scorecard", "open_scorecard"), + ], + action_mode="approval_required", + proof_impact=["daily_growth_plan_created", "channel_insight_created"], + ), + _card( + card_id=f"{r}_opp_1", + role=r, + ctype=CardType.OPPORTUNITY.value, + title_ar="🟢 فرصة — قطاع تدريب", + why_now_ar="إعلان توظيف مبيعات + توسع فرع.", + recommended_action_ar="مسودة إيميل تعريفي + عرض Diagnostic (بدون إرسال حي).", + risk_level="low", + buttons=[ + _btn("اعتمد المسودة", "approve_draft"), + _btn("عدّل", "edit_draft"), + _btn("تخطي", "skip"), + ], + action_mode="draft_only", + proof_impact=["opportunity_created", "draft_created"], + ), + ] + + +def _agency_cards() -> list[dict[str, Any]]: + r = UserRole.AGENCY_PARTNER.value + return [ + _card( + card_id=f"{r}_client_1", + role=r, + ctype=CardType.DELIVERY.value, + title_ar="➕ عميل جديد للوكالة", + why_now_ar="إضافة عميل تفعّل Diagnostic وProof Pack باسم الوكالة.", + recommended_action_ar="شغّل intake قصير ثم Diagnostic على القطاع المختار.", + risk_level="low", + buttons=[ + _btn("ابدأ intake", "start_intake"), + _btn("شغّل diagnostic", "run_diagnostic"), + _btn("تخطي", "skip"), + ], + action_mode="suggest_only", + proof_impact=["partner_client_onboarded", "opportunity_created"], + ), + _card( + card_id=f"{r}_proof_1", + role=r, + ctype=CardType.PROOF.value, + title_ar="📊 Co-branded Proof Pack", + why_now_ar="العميل أنهى أسبوعاً؛ جاهز لتسليم تقرير مختوم باسم الوكالة.", + recommended_action_ar="راجع الأرقام واطلب موافقة العميل قبل الإرسال.", + risk_level="low", + buttons=[ + _btn("اعرض المسودة", "open_proof_draft"), + _btn("اطلب موافقة", "request_client_approval"), + _btn("تخطي", "skip"), + ], + action_mode="approval_required", + proof_impact=["proof_generated", "partner_revenue_event"], + ), + ] + + +def _support_cards() -> list[dict[str, Any]]: + r = UserRole.SUPPORT.value + return [ + _card( + card_id=f"{r}_ticket_1", + role=r, + ctype=CardType.SUPPORT.value, + title_ar="🛠️ طلب دعم — ربط Gmail", + why_now_ar="العميل لم يكمل OAuth؛ التذكرة P2.", + recommended_action_ar="أرسل رابط الدليل + تحقق من صلاحيات المسودة فقط.", + risk_level="low", + buttons=[ + _btn("افتح دليل الربط", "open_connector_doc"), + _btn("صعّد", "escalate_p1"), + _btn("أغلق", "resolve"), + ], + action_mode="suggest_only", + proof_impact=["support_ticket_created", "sla_started"], + ), + ] + + +def _delivery_cards() -> list[dict[str, Any]]: + r = UserRole.SERVICE_DELIVERY.value + return [ + _card( + card_id=f"{r}_sla_1", + role=r, + ctype=CardType.DELIVERY.value, + title_ar="⏱️ تسليم — بيانات ناقصة", + why_now_ar="Pilot يوم ٣: لم يُرفع ملف القطاع.", + recommended_action_ar="طلب الملف قبل توليد الفرص التالية.", + risk_level="medium", + buttons=[ + _btn("اطلب بيانات", "draft_data_request"), + _btn("إشعار العميل", "draft_client_nudge"), + _btn("تخطي", "skip"), + ], + action_mode="draft_only", + proof_impact=["delivery_blocked_input", "draft_created"], + ), + _card( + card_id=f"{r}_proof_done", + role=r, + ctype=CardType.PROOF.value, + title_ar="✅ Proof Pack — جاهز للتسليم", + why_now_ar="اكتملت المهام المتفق عليها للأسبوع.", + recommended_action_ar="مراجعة داخلية ثم إرسال يدوي للعميل (بدون إرسال تلقائي).", + risk_level="low", + buttons=[ + _btn("اعرض الملخص", "open_proof"), + _btn("سجّل التسليم", "log_delivery"), + _btn("تخطي", "skip"), + ], + action_mode="approval_required", + proof_impact=["proof_generated", "approval_requested"], + ), + ] + + +_ROLE_BUILDERS: dict[str, list[dict[str, Any]]] = { + UserRole.CEO.value: _ceo_cards(), + UserRole.SALES_MANAGER.value: _sales_cards(), + UserRole.GROWTH_MANAGER.value: _growth_cards(), + UserRole.AGENCY_PARTNER.value: _agency_cards(), + UserRole.SUPPORT.value: _support_cards(), + UserRole.SERVICE_DELIVERY.value: _delivery_cards(), +} + + +def build_role_command_feed(role: str) -> dict[str, Any]: + """Return up to MAX_CARDS_VISIBLE normalized cards for ``role``.""" + cards = list(_ROLE_BUILDERS.get(role, [])) + visible = cards[:MAX_CARDS_VISIBLE] + return { + "role": role, + "cards": visible, + "card_count": len(visible), + "max_visible": MAX_CARDS_VISIBLE, + "demo": True, + } + + +def build_whatsapp_daily_brief_lines(role: str) -> list[str]: + """Short lines suitable for WhatsApp (approval-first; no auto-send).""" + feed = build_role_command_feed(role) + lines: list[str] = [] + for c in feed["cards"][:3]: + lines.append(f"• {c['title_ar']}") + if not lines: + lines = ["• لا توجد قرارات معروضة لهذا الدور."] + lines.append("") + lines.append("افتح لوحة الموافقات للتفاصيل — لا يُرسل شيء تلقائياً من المنصة.") + return lines diff --git a/dealix/auto_client_acquisition/revenue_company_os/cards.py b/dealix/auto_client_acquisition/revenue_company_os/cards.py new file mode 100644 index 00000000..0ed88465 --- /dev/null +++ b/dealix/auto_client_acquisition/revenue_company_os/cards.py @@ -0,0 +1,109 @@ +"""Role-Based Revenue Command Cards — schema, validation, safety constants. + +No live send, no scraping, no cold WhatsApp. Cards are decision units (max 3 buttons). +""" + +from __future__ import annotations + +from enum import Enum +from typing import Any, Literal, TypedDict + +MAX_CARD_BUTTONS = 3 +MAX_CARDS_VISIBLE = 7 + +RiskLevel = Literal["low", "medium", "high"] +ActionMode = Literal["approval_required", "draft_only", "suggest_only", "blocked"] + + +class UserRole(str, Enum): + """Operational personas (one interface, role-scoped feeds).""" + + CEO = "ceo" + SALES_MANAGER = "sales_manager" + GROWTH_MANAGER = "growth_manager" + AGENCY_PARTNER = "agency_partner" + SUPPORT = "support" + SERVICE_DELIVERY = "service_delivery" + + +class CardType(str, Enum): + OPPORTUNITY = "opportunity" + PARTNER = "partner" + DEAL_FOLLOWUP = "deal_followup" + NEGOTIATION = "negotiation" + CLOSE = "close" + PROOF = "proof" + SUPPORT = "support" + CEO_BRIEF = "ceo_brief" + GROWTH_PLAN = "growth_plan" + RISK = "risk" + DELIVERY = "delivery" + + +class ForbiddenPattern(str, Enum): + """Substrings that must not appear in actions or button machine keys.""" + + LINKEDIN_SCRAPE = "linkedin_scrape" + COLD_WHATSAPP = "cold_whatsapp" + WHATSAPP_BLAST = "whatsapp_blast" + LIVE_GMAIL_SEND = "live_gmail_send" + MOYASAR_CHARGE = "moyasar_charge" + + +FORBIDDEN_SUBSTRINGS: tuple[str, ...] = tuple(p.value for p in ForbiddenPattern) + + +class CardButton(TypedDict, total=False): + label_ar: str + action: str + + +def _buttons_payload(buttons: list[CardButton]) -> list[CardButton]: + if len(buttons) > MAX_CARD_BUTTONS: + return buttons[:MAX_CARD_BUTTONS] + return buttons + + +def assert_safe_card_copy(card: dict[str, Any]) -> None: + """Raise ValueError if copy suggests blocked automation (tests + CI guard).""" + blob = " ".join( + str(x).lower() + for x in ( + card.get("recommended_action_ar"), + card.get("why_now_ar"), + card.get("title_ar"), + " ".join(b.get("action", "") + " " + b.get("label_ar", "") for b in card.get("buttons") or []), + ) + if x + ) + for bad in FORBIDDEN_SUBSTRINGS: + if bad in blob: + raise ValueError(f"unsafe card copy references forbidden pattern: {bad}") + + +def normalize_card(card: dict[str, Any]) -> dict[str, Any]: + """Clamp buttons, ensure lists for proof_impact, run safety assert.""" + out = dict(card) + buttons = list(out.get("buttons") or []) + out["buttons"] = _buttons_payload(buttons) # type: ignore[arg-type] + pi = out.get("proof_impact") + if pi is None: + out["proof_impact"] = [] + elif isinstance(pi, str): + out["proof_impact"] = [pi] + elif isinstance(pi, list): + out["proof_impact"] = [str(x) for x in pi] + else: + out["proof_impact"] = [] + assert_safe_card_copy(out) + return out + + +def normalize_role_param(role: str | None) -> str: + r = (role or "").strip().lower().replace("-", "_") + aliases = {"agency": "agency_partner", "delivery": "service_delivery", "cs": "support"} + return aliases.get(r, r) + + +def is_known_role(role: str) -> bool: + return role in {e.value for e in UserRole} diff --git a/dealix/landing/command-center.html b/dealix/landing/command-center.html index 1f468bfd..c91a6e6f 100644 --- a/dealix/landing/command-center.html +++ b/dealix/landing/command-center.html @@ -553,6 +553,26 @@ + +
+
+
+

كروت القرار حسب الدور — Revenue Command

+

واجهة واحدة، تغذية مختلفة لكل دور. البيانات من API (عند تشغيل الخادم على نفس النطاق).

+
+
جاري التحميل…
+
+
+ + + + + + +
+
+
+

🎯 3 قرارات يجب اتخاذها اليوم

@@ -578,8 +598,51 @@
📌 ملاحظة: هذا preview للوحة Dealix — البيانات في هذه النسخة العامة افتراضية لأغراض العرض. عند الاشتراك، اللوحة تُربط مباشرة بـ Saudi Revenue Graph الخاص بشركتك وتُحدّث في الوقت الحقيقي عبر /api/v1/command-center/snapshot. + كروت الأدوار تُحمّل من /api/v1/cards/feed (موافقة أولاً — لا إرسال تلقائي). كل التوصيات مبنية على Pulse القطاعي + بيانات شركتك التراكمية + 11 AI Agents يعملون 24/7.
+ diff --git a/dealix/scripts/smoke_inprocess.py b/dealix/scripts/smoke_inprocess.py index e7a0e9ea..21698eee 100644 --- a/dealix/scripts/smoke_inprocess.py +++ b/dealix/scripts/smoke_inprocess.py @@ -37,6 +37,9 @@ PATHS = [ "/api/v1/operator/whatsapp/daily-brief", "/api/v1/operator/tools/matrix", "/api/v1/revenue-os/company-os/command-feed/demo", + "/api/v1/cards/feed?role=ceo", + "/api/v1/cards/feed?role=sales_manager", + "/api/v1/cards/whatsapp/daily-brief?role=growth_manager", "/api/v1/revenue-os/company-os/work-units/demo", "/api/v1/revenue-os/company-os/self-improvement/weekly-report", "/api/v1/customer-ops/onboarding/checklist", diff --git a/dealix/tests/test_role_based_cards.py b/dealix/tests/test_role_based_cards.py new file mode 100644 index 00000000..ca002e2f --- /dev/null +++ b/dealix/tests/test_role_based_cards.py @@ -0,0 +1,140 @@ +"""Role-based Revenue Command Cards — schema, safety, API.""" + +from __future__ import annotations + +import pytest +from fastapi.testclient import TestClient + +from api.main import create_app +from auto_client_acquisition.revenue_company_os.card_factory import build_role_command_feed +from auto_client_acquisition.revenue_company_os.cards import ( + UserRole, + assert_safe_card_copy, + is_known_role, + normalize_card, + normalize_role_param, +) + + +def _every_card(role: str) -> list[dict]: + return build_role_command_feed(role)["cards"] + + +@pytest.mark.parametrize("role", [e.value for e in UserRole]) +def test_each_role_has_cards(role: str) -> None: + cards = _every_card(role) + assert len(cards) >= 1 + + +@pytest.mark.parametrize("role", [e.value for e in UserRole]) +def test_buttons_max_three(role: str) -> None: + for c in _every_card(role): + assert len(c.get("buttons") or []) <= 3 + + +@pytest.mark.parametrize("role", [e.value for e in UserRole]) +def test_arabic_titles_present(role: str) -> None: + for c in _every_card(role): + t = c.get("title_ar") + assert isinstance(t, str) and len(t.strip()) >= 2 + + +def test_no_forbidden_patterns_in_demo_cards() -> None: + for role in UserRole: + for c in _every_card(role.value): + blob = " ".join( + str(x) + for x in ( + c.get("recommended_action_ar"), + c.get("why_now_ar"), + c.get("title_ar"), + ) + if x + ).lower() + assert "linkedin_scrape" not in blob + assert "cold_whatsapp" not in blob + + +def test_normalize_rejects_unsafe_card() -> None: + bad = { + "card_id": "x", + "tenant_id": "t", + "role": "ceo", + "type": "risk", + "title_ar": "bad", + "why_now_ar": "bad", + "recommended_action_ar": "use linkedin_scrape now", + "risk_level": "high", + "buttons": [{"label_ar": "ok", "action": "a"}], + "action_mode": "blocked", + "proof_impact": [], + "status": "pending", + } + with pytest.raises(ValueError): + normalize_card(bad) + + +def test_normalize_role_param_aliases() -> None: + assert normalize_role_param("Agency") == "agency_partner" + assert normalize_role_param("SERVICE-DELIVERY") == "service_delivery" + + +def test_is_known_role() -> None: + assert is_known_role("ceo") + assert not is_known_role("cfo") + + +def test_decision_requires_approval_returns_draft_only() -> None: + client = TestClient(create_app()) + cid = "sales_manager_deal_1" + r = client.post( + f"/api/v1/cards/{cid}/decision", + json={"action": "approve", "button_action": "draft_followup_email"}, + ) + assert r.status_code == 200 + data = r.json() + assert data.get("execution_mode") == "draft_only" + assert "draft_export_ar" in data + assert "proof_events" in data + assert any("decision_recorded" in x for x in data["proof_events"]) + + +def test_unknown_card_returns_404() -> None: + client = TestClient(create_app()) + r = client.post("/api/v1/cards/unknown_id/decision", json={"action": "skip"}) + assert r.status_code == 404 + + +def test_feed_unknown_role_400() -> None: + client = TestClient(create_app()) + r = client.get("/api/v1/cards/feed?role=cfo") + assert r.status_code == 400 + body = r.json() + detail = body.get("detail", body) + if isinstance(detail, dict): + assert detail.get("error") == "unknown_role" or "allowed" in detail + else: + assert "unknown" in str(detail).lower() + + +def test_whatsapp_brief_no_auto_send_flag() -> None: + client = TestClient(create_app()) + r = client.get("/api/v1/cards/whatsapp/daily-brief?role=ceo") + assert r.status_code == 200 + body = r.json() + assert body.get("no_auto_send") is True + assert isinstance(body.get("lines_ar"), list) + + +def test_api_feed_matches_factory() -> None: + client = TestClient(create_app()) + for role in ("ceo", "agency_partner"): + api = client.get(f"/api/v1/cards/feed?role={role}").json() + direct = build_role_command_feed(role) + assert api["card_count"] == direct["card_count"] + + +def test_assert_safe_card_copy_passes_demo() -> None: + for role in UserRole: + for c in _every_card(role.value): + assert_safe_card_copy(c)