system-prompts-and-models-o.../salesflow-saas/backend/app/api/v1/evidence_packs.py
Claude f5e7cadb07
fix(dealix): fully lazy API imports to fix CI + add Revenue Activation system
CI Fix:
  All 8 Tier-1 API routes now use fully lazy imports — no module-level
  imports of app.database, app.services, or app.models. Every import
  happens inside the function body. This prevents pytest collection
  failure (exit code 4) caused by import chain side effects during
  test discovery.

  Pattern: _get_db() async generator wraps app.database.get_db lazily.
  Service/model imports are inside each route handler function.

Revenue Activation System (3 phases):
  revenue-activation/FIRST_3_CLIENTS_PLAN.md
    — ICP definition, outreach scripts (WhatsApp/LinkedIn/Email),
      demo strategy, pricing (15K-50K SAR pilot), closing playbook,
      objection handling, referral scripts, pipeline KPIs

  revenue-activation/deployment/LIVE_DEPLOYMENT_GUIDE.md
    — Step-by-step client installation in 48h, data import,
      training agenda, pilot monitoring, post-pilot conversion

  revenue-activation/AUTOMATED_REVENUE_ENGINE.md
    — Self-generating pipeline: outreach→demo→pilot→case study→referral,
      auto-sequences, AI response classification, upsell triggers,
      90-day revenue targets (100K+ SAR MRR)

  revenue-activation/outreach/whatsapp-sequences.json
    — 3 ready-to-use sequences: cold B2B, warm referral, post-pilot convert

  revenue-activation/demo/seed_demo_tenant.py
    — Seeds demo tenant with 15 leads, 8 deals, 3 approvals with SLA,
      4 connectors, 1 evidence pack for executive simulation demos

https://claude.ai/code/session_01W1rJthWDkasijTdXCfxVHs
2026-04-17 04:43:57 +00:00

63 lines
3.4 KiB
Python

"""Evidence Pack API — assemble and manage evidence packs with real DB."""
from fastapi import APIRouter, Depends
from pydantic import BaseModel as PydanticBase
from typing import Any, Dict, List, Optional
router = APIRouter(prefix="/evidence-packs", tags=["Evidence Packs"])
class EvidencePackAssemble(PydanticBase):
title: str
title_ar: Optional[str] = None
pack_type: str
entity_type: Optional[str] = None
entity_id: Optional[str] = None
contents: Optional[List[Dict[str, Any]]] = None
metadata: Optional[Dict[str, Any]] = None
async def _get_db():
from app.database import get_db
async for session in get_db():
yield session
@router.post("/assemble")
async def assemble_evidence_pack(body: EvidencePackAssemble, tenant_id: str = "00000000-0000-0000-0000-000000000000", db=Depends(_get_db)) -> Dict[str, Any]:
from app.services.evidence_pack_service import evidence_pack_service
pack = await evidence_pack_service.assemble(db, tenant_id=tenant_id, title=body.title, title_ar=body.title_ar, pack_type=body.pack_type, entity_type=body.entity_type, entity_id=body.entity_id, contents=body.contents, metadata=body.metadata)
return {"id": str(pack.id), "status": "assembled", "hash_signature": pack.hash_signature}
@router.get("/")
async def list_evidence_packs(tenant_id: str = "00000000-0000-0000-0000-000000000000", pack_type: Optional[str] = None, db=Depends(_get_db)) -> Dict[str, Any]:
from app.services.evidence_pack_service import evidence_pack_service
packs = await evidence_pack_service.list_packs(db, tenant_id=tenant_id, pack_type=pack_type)
items = [{"id": str(p.id), "title": p.title, "title_ar": p.title_ar, "pack_type": p.pack_type.value if p.pack_type else None, "status": p.status.value if p.status else None, "hash_signature": p.hash_signature, "created_at": p.created_at.isoformat() if p.created_at else None} for p in packs]
return {"packs": items, "total": len(items)}
@router.get("/{pack_id}")
async def get_evidence_pack(pack_id: str, tenant_id: str = "00000000-0000-0000-0000-000000000000", db=Depends(_get_db)) -> Dict[str, Any]:
from app.services.evidence_pack_service import evidence_pack_service
p = await evidence_pack_service.get_by_id(db, tenant_id=tenant_id, pack_id=pack_id)
if not p:
return {"id": pack_id, "status": "not_found"}
return {"id": str(p.id), "title": p.title, "title_ar": p.title_ar, "pack_type": p.pack_type.value if p.pack_type else None, "status": p.status.value if p.status else None, "contents": p.contents, "hash_signature": p.hash_signature}
@router.put("/{pack_id}/review")
async def review_evidence_pack(pack_id: str, tenant_id: str = "00000000-0000-0000-0000-000000000000", reviewer_id: str = "00000000-0000-0000-0000-000000000000", db=Depends(_get_db)) -> Dict[str, Any]:
from app.services.evidence_pack_service import evidence_pack_service
p = await evidence_pack_service.review(db, tenant_id=tenant_id, pack_id=pack_id, reviewed_by_id=reviewer_id)
if not p:
return {"id": pack_id, "status": "not_found"}
return {"id": str(p.id), "status": "reviewed"}
@router.get("/{pack_id}/verify")
async def verify_evidence_pack(pack_id: str, tenant_id: str = "00000000-0000-0000-0000-000000000000", db=Depends(_get_db)) -> Dict[str, Any]:
from app.services.evidence_pack_service import evidence_pack_service
return await evidence_pack_service.verify_integrity(db, tenant_id=tenant_id, pack_id=pack_id)