# LeadPulse API — Python dependencies
fastapi>=0.110.0
uvicorn[standard]>=0.29.0
sqlalchemy>=2.0.0
pydantic>=2.7.0
pydantic-settings>=2.2.0
python-multipart>=0.0.9
pytest>=8.0.0
httpx>=0.27.0
pytest-asyncio>=0.23.0

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from database import Base, engine
from routers import leads, export


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create tables on startup
    Base.metadata.create_all(bind=engine)
    yield


app = FastAPI(
    title="LeadPulse API",
    description="B2B lead management & scoring API",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(leads.router, prefix="/leads", tags=["leads"])
app.include_router(export.router, prefix="/leads", tags=["export"])


@app.get("/")
def health():
    return {"status": "ok", "service": "leadpulse-api", "version": "1.0.0"}

from sqlalchemy import Column, Integer, String, Float, DateTime, Enum
from sqlalchemy.sql import func
import enum
from database import Base


class LeadStatus(str, enum.Enum):
    new = "new"
    contacted = "contacted"
    qualified = "qualified"
    lost = "lost"


class Lead(Base):
    __tablename__ = "leads"

    id = Column(Integer, primary_key=True, index=True)
    nombre = Column(String(255), nullable=False)
    empresa = Column(String(255), nullable=False)
    email = Column(String(255), unique=True, index=True, nullable=False)
    cargo = Column(String(255), nullable=True)
    linkedin_url = Column(String(500), nullable=True)
    score = Column(Float, default=0.0)
    estado = Column(
        Enum(LeadStatus),
        default=LeadStatus.new,
        nullable=False,
    )
    created_at = Column(DateTime, server_default=func.now(), nullable=False)

from models import Lead

# Cargo tiers for B2B scoring
_C_LEVEL_TITLES = {"ceo", "cto", "coo", "cfo", "cpo", "chief", "founder", "co-founder"}
_DIRECTOR_TITLES = {"director", "vp", "vice president", "head of"}
_MANAGER_TITLES = {"manager", "lead", "principal"}


def compute_score(lead: Lead) -> float:
    """Calculate lead score (0–100) based on job title and LinkedIn presence."""
    base_score: float = 40.0  # default for unknown roles

    if lead.cargo:
        title_lower = lead.cargo.lower()
        if any(kw in title_lower for kw in _C_LEVEL_TITLES):
            base_score = 90.0
        elif any(kw in title_lower for kw in _DIRECTOR_TITLES):
            base_score = 75.0
        elif any(kw in title_lower for kw in _MANAGER_TITLES):
            base_score = 60.0

    # LinkedIn bonus: verified lead identity
    if lead.linkedin_url and lead.linkedin_url.strip():
        base_score = min(100.0, base_score + 10.0)

    return round(base_score, 2)

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from typing import Optional
from database import get_db
from models import Lead, LeadStatus
from schemas import LeadCreate, LeadUpdate, LeadResponse
from auth import verify_api_key
from scoring import compute_score

router = APIRouter(dependencies=[Depends(verify_api_key)])


@router.get("", response_model=list[LeadResponse])
def list_leads(
    estado: Optional[LeadStatus] = Query(None),
    score_min: Optional[float] = Query(None, ge=0, le=100),
    skip: int = Query(0, ge=0),
    limit: int = Query(50, ge=1, le=500),
    db: Session = Depends(get_db),
):
    q = db.query(Lead)
    if estado:
        q = q.filter(Lead.estado == estado)
    if score_min is not None:
        q = q.filter(Lead.score >= score_min)
    return q.offset(skip).limit(limit).all()


@router.post("", response_model=LeadResponse, status_code=status.HTTP_201_CREATED)
def create_lead(payload: LeadCreate, db: Session = Depends(get_db)):
    existing = db.query(Lead).filter(Lead.email == payload.email).first()
    if existing:
        raise HTTPException(
            status_code=409, detail=f"Lead with email {payload.email!r} already exists"
        )
    lead = Lead(**payload.model_dump())
    lead.score = compute_score(lead)
    db.add(lead)
    db.commit()
    db.refresh(lead)
    return lead


@router.get("/{lead_id}", response_model=LeadResponse)
def get_lead(lead_id: int, db: Session = Depends(get_db)):
    lead = db.query(Lead).filter(Lead.id == lead_id).first()
    if not lead:
        raise HTTPException(status_code=404, detail="Lead not found")
    return lead


@router.put("/{lead_id}", response_model=LeadResponse)
def update_lead(lead_id: int, payload: LeadUpdate, db: Session = Depends(get_db)):
    lead = db.query(Lead).filter(Lead.id == lead_id).first()
    if not lead:
        raise HTTPException(status_code=404, detail="Lead not found")
    for field, value in payload.model_dump(exclude_unset=True).items():
        setattr(lead, field, value)
    lead.score = compute_score(lead)
    db.commit()
    db.refresh(lead)
    return lead


@router.delete("/{lead_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_lead(lead_id: int, db: Session = Depends(get_db)):
    lead = db.query(Lead).filter(Lead.id == lead_id).first()
    if not lead:
        raise HTTPException(status_code=404, detail="Lead not found")
    db.delete(lead)
    db.commit()


@router.post("/{lead_id}/score", response_model=LeadResponse)
def rescore_lead(lead_id: int, db: Session = Depends(get_db)):
    lead = db.query(Lead).filter(Lead.id == lead_id).first()
    if not lead:
        raise HTTPException(status_code=404, detail="Lead not found")
    lead.score = compute_score(lead)
    db.commit()
    db.refresh(lead)
    return lead

import csv
import io
from fastapi import APIRouter, Depends, Query
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from typing import Optional
from database import get_db
from models import Lead, LeadStatus
from auth import verify_api_key

router = APIRouter(dependencies=[Depends(verify_api_key)])

_CSV_FIELDS = ["id", "nombre", "empresa", "email", "cargo", "linkedin_url", "score", "estado", "created_at"]


@router.get("/export/csv")
def export_csv(
    estado: Optional[LeadStatus] = Query(None),
    score_min: Optional[float] = Query(None, ge=0, le=100),
    db: Session = Depends(get_db),
):
    q = db.query(Lead)
    if estado:
        q = q.filter(Lead.estado == estado)
    if score_min is not None:
        q = q.filter(Lead.score >= score_min)
    rows = q.order_by(Lead.score.desc()).all()

    output = io.StringIO()
    writer = csv.DictWriter(output, fieldnames=_CSV_FIELDS)
    writer.writeheader()
    for lead in rows:
        writer.writerow({f: getattr(lead, f) for f in _CSV_FIELDS})

    output.seek(0)
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": 'attachment; filename="leads.csv"'},
    )

from fastapi import Header, HTTPException, status
from typing import Annotated
import os

_API_KEY = os.environ.get("API_KEY", "")


def verify_api_key(x_api_key: Annotated[str, Header(alias="X-API-Key")] = "") -> None:
    """FastAPI dependency — raises 401 if API key is missing or invalid."""
    if not _API_KEY:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="API_KEY env var not configured",
        )
    if x_api_key != _API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing X-API-Key header",
            headers={"WWW-Authenticate": "ApiKey"},
        )

import pytest
from httpx import AsyncClient
from main import app

HEADERS = {"X-API-Key": "test-key-1234"}

@pytest.mark.asyncio
async def test_create_lead(client: AsyncClient):
    resp = await client.post("/leads", json={
        "nombre": "Ana García",
        "empresa": "TechScale SL",
        "email": "ana@techscale.io",
        "cargo": "CEO",
        "linkedin_url": "https://linkedin.com/in/anagarcia",
    }, headers=HEADERS)
    assert resp.status_code == 201
    data = resp.json()
    assert data["score"] == 100.0  # CEO (90) + LinkedIn (+10)
    assert data["estado"] == "new"

@pytest.mark.asyncio
async def test_list_with_score_filter(client: AsyncClient):
    resp = await client.get("/leads?score_min=80", headers=HEADERS)
    assert resp.status_code == 200
    for lead in resp.json():
        assert lead["score"] >= 80

@pytest.mark.asyncio
async def test_duplicate_email_returns_409(client: AsyncClient):
    payload = {"nombre": "X", "empresa": "Y", "email": "dup@test.com", "cargo": "Manager"}
    await client.post("/leads", json=payload, headers=HEADERS)
    resp = await client.post("/leads", json=payload, headers=HEADERS)
    assert resp.status_code == 409

@pytest.mark.asyncio
async def test_unauthorized_returns_401(client: AsyncClient):
    resp = await client.get("/leads")
    assert resp.status_code == 401

@pytest.mark.asyncio
async def test_export_csv(client: AsyncClient):
    resp = await client.get("/leads/export/csv", headers=HEADERS)
    assert resp.status_code == 200
    assert "text/csv" in resp.headers["content-type"]
    assert "nombre" in resp.text

FROM python:3.12-slim

WORKDIR /app

# Install dependencies first (layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy source
COPY . .

# Railway injects PORT env var
ENV PORT=8000
EXPOSE $PORT

CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port ${PORT}"]

# === LeadPulse API — Environment Variables ===
# Copy to .env and fill in values before running

# Authentication: shared secret for X-API-Key header
API_KEY=your-secret-api-key-here

# SQLAlchemy database URL
# SQLite (development):  sqlite:///./leads.db
# PostgreSQL (prod):      postgresql://user:pass@host:5432/leadpulse
DATABASE_URL=sqlite:///./leads.db

# Server port (Railway injects this automatically)
PORT=8000