maskanx_crm_backend/app/cli.py

import argparse
from datetime import UTC, datetime, timedelta
from decimal import Decimal

from sqlalchemy import delete, select
from sqlalchemy.orm import selectinload

from app.core.config import get_settings
from app.core.database import SessionLocal
from app.core.security import hash_password
from app.models import (
    Activity,
    Contact,
    Lead,
    LeadSource,
    LeadType,
    Organization,
    Pipeline,
    Product,
    Quote,
    Stage,
    Tenant,
    User,
)


def get_or_create(db, model, defaults: dict, **lookup):
    instance = db.scalar(select(model).filter_by(**lookup))
    if instance:
        return instance
    instance = model(**lookup, **defaults)
    db.add(instance)
    db.flush()
    return instance


def seed(with_demo: bool) -> None:
    settings = get_settings()
    if (
        settings.environment.lower() == "production"
        and settings.bootstrap_admin_password == "change-this-before-first-run"
    ):
        raise RuntimeError(
            "Set MASKAN_CRM_BOOTSTRAP_ADMIN_PASSWORD before production seeding.",
        )

    with SessionLocal() as db:
        tenant = get_or_create(
            db,
            Tenant,
            {
                "name": settings.bootstrap_company,
                "mode": settings.mode,
                "settings": {"currency": "USD", "timezone": "Asia/Kolkata"},
            },
            slug=settings.bootstrap_workspace,
        )
        owner = db.scalar(
            select(User).where(
                User.tenant_id == tenant.id,
                User.email == settings.bootstrap_admin_email.lower(),
            ),
        )
        if owner is None:
            owner = User(
                tenant_id=tenant.id,
                email=settings.bootstrap_admin_email.lower(),
                full_name="Maskan CRM Owner",
                password_hash=hash_password(settings.bootstrap_admin_password),
                role="owner",
            )
            db.add(owner)
            db.flush()

        pipeline = db.scalar(
            select(Pipeline)
            .where(Pipeline.tenant_id == tenant.id, Pipeline.is_default.is_(True))
            .options(selectinload(Pipeline.stages)),
        )
        if pipeline is None:
            pipeline = Pipeline(
                tenant_id=tenant.id,
                name="Sales Pipeline",
                is_default=True,
            )
            db.add(pipeline)
            db.flush()

        stage_specs = [
            ("New", 1, 10, "#64748b"),
            ("Qualified", 2, 30, "#2563eb"),
            ("Proposal", 3, 60, "#7c3aed"),
            ("Negotiation", 4, 80, "#d97706"),
            ("Won", 5, 100, "#059669"),
        ]
        stages = {}
        for name, position, probability, color in stage_specs:
            stages[name] = get_or_create(
                db,
                Stage,
                {
                    "tenant_id": tenant.id,
                    "position": position,
                    "probability": probability,
                    "color": color,
                },
                pipeline_id=pipeline.id,
                name=name,
            )

        sources = {}
        for name in ["MaskanX", "LinkedIn", "Meta Ads", "Website", "Referral"]:
            sources[name] = get_or_create(
                db,
                LeadSource,
                {"tenant_id": tenant.id},
                name=name,
            )

        lead_types = {}
        for name in ["New Business", "Expansion", "Renewal"]:
            lead_types[name] = get_or_create(
                db,
                LeadType,
                {"tenant_id": tenant.id},
                name=name,
            )

        if with_demo:
            seed_demo_data(
                db,
                tenant=tenant,
                owner=owner,
                pipeline=pipeline,
                stages=stages,
                sources=sources,
                lead_types=lead_types,
            )

        db.commit()
        print(f"Seed complete for workspace '{tenant.slug}'.")


def undo_seed() -> None:
    settings = get_settings()
    if settings.environment.lower() == "production":
        raise RuntimeError("Seed undo is disabled in production.")

    with SessionLocal() as db:
        result = db.execute(
            delete(Tenant).where(Tenant.slug == settings.bootstrap_workspace),
        )
        db.commit()
        if result.rowcount:
            print(
                f"Seed data removed for workspace "
                f"'{settings.bootstrap_workspace}'.",
            )
        else:
            print("No bootstrap seed data was found.")


def seed_demo_data(
    db,
    *,
    tenant: Tenant,
    owner: User,
    pipeline: Pipeline,
    stages: dict[str, Stage],
    sources: dict[str, LeadSource],
    lead_types: dict[str, LeadType],
) -> None:
    company = get_or_create(
        db,
        Organization,
        {
            "tenant_id": tenant.id,
            "industry": "Software",
            "website": "https://example.com",
            "owner_id": owner.id,
        },
        name="Northstar Labs",
    )
    contact = db.scalar(
        select(Contact).where(
            Contact.tenant_id == tenant.id,
            Contact.primary_email == "maya@example.com",
        ),
    )
    if contact is None:
        contact = Contact(
            tenant_id=tenant.id,
            first_name="Maya",
            last_name="Shah",
            job_title="Chief Technology Officer",
            primary_email="maya@example.com",
            emails=[{"label": "work", "value": "maya@example.com", "primary": True}],
            phones=[{"label": "work", "value": "+91 90000 00000", "primary": True}],
            lifecycle_stage="opportunity",
            lead_source="LinkedIn",
            score=82,
            organization_id=company.id,
            owner_id=owner.id,
        )
        db.add(contact)
        db.flush()

    lead_specs = [
        ("AI workflow modernization", "Qualified", "25000", "LinkedIn", 82),
        ("Customer support automation", "Proposal", "18000", "Website", 76),
        ("Custom SaaS build", "New", "42000", "Meta Ads", 61),
    ]
    for index, (title, stage_name, value, source_name, score) in enumerate(lead_specs):
        existing = db.scalar(
            select(Lead).where(
                Lead.tenant_id == tenant.id,
                Lead.title == title,
            ),
        )
        if existing is None:
            db.add(
                Lead(
                    tenant_id=tenant.id,
                    title=title,
                    description="Demo opportunity for local development.",
                    value=Decimal(value),
                    currency="USD",
                    score=score,
                    position=index + 1,
                    contact_id=contact.id,
                    organization_id=company.id,
                    owner_id=owner.id,
                    pipeline_id=pipeline.id,
                    stage_id=stages[stage_name].id,
                    source_id=sources[source_name].id,
                    type_id=lead_types["New Business"].id,
                ),
            )

    if db.scalar(
        select(Activity).where(
            Activity.tenant_id == tenant.id,
            Activity.title == "Discovery call with Northstar Labs",
        ),
    ) is None:
        db.add(
            Activity(
                tenant_id=tenant.id,
                activity_type="call",
                title="Discovery call with Northstar Labs",
                details="Review automation priorities and success metrics.",
                starts_at=datetime.now(UTC) + timedelta(days=1),
                ends_at=datetime.now(UTC) + timedelta(days=1, minutes=30),
                due_at=datetime.now(UTC) + timedelta(days=1),
                owner_id=owner.id,
                contact_id=contact.id,
                organization_id=company.id,
            ),
        )

    if db.scalar(
        select(Product).where(
            Product.tenant_id == tenant.id,
            Product.sku == "AI-AUTOMATION",
        ),
    ) is None:
        db.add(
            Product(
                tenant_id=tenant.id,
                sku="AI-AUTOMATION",
                name="AI Automation Implementation",
                description="Discovery, design, implementation, and rollout.",
                quantity=100,
                price=Decimal("15000"),
                currency="USD",
            ),
        )

    if db.scalar(
        select(Quote).where(
            Quote.tenant_id == tenant.id,
            Quote.number == "Q-1001",
        ),
    ) is None:
        db.add(
            Quote(
                tenant_id=tenant.id,
                number="Q-1001",
                subject="AI workflow modernization proposal",
                status="draft",
                currency="USD",
                subtotal=Decimal("25000"),
                grand_total=Decimal("25000"),
                contact_id=contact.id,
                organization_id=company.id,
                owner_id=owner.id,
                expires_at=datetime.now(UTC) + timedelta(days=30),
            ),
        )


def main() -> None:
    parser = argparse.ArgumentParser(prog="maskan-crm")
    subparsers = parser.add_subparsers(dest="command", required=True)
    seed_parser = subparsers.add_parser("seed")
    seed_parser.add_argument("--without-demo", action="store_true")
    subparsers.add_parser("seed-undo")
    args = parser.parse_args()

    if args.command == "seed":
        seed(with_demo=not args.without_demo and get_settings().seed_demo)
    elif args.command == "seed-undo":
        undo_seed()


if __name__ == "__main__":
    main()