Spaces:
Sleeping
Sleeping
File size: 3,348 Bytes
76089f2 3a19693 76089f2 3a19693 76089f2 3a19693 76089f2 3a19693 76089f2 3a19693 76089f2 3a19693 76089f2 3a19693 76089f2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | import io
import pandas as pd
from fastapi import APIRouter, Depends, Query, HTTPException
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from typing import Optional
from database import get_db
from models import Lead
from auth import decode_token
router = APIRouter(prefix="/export", tags=["export"])
def _verify_token_param(token: Optional[str], db: Session) -> None:
"""Allow auth via ?token= query param for direct download links."""
if not token:
raise HTTPException(status_code=401, detail="Not authenticated")
payload = decode_token(token)
if not payload:
raise HTTPException(status_code=401, detail="Invalid or expired token")
EXPORT_FIELDS = [
"id", "company_name", "contact_name", "email", "phone", "website",
"industry", "company_size", "location", "linkedin_url",
"source", "status", "notes", "created_at",
]
def leads_to_df(leads: list[Lead]) -> pd.DataFrame:
rows = []
for lead in leads:
row = {f: getattr(lead, f, None) for f in EXPORT_FIELDS}
if lead.custom_fields:
for k, v in lead.custom_fields.items():
row[f"custom_{k}"] = v
rows.append(row)
return pd.DataFrame(rows)
def get_filtered_leads(
db: Session,
status: Optional[str],
source: Optional[str],
industry: Optional[str],
ids: Optional[str],
) -> list[Lead]:
query = db.query(Lead)
if ids:
id_list = [int(i) for i in ids.split(",") if i.strip().isdigit()]
query = query.filter(Lead.id.in_(id_list))
if status:
query = query.filter(Lead.status == status)
if source:
query = query.filter(Lead.source == source)
if industry:
query = query.filter(Lead.industry.ilike(f"%{industry}%"))
return query.order_by(Lead.created_at.desc()).all()
@router.get("/csv")
def export_csv(
token: Optional[str] = Query(None),
status: Optional[str] = None,
source: Optional[str] = None,
industry: Optional[str] = None,
ids: Optional[str] = Query(None, description="Comma-separated lead IDs"),
db: Session = Depends(get_db),
):
_verify_token_param(token, db)
leads = get_filtered_leads(db, status, source, industry, ids)
df = leads_to_df(leads)
buf = io.StringIO()
df.to_csv(buf, index=False)
buf.seek(0)
return StreamingResponse(
iter([buf.getvalue()]),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=leads_export.csv"},
)
@router.get("/excel")
def export_excel(
token: Optional[str] = Query(None),
status: Optional[str] = None,
source: Optional[str] = None,
industry: Optional[str] = None,
ids: Optional[str] = Query(None, description="Comma-separated lead IDs"),
db: Session = Depends(get_db),
):
_verify_token_param(token, db)
leads = get_filtered_leads(db, status, source, industry, ids)
df = leads_to_df(leads)
buf = io.BytesIO()
with pd.ExcelWriter(buf, engine="openpyxl") as writer:
df.to_excel(writer, index=False, sheet_name="Leads")
buf.seek(0)
return StreamingResponse(
iter([buf.read()]),
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": "attachment; filename=leads_export.xlsx"},
)
|