{message_html}
- - - - """ - # Initialize Wix service wix_service = WixService() @@ -121,34 +66,38 @@ def _resolve_valid_wix_token(current_user: dict) -> Dict[str, Any]: if not expired_tokens: raise HTTPException(status_code=401, detail="Wix account not connected") - latest = expired_tokens[0] - refresh_token = latest.get("refresh_token") - if not refresh_token: - raise HTTPException(status_code=401, detail="Wix token expired and cannot be refreshed") - try: - refreshed = wix_service.refresh_access_token(refresh_token) - except Exception as exc: - raise _map_wix_error(exc, "Failed to refresh Wix access token") + for candidate in expired_tokens: + refresh_token = candidate.get("refresh_token") + token_id = candidate.get("id") + if not refresh_token: + continue + try: + refreshed = wix_service.refresh_access_token(refresh_token) + except Exception as exc: + continue - wix_oauth_service.update_tokens( - user_id=user_id, - access_token=refreshed.get("access_token"), - refresh_token=refreshed.get("refresh_token", refresh_token), - expires_in=refreshed.get("expires_in"), - ) + wix_oauth_service.update_tokens( + user_id=user_id, + access_token=refreshed.get("access_token"), + refresh_token=refreshed.get("refresh_token", refresh_token), + expires_in=refreshed.get("expires_in"), + token_id=token_id, + ) - return { - "access_token": refreshed.get("access_token"), - "refresh_token": refreshed.get("refresh_token", refresh_token), - "member_id": latest.get("member_id"), - "site_id": latest.get("site_id"), - } + return { + "access_token": refreshed.get("access_token"), + "refresh_token": refreshed.get("refresh_token", refresh_token), + "member_id": candidate.get("member_id"), + "site_id": candidate.get("site_id"), + } + + raise HTTPException(status_code=401, detail="Wix token expired and cannot be refreshed") class WixAuthRequest(BaseModel): """Request model for Wix authentication""" code: str - state: Optional[str] = None + state: str class WixPublishRequest(BaseModel): @@ -377,7 +326,7 @@ async def handle_oauth_callback_get(code: str, state: Optional[str] = None, requ "permissions": permissions } - html = _build_oauth_callback_html( + html = build_oauth_callback_html( payload=payload, title="Wix Connected", heading="Connection Successful", @@ -389,8 +338,8 @@ async def handle_oauth_callback_get(code: str, state: Optional[str] = None, requ }) except Exception as e: logger.error(f"Wix OAuth GET callback failed: {e}") - html = _build_oauth_callback_html( - payload={"type": "WIX_OAUTH_ERROR", "success": False, "error": _sanitize_error_message(e)}, + html = build_oauth_callback_html( + payload={"type": "WIX_OAUTH_ERROR", "success": False, "error": sanitize_error(e)}, title="Wix Connection Failed", heading="Connection Failed", message="There was an issue connecting your Wix account. You can close this window and try again." @@ -420,19 +369,17 @@ async def get_connection_status(current_user: dict = Depends(get_current_user)) } except HTTPException as e: if e.status_code == 401: - return {"connected": False, "has_permissions": False} + return {"connected": False, "has_permissions": False, "error": "Wix account not connected"} raise except Exception as e: logger.error(f"Failed to check connection status: {e}") - return {"connected": False, "has_permissions": False} + return {"connected": False, "has_permissions": False, "error": "Unable to check Wix connection"} @router.get("/status") async def get_wix_status(current_user: dict = Depends(get_current_user)) -> Dict[str, Any]: """ Get Wix connection status (similar to GSC/WordPress pattern) - Note: Wix tokens are stored in frontend sessionStorage, so we can't directly check them here. - The frontend will check sessionStorage and update the UI accordingly. """ try: token_info = _resolve_valid_wix_token(current_user) @@ -671,8 +618,8 @@ async def get_test_authorization_url(state: Optional[str] = None, _: Dict[str, A "message": "WIX_CLIENT_ID not configured. Please set it in your .env file to get a real authorization URL." } - auth_url = wix_service.get_authorization_url(state) - return {"url": auth_url, "state": state or "test_state"} + auth_payload = wix_service.get_authorization_url(state) + return {"url": auth_payload.get("authorization_url", ""), "state": state or "test_state"} except Exception as e: logger.error(f"TEST: Failed to generate authorization URL: {e}") raise HTTPException(status_code=500, detail=str(e)) @@ -699,28 +646,44 @@ async def test_publish_to_wix(request: WixPublishRequest, _: Dict[str, Any] = De @router.post("/refresh-token") -async def refresh_wix_token(request: Dict[str, Any]) -> Dict[str, Any]: +async def refresh_wix_token(current_user: dict = Depends(get_current_user)) -> Dict[str, Any]: """ - Refresh Wix access token using refresh token + Refresh Wix access token using stored refresh token. Args: - request: Dict containing refresh_token + current_user: Current authenticated user Returns: New token information with access_token, refresh_token, expires_in """ try: - refresh_token = request.get("refresh_token") - if not refresh_token: - raise HTTPException(status_code=400, detail="Missing refresh_token") + user_id = _get_current_user_id(current_user) + token_status = wix_oauth_service.get_user_token_status(user_id) + all_tokens = token_status.get("active_tokens", []) + token_status.get("expired_tokens", []) + + refresh_token = None + token_id = None + for t in all_tokens: + if t.get("refresh_token"): + refresh_token = t["refresh_token"] + token_id = t["id"] + break + + if not refresh_token: + raise HTTPException(status_code=400, detail="No refresh token found. Please reconnect your Wix account.") - # Refresh the token new_tokens = wix_service.refresh_access_token(refresh_token) + wix_oauth_service.update_tokens( + user_id=user_id, + access_token=new_tokens.get("access_token"), + refresh_token=new_tokens.get("refresh_token", refresh_token), + expires_in=new_tokens.get("expires_in"), + token_id=token_id, + ) + return { "success": True, - "access_token": new_tokens.get("access_token"), - "refresh_token": new_tokens.get("refresh_token"), "expires_in": new_tokens.get("expires_in"), "token_type": new_tokens.get("token_type", "Bearer") } @@ -728,7 +691,7 @@ async def refresh_wix_token(request: Dict[str, Any]) -> Dict[str, Any]: raise except Exception as e: logger.error(f"Failed to refresh Wix token: {e}") - raise HTTPException(status_code=500, detail=f"Failed to refresh token: {str(e)}") + raise _map_wix_error(e, "Failed to refresh token") @qa_router.post("/publish/real") @@ -800,7 +763,6 @@ async def test_publish_real(payload: Dict[str, Any], _: Dict[str, Any] = Depends "post_id": (result.get("draftPost") or result.get("post") or {}).get("id"), "url": (result.get("draftPost") or result.get("post") or {}).get("url"), "message": "Blog post published to Wix", - "raw": result, } except HTTPException: raise diff --git a/backend/api/youtube/router.py b/backend/api/youtube/router.py index 9a6ea9d1..22edbd32 100644 --- a/backend/api/youtube/router.py +++ b/backend/api/youtube/router.py @@ -459,20 +459,21 @@ async def start_video_render( try: user_id = require_authenticated_user(current_user) - # Validate subscription limits - pricing_service = PricingService(db) - validate_scene_animation_operation( - pricing_service=pricing_service, - user_id=user_id - ) - - # Filter enabled scenes + # Filter enabled scenes FIRST so we can validate credits for the actual count enabled_scenes = [s for s in request.scenes if s.get("enabled", True)] if not enabled_scenes: return VideoRenderResponse( success=False, message="No enabled scenes to render" ) + + # Validate subscription limits for ALL scenes in the batch + pricing_service = PricingService(db) + validate_scene_animation_operation( + pricing_service=pricing_service, + user_id=user_id, + scene_count=len(enabled_scenes), + ) # VALIDATION: Pre-validate scenes before creating task to prevent wasted API calls validation_errors = [] diff --git a/backend/app.py b/backend/app.py index 779fda35..5a1fb85e 100644 --- a/backend/app.py +++ b/backend/app.py @@ -672,6 +672,9 @@ if _is_full_mode(): # Include Bing Analytics Storage router to expose storage-backed endpoints from routers.bing_analytics_storage import router as bing_analytics_storage_router app.include_router(bing_analytics_storage_router) + # Include SEO Tools router with enterprise audit and GSC analysis + if seo_tools_router: + app.include_router(seo_tools_router) if images_router: app.include_router(images_router) if image_studio_router: diff --git a/backend/env_template.txt b/backend/env_template.txt index 4566cc6d..36558464 100644 --- a/backend/env_template.txt +++ b/backend/env_template.txt @@ -21,6 +21,11 @@ FRONTEND_URL=https://alwrity-ai.vercel.app # Example: OAUTH_CALLBACK_ALLOWED_ORIGINS=https://alwrity-ai.vercel.app,http://localhost:3000 OAUTH_CALLBACK_ALLOWED_ORIGINS= +# OAuth Token Encryption (Fernet key - generate with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())") +# Used by both WordPress and Wix OAuth token encryption at rest. +# WORDPRESS_TOKEN_ENCRYPTION_KEY and WIX_TOKEN_ENCRYPTION_KEY can override per-provider. +OAUTH_TOKEN_ENCRYPTION_KEY= + # OAuth Redirect URIs (Using environment variable for flexibility) GSC_REDIRECT_URI=${FRONTEND_URL}/gsc/callback WORDPRESS_REDIRECT_URI=${FRONTEND_URL}/wp/callback diff --git a/backend/models/backlink_outreach_models.py b/backend/models/backlink_outreach_models.py index f4c8f06e..53ca15e4 100644 --- a/backend/models/backlink_outreach_models.py +++ b/backend/models/backlink_outreach_models.py @@ -1,7 +1,7 @@ """DB models for production backlink outreach tracking.""" from datetime import datetime -from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Index, Boolean +from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Index, Boolean, Date from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() @@ -39,8 +39,12 @@ class OutreachAttempt(Base): lead_id = Column(String(64), ForeignKey("backlink_leads.id"), nullable=False, index=True) campaign_id = Column(String(64), ForeignKey("backlink_campaigns.id"), nullable=False, index=True) idempotency_key = Column(String(128), nullable=False, unique=True, index=True) + sender_email = Column(String(255), nullable=True) + subject = Column(String(512), nullable=True) + body = Column(Text, nullable=True) status = Column(String(32), nullable=False, default="queued", index=True) decision_reason = Column(Text, nullable=True) + sent_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=datetime.utcnow, index=True) @@ -48,6 +52,8 @@ class OutreachReply(Base): __tablename__ = "backlink_replies" id = Column(String(64), primary_key=True) attempt_id = Column(String(64), ForeignKey("backlink_outreach_attempts.id"), nullable=False, index=True) + from_email = Column(String(255), nullable=True) + subject = Column(String(512), nullable=True) received_at = Column(DateTime, default=datetime.utcnow, index=True) classification = Column(String(32), nullable=False, default="replied") body = Column(Text, nullable=True) @@ -57,9 +63,72 @@ class FollowUpSchedule(Base): __tablename__ = "backlink_followup_schedules" id = Column(String(64), primary_key=True) attempt_id = Column(String(64), ForeignKey("backlink_outreach_attempts.id"), nullable=False, index=True) + subject = Column(String(512), nullable=True) + body = Column(Text, nullable=True) scheduled_for = Column(DateTime, nullable=False, index=True) sent = Column(Boolean, default=False, index=True) +class EmailTemplate(Base): + __tablename__ = "backlink_email_templates" + id = Column(String(64), primary_key=True) + user_id = Column(String(255), nullable=False, index=True) + name = Column(String(128), nullable=False) + subject_template = Column(String(512), nullable=False) + body_template = Column(Text, nullable=False) + variables = Column(Text, nullable=True) + created_at = Column(DateTime, default=datetime.utcnow) + + +class SuppressedRecipient(Base): + __tablename__ = "backlink_suppressed_recipients" + id = Column(String(64), primary_key=True) + email = Column(String(255), nullable=False, index=True) + domain = Column(String(255), nullable=True) + reason = Column(String(128), nullable=True) + user_id = Column(String(255), nullable=True) + created_at = Column(DateTime, default=datetime.utcnow) + + +class SentIdempotencyKey(Base): + __tablename__ = "backlink_sent_idempotency_keys" + id = Column(String(64), primary_key=True) + idempotency_key = Column(String(128), nullable=False, unique=True, index=True) + user_id = Column(String(255), nullable=False) + created_at = Column(DateTime, default=datetime.utcnow) + + +class AuditLogEntry(Base): + __tablename__ = "backlink_audit_logs" + id = Column(String(64), primary_key=True) + user_id = Column(String(255), nullable=False, index=True) + campaign_id = Column(String(64), nullable=True) + event = Column(String(64), nullable=False, index=True) + recipient = Column(String(255), nullable=True) + allowed = Column(Boolean, nullable=True) + reasons = Column(Text, nullable=True) + override = Column(Boolean, default=False) + created_at = Column(DateTime, default=datetime.utcnow, index=True) + + +class SendCounterUser(Base): + __tablename__ = "backlink_send_counters_user" + id = Column(String(64), primary_key=True) + user_id = Column(String(255), nullable=False, index=True) + date = Column(Date, nullable=False, index=True) + count = Column(Integer, default=0) + + +class SendCounterDomain(Base): + __tablename__ = "backlink_send_counters_domain" + id = Column(String(64), primary_key=True) + domain = Column(String(255), nullable=False, index=True) + date = Column(Date, nullable=False, index=True) + count = Column(Integer, default=0) + + Index("idx_backlink_campaign_user_date", BacklinkCampaign.user_id, BacklinkCampaign.created_at) Index("idx_backlink_attempt_campaign_date", OutreachAttempt.campaign_id, OutreachAttempt.created_at) +Index("idx_backlink_suppressed_email", SuppressedRecipient.email, SuppressedRecipient.user_id) +Index("idx_backlink_counter_user_date", SendCounterUser.user_id, SendCounterUser.date, unique=True) +Index("idx_backlink_counter_domain_date", SendCounterDomain.domain, SendCounterDomain.date, unique=True) diff --git a/backend/routers/backlink_outreach.py b/backend/routers/backlink_outreach.py index b065ab2f..1900f38b 100644 --- a/backend/routers/backlink_outreach.py +++ b/backend/routers/backlink_outreach.py @@ -1,47 +1,97 @@ -"""Backlink outreach router.""" +"""Backlink outreach router with Clerk auth.""" -from fastapi import APIRouter, Query, HTTPException +from typing import Dict, Any +from fastapi import APIRouter, Depends, Query, HTTPException +from fastapi.responses import Response from services.backlink_outreach_models import ( BacklinkDiscoveryResponse, BacklinkKeywordInput, DeepKeywordInput, LeadCreateRequest, LeadStatusUpdateRequest, PolicyValidationRequest, PolicyValidationResponse, + SendOutreachRequest, SendOutreachResponse, + OutreachAttemptListResponse, OutreachAttemptRecord, + OutreachReplyListResponse, OutreachReplyRecord, + ScheduleFollowUpRequest, FollowUpScheduleRecord, + EmailTemplateRequest, EmailTemplateRecord, + GenerateEmailRequest, GeneratedEmailResponse, + PersonalizeEmailRequest, SubjectLinesRequest, SubjectLinesResponse, + FollowUpRequest, + BacklinkReportingSnapshot, + CampaignAnalyticsResponse, CampaignVolumeResponse, + ConversionFunnelResponse, BulkStatusUpdateRequest, BulkStatusUpdateResponse, + SuppressionAddRequest, ) from services.backlink_outreach_service import backlink_outreach_service from services.backlink_outreach_storage import BacklinkOutreachStorageService +from services.backlink_outreach_sender import backlink_outreach_sender +from services.backlink_outreach_reply_monitor import backlink_outreach_reply_monitor +from services.backlink_outreach_template_generator import ( + generate_outreach_email, + generate_personalized_email, + generate_subject_lines, + generate_follow_up, +) +from middleware.auth_middleware import get_current_user from pydantic import BaseModel, Field router = APIRouter(prefix="/api/backlink-outreach", tags=["backlink-outreach"]) class BacklinkCampaignCreateRequest(BaseModel): - user_id: str = Field(..., min_length=1) workspace_id: str = Field(..., min_length=1) name: str = Field(..., min_length=3) +def _resolve_user_id(current_user: Dict[str, Any]) -> str: + return current_user.get("id") or current_user.get("clerk_user_id") or "default" + + +# -- Auth-Required Endpoints -- + @router.get("/modules") -async def get_backlink_module_registry(): +async def get_backlink_module_registry( + current_user: Dict[str, Any] = Depends(get_current_user), +): return {"feature": "backlink_outreach", "modules": backlink_outreach_service.list_backlink_modules()} @router.get("/query-templates") -async def get_backlink_query_templates(keyword: str = Query(..., min_length=1)): +async def get_backlink_query_templates( + keyword: str = Query(..., min_length=1), + current_user: Dict[str, Any] = Depends(get_current_user), +): return {"keyword": keyword, "queries": backlink_outreach_service.generate_guest_post_queries(keyword)} @router.post("/discover", response_model=BacklinkDiscoveryResponse) -async def discover_backlink_opportunities(payload: BacklinkKeywordInput): +async def discover_backlink_opportunities( + payload: BacklinkKeywordInput, + current_user: Dict[str, Any] = Depends(get_current_user), +): return backlink_outreach_service.discover_opportunities(payload.keyword, payload.max_results) +@router.get("/migration-coverage") +async def get_backlink_migration_coverage( + current_user: Dict[str, Any] = Depends(get_current_user), +): + return backlink_outreach_service.get_migration_coverage() + + +# -- Auth-Required Endpoints -- + @router.post("/discover/deep") -async def discover_deep_backlink_opportunities(payload: DeepKeywordInput): +async def discover_deep_backlink_opportunities( + payload: DeepKeywordInput, + current_user: Dict[str, Any] = Depends(get_current_user), +): """Enhanced discovery using Exa neural search + DuckDuckGo with full-page scraping.""" + user_id = _resolve_user_id(current_user) result = await backlink_outreach_service.deep_discover(payload.keyword, payload.max_results) if payload.campaign_id: storage = BacklinkOutreachStorageService() - user_id = "default" + saved = 0 + save_failed = 0 for opp in result.get("opportunities", []): try: storage.add_lead( @@ -55,26 +105,42 @@ async def discover_deep_backlink_opportunities(payload: DeepKeywordInput): confidence_score=opp.get("confidence_score", 0.0), discovery_source=opp.get("discovery_source", "duckduckgo"), ) + saved += 1 except Exception: - continue + save_failed += 1 + result["saved_to_campaign"] = saved + result["save_failed"] = save_failed return result @router.post("/campaigns") -async def create_backlink_campaign(payload: BacklinkCampaignCreateRequest): +async def create_backlink_campaign( + payload: BacklinkCampaignCreateRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + user_id = _resolve_user_id(current_user) storage = BacklinkOutreachStorageService() - return storage.create_campaign(payload.user_id, payload.workspace_id, payload.name) + return storage.create_campaign(user_id, payload.workspace_id, payload.name) @router.get("/campaigns") -async def list_backlink_campaigns(user_id: str, workspace_id: str, limit: int = 50): +async def list_backlink_campaigns( + workspace_id: str = Query(None), + limit: int = 50, + current_user: Dict[str, Any] = Depends(get_current_user), +): + user_id = _resolve_user_id(current_user) storage = BacklinkOutreachStorageService() - return {"campaigns": storage.list_campaigns(user_id, workspace_id, limit)} + return {"campaigns": storage.list_campaigns(user_id, workspace_id or user_id, limit)} @router.get("/campaigns/{campaign_id}") -async def get_backlink_campaign(campaign_id: str, user_id: str = Query(...)): +async def get_backlink_campaign( + campaign_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): """Get campaign detail with leads.""" + user_id = _resolve_user_id(current_user) storage = BacklinkOutreachStorageService() campaign = storage.get_campaign(campaign_id, user_id) if not campaign: @@ -84,22 +150,30 @@ async def get_backlink_campaign(campaign_id: str, user_id: str = Query(...)): @router.get("/campaigns/{campaign_id}/leads") async def list_campaign_leads( - campaign_id: str, user_id: str = Query(...), status: str = Query(None) + campaign_id: str, + status: str = Query(None), + current_user: Dict[str, Any] = Depends(get_current_user), ): """List leads for a campaign, optionally filtered by status.""" + user_id = _resolve_user_id(current_user) storage = BacklinkOutreachStorageService() leads = storage.list_leads(campaign_id, user_id, status=status or None) return {"leads": leads, "total": len(leads)} @router.post("/campaigns/{campaign_id}/leads") -async def add_campaign_lead(campaign_id: str, payload: LeadCreateRequest): +async def add_campaign_lead( + campaign_id: str, + payload: LeadCreateRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): """Add a single lead to a campaign.""" + user_id = _resolve_user_id(current_user) storage = BacklinkOutreachStorageService() try: lead = storage.add_lead( - campaign_id=payload.campaign_id, - user_id="default", + campaign_id=campaign_id, + user_id=user_id, url=payload.url, domain=payload.domain, page_title=payload.page_title or "", @@ -110,29 +184,480 @@ async def add_campaign_lead(campaign_id: str, payload: LeadCreateRequest): ) return lead except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail="Failed to add lead") + + +@router.post("/leads/bulk-status", response_model=BulkStatusUpdateResponse) +async def bulk_update_lead_status( + payload: BulkStatusUpdateRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Bulk update lead statuses.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + updated = 0 + failed: list[str] = [] + for lid in payload.lead_ids: + try: + lead = storage.update_lead_status(lid, user_id, payload.status, payload.notes) + if lead: + updated += 1 + else: + failed.append(lid) + except Exception: + failed.append(lid) + return BulkStatusUpdateResponse(updated=updated, failed=failed) @router.patch("/leads/{lead_id}/status") -async def update_lead_status(lead_id: str, payload: LeadStatusUpdateRequest): +async def update_lead_status( + lead_id: str, + payload: LeadStatusUpdateRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): """Update lead status (discovered -> contacted -> replied -> placed).""" + user_id = _resolve_user_id(current_user) storage = BacklinkOutreachStorageService() - lead = storage.update_lead_status(lead_id, "default", payload.status, payload.notes) + lead = storage.update_lead_status(lead_id, user_id, payload.status, payload.notes) if not lead: raise HTTPException(status_code=404, detail="Lead not found") return lead @router.post("/policy-validate", response_model=PolicyValidationResponse) -async def validate_outreach_policy(payload: PolicyValidationRequest): +async def validate_outreach_policy( + payload: PolicyValidationRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): return backlink_outreach_service.validate_send_policy(payload) -@router.get("/reporting") -async def get_backlink_reporting_snapshot(): - return backlink_outreach_service.get_reporting_snapshot() +@router.get("/reporting", response_model=BacklinkReportingSnapshot) +async def get_backlink_reporting_snapshot( + current_user: Dict[str, Any] = Depends(get_current_user), +): + user_id = _resolve_user_id(current_user) + return backlink_outreach_service.get_reporting_snapshot(user_id=user_id) -@router.get("/migration-coverage") -async def get_backlink_migration_coverage(): - return backlink_outreach_service.get_migration_coverage() +# -- Outreach Attempts -- + +@router.post("/send-outreach", response_model=SendOutreachResponse) +async def send_outreach( + payload: SendOutreachRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Validate policy, record attempt, personalize, and send email.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + subject = payload.subject + body = payload.body + + if payload.template_id: + tmpl = storage.get_template(payload.template_id, user_id) + if tmpl: + variables = payload.template_variables or {} + subject = backlink_outreach_sender.personalize(tmpl.get("subject_template", subject), variables) + body = backlink_outreach_sender.personalize(tmpl.get("body_template", body), variables) + + result = backlink_outreach_service.send_outreach( + SendOutreachRequest( + lead_id=payload.lead_id, + campaign_id=payload.campaign_id, + user_id=user_id, + workspace_id=payload.workspace_id, + sender_email=payload.sender_email, + subject=subject, + body=body, + idempotency_key=payload.idempotency_key, + ) + ) + + lead_email = "" + if result.attempt_id: + lead = storage.get_lead(payload.lead_id, user_id=user_id) + lead_email = (lead.get("email") or "") if lead else "" + + if result.policy_allowed and lead_email: + sent = await backlink_outreach_sender.send_email( + to_email=lead_email, + subject=subject, + body=body, + ) + status = "sent" if sent else "failed" + storage.update_attempt_status(result.attempt_id, status, user_id=user_id) + result.status = status + if sent: + storage.mark_idempotency(payload.idempotency_key, user_id) + storage.increment_user_send_counter(user_id) + domain = lead_email.split("@")[-1] if "@" in lead_email else "unknown" + storage.increment_domain_send_counter(domain, user_id=user_id) + elif result.policy_allowed and not lead_email: + storage.update_attempt_status(result.attempt_id, "failed", user_id=user_id) + result.status = "failed" + result.policy_reasons = (result.policy_reasons or []) + ["lead_has_no_email"] + + return result + + +@router.get("/campaigns/{campaign_id}/attempts", response_model=OutreachAttemptListResponse) +async def list_campaign_attempts( + campaign_id: str, + limit: int = Query(50), + current_user: Dict[str, Any] = Depends(get_current_user), +): + """List outreach attempts for a campaign.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + attempts = storage.list_attempts(campaign_id, limit, user_id=user_id) + return {"attempts": attempts, "total": len(attempts)} + + +# -- Replies -- + +@router.get("/campaigns/{campaign_id}/replies", response_model=OutreachReplyListResponse) +async def list_campaign_replies( + campaign_id: str, + limit: int = Query(50), + current_user: Dict[str, Any] = Depends(get_current_user), +): + """List received replies for a campaign.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + replies = storage.list_replies(campaign_id, limit, user_id=user_id) + return {"replies": replies, "total": len(replies)} + + +@router.post("/replies/poll") +async def poll_replies( + sent_from_email: str = Query(..., min_length=3), + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Poll IMAP inbox for new replies and store them.""" + user_id = _resolve_user_id(current_user) + if not backlink_outreach_reply_monitor.is_configured(): + raise HTTPException(status_code=503, detail="IMAP not configured") + + storage = BacklinkOutreachStorageService() + raw_replies = await backlink_outreach_reply_monitor.poll_replies(sent_from_email) + stored = [] + skipped = 0 + failed = 0 + for raw in raw_replies: + try: + from_email = raw.get("from_email", "") + subject = raw.get("subject", "") + if storage.reply_exists(from_email, subject, user_id=user_id): + skipped += 1 + continue + attempt_id = storage.find_attempt_by_from_email(from_email, user_id=user_id) or "" + reply = storage.add_reply( + attempt_id=attempt_id, + from_email=from_email, + subject=subject, + body=raw.get("body", ""), + classification=raw.get("classification", "replied"), + user_id=user_id, + ) + stored.append(reply) + except Exception: + failed += 1 + return {"polled": len(raw_replies), "stored": len(stored), "skipped": skipped, "failed": failed, "replies": stored} + + +# -- Follow-ups -- + +@router.post("/campaigns/{campaign_id}/schedule-followup") +async def schedule_followup( + campaign_id: str, + payload: ScheduleFollowUpRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Schedule a follow-up for an outreach attempt.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + sched = storage.schedule_followup( + attempt_id=payload.attempt_id, + scheduled_for=payload.scheduled_for, + subject=payload.subject or "", + body=payload.body or "", + user_id=user_id, + ) + return {"campaign_id": campaign_id, "schedule": sched} + + +@router.get("/campaigns/{campaign_id}/followups") +async def list_followups( + campaign_id: str, + limit: int = Query(50), + current_user: Dict[str, Any] = Depends(get_current_user), +): + """List scheduled follow-ups for a campaign.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + followups = storage.list_followups(campaign_id, limit, user_id=user_id) + return {"followups": followups, "total": len(followups)} + + +# -- Email Templates -- + +@router.post("/templates") +async def create_template( + payload: EmailTemplateRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Create an email template.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + return storage.create_template( + user_id=user_id, + name=payload.name, + subject_template=payload.subject_template, + body_template=payload.body_template, + variables=payload.variables, + ) + + +@router.get("/templates") +async def list_templates( + current_user: Dict[str, Any] = Depends(get_current_user), +): + """List email templates for the authenticated user.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + return {"templates": storage.list_templates(user_id)} + + +@router.get("/templates/{template_id}") +async def get_template( + template_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Get a specific email template.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + tmpl = storage.get_template(template_id, user_id) + if not tmpl: + raise HTTPException(status_code=404, detail="Template not found") + return tmpl + + +@router.delete("/templates/{template_id}") +async def delete_template( + template_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Delete an email template.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + if not storage.delete_template(template_id, user_id): + raise HTTPException(status_code=404, detail="Template not found") + return {"deleted": True} + + +@router.post("/templates/generate", response_model=GeneratedEmailResponse) +async def generate_email_template( + payload: GenerateEmailRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Generate an outreach email using AI.""" + user_id = _resolve_user_id(current_user) + existing_body = None + if payload.existing_template_id: + storage = BacklinkOutreachStorageService() + tmpl = storage.get_template(payload.existing_template_id, user_id) + if tmpl: + existing_body = tmpl.get("body_template") + + result = generate_outreach_email( + topic=payload.topic, + target_site=payload.target_site, + tone=payload.tone, + user_id=user_id, + existing_body=existing_body, + ) + return result + + +@router.post("/generate/personalized", response_model=GeneratedEmailResponse) +async def generate_personalized_email_endpoint( + payload: PersonalizeEmailRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Personalize an outreach email for a specific lead.""" + user_id = _resolve_user_id(current_user) + result = generate_personalized_email( + lead_name=payload.lead_name, + lead_site=payload.lead_site, + lead_content_topic=payload.lead_content_topic, + pitch_topic=payload.pitch_topic, + existing_body=payload.existing_body, + user_id=user_id, + ) + return result + + +@router.post("/generate/subject-lines", response_model=SubjectLinesResponse) +async def generate_subject_lines_endpoint( + payload: SubjectLinesRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Generate subject line suggestions for an email body.""" + user_id = _resolve_user_id(current_user) + subjects = generate_subject_lines( + body=payload.body, + count=payload.count, + user_id=user_id, + ) + return {"subjects": subjects} + + +@router.post("/generate/follow-up", response_model=GeneratedEmailResponse) +async def generate_follow_up_endpoint( + payload: FollowUpRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Generate a follow-up email for an outreach attempt.""" + user_id = _resolve_user_id(current_user) + result = generate_follow_up( + original_subject=payload.original_subject, + original_body=payload.original_body, + days_elapsed=payload.days_elapsed, + reply_context=payload.reply_context, + user_id=user_id, + ) + return result + + +# -- Suppression -- + +@router.get("/suppression") +async def list_suppression( + current_user: Dict[str, Any] = Depends(get_current_user), +): + """List suppressed recipients.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + return {"suppressed": storage.list_suppressed(user_id)} + + +@router.post("/suppression") +async def add_suppression( + payload: SuppressionAddRequest, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Add a recipient to the suppression list.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + return storage.add_suppressed(email=payload.email, domain=payload.domain, reason=payload.reason, user_id=user_id) + + +@router.get("/campaigns/{campaign_id}/analytics/volume", response_model=CampaignVolumeResponse) +async def get_campaign_analytics_volume( + campaign_id: str, + days: int = Query(30, ge=1, le=365), + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Get daily send volume for a campaign over the last N days.""" + user_id = _resolve_user_id(current_user) + return backlink_outreach_service.get_campaign_volume(campaign_id, days, user_id=user_id) + + +@router.get("/campaigns/{campaign_id}/analytics/funnel", response_model=ConversionFunnelResponse) +async def get_campaign_analytics_funnel( + campaign_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Get conversion funnel (lead status breakdown) for a campaign.""" + user_id = _resolve_user_id(current_user) + return backlink_outreach_service.get_campaign_funnel(campaign_id, user_id=user_id) + + +@router.get("/campaigns/{campaign_id}/export/leads") +async def export_campaign_leads_csv( + campaign_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Export campaign leads as CSV.""" + user_id = _resolve_user_id(current_user) + csv_content = backlink_outreach_service.export_leads_csv(campaign_id, user_id=user_id) + return Response(content=csv_content, media_type="text/csv", + headers={"Content-Disposition": f"attachment; filename=leads_{campaign_id}.csv"}) + + +@router.get("/campaigns/{campaign_id}/export/attempts") +async def export_campaign_attempts_csv( + campaign_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Export campaign outreach attempts as CSV.""" + user_id = _resolve_user_id(current_user) + csv_content = backlink_outreach_service.export_attempts_csv(campaign_id, user_id=user_id) + return Response(content=csv_content, media_type="text/csv", + headers={"Content-Disposition": f"attachment; filename=attempts_{campaign_id}.csv"}) + + +@router.get("/campaigns/{campaign_id}/export/replies") +async def export_campaign_replies_csv( + campaign_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Export campaign replies as CSV.""" + user_id = _resolve_user_id(current_user) + csv_content = backlink_outreach_service.export_replies_csv(campaign_id, user_id=user_id) + return Response(content=csv_content, media_type="text/csv", + headers={"Content-Disposition": f"attachment; filename=replies_{campaign_id}.csv"}) + + +# -- Audit Log -- + +@router.get("/audit-logs") +async def list_audit_logs( + campaign_id: str = Query(None), + limit: int = Query(100), + current_user: Dict[str, Any] = Depends(get_current_user), +): + """List audit log entries, optionally filtered by campaign.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + return {"logs": storage.list_audit_logs(campaign_id or None, limit, user_id=user_id)} + + +# -- Analytics -- + +@router.get("/campaigns/{campaign_id}/analytics", response_model=CampaignAnalyticsResponse) +async def get_campaign_analytics( + campaign_id: str, + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Get campaign analytics: send volume, response/placement rates, reply breakdown.""" + user_id = _resolve_user_id(current_user) + storage = BacklinkOutreachStorageService() + campaign = storage.get_campaign(campaign_id, user_id) + if not campaign: + raise HTTPException(status_code=404, detail="Campaign not found") + + attempts = storage.list_attempts(campaign_id, user_id=user_id) + replies = storage.list_replies(campaign_id, user_id=user_id) + leads = storage.list_leads_all(campaign_id, user_id=user_id) + + total_sent = sum(1 for a in attempts if a.get("status") == "sent") + total_blocked = sum(1 for a in attempts if a.get("status") == "blocked") + total_replied = len(replies) + total_placed = sum(1 for l in leads if l.get("status") == "placed") + + reply_classification = {} + for r in replies: + cls = r.get("classification", "replied") + reply_classification[cls] = reply_classification.get(cls, 0) + 1 + + return CampaignAnalyticsResponse( + campaign_id=campaign_id, + lead_count=campaign.get("lead_count", 0), + send_volume=total_sent, + blocked_count=total_blocked, + reply_count=total_replied, + response_rate=round(total_replied / total_sent, 4) if total_sent > 0 else 0.0, + placement_rate=round(total_placed / campaign.get("lead_count", 1), 4) if campaign.get("lead_count", 0) > 0 else 0.0, + reply_classification=reply_classification, + ) \ No newline at end of file diff --git a/backend/routers/image_studio/save.py b/backend/routers/image_studio/save.py index f4fb722c..3fd83b6d 100644 --- a/backend/routers/image_studio/save.py +++ b/backend/routers/image_studio/save.py @@ -63,8 +63,8 @@ async def save_to_library( file_path = assets_dir / filename file_path.write_bytes(image_bytes) - # Build serving URL (assets_serving.py serves /{user_id}/avatars/{filename}) - file_url = f"/api/assets/{safe_user}/avatars/{filename}" + # Build serving URL (assets_serving.py serves /{user_id}/images/{filename}) + file_url = f"/api/assets/{safe_user}/images/{filename}" # Save to unified asset library via existing utility from utils.asset_tracker import save_asset_to_library diff --git a/backend/routers/wordpress.py b/backend/routers/wordpress.py index b79b0afd..0dc98150 100644 --- a/backend/routers/wordpress.py +++ b/backend/routers/wordpress.py @@ -87,7 +87,7 @@ async def get_wordpress_status(user: dict = Depends(get_current_user)): logger.info(f"Checking WordPress status for user: {user_id}") # Get user's WordPress sites -sites = wp_service.get_user_sites(user_id) + sites = wp_service.get_user_sites(user_id) if sites: site_responses = [ diff --git a/backend/routers/wordpress_oauth.py b/backend/routers/wordpress_oauth.py index 0204dfc5..614cc41c 100644 --- a/backend/routers/wordpress_oauth.py +++ b/backend/routers/wordpress_oauth.py @@ -8,11 +8,12 @@ from fastapi.responses import RedirectResponse, HTMLResponse, JSONResponse from typing import Dict, Any, Optional from pydantic import BaseModel from loguru import logger -import json -import os -from urllib.parse import urlparse from services.integrations.wordpress_oauth import WordPressOAuthService +from services.integrations.oauth_callback_utils import ( + build_oauth_callback_html, + sanitize_string, +) from middleware.auth_middleware import get_current_user router = APIRouter(prefix="/wp", tags=["WordPress OAuth"]) @@ -20,65 +21,6 @@ router = APIRouter(prefix="/wp", tags=["WordPress OAuth"]) # Initialize OAuth service oauth_service = WordPressOAuthService() - -def _sanitize_string(value: Any, max_len: int = 500) -> str: - if value is None: - return "" - return " ".join(str(value).split())[:max_len] - - -def _normalize_origin(url: Optional[str]) -> Optional[str]: - if not url: - return None - parsed = urlparse(url.strip()) - if parsed.scheme not in {"http", "https"} or not parsed.netloc: - return None - return f"{parsed.scheme}://{parsed.netloc}" - - -def _trusted_frontend_origin() -> Optional[str]: - origins_env = os.getenv("OAUTH_CALLBACK_ALLOWED_ORIGINS", "") - configured_origins = [ - _normalize_origin(origin) - for origin in origins_env.split(",") - if origin.strip() - ] - configured_origins = [origin for origin in configured_origins if origin] - if configured_origins: - return configured_origins[0] - return _normalize_origin(os.getenv("FRONTEND_URL")) - - -def _oauth_callback_html(payload: Dict[str, Any], title: str, heading: str, message: str) -> str: - payload_json = json.dumps(payload) - target_origin = json.dumps(_trusted_frontend_origin() or "") - heading_html = heading.replace("&", "&").replace("<", "<").replace(">", ">") - message_html = message.replace("&", "&").replace("<", "<").replace(">", ">") - return f""" - - -{message_html}
- - - - """ - # Pydantic Models class WordPressOAuthResponse(BaseModel): auth_url: str @@ -140,8 +82,8 @@ async def handle_wordpress_callback( status_code=status.HTTP_400_BAD_REQUEST, content={"success": False, "error": error} ) - html_content = _oauth_callback_html( - payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": _sanitize_string(error)}, + html_content = build_oauth_callback_html( + payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": sanitize_string(error)}, title="WordPress.com Connection Failed", heading="Connection Failed", message="There was an error connecting to WordPress.com. You can close this window and try again." @@ -158,7 +100,7 @@ async def handle_wordpress_callback( status_code=status.HTTP_400_BAD_REQUEST, content={"success": False, "error": "Missing parameters"} ) - html_content = _oauth_callback_html( + html_content = build_oauth_callback_html( payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": "Missing parameters"}, title="WordPress.com Connection Failed", heading="Connection Failed", @@ -179,7 +121,7 @@ async def handle_wordpress_callback( status_code=status.HTTP_400_BAD_REQUEST, content={"success": False, "error": "Token exchange failed"} ) - html_content = _oauth_callback_html( + html_content = build_oauth_callback_html( payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": "Token exchange failed"}, title="WordPress.com Connection Failed", heading="Connection Failed", @@ -201,12 +143,12 @@ async def handle_wordpress_callback( } ) - html_content = _oauth_callback_html( + html_content = build_oauth_callback_html( payload={ "type": "WPCOM_OAUTH_SUCCESS", "success": True, - "blogUrl": _sanitize_string(blog_url, 300), - "blogId": _sanitize_string(blog_id, 128) + "blogUrl": sanitize_string(blog_url, 300), + "blogId": sanitize_string(blog_id, 128) }, title="WordPress.com Connection Successful", heading="Connection Successful", @@ -220,7 +162,7 @@ async def handle_wordpress_callback( except Exception as e: logger.error(f"Error handling WordPress OAuth callback: {e}") - html_content = _oauth_callback_html( + html_content = build_oauth_callback_html( payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": "Callback error"}, title="WordPress.com Connection Failed", heading="Connection Failed", diff --git a/backend/scripts/cap_basic_plan_usage.py b/backend/scripts/cap_basic_plan_usage.py index 1baa5769..8a0d0f59 100644 --- a/backend/scripts/cap_basic_plan_usage.py +++ b/backend/scripts/cap_basic_plan_usage.py @@ -43,7 +43,7 @@ def cap_basic_plan_usage(): # New limits new_call_limit = basic_plan.gemini_calls_limit # Should be 10 new_token_limit = basic_plan.gemini_tokens_limit # Should be 2000 - new_image_limit = basic_plan.stability_calls_limit # Should be 5 + new_image_limit = basic_plan.stability_calls_limit # 25 logger.info(f"π Basic Plan Limits:") logger.info(f" Calls: {new_call_limit}") diff --git a/backend/scripts/update_basic_plan_limits.py b/backend/scripts/update_basic_plan_limits.py index 3a5ec090..dc101fc5 100644 --- a/backend/scripts/update_basic_plan_limits.py +++ b/backend/scripts/update_basic_plan_limits.py @@ -75,8 +75,14 @@ def update_basic_plan_limits(): basic_plan.anthropic_tokens_limit = 20000 basic_plan.mistral_tokens_limit = 20000 - # Update image generation limit to 5 - basic_plan.stability_calls_limit = 5 + # Update image generation limit to 25 (minimum 10 for podcast workflows) + basic_plan.stability_calls_limit = 25 + + # Update image edit limit to 25 (podcast episode covers + scene images) + basic_plan.image_edit_calls_limit = 25 + + # Update audio generation limit to 100 (TTS for podcast narration) + basic_plan.audio_calls_limit = 100 # Update timestamp basic_plan.updated_at = datetime.now(timezone.utc) @@ -84,7 +90,9 @@ def update_basic_plan_limits(): logger.info("\nπ New Basic plan limits:") logger.info(f" LLM Calls (all providers): 10") logger.info(f" LLM Tokens (all providers): 20000 (increased from 5000)") - logger.info(f" Images: 5") + logger.info(f" Images (stability): 25") + logger.info(f" Image Edits: 25") + logger.info(f" Audio Calls: 100") # Count and get affected users user_subscriptions = db.query(UserSubscription).filter( diff --git a/backend/services/backlink_outreach_models.py b/backend/services/backlink_outreach_models.py index 823107cb..8f7e0873 100644 --- a/backend/services/backlink_outreach_models.py +++ b/backend/services/backlink_outreach_models.py @@ -106,22 +106,138 @@ class CampaignDetailResponse(BaseModel): leads: List[LeadRecord] = Field(default_factory=list) +class GenerateEmailRequest(BaseModel): + topic: str = Field(..., min_length=2, max_length=500) + target_site: Optional[str] = Field(None, description="Target website for guest post pitch") + tone: str = Field(default="professional", pattern="^(professional|friendly|casual|formal)$") + existing_template_id: Optional[str] = None + + class GeneratedEmailResponse(BaseModel): subject: str body: str +class PersonalizeEmailRequest(BaseModel): + lead_name: str = Field(..., min_length=1, max_length=200) + lead_site: str = Field(..., min_length=1, max_length=500) + lead_content_topic: str = Field(..., min_length=1, max_length=500) + pitch_topic: str = Field(..., min_length=2, max_length=500) + existing_body: str = Field(default="", max_length=10000) + + +class SubjectLinesRequest(BaseModel): + body: str = Field(..., min_length=10, max_length=10000) + count: int = Field(default=5, ge=1, le=10) + + +class SubjectLinesResponse(BaseModel): + subjects: list[str] + + +class FollowUpRequest(BaseModel): + original_subject: str = Field(..., min_length=1, max_length=500) + original_body: str = Field(..., min_length=10, max_length=10000) + days_elapsed: int = Field(default=7, ge=1, le=90) + reply_context: str = Field(default="", max_length=2000) + + class OutreachStatusRecord(BaseModel): opportunity_url: HttpUrl status: str notes: Optional[str] = None +class SendOutreachRequest(BaseModel): + lead_id: str = Field(..., min_length=1) + campaign_id: str = Field(..., min_length=1) + user_id: str = Field(..., min_length=1) + workspace_id: str = Field(default="default") + sender_email: str = Field(..., min_length=3) + subject: str = Field(..., min_length=1) + body: str = Field(..., min_length=1) + idempotency_key: str = Field(..., min_length=8) + template_id: Optional[str] = Field(None, description="Optional template ID for personalization") + template_variables: Optional[dict] = Field(None, description="Variable values for template personalization") + + +class SendOutreachResponse(BaseModel): + attempt_id: str + status: str + policy_allowed: bool + policy_reasons: List[str] = Field(default_factory=list) + + +class OutreachAttemptRecord(BaseModel): + attempt_id: str + lead_id: str + campaign_id: str + idempotency_key: str + sender_email: Optional[str] = None + subject: Optional[str] = None + status: str = "queued" + decision_reason: Optional[str] = None + sent_at: Optional[str] = None + created_at: Optional[str] = None + + +class OutreachAttemptListResponse(BaseModel): + attempts: List[OutreachAttemptRecord] + total: int + + +class OutreachReplyRecord(BaseModel): + reply_id: str + attempt_id: str + from_email: Optional[str] = None + subject: Optional[str] = None + received_at: Optional[str] = None + classification: str = "replied" + body: Optional[str] = None + + +class OutreachReplyListResponse(BaseModel): + replies: List[OutreachReplyRecord] + total: int + + +class ScheduleFollowUpRequest(BaseModel): + attempt_id: str = Field(..., min_length=1) + scheduled_for: str = Field(..., min_length=1) + subject: Optional[str] = None + body: Optional[str] = None + + +class FollowUpScheduleRecord(BaseModel): + schedule_id: str + attempt_id: str + subject: Optional[str] = None + scheduled_for: str + sent: bool = False + + +class EmailTemplateRequest(BaseModel): + name: str = Field(..., min_length=1) + subject_template: str = Field(..., min_length=1) + body_template: str = Field(..., min_length=1) + variables: Optional[List[str]] = None + + +class EmailTemplateRecord(BaseModel): + template_id: str + user_id: str + name: str + subject_template: str + body_template: str + variables: Optional[List[str]] = None + created_at: Optional[str] = None + + class PolicyValidationRequest(BaseModel): user_id: str = Field(..., min_length=1) workspace_id: str = Field(..., min_length=1) campaign_id: str = Field(..., min_length=1) - recipient_email: EmailStr + recipient_email: str = Field(..., min_length=1) recipient_domain: str recipient_region: str = Field(default="unknown") legal_basis: str = Field(..., min_length=2) @@ -135,3 +251,61 @@ class PolicyValidationResponse(BaseModel): allowed: bool reasons: List[str] = Field(default_factory=list) final_status: str + + +# -- Analytics & Reporting Models -- + +class CampaignAnalyticsResponse(BaseModel): + campaign_id: str + lead_count: int = 0 + send_volume: int = 0 + blocked_count: int = 0 + reply_count: int = 0 + response_rate: float = 0.0 + placement_rate: float = 0.0 + reply_classification: Dict[str, int] = Field(default_factory=dict) + + +class BacklinkReportingSnapshot(BaseModel): + send_volume: int = 0 + decision_events: int = 0 + response_rate: float = 0.0 + placement_conversion: float = 0.0 + + +class CampaignVolumePoint(BaseModel): + date: str + count: int = 0 + + +class CampaignVolumeResponse(BaseModel): + campaign_id: str + days: int = 30 + volume: List[CampaignVolumePoint] = Field(default_factory=list) + + +class FunnelStage(BaseModel): + status: str + count: int = 0 + + +class ConversionFunnelResponse(BaseModel): + campaign_id: str + stages: List[FunnelStage] = Field(default_factory=list) + + +class BulkStatusUpdateRequest(BaseModel): + lead_ids: List[str] = Field(..., min_length=1) + status: str = Field(..., min_length=1) + notes: Optional[str] = None + + +class BulkStatusUpdateResponse(BaseModel): + updated: int = 0 + failed: List[str] = Field(default_factory=list) + + +class SuppressionAddRequest(BaseModel): + email: str = Field(..., min_length=3) + reason: str = Field(default="") + domain: str = Field(default="") diff --git a/backend/services/backlink_outreach_reply_monitor.py b/backend/services/backlink_outreach_reply_monitor.py new file mode 100644 index 00000000..3f465fa9 --- /dev/null +++ b/backend/services/backlink_outreach_reply_monitor.py @@ -0,0 +1,164 @@ +"""IMAP-based reply monitoring for backlink outreach.""" + +from __future__ import annotations + +import os +import asyncio +import imaplib +import email as email_lib +from email.utils import parsedate_to_datetime +from typing import List, Optional +from loguru import logger + + +IMAP_HOST = os.getenv("IMAP_HOST", "imap.gmail.com") +IMAP_PORT = int(os.getenv("IMAP_PORT", "993")) +IMAP_USERNAME = os.getenv("IMAP_USERNAME", "") +IMAP_PASSWORD = os.getenv("IMAP_PASSWORD", "") +IMAP_FOLDER = os.getenv("IMAP_FOLDER", "INBOX") +IMAP_FETCH_LIMIT = int(os.getenv("IMAP_FETCH_LIMIT", "50")) + +# Search keywords for auto-classification +INTERESTED_KEYWORDS = [ + "interested", "let's discuss", "sounds good", "would love to", "yes", + "sure", "tell me more", "looks good", "happy to", "let's do it", + "sign me up", "count me in", "proceed", "approved", +] +NOT_INTERESTED_KEYWORDS = [ + "not interested", "unsubscribe", "no thanks", "remove me", "stop", + "don't contact", "spam", "not relevant", "no longer interested", + "please stop", "do not email", +] +OUT_OF_OFFICE_KEYWORDS = [ + "out of office", "vacation", "on leave", "away from", "return on", + "not in the office", "will be back", +] + + +class BacklinkOutreachReplyMonitor: + def __init__(self): + self._host = IMAP_HOST + self._port = IMAP_PORT + self._username = IMAP_USERNAME + self._password = IMAP_PASSWORD + self._folder = IMAP_FOLDER + self._fetch_limit = IMAP_FETCH_LIMIT + + def is_configured(self) -> bool: + return bool(self._username and self._password) + + async def poll_replies(self, sent_from_email: str) -> List[dict]: + """Poll IMAP inbox for replies to a specific sender address.""" + if not self.is_configured(): + logger.warning("IMAP not configured: set IMAP_USERNAME and IMAP_PASSWORD") + return [] + + loop = asyncio.get_running_loop() + + def _poll() -> List[dict]: + try: + mail = imaplib.IMAP4_SSL(self._host, self._port) + mail.login(self._username, self._password) + mail.select(self._folder) + + safe_email = sent_from_email.replace('"', "").replace("\\", "") + search_criteria = f'(TO "{safe_email}")' + status, message_ids = mail.search(None, search_criteria) + if status != "OK": + return [] + + ids = message_ids[0].split() if message_ids[0] else [] + if not ids: + return [] + + ids = ids[-self._fetch_limit:] + + replies = [] + for mid in ids: + status, msg_data = mail.fetch(mid, "(RFC822)") + if status != "OK": + continue + + raw_email = msg_data[0][1] if msg_data else None + if not raw_email: + continue + + parsed = email_lib.message_from_bytes(raw_email) + reply = self._parse_reply(parsed) + if reply: + replies.append(reply) + + mail.logout() + return replies + except imaplib.IMAP4.error as e: + logger.error(f"IMAP error: {e}") + return [] + except Exception as e: + logger.error(f"Unexpected IMAP error: {e}") + return [] + + return await loop.run_in_executor(None, _poll) + + def _parse_reply(self, parsed_msg) -> Optional[dict]: + try: + from_email = parsed_msg.get("From", "") + subject = parsed_msg.get("Subject", "") + received_at = parsed_msg.get("Date", "") + + # Extract body + body = "" + if parsed_msg.is_multipart(): + for part in parsed_msg.walk(): + content_type = part.get_content_type() + if content_type == "text/plain": + try: + body = part.get_payload(decode=True).decode("utf-8", errors="ignore") + break + except Exception: + continue + else: + try: + body = parsed_msg.get_payload(decode=True).decode("utf-8", errors="ignore") + except Exception: + body = str(parsed_msg.get_payload()) + + classification = self._classify_reply(body, subject) + + # Parse date + try: + dt = parsedate_to_datetime(received_at) + received_at_iso = dt.isoformat() if dt else None + except Exception: + received_at_iso = None + + return { + "from_email": from_email, + "subject": subject, + "body": body[:5000], + "classification": classification, + "received_at": received_at_iso, + } + except Exception as e: + logger.error(f"Failed to parse reply: {e}") + return None + + @staticmethod + def _classify_reply(body: str, subject: str) -> str: + text = f"{subject} {body}".lower() + + for kw in OUT_OF_OFFICE_KEYWORDS: + if kw in text: + return "out_of_office" + + for kw in NOT_INTERESTED_KEYWORDS: + if kw in text: + return "not_interested" + + for kw in INTERESTED_KEYWORDS: + if kw in text: + return "interested" + + return "replied" + + +backlink_outreach_reply_monitor = BacklinkOutreachReplyMonitor() \ No newline at end of file diff --git a/backend/services/backlink_outreach_sender.py b/backend/services/backlink_outreach_sender.py new file mode 100644 index 00000000..4eba9061 --- /dev/null +++ b/backend/services/backlink_outreach_sender.py @@ -0,0 +1,90 @@ +"""Email sender for backlink outreach via SMTP.""" + +from __future__ import annotations + +import os +import ssl +import smtplib +import asyncio +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from typing import Optional +from loguru import logger + + +SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com") +SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) +SMTP_USERNAME = os.getenv("SMTP_USERNAME", "") +SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "") +SMTP_FROM_EMAIL = os.getenv("SMTP_FROM_EMAIL", SMTP_USERNAME) +SMTP_USE_TLS = os.getenv("SMTP_USE_TLS", "true").lower() in ("true", "1", "yes") +SMTP_VERIFY_TLS = os.getenv("SMTP_VERIFY_TLS", "true").lower() in ("true", "1", "yes") +SMTP_SEND_TIMEOUT = int(os.getenv("SMTP_SEND_TIMEOUT", "30")) + + +class BacklinkOutreachSender: + def __init__(self): + self._host = SMTP_HOST + self._port = SMTP_PORT + self._username = SMTP_USERNAME + self._password = SMTP_PASSWORD + self._from_email = SMTP_FROM_EMAIL or SMTP_USERNAME + self._use_tls = SMTP_USE_TLS + self._verify_tls = SMTP_VERIFY_TLS + self._timeout = SMTP_SEND_TIMEOUT + + def is_configured(self) -> bool: + return bool(self._username and self._password) + + async def send_email( + self, + to_email: str, + subject: str, + body: str, + from_email: Optional[str] = None, + ) -> bool: + if not self.is_configured(): + logger.error("SMTP not configured: set SMTP_USERNAME and SMTP_PASSWORD") + return False + + sender = from_email or self._from_email + + msg = MIMEMultipart("alternative") + msg["From"] = sender + msg["To"] = to_email + msg["Subject"] = subject + msg.attach(MIMEText(body, "plain")) + + loop = asyncio.get_running_loop() + + def _send() -> bool: + try: + tls_context = ssl.create_default_context() + if not self._verify_tls: + tls_context.check_hostname = False + tls_context.verify_mode = ssl.CERT_NONE + with smtplib.SMTP(self._host, self._port, timeout=self._timeout) as server: + if self._use_tls: + server.starttls(context=tls_context) + server.ehlo() + server.login(self._username, self._password) + server.sendmail(sender, [to_email], msg.as_string()) + logger.info(f"Email sent to {to_email}: {subject[:60]}") + return True + except smtplib.SMTPException as e: + logger.error(f"SMTP error sending to {to_email}: {e}") + return False + except Exception as e: + logger.error(f"Unexpected error sending to {to_email}: {e}") + return False + + return await loop.run_in_executor(None, _send) + + def personalize(self, template: str, variables: dict) -> str: + """Replace {placeholder} variables in a template string.""" + for key, value in variables.items(): + template = template.replace(f"{{{key}}}", str(value)) + return template + + +backlink_outreach_sender = BacklinkOutreachSender() \ No newline at end of file diff --git a/backend/services/backlink_outreach_service.py b/backend/services/backlink_outreach_service.py index c80e6b52..da8846cd 100644 --- a/backend/services/backlink_outreach_service.py +++ b/backend/services/backlink_outreach_service.py @@ -3,24 +3,25 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional import re import time import requests from bs4 import BeautifulSoup -from services.backlink_outreach_models import OpportunityContactInfo, OpportunityRecord, PolicyValidationRequest, PolicyValidationResponse +import csv +import io +from services.backlink_outreach_models import ( + OpportunityContactInfo, OpportunityRecord, + PolicyValidationRequest, PolicyValidationResponse, + SendOutreachRequest, SendOutreachResponse, + CampaignVolumeResponse, CampaignVolumePoint, + ConversionFunnelResponse, FunnelStage, +) +from services.backlink_outreach_storage import BacklinkOutreachStorageService - - -# Temporary in-memory control plane until DB wiring is complete -SUPPRESSION_LIST = set() -SENT_IDEMPOTENCY_KEYS = set() -AUDIT_LOGS: list[dict] = [] -SEND_COUNTERS_BY_USER: dict[str, int] = {} -SEND_COUNTERS_BY_DOMAIN: dict[str, int] = {} DEFAULT_USER_DAILY_CAP = 100 DEFAULT_DOMAIN_DAILY_CAP = 20 @@ -140,8 +141,12 @@ class BacklinkOutreachService: return min(1.0, 0.35 + (0.13 * hits)) + def _get_storage(self) -> BacklinkOutreachStorageService: + return BacklinkOutreachStorageService() + def validate_send_policy(self, payload: PolicyValidationRequest) -> PolicyValidationResponse: reasons: List[str] = [] + storage = self._get_storage() if payload.workspace_id.startswith("new-") and not payload.approved_by_human: reasons.append("human_review_required_for_new_workspace") @@ -149,19 +154,17 @@ class BacklinkOutreachService: reasons.append("invalid_legal_basis") if payload.recipient_region.lower() in {"eu", "eea"} and payload.legal_basis.lower() != "consent": reasons.append("region_requires_explicit_consent") - if not payload.unsubscribe_url: - reasons.append("unsubscribe_url_required") + if len(payload.sender_identity.strip()) < 3: reasons.append("sender_identity_required") - recipient_key = f"{payload.recipient_email.lower()}::{payload.recipient_domain.lower()}" - if recipient_key in SUPPRESSION_LIST: + if storage.is_suppressed(str(payload.recipient_email), payload.recipient_domain, user_id=payload.user_id): reasons.append("recipient_suppressed") - if payload.idempotency_key in SENT_IDEMPOTENCY_KEYS: + if storage.check_idempotency(payload.idempotency_key, user_id=payload.user_id): reasons.append("duplicate_idempotency_key") - user_count = SEND_COUNTERS_BY_USER.get(payload.user_id, 0) - domain_count = SEND_COUNTERS_BY_DOMAIN.get(payload.recipient_domain.lower(), 0) + user_count = storage.get_user_send_count(payload.user_id) + domain_count = storage.get_domain_send_count(payload.recipient_domain, user_id=payload.user_id) if user_count >= DEFAULT_USER_DAILY_CAP: reasons.append("user_daily_cap_exceeded") if domain_count >= DEFAULT_DOMAIN_DAILY_CAP: @@ -170,33 +173,156 @@ class BacklinkOutreachService: allowed = len(reasons) == 0 final_status = "approved" if allowed else "blocked" - AUDIT_LOGS.append({ - "event": "policy_check", - "user_id": payload.user_id, - "campaign_id": payload.campaign_id, - "recipient": str(payload.recipient_email), - "allowed": allowed, - "reasons": reasons, - "override": payload.approved_by_human, - }) - - if allowed: - SENT_IDEMPOTENCY_KEYS.add(payload.idempotency_key) - SEND_COUNTERS_BY_USER[payload.user_id] = user_count + 1 - SEND_COUNTERS_BY_DOMAIN[payload.recipient_domain.lower()] = domain_count + 1 + storage.add_audit_log( + event="policy_check", + user_id=payload.user_id, + campaign_id=payload.campaign_id, + recipient=str(payload.recipient_email), + allowed=allowed, + reasons=reasons, + override=payload.approved_by_human, + ) return PolicyValidationResponse(allowed=allowed, reasons=reasons, final_status=final_status) - def get_reporting_snapshot(self) -> Dict[str, Any]: - total_decisions = len(AUDIT_LOGS) - approved = sum(1 for row in AUDIT_LOGS if row.get("allowed")) + EU_DOMAIN_SUFFIXES = (".de", ".fr", ".it", ".es", ".nl", ".be", ".at", ".se", ".dk", ".fi", ".pt", ".ie", ".gr", ".pl", ".cz", ".ro", ".hu", ".bg", ".hr", ".sk", ".si", ".ee", ".lv", ".lt", ".lu", ".mt", ".cy") + + def _infer_region(self, domain: str) -> str: + d = domain.lower() + if any(d.endswith(s) or d.endswith(s + "/") for s in self.EU_DOMAIN_SUFFIXES): + return "eu" + if d.endswith(".uk"): + return "uk" + if d.endswith(".ca"): + return "ca" + if d.endswith(".au"): + return "au" + return "unknown" + + def send_outreach(self, request: SendOutreachRequest) -> SendOutreachResponse: + storage = self._get_storage() + lead = storage.get_lead(request.lead_id, user_id=request.user_id) + if not lead: + return SendOutreachResponse(attempt_id="", status="failed", policy_allowed=False, policy_reasons=["lead_not_found"]) + + domain = lead.get("domain", request.sender_email.split("@")[-1] if "@" in request.sender_email else "unknown") + recipient_region = self._infer_region(domain) + legal_basis = "consent" if recipient_region == "eu" else "legitimate_interest" + + policy_req = PolicyValidationRequest( + user_id=request.user_id, + workspace_id=request.workspace_id, + campaign_id=request.campaign_id, + recipient_email=lead.get("email", ""), + recipient_domain=domain, + recipient_region=recipient_region, + legal_basis=legal_basis, + approved_by_human=False, + unsubscribe_url=None, + sender_identity=request.sender_email, + idempotency_key=request.idempotency_key, + ) + policy = self.validate_send_policy(policy_req) + + attempt = storage.add_attempt( + lead_id=request.lead_id, + campaign_id=request.campaign_id, + idempotency_key=request.idempotency_key, + sender_email=request.sender_email, + subject=request.subject, + body=request.body, + status="approved" if policy.allowed else "blocked", + decision_reason="; ".join(policy.reasons) if policy.reasons else None, + user_id=request.user_id, + ) + + return SendOutreachResponse( + attempt_id=attempt.get("attempt_id", ""), + status=attempt.get("status", "failed"), + policy_allowed=policy.allowed, + policy_reasons=policy.reasons, + ) + + def get_reporting_snapshot(self, user_id: str = "default") -> Dict[str, Any]: + storage = self._get_storage() + campaigns = storage.list_campaigns(user_id, user_id, limit=100) + total_sent = 0 + total_replied = 0 + total_placed = 0 + total_leads = 0 + for c in campaigns: + cid = c["campaign_id"] + attempts = storage.list_attempts(cid, limit=10000, user_id=user_id) + leads = storage.list_leads_all(cid, user_id=user_id) + total_sent += sum(1 for a in attempts if a.get("status") == "sent") + total_replied += storage.count_replies(cid, user_id=user_id) + total_placed += sum(1 for l in leads if l.get("status") == "placed") + total_leads += len(leads) + logs = storage.list_audit_logs("", limit=1000, user_id=user_id) return { - "send_volume": approved, - "decision_events": total_decisions, - "response_rate": 0.0, - "placement_conversion": 0.0, + "send_volume": total_sent, + "decision_events": len(logs), + "response_rate": round(total_replied / total_sent, 4) if total_sent > 0 else 0.0, + "placement_conversion": round(total_placed / total_leads, 4) if total_leads > 0 else 0.0, } + def get_campaign_volume(self, campaign_id: str, days: int = 30, user_id: str = "default") -> CampaignVolumeResponse: + storage = self._get_storage() + points = storage.get_send_volume_by_day(campaign_id, days, user_id=user_id) + return CampaignVolumeResponse( + campaign_id=campaign_id, days=days, + volume=[CampaignVolumePoint(**p) for p in points], + ) + + def get_campaign_funnel(self, campaign_id: str, user_id: str = "default") -> ConversionFunnelResponse: + storage = self._get_storage() + stages = storage.get_lead_status_counts(campaign_id, user_id=user_id) + return ConversionFunnelResponse( + campaign_id=campaign_id, + stages=[FunnelStage(**s) for s in stages], + ) + + CSV_LEAD_FIELDS = ["lead_id", "campaign_id", "domain", "page_title", "email", "status", "discovery_source", "created_at"] + CSV_ATTEMPT_FIELDS = ["attempt_id", "lead_id", "campaign_id", "sender_email", "subject", "status", "sent_at", "created_at"] + CSV_REPLY_FIELDS = ["reply_id", "attempt_id", "from_email", "subject", "classification", "received_at"] + + @staticmethod + def _sanitize_csv_value(value: Any) -> str: + s = str(value) if value is not None else "" + if s and s[0] in ("=", "+", "-", "@", "\t", "\r"): + s = "'" + s + return s + + def export_leads_csv(self, campaign_id: str, user_id: str = "default") -> str: + storage = self._get_storage() + leads = storage.list_leads_all(campaign_id, user_id=user_id) + output = io.StringIO() + writer = csv.DictWriter(output, fieldnames=self.CSV_LEAD_FIELDS, extrasaction="ignore") + writer.writeheader() + for row in leads: + writer.writerows([{k: self._sanitize_csv_value(v) for k, v in row.items()}]) + return output.getvalue() + + def export_attempts_csv(self, campaign_id: str, user_id: str = "default") -> str: + storage = self._get_storage() + attempts = storage.list_attempts_all(campaign_id, user_id=user_id) + output = io.StringIO() + writer = csv.DictWriter(output, fieldnames=self.CSV_ATTEMPT_FIELDS, extrasaction="ignore") + writer.writeheader() + for row in attempts: + writer.writerows([{k: self._sanitize_csv_value(v) for k, v in row.items()}]) + return output.getvalue() + + def export_replies_csv(self, campaign_id: str, user_id: str = "default") -> str: + storage = self._get_storage() + replies = storage.list_replies_all(campaign_id, user_id=user_id) + output = io.StringIO() + writer = csv.DictWriter(output, fieldnames=self.CSV_REPLY_FIELDS, extrasaction="ignore") + writer.writeheader() + for row in replies: + writer.writerows([{k: self._sanitize_csv_value(v) for k, v in row.items()}]) + return output.getvalue() + async def deep_discover(self, keyword: str, max_results: int = 15) -> Dict[str, Any]: """Enhanced discovery using Exa neural search + DuckDuckGo with full-page scraping.""" from services.backlink_outreach_scraper import BacklinkOutreachScraper @@ -212,9 +338,15 @@ class BacklinkOutreachService: "typed opportunity records and confidence score", "deep webpage scraping + contact-page extraction via Exa", "quality scoring and guest-post signal detection", + "DB-backed policy validation with suppression & idempotency", + "outreach attempt recording + status lifecycle", + "SMTP email sending via backlink_outreach_sender", + "IMAP reply polling with auto-classification", + "follow-up scheduling with sent tracking", + "email template CRUD + AI generation (llm_text_gen)", + "personalized send via template variables", ] planned = [ - "email sending automation + response tracking", "follow-up orchestration and campaign analytics", ] return { diff --git a/backend/services/backlink_outreach_storage.py b/backend/services/backlink_outreach_storage.py index 97e2fc28..b7498aca 100644 --- a/backend/services/backlink_outreach_storage.py +++ b/backend/services/backlink_outreach_storage.py @@ -2,13 +2,18 @@ from __future__ import annotations -from datetime import datetime +from datetime import datetime, date from uuid import uuid4 from typing import List, Optional -from sqlalchemy import text as sql_text +from sqlalchemy import text as sql_text, func as sa_func from services.database import get_session_for_user -from models.backlink_outreach_models import Base, BacklinkCampaign, BacklinkLead +from models.backlink_outreach_models import ( + Base, BacklinkCampaign, BacklinkLead, + OutreachAttempt, OutreachReply, FollowUpSchedule, EmailTemplate, + SuppressedRecipient, SentIdempotencyKey, AuditLogEntry, + SendCounterUser, SendCounterDomain, +) class BacklinkOutreachStorageService: @@ -29,11 +34,14 @@ class BacklinkOutreachStorageService: def _migrate_lead_columns(self, db) -> None: """Add new columns to backlink_leads if they don't exist (dev migration).""" try: + valid_columns = {"url", "page_title", "snippet", "confidence_score", "discovery_source", "notes"} for col in self._NEW_LEAD_COLUMNS: + if col not in valid_columns: + continue + safe_col = col.replace('"', "").replace(";", "") db.execute(sql_text( - f"ALTER TABLE backlink_leads ADD COLUMN IF NOT EXISTS {col} TEXT" + f"ALTER TABLE backlink_leads ADD COLUMN IF NOT EXISTS \"{safe_col}\" TEXT" )) - # confidence_score is Float, add separately db.execute(sql_text( "ALTER TABLE backlink_leads ADD COLUMN IF NOT EXISTS confidence_score FLOAT DEFAULT 0.0" )) @@ -198,6 +206,7 @@ class BacklinkOutreachStorageService: def update_lead_status( self, lead_id: str, user_id: str, status: str, notes: Optional[str] = None ) -> Optional[dict]: + self._ensure_tables(user_id) db = get_session_for_user(user_id) if not db: return None @@ -229,3 +238,696 @@ class BacklinkOutreachStorageService: "notes": lead.notes, "created_at": lead.created_at.isoformat() if lead.created_at else None, } + + # -- Outreach Attempt CRUD -- + + def add_attempt( + self, + lead_id: str, + campaign_id: str, + idempotency_key: str, + sender_email: str = "", + subject: str = "", + body: str = "", + status: str = "queued", + decision_reason: Optional[str] = None, + user_id: str = "default", + ) -> dict: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + raise RuntimeError("Database session unavailable") + try: + attempt = OutreachAttempt( + id=f"att_{uuid4().hex[:16]}", + lead_id=lead_id, + campaign_id=campaign_id, + idempotency_key=idempotency_key, + sender_email=sender_email, + subject=subject, + body=body, + status=status, + decision_reason=decision_reason, + created_at=datetime.utcnow(), + ) + db.add(attempt) + db.commit() + return self._attempt_to_dict(attempt) + finally: + db.close() + + def list_attempts(self, campaign_id: str, limit: int = 50, user_id: str = "default") -> List[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(OutreachAttempt) + .filter(OutreachAttempt.campaign_id == campaign_id) + .order_by(OutreachAttempt.created_at.desc()) + .limit(limit) + .all() + ) + return [self._attempt_to_dict(r) for r in rows] + finally: + db.close() + + def update_attempt_status(self, attempt_id: str, status: str, decision_reason: Optional[str] = None, user_id: str = "default") -> Optional[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return None + try: + attempt = db.query(OutreachAttempt).filter(OutreachAttempt.id == attempt_id).first() + if not attempt: + return None + attempt.status = status + if decision_reason is not None: + attempt.decision_reason = decision_reason + if status == "sent": + attempt.sent_at = datetime.utcnow() + db.commit() + return self._attempt_to_dict(attempt) + finally: + db.close() + + @staticmethod + def _attempt_to_dict(attempt) -> dict: + return { + "attempt_id": attempt.id, + "lead_id": attempt.lead_id, + "campaign_id": attempt.campaign_id, + "idempotency_key": attempt.idempotency_key, + "sender_email": attempt.sender_email or "", + "subject": attempt.subject or "", + "status": attempt.status, + "decision_reason": attempt.decision_reason, + "sent_at": attempt.sent_at.isoformat() if attempt.sent_at else None, + "created_at": attempt.created_at.isoformat() if attempt.created_at else None, + } + + def find_attempt_by_from_email(self, from_email: str, user_id: str = "default") -> Optional[str]: + """Find the most recent attempt_id for a given sender email (lead).""" + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return None + try: + from sqlalchemy import desc + attempt = ( + db.query(OutreachAttempt) + .join(BacklinkLead, OutreachAttempt.lead_id == BacklinkLead.id) + .filter(BacklinkLead.email == from_email) + .order_by(desc(OutreachAttempt.created_at)) + .first() + ) + return attempt.id if attempt else None + finally: + db.close() + + # -- Outreach Reply CRUD -- + + def reply_exists(self, from_email: str, subject: str, user_id: str = "default") -> bool: + """Check if a reply with this from_email+subject already exists.""" + db = get_session_for_user(user_id) + if not db: + return False + try: + exists = ( + db.query(OutreachReply.id) + .filter(OutreachReply.from_email == from_email, OutreachReply.subject == subject) + .first() + ) + return exists is not None + finally: + db.close() + + def add_reply( + self, + attempt_id: str, + from_email: str = "", + subject: str = "", + body: str = "", + classification: str = "replied", + user_id: str = "default", + ) -> dict: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + raise RuntimeError("Database session unavailable") + try: + reply = OutreachReply( + id=f"rep_{uuid4().hex[:16]}", + attempt_id=attempt_id, + from_email=from_email, + subject=subject, + body=body, + classification=classification, + received_at=datetime.utcnow(), + ) + db.add(reply) + db.commit() + return self._reply_to_dict(reply) + finally: + db.close() + + def list_replies(self, campaign_id: str, limit: int = 50, user_id: str = "default") -> List[dict]: + """List replies by joining through attempts to filter by campaign.""" + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(OutreachReply) + .join(OutreachAttempt, OutreachReply.attempt_id == OutreachAttempt.id) + .filter(OutreachAttempt.campaign_id == campaign_id) + .order_by(OutreachReply.received_at.desc()) + .limit(limit) + .all() + ) + return [self._reply_to_dict(r) for r in rows] + finally: + db.close() + + @staticmethod + def _reply_to_dict(reply) -> dict: + return { + "reply_id": reply.id, + "attempt_id": reply.attempt_id, + "from_email": reply.from_email or "", + "subject": reply.subject or "", + "received_at": reply.received_at.isoformat() if reply.received_at else None, + "classification": reply.classification, + "body": reply.body or "", + } + + # -- Follow-Up Schedule CRUD -- + + def schedule_followup( + self, + attempt_id: str, + scheduled_for: str, + subject: str = "", + body: str = "", + user_id: str = "default", + ) -> dict: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + raise RuntimeError("Database session unavailable") + try: + sched = FollowUpSchedule( + id=f"fu_{uuid4().hex[:16]}", + attempt_id=attempt_id, + subject=subject or None, + body=body or None, + scheduled_for=datetime.fromisoformat(scheduled_for) if isinstance(scheduled_for, str) else scheduled_for, + sent=False, + ) + db.add(sched) + db.commit() + return self._followup_to_dict(sched) + finally: + db.close() + + def list_followups(self, campaign_id: str, limit: int = 50, user_id: str = "default") -> List[dict]: + """List follow-ups by joining through attempts to filter by campaign.""" + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(FollowUpSchedule) + .join(OutreachAttempt, FollowUpSchedule.attempt_id == OutreachAttempt.id) + .filter(OutreachAttempt.campaign_id == campaign_id) + .order_by(FollowUpSchedule.scheduled_for.asc()) + .limit(limit) + .all() + ) + return [self._followup_to_dict(r) for r in rows] + finally: + db.close() + + def mark_followup_sent(self, schedule_id: str, user_id: str = "default") -> Optional[dict]: + db = get_session_for_user(user_id) + if not db: + return None + try: + sched = db.query(FollowUpSchedule).filter(FollowUpSchedule.id == schedule_id).first() + if not sched: + return None + sched.sent = True + db.commit() + return self._followup_to_dict(sched) + finally: + db.close() + + @staticmethod + def _followup_to_dict(sched) -> dict: + return { + "schedule_id": sched.id, + "attempt_id": sched.attempt_id, + "subject": sched.subject or "", + "scheduled_for": sched.scheduled_for.isoformat() if sched.scheduled_for else None, + "sent": sched.sent, + } + + # -- Email Template CRUD -- + + def create_template( + self, + user_id: str, + name: str, + subject_template: str, + body_template: str, + variables: Optional[List[str]] = None, + ) -> dict: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + raise RuntimeError("Database session unavailable") + try: + tmpl = EmailTemplate( + id=f"tpl_{uuid4().hex[:16]}", + user_id=user_id, + name=name, + subject_template=subject_template, + body_template=body_template, + variables=",".join(variables) if variables else None, + created_at=datetime.utcnow(), + ) + db.add(tmpl) + db.commit() + return self._template_to_dict(tmpl) + finally: + db.close() + + def list_templates(self, user_id: str, limit: int = 50) -> List[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(EmailTemplate) + .filter(EmailTemplate.user_id == user_id) + .order_by(EmailTemplate.created_at.desc()) + .limit(limit) + .all() + ) + return [self._template_to_dict(r) for r in rows] + finally: + db.close() + + def get_template(self, template_id: str, user_id: str) -> Optional[dict]: + db = get_session_for_user(user_id) + if not db: + return None + try: + tmpl = ( + db.query(EmailTemplate) + .filter(EmailTemplate.id == template_id, EmailTemplate.user_id == user_id) + .first() + ) + if not tmpl: + return None + return self._template_to_dict(tmpl) + finally: + db.close() + + def delete_template(self, template_id: str, user_id: str) -> bool: + db = get_session_for_user(user_id) + if not db: + return False + try: + tmpl = ( + db.query(EmailTemplate) + .filter(EmailTemplate.id == template_id, EmailTemplate.user_id == user_id) + .first() + ) + if not tmpl: + return False + db.delete(tmpl) + db.commit() + return True + finally: + db.close() + + @staticmethod + def _template_to_dict(tmpl) -> dict: + return { + "template_id": tmpl.id, + "user_id": tmpl.user_id, + "name": tmpl.name, + "subject_template": tmpl.subject_template, + "body_template": tmpl.body_template, + "variables": tmpl.variables.split(",") if tmpl.variables else [], + "created_at": tmpl.created_at.isoformat() if tmpl.created_at else None, + } + + # -- Suppression List -- + + def add_suppressed(self, email: str, user_id: str = "default", domain: str = "", reason: str = "") -> dict: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + raise RuntimeError("Database session unavailable") + try: + entry = SuppressedRecipient( + id=f"sup_{uuid4().hex[:16]}", + email=email.lower(), + domain=domain.lower() if domain else email.split("@")[-1].lower(), + reason=reason, + user_id=user_id, + created_at=datetime.utcnow(), + ) + db.add(entry) + db.commit() + return {"id": entry.id, "email": entry.email, "reason": entry.reason} + finally: + db.close() + + def is_suppressed(self, email: str, domain: str = "", user_id: str = "default") -> bool: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return False + try: + email_lower = email.lower() + domain_lower = domain.lower() if domain else email.split("@")[-1].lower() + exists = ( + db.query(SuppressedRecipient.id) + .filter( + (SuppressedRecipient.email == email_lower) | + (SuppressedRecipient.domain == domain_lower) + ) + .first() + ) + return exists is not None + finally: + db.close() + + def list_suppressed(self, user_id: str = "default", limit: int = 100) -> List[dict]: + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(SuppressedRecipient) + .order_by(SuppressedRecipient.created_at.desc()) + .limit(limit) + .all() + ) + return [{"id": r.id, "email": r.email, "domain": r.domain, "reason": r.reason, "created_at": r.created_at.isoformat() if r.created_at else None} for r in rows] + finally: + db.close() + + # -- Idempotency -- + + def check_idempotency(self, idempotency_key: str, user_id: str = "default") -> bool: + """Returns True if key already exists (duplicate).""" + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return False + try: + exists = ( + db.query(SentIdempotencyKey.id) + .filter(SentIdempotencyKey.idempotency_key == idempotency_key) + .first() + ) + return exists is not None + finally: + db.close() + + def mark_idempotency(self, idempotency_key: str, user_id: str = "default") -> dict: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + raise RuntimeError("Database session unavailable") + try: + entry = SentIdempotencyKey( + id=f"idm_{uuid4().hex[:16]}", + idempotency_key=idempotency_key, + user_id=user_id, + created_at=datetime.utcnow(), + ) + db.add(entry) + db.commit() + return {"idempotency_key": idempotency_key} + finally: + db.close() + + # -- Send Counters -- + + def _today(self) -> date: + return date.today() + + def increment_user_send_counter(self, user_id: str) -> int: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return 0 + try: + today = self._today() + row_id = f"scu_{uuid4().hex[:16]}" + db.execute(sql_text( + "INSERT INTO backlink_send_counters_user (id, user_id, date, count) " + "VALUES (:id, :uid, :dt, 1) " + "ON CONFLICT (user_id, date) DO UPDATE SET count = count + 1" + ), {"id": row_id, "uid": user_id, "dt": today}) + db.commit() + result = db.query(SendCounterUser.count).filter( + SendCounterUser.user_id == user_id, SendCounterUser.date == today + ).first() + return result[0] if result else 0 + finally: + db.close() + + def get_user_send_count(self, user_id: str) -> int: + db = get_session_for_user(user_id) + if not db: + return 0 + try: + today = self._today() + row = ( + db.query(SendCounterUser.count) + .filter(SendCounterUser.user_id == user_id, SendCounterUser.date == today) + .first() + ) + return row[0] if row else 0 + finally: + db.close() + + def increment_domain_send_counter(self, domain: str, user_id: str = "default") -> int: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return 0 + try: + today = self._today() + domain_lower = domain.lower() + row_id = f"scd_{uuid4().hex[:16]}" + db.execute(sql_text( + "INSERT INTO backlink_send_counters_domain (id, domain, date, count) " + "VALUES (:id, :dom, :dt, 1) " + "ON CONFLICT (domain, date) DO UPDATE SET count = count + 1" + ), {"id": row_id, "dom": domain_lower, "dt": today}) + db.commit() + result = db.query(SendCounterDomain.count).filter( + SendCounterDomain.domain == domain_lower, SendCounterDomain.date == today + ).first() + return result[0] if result else 0 + finally: + db.close() + + def get_domain_send_count(self, domain: str, user_id: str = "default") -> int: + db = get_session_for_user(user_id) + if not db: + return 0 + try: + today = self._today() + row = ( + db.query(SendCounterDomain.count) + .filter(SendCounterDomain.domain == domain.lower(), SendCounterDomain.date == today) + .first() + ) + return row[0] if row else 0 + finally: + db.close() + + # -- Audit Log -- + + def add_audit_log( + self, + event: str, + user_id: str, + campaign_id: str = "", + recipient: str = "", + allowed: bool = False, + reasons: Optional[List[str]] = None, + override: bool = False, + ) -> dict: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + raise RuntimeError("Database session unavailable") + try: + entry = AuditLogEntry( + id=f"aud_{uuid4().hex[:16]}", + user_id=user_id, + campaign_id=campaign_id or None, + event=event, + recipient=recipient or None, + allowed=allowed, + reasons=";".join(reasons) if reasons else None, + override=override, + created_at=datetime.utcnow(), + ) + db.add(entry) + db.commit() + return {"id": entry.id, "event": entry.event, "allowed": entry.allowed} + finally: + db.close() + + def list_audit_logs(self, campaign_id: Optional[str] = None, limit: int = 100, user_id: str = "default") -> List[dict]: + db = get_session_for_user(user_id) + if not db: + return [] + try: + q = db.query(AuditLogEntry) + if campaign_id: + q = q.filter(AuditLogEntry.campaign_id == campaign_id) + rows = q.order_by(AuditLogEntry.created_at.desc()).limit(limit).all() + return [ + { + "id": r.id, + "event": r.event, + "recipient": r.recipient, + "allowed": r.allowed, + "reasons": r.reasons.split(";") if r.reasons else [], + "override": r.override, + "created_at": r.created_at.isoformat() if r.created_at else None, + } + for r in rows + ] + finally: + db.close() + + # -- Analytics -- + + def get_send_volume_by_day(self, campaign_id: str, days: int = 30, user_id: str = "default") -> List[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + from datetime import timedelta + cutoff = datetime.utcnow() - timedelta(days=days) + rows = ( + db.query(sa_func.date(OutreachAttempt.sent_at).label("date"), sa_func.count(OutreachAttempt.id).label("count")) + .filter(OutreachAttempt.campaign_id == campaign_id, OutreachAttempt.status == "sent", OutreachAttempt.sent_at >= cutoff) + .group_by(sa_func.date(OutreachAttempt.sent_at)) + .order_by(sa_func.date(OutreachAttempt.sent_at).asc()) + .all() + ) + return [{"date": str(r.date), "count": r.count} for r in rows] + finally: + db.close() + + def get_lead_status_counts(self, campaign_id: str, user_id: str = "default") -> List[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(BacklinkLead.status, sa_func.count(BacklinkLead.id).label("count")) + .filter(BacklinkLead.campaign_id == campaign_id) + .group_by(BacklinkLead.status) + .order_by(BacklinkLead.status.asc()) + .all() + ) + return [{"status": r.status, "count": r.count} for r in rows] + finally: + db.close() + + def list_attempts_all(self, campaign_id: str, user_id: str = "default") -> List[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(OutreachAttempt) + .filter(OutreachAttempt.campaign_id == campaign_id) + .order_by(OutreachAttempt.created_at.desc()) + .all() + ) + return [self._attempt_to_dict(r) for r in rows] + finally: + db.close() + + def list_replies_all(self, campaign_id: str, user_id: str = "default") -> List[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(OutreachReply) + .join(OutreachAttempt, OutreachReply.attempt_id == OutreachAttempt.id) + .filter(OutreachAttempt.campaign_id == campaign_id) + .order_by(OutreachReply.received_at.desc()) + .all() + ) + return [self._reply_to_dict(r) for r in rows] + finally: + db.close() + + def count_replies(self, campaign_id: str, user_id: str = "default") -> int: + db = get_session_for_user(user_id) + if not db: + return 0 + try: + return ( + db.query(OutreachReply.id) + .join(OutreachAttempt, OutreachReply.attempt_id == OutreachAttempt.id) + .filter(OutreachAttempt.campaign_id == campaign_id) + .count() + ) + finally: + db.close() + + def list_leads_all(self, campaign_id: str, user_id: str = "default") -> List[dict]: + self._ensure_tables(user_id) + db = get_session_for_user(user_id) + if not db: + return [] + try: + rows = ( + db.query(BacklinkLead) + .filter(BacklinkLead.campaign_id == campaign_id) + .order_by(BacklinkLead.created_at.desc()) + .all() + ) + return [self._lead_to_dict(r) for r in rows] + finally: + db.close() + + # -- Policy Helpers (composite checks) -- + + def get_lead(self, lead_id: str, user_id: str = "default") -> Optional[dict]: + db = get_session_for_user(user_id) + if not db: + return None + try: + lead = db.query(BacklinkLead).filter(BacklinkLead.id == lead_id).first() + if not lead: + return None + return self._lead_to_dict(lead) + finally: + db.close() diff --git a/backend/services/backlink_outreach_template_generator.py b/backend/services/backlink_outreach_template_generator.py new file mode 100644 index 00000000..5f0db2b1 --- /dev/null +++ b/backend/services/backlink_outreach_template_generator.py @@ -0,0 +1,307 @@ +"""AI-powered outreach email template generation.""" + +from __future__ import annotations + +import json +import re +from typing import List, Optional +from loguru import logger + +from services.llm_providers.main_text_generation import llm_text_gen + + +SYSTEM_PROMPT = """You are an expert outreach copywriter specializing in guest post and backlink pitch emails. +Write concise, personalized outreach emails that get high response rates. +Follow these rules: +- Be specific about why you're reaching out (mention their content) +- Keep it under 200 words +- Include a clear call to action +- Sound human, not templated +- Never use spammy phrases +- Output ONLY valid JSON with "subject" and "body" keys""" + +SUBJECT_LINES_PROMPT = """You are an expert email subject line writer. +Given an outreach email body, generate subject lines that are: +- Intriguing but not clickbait +- Personalized when possible +- Under 60 characters +- Varied in style (question, curiosity, value-prop) +Output ONLY valid JSON with a "subjects" key containing an array of strings.""" + +FOLLOW_UP_PROMPT = """You are an expert outreach copywriter. +Write a polite follow-up email for a guest post pitch that hasn't received a response. +Rules: +- Reference the original email without repeating it verbatim +- Keep it shorter than the original (under 100 words) +- Add a new angle or piece of value +- Include a clear call to action +- Sound human and respectful, never pushy +- Output ONLY valid JSON with "subject" and "body" keys""" + +PERSONALIZATION_PROMPT = """You are an expert outreach personalization specialist. +Given a lead's information and a draft outreach email, personalize it for that specific lead. +Rules: +- Mention their specific content or website +- Reference something relevant from their site +- Keep the core pitch but make it feel custom-written +- Under 200 words +- Output ONLY valid JSON with "subject" and "body" keys""" + + +def generate_outreach_email( + topic: str, + target_site: Optional[str] = None, + tone: str = "professional", + user_id: str = "default", + existing_body: Optional[str] = None, +) -> dict: + """Generate an outreach email using the LLM. + + Args: + topic: The topic/keyword to pitch. + target_site: Optional target website name/URL. + tone: professional, friendly, casual, or formal. + user_id: Clerk user ID for subscription check. + existing_body: If provided, rewrite/improve this existing template. + + Returns: + dict with "subject" and "body" keys. + """ + if existing_body: + prompt = ( + f"Rewrite and improve the following outreach email for a {tone} tone. " + f"Topic: {topic}. " + f"{f'Target website: {target_site}. ' if target_site else ''}" + f"Keep the core message but make it more effective. " + f"Original email:\n\n{existing_body}\n\n" + f"Return ONLY valid JSON with 'subject' and 'body' keys." + ) + else: + prompt = ( + f"Write a {tone} outreach email for a guest post opportunity about: {topic}. " + f"{f'We are pitching this to: {target_site}. ' if target_site else ''}" + f"Mention specific value the guest post would bring to their audience. " + f"Return ONLY valid JSON with 'subject' and 'body' keys." + ) + + try: + raw = llm_text_gen( + prompt=prompt, + system_prompt=SYSTEM_PROMPT, + user_id=user_id, + temperature=0.7, + ) + + result = _parse_json_response(raw) + if result: + return result + + return _fallback_extract(raw, topic) + + except Exception as e: + logger.error(f"Failed to generate outreach email: {e}") + return { + "subject": f"Guest post opportunity: {topic}", + "body": f"Hi there,\n\nI came across your site and I'd love to contribute a guest post about {topic}. " + f"Please let me know if you're open to submissions.\n\nBest regards", + } + + +def generate_personalized_email( + lead_name: str, + lead_site: str, + lead_content_topic: str, + pitch_topic: str, + existing_body: str = "", + user_id: str = "default", +) -> dict: + """Personalize an outreach email for a specific lead. + + Args: + lead_name: Contact name or site owner name. + lead_site: The lead's website URL. + lead_content_topic: Topic of relevant content on their site. + pitch_topic: The topic we want to pitch. + existing_body: Optional draft to personalize further. + user_id: Clerk user ID for subscription check. + + Returns: + dict with "subject" and "body" keys. + """ + if existing_body: + prompt = ( + f"Personalize this outreach email for {lead_name} from {lead_site}. " + f"They have content about '{lead_content_topic}'. " + f"We want to pitch: {pitch_topic}. " + f"Mention something specific about their content on {lead_content_topic} " + f"to show we've done our research. " + f"Draft email to personalize:\n\n{existing_body}\n\n" + f"Return ONLY valid JSON with 'subject' and 'body' keys." + ) + else: + prompt = ( + f"Write a personalized outreach email to {lead_name} at {lead_site}. " + f"They have published content about '{lead_content_topic}'. " + f"We want to pitch a guest post about: {pitch_topic}. " + f"Reference their article on {lead_content_topic} and explain how our pitch " + f"would provide value to their audience. " + f"Return ONLY valid JSON with 'subject' and 'body' keys." + ) + + try: + raw = llm_text_gen( + prompt=prompt, + system_prompt=PERSONALIZATION_PROMPT, + user_id=user_id, + temperature=0.7, + ) + result = _parse_json_response(raw) + if result: + return result + return _fallback_extract(raw, pitch_topic) + except Exception as e: + logger.error(f"Failed to personalize email: {e}") + return {"subject": f"Question about your content on {lead_content_topic}", "body": existing_body or f"Hi {lead_name},\n\nI enjoyed your article about {lead_content_topic}..."} + + +def generate_subject_lines( + body: str, + count: int = 5, + user_id: str = "default", +) -> List[str]: + """Generate subject line suggestions for an email body. + + Args: + body: The email body to generate subject lines for. + count: Number of subject lines to generate. + user_id: Clerk user ID for subscription check. + + Returns: + List of subject line strings. + """ + prompt = ( + f"Generate {count} subject lines for the following outreach email. " + f"Make them varied in style and optimized for open rates.\n\n" + f"Email body:\n{body}\n\n" + f"Return ONLY valid JSON with a 'subjects' key containing an array of strings." + ) + + try: + raw = llm_text_gen( + prompt=prompt, + system_prompt=SUBJECT_LINES_PROMPT, + user_id=user_id, + temperature=0.8, + ) + if raw: + text = raw.strip() + if text.startswith("```"): + text = re.sub(r"^```(?:json)?\s*", "", text) + text = re.sub(r"\s*```$", "", text) + try: + data = json.loads(text) + if isinstance(data, dict) and "subjects" in data and isinstance(data["subjects"], list): + return [s.strip() for s in data["subjects"][:count]] + except json.JSONDecodeError: + pass + lines = [l.strip("- ").strip() for l in raw.strip().split("\n") if l.strip() and not l.strip().startswith("```")] + return [l for l in lines if len(l) > 10][:count] + except Exception as e: + logger.error(f"Failed to generate subject lines: {e}") + return [f"Guest post opportunity", f"Question about your content", f"Collaboration idea"] + + +def generate_follow_up( + original_subject: str, + original_body: str, + days_elapsed: int = 7, + reply_context: str = "", + user_id: str = "default", +) -> dict: + """Generate a follow-up email for an outreach that hasn't received a response. + + Args: + original_subject: Subject line of the original email. + original_body: Body of the original email. + days_elapsed: Number of days since the original was sent. + reply_context: If the recipient replied, context of their reply. + user_id: Clerk user ID for subscription check. + + Returns: + dict with "subject" and "body" keys. + """ + if reply_context: + prompt = ( + f"The recipient replied with: '{reply_context}'. " + f"Write a follow-up email that addresses their response and keeps the conversation moving. " + f"Original subject: {original_subject}.\n\n" + f"Original email:\n{original_body}\n\n" + f"Return ONLY valid JSON with 'subject' and 'body' keys." + ) + else: + prompt = ( + f"Write a polite follow-up email. {days_elapsed} days have passed since the original email. " + f"Do not apologize for following up. Add a new piece of value or angle. " + f"Original subject: {original_subject}.\n\n" + f"Original email:\n{original_body}\n\n" + f"Return ONLY valid JSON with 'subject' and 'body' keys." + ) + + try: + raw = llm_text_gen( + prompt=prompt, + system_prompt=FOLLOW_UP_PROMPT, + user_id=user_id, + temperature=0.7, + ) + result = _parse_json_response(raw) + if result: + return result + return _fallback_extract(raw, original_subject) + except Exception as e: + logger.error(f"Failed to generate follow-up: {e}") + return { + "subject": f"Re: {original_subject}", + "body": f"Hi there,\n\nI wanted to follow up on my previous email. " + f"I'd love to hear your thoughts when you have a moment.\n\nBest regards", + } + + +def _parse_json_response(raw: str) -> Optional[dict]: + """Try to parse JSON from LLM response, handling markdown fences.""" + if not raw: + return None + + text = raw.strip() + + if text.startswith("```"): + text = re.sub(r"^```(?:json)?\s*", "", text) + text = re.sub(r"\s*```$", "", text) + + try: + data = json.loads(text) + if isinstance(data, dict) and "subject" in data and "body" in data: + return {"subject": data["subject"].strip(), "body": data["body"].strip()} + except json.JSONDecodeError: + pass + + return None + + +def _fallback_extract(raw: str, topic: str) -> dict: + """Fallback: try to extract subject line and body from unstructured text.""" + lines = [l.strip() for l in raw.strip().split("\n") if l.strip()] + subject = topic + body_lines = [] + + for i, line in enumerate(lines): + lower = line.lower() + if lower.startswith("subject") or lower.startswith("subject:"): + subject = line.split(":", 1)[-1].strip() + elif lower.startswith("body") or lower.startswith("body:"): + body_lines.append(line.split(":", 1)[-1].strip()) + else: + body_lines.append(line) + + body = "\n".join(body_lines) if body_lines else raw + return {"subject": subject, "body": body} \ No newline at end of file diff --git a/backend/services/integrations/oauth_callback_utils.py b/backend/services/integrations/oauth_callback_utils.py new file mode 100644 index 00000000..b0cfa28a --- /dev/null +++ b/backend/services/integrations/oauth_callback_utils.py @@ -0,0 +1,79 @@ +""" +Shared OAuth callback utilities for Wix and WordPress integrations. + +Provides hardened postMessage-based HTML callback generation, origin +validation, and string sanitization used across OAuth callback routes. +""" + +import json +import os +from typing import Any, Optional +from urllib.parse import urlparse + + +def sanitize_string(value: Any, max_len: int = 500) -> str: + if value is None: + return "" + return " ".join(str(value).split())[:max_len] + + +def sanitize_error(error: Exception, max_len: int = 500) -> str: + return sanitize_string(error, max_len) + + +def normalize_origin(url: Optional[str]) -> Optional[str]: + if not url: + return None + parsed = urlparse(url.strip()) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + return None + return f"{parsed.scheme}://{parsed.netloc}" + + +def trusted_frontend_origin() -> Optional[str]: + origins_env = os.getenv("OAUTH_CALLBACK_ALLOWED_ORIGINS", "") + configured = [ + origin + for origin in (normalize_origin(o) for o in origins_env.split(",") if o.strip()) + if origin is not None + ] + if configured: + return configured[0] + return normalize_origin(os.getenv("FRONTEND_URL")) + + +def build_oauth_callback_html( + payload: dict, + title: str, + heading: str, + message: str, +) -> str: + trusted_origin = trusted_frontend_origin() + payload_json = json.dumps(payload) + target_origin_json = json.dumps(trusted_origin or "") + heading_html = heading.replace("&", "&").replace("<", "<").replace(">", ">") + message_html = message.replace("&", "&").replace("<", "<").replace(">", ">") + return f""" + + +{message_html}
+ + + + """ diff --git a/backend/services/integrations/wix_oauth.py b/backend/services/integrations/wix_oauth.py index 9e3a3e3b..107cc0ab 100644 --- a/backend/services/integrations/wix_oauth.py +++ b/backend/services/integrations/wix_oauth.py @@ -8,7 +8,7 @@ import sqlite3 from typing import Optional, Dict, Any, List from datetime import datetime, timedelta from loguru import logger - +from cryptography.fernet import Fernet, InvalidToken from services.database import get_user_db_path @@ -17,6 +17,66 @@ class WixOAuthService: def __init__(self, db_path: Optional[str] = None): self.db_path = db_path + self.token_encryption_key = ( + os.getenv("WIX_TOKEN_ENCRYPTION_KEY") + or os.getenv("OAUTH_TOKEN_ENCRYPTION_KEY") + ) + self._fernet = self._initialize_fernet() + self._migration_done: set = set() + + def _initialize_fernet(self) -> Optional[Fernet]: + if not self.token_encryption_key: + logger.error("Wix token encryption key is not configured.") + return None + try: + return Fernet(self.token_encryption_key.encode("utf-8")) + except Exception: + logger.error("Wix token encryption key is invalid.") + return None + + def _encrypt_token(self, token: Optional[str]) -> Optional[str]: + if not token: + return None + if not self._fernet: + raise ValueError("Token encryption is unavailable: missing/invalid managed key") + return self._fernet.encrypt(token.encode("utf-8")).decode("utf-8") + + def _decrypt_token(self, token_blob: Optional[str]) -> Optional[str]: + if not token_blob: + return None + if not self._fernet: + raise ValueError("Token decryption is unavailable: missing/invalid managed key") + return self._fernet.decrypt(token_blob.encode("utf-8")).decode("utf-8") + + def _is_likely_encrypted_blob(self, value: Optional[str]) -> bool: + return bool(value and value.startswith("gAAAAA")) + + def _migrate_plaintext_tokens_if_needed(self, conn: sqlite3.Connection, user_id: str) -> None: + if not self._fernet or user_id in self._migration_done: + return + cursor = conn.cursor() + cursor.execute( + "SELECT id, access_token, refresh_token FROM wix_oauth_tokens WHERE user_id = ?", + (user_id,), + ) + rows = cursor.fetchall() + migrated = 0 + for token_id, access_token, refresh_token in rows: + needs_access = access_token and not self._is_likely_encrypted_blob(access_token) + needs_refresh = refresh_token and not self._is_likely_encrypted_blob(refresh_token) + if not (needs_access or needs_refresh): + continue + enc_access = self._encrypt_token(access_token) if needs_access else access_token + enc_refresh = self._encrypt_token(refresh_token) if needs_refresh else refresh_token + cursor.execute( + "UPDATE wix_oauth_tokens SET access_token = ?, refresh_token = ?, updated_at = datetime('now') WHERE id = ? AND user_id = ?", + (enc_access, enc_refresh, token_id, user_id), + ) + migrated += 1 + if migrated: + conn.commit() + logger.info(f"Wix OAuth token migration completed for user {user_id}; rows migrated={migrated}") + self._migration_done.add(user_id) def _get_db_path(self, user_id: str) -> str: if self.db_path: @@ -173,13 +233,16 @@ class WixOAuthService: if expires_in: expires_at = datetime.now() + timedelta(seconds=expires_in) + encrypted_access = self._encrypt_token(access_token) + encrypted_refresh = self._encrypt_token(refresh_token) if refresh_token else None + with sqlite3.connect(db_path) as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO wix_oauth_tokens (user_id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - ''', (user_id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id)) + ''', (user_id, encrypted_access, encrypted_refresh, token_type, expires_at, expires_in, scope, site_id, member_id)) conn.commit() logger.info(f"Wix OAuth: Token inserted into database for user {user_id}") @@ -200,6 +263,7 @@ class WixOAuthService: return [] with sqlite3.connect(db_path) as conn: + self._migrate_plaintext_tokens_if_needed(conn, user_id) cursor = conn.cursor() cursor.execute(''' SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, created_at @@ -210,10 +274,29 @@ class WixOAuthService: tokens = [] for row in cursor.fetchall(): + access_token_val = row[1] + refresh_token_val = row[2] + try: + decrypted_access = ( + self._decrypt_token(access_token_val) + if self._is_likely_encrypted_blob(access_token_val) + else access_token_val + ) + except InvalidToken: + logger.error(f"Failed to decrypt Wix access token for user {user_id}, token_id={row[0]}") + continue + try: + decrypted_refresh = ( + self._decrypt_token(refresh_token_val) + if self._is_likely_encrypted_blob(refresh_token_val) + else refresh_token_val + ) + except InvalidToken: + decrypted_refresh = None tokens.append({ "id": row[0], - "access_token": row[1], - "refresh_token": row[2], + "access_token": decrypted_access, + "refresh_token": decrypted_refresh, "token_type": row[3], "expires_at": row[4], "expires_in": row[5], @@ -248,9 +331,9 @@ class WixOAuthService: } with sqlite3.connect(db_path) as conn: + self._migrate_plaintext_tokens_if_needed(conn, user_id) cursor = conn.cursor() - # Get all tokens (active and expired) cursor.execute(''' SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, created_at, is_active FROM wix_oauth_tokens @@ -263,10 +346,29 @@ class WixOAuthService: expired_tokens = [] for row in cursor.fetchall(): + access_token_val = row[1] + refresh_token_val = row[2] + try: + decrypted_access = ( + self._decrypt_token(access_token_val) + if self._is_likely_encrypted_blob(access_token_val) + else access_token_val + ) + except InvalidToken: + decrypted_access = None + try: + decrypted_refresh = ( + self._decrypt_token(refresh_token_val) + if self._is_likely_encrypted_blob(refresh_token_val) + else refresh_token_val + ) + except InvalidToken: + decrypted_refresh = None + token_data = { "id": row[0], - "access_token": row[1], - "refresh_token": row[2], + "access_token": decrypted_access, + "refresh_token": decrypted_refresh, "token_type": row[3], "expires_at": row[4], "expires_in": row[5], @@ -331,34 +433,46 @@ class WixOAuthService: user_id: str, access_token: str, refresh_token: Optional[str] = None, - expires_in: Optional[int] = None + expires_in: Optional[int] = None, + token_id: Optional[int] = None ) -> bool: """Update tokens for a user (e.g., after refresh).""" try: - # Ensure DB initialized for this user self._init_db(user_id) db_path = self._get_db_path(user_id) expires_at = None if expires_in: expires_at = datetime.now() + timedelta(seconds=expires_in) + + encrypted_access = self._encrypt_token(access_token) + encrypted_refresh = self._encrypt_token(refresh_token) if refresh_token else None with sqlite3.connect(db_path) as conn: + self._migrate_plaintext_tokens_if_needed(conn, user_id) cursor = conn.cursor() - if refresh_token: - cursor.execute(''' - UPDATE wix_oauth_tokens - SET access_token = ?, refresh_token = ?, expires_at = ?, expires_in = ?, - is_active = TRUE, updated_at = datetime('now') - WHERE user_id = ? AND refresh_token = ? - ''', (access_token, refresh_token, expires_at, expires_in, user_id, refresh_token)) + if token_id: + if encrypted_refresh: + cursor.execute(''' + UPDATE wix_oauth_tokens + SET access_token = ?, refresh_token = ?, expires_at = ?, expires_in = ?, + is_active = TRUE, updated_at = datetime('now') + WHERE user_id = ? AND id = ? + ''', (encrypted_access, encrypted_refresh, expires_at, expires_in, user_id, token_id)) + else: + cursor.execute(''' + UPDATE wix_oauth_tokens + SET access_token = ?, expires_at = ?, expires_in = ?, + is_active = TRUE, updated_at = datetime('now') + WHERE user_id = ? AND id = ? + ''', (encrypted_access, expires_at, expires_in, user_id, token_id)) else: cursor.execute(''' UPDATE wix_oauth_tokens SET access_token = ?, expires_at = ?, expires_in = ?, is_active = TRUE, updated_at = datetime('now') WHERE user_id = ? AND id = (SELECT id FROM wix_oauth_tokens WHERE user_id = ? ORDER BY created_at DESC LIMIT 1) - ''', (access_token, expires_at, expires_in, user_id, user_id)) + ''', (encrypted_access, expires_at, expires_in, user_id, user_id)) conn.commit() logger.info(f"Wix OAuth: Tokens updated for user {user_id}") diff --git a/backend/services/research/trends/google_trends_service.py b/backend/services/research/trends/google_trends_service.py index 96b20f8c..af14273d 100644 --- a/backend/services/research/trends/google_trends_service.py +++ b/backend/services/research/trends/google_trends_service.py @@ -343,7 +343,7 @@ class GoogleTrendsService: logger.info( f"[Trends] ===== DONE analyze_trends ===== total={total_ms}ms " f"iot={len(interest_over_time)} ibr={len(interest_by_region)} " - f"rt_top={rt_top} rq_top={rq_top}" + f"rt_top={len(related_topics.get('top', []))} rq_top={len(related_queries.get('top', []))}" ) result = { diff --git a/backend/services/subscription/preflight_validator.py b/backend/services/subscription/preflight_validator.py index 0dc2ab3e..eb33a3dd 100644 --- a/backend/services/subscription/preflight_validator.py +++ b/backend/services/subscription/preflight_validator.py @@ -548,9 +548,11 @@ def validate_video_generation_operations( def validate_scene_animation_operation( pricing_service: PricingService, user_id: str, + scene_count: int = 1, ) -> None: """ Validate the per-scene animation workflow before API calls. + Validates that the user has sufficient credits for *all* scenes in the batch. """ try: operations_to_validate = [ @@ -560,6 +562,7 @@ def validate_scene_animation_operation( 'actual_provider_name': 'wavespeed', 'operation_type': 'scene_animation', } + for _ in range(scene_count) ] can_proceed, message, error_details = pricing_service.check_comprehensive_limits( @@ -581,9 +584,8 @@ def validate_scene_animation_operation( } ) - logger.info(f"[Pre-flight Validator] β Scene animation validated for user {user_id}") - # Validation passed - no return needed (function raises HTTPException if validation fails) - + logger.info(f"[Pre-flight Validator] β Scene animation validated for user {user_id} ({scene_count} scene(s))") + except HTTPException: raise except Exception as e: @@ -730,9 +732,11 @@ def validate_video_generation_operations( def validate_scene_animation_operation( pricing_service: PricingService, user_id: str, + scene_count: int = 1, ) -> None: """ Validate the per-scene animation workflow before API calls. + Validates that the user has sufficient credits for *all* scenes in the batch. """ try: operations_to_validate = [ @@ -742,6 +746,7 @@ def validate_scene_animation_operation( 'actual_provider_name': 'wavespeed', 'operation_type': 'scene_animation', } + for _ in range(scene_count) ] can_proceed, message, error_details = pricing_service.check_comprehensive_limits( @@ -763,7 +768,7 @@ def validate_scene_animation_operation( } ) - logger.info(f"[Pre-flight Validator] β Scene animation validated for user {user_id}") + logger.info(f"[Pre-flight Validator] β Scene animation validated for user {user_id} ({scene_count} scene(s))") except HTTPException: raise diff --git a/backend/services/subscription/pricing_service.py b/backend/services/subscription/pricing_service.py index 341bf652..39dae950 100644 --- a/backend/services/subscription/pricing_service.py +++ b/backend/services/subscription/pricing_service.py @@ -566,10 +566,10 @@ class PricingService: "firecrawl_calls_limit": 0, # DISABLED: Firecrawl not in Free tier "stability_calls_limit": 3, # 3 images - enough to try the product "exa_calls_limit": 10, # 10 research queries - enough to try the product - "video_calls_limit": 0, # DISABLED: Video generation not in Free tier + "video_calls_limit": 2, # 2 video renders - try podcast video on Free "image_edit_calls_limit": 5, # 5 image edits - enough to try the product "audio_calls_limit": 5, # 5 audio clips - enough to try the product - "wavespeed_calls_limit": 0, # DISABLED: WaveSpeed not included in Free tier + "wavespeed_calls_limit": 0, # 0 = unlimited for Free; video controlled via video_calls_limit "gemini_tokens_limit": 50000, "openai_tokens_limit": 0, # DISABLED "anthropic_tokens_limit": 0, # DISABLED diff --git a/docs-site/docs/features/backlink-outreach/analytics.md b/docs-site/docs/features/backlink-outreach/analytics.md new file mode 100644 index 00000000..a36cbf67 --- /dev/null +++ b/docs-site/docs/features/backlink-outreach/analytics.md @@ -0,0 +1,181 @@ +# Analytics + +Track campaign performance with built-in analytics including send volume trends, conversion funnels, reply classification breakdowns, and CSV exports. + +## Dashboard Overview + +The analytics tab provides a comprehensive view of your outreach performance: + +```mermaid +flowchart LR + A[Campaign Analytics] --> B[Volume Trends] + A --> C[Conversion Funnel] + A --> D[Reply Classification] + A --> E[Response Rate] + A --> F[Placement Rate] + A --> G[CSV Exports] + + style A fill:#e3f2fd + style B fill:#e8f5e8 + style G fill:#fff3e0 +``` + +## Metrics + +### Send Volume Trends + +A line chart showing daily email send volume over a configurable time window (7, 14, 30, or 90 days). + +- **X-axis**: Date. +- **Y-axis**: Number of emails sent. +- **Use case**: Spot trends, ensure consistent outreach cadence, stay within daily caps. + +### Conversion Funnel + +A bar chart showing lead counts at each status stage: + +| Stage | Description | +|---|---| +| Discovered | Total leads found. | +| Contacted | Leads that received an outreach email. | +| Replied | Leads that responded (interested or neutral). | +| Placed | Leads that resulted in a published backlink. | + +- **Use case**: Identify bottlenecks in your outreach pipeline. + +### Reply Classification + +A breakdown of auto-classified replies: + +| Classification | Color | Meaning | +|---|---|---| +| Interested | Green | Positive response β follow up! | +| Not interested | Red | Declined β auto-suppressed. | +| Out of office | Yellow | Auto-responder β schedule follow-up. | +| Replied | Blue | General response β needs review. | + +### Response Rate + +Percentage of sent emails that received any reply: + +``` +Response Rate = (Total Replies / Total Sent) Γ 100 +``` + +### Placement Rate + +Percentage of contacted leads that resulted in a published backlink: + +``` +Placement Rate = (Placed Leads / Contacted Leads) Γ 100 +``` + +## Analytics API + +### Campaign Analytics + +**API:** `GET /api/v1/backlink-outreach/campaigns/{campaign_id}/analytics` + +**Query parameters:** + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `days` | int | `30` | Number of days to include in trends. | + +**Response:** + +```json +{ + "total_leads": 150, + "leads_by_status": { + "discovered": 80, + "contacted": 45, + "replied": 18, + "placed": 7, + "bounced": 5 + }, + "total_attempts": 52, + "total_replies": 23, + "replies_by_classification": { + "interested": 12, + "not_interested": 5, + "out_of_office": 3, + "replied": 3 + }, + "response_rate": 0.44, + "placement_rate": 0.16, + "daily_send_volume": [ + {"date": "2025-01-15", "count": 8}, + {"date": "2025-01-16", "count": 12} + ] +} +``` + +### Reporting Snapshot + +Cross-campaign analytics across all campaigns for the authenticated user. + +**API:** `GET /api/v1/backlink-outreach/reporting/snapshot` + +**Response:** + +```json +{ + "total_campaigns": 5, + "total_sends": 342, + "total_replies": 87, + "total_placements": 14, + "overall_response_rate": 0.25, + "overall_placement_rate": 0.04 +} +``` + +!!! note "Reply counting" + The reporting snapshot counts `OutreachReply` records (not `status == "replied"` on attempts). This ensures accuracy β a lead marked "replied" manually without an actual reply record won't inflate the count. + +## CSV Exports + +Export campaign data as CSV files for CRM import, spreadsheet analysis, or client reporting. + +### Export Leads + +**API:** `GET /api/v1/backlink-outreach/campaigns/{campaign_id}/export/leads` + +### Export Attempts + +**API:** `GET /api/v1/backlink-outreach/campaigns/{campaign_id}/export/attempts` + +### Export Replies + +**API:** `GET /api/v1/backlink-outreach/campaigns/{campaign_id}/export/replies` + +### CSV Safety + +All exports include these safety measures: + +| Measure | Purpose | +|---|---| +| Explicit fieldnames | Only expected columns are included. | +| `extrasaction="ignore"` | Unexpected fields are silently dropped. | +| Formula injection sanitization | Cells starting with `=`, `+`, `-`, `@` are prefixed with a single quote to prevent formula injection in spreadsheets. | + +!!! warning "Export loading" + Exports may take a few seconds for large campaigns. The UI shows an "Exporting..." state with a disabled button while the download is in progress. + +## UI Features + +### Time Window Selector + +Choose from 7, 14, 30, or 90 days for trend charts. The analytics data is re-fetched when the window changes. + +### Separate Loading States + +Each data section (attempts, replies, analytics) has its own loading indicator, so slow analytics queries don't block the entire page. + +### Error Handling + +If analytics or export requests fail, a toast notification shows the error message. On 5xx server errors, the store automatically retries read operations once with exponential backoff. + +--- + +*Next: [API Reference](api-reference.md) β full endpoint documentation.* diff --git a/docs-site/docs/features/backlink-outreach/api-reference.md b/docs-site/docs/features/backlink-outreach/api-reference.md new file mode 100644 index 00000000..391725ff --- /dev/null +++ b/docs-site/docs/features/backlink-outreach/api-reference.md @@ -0,0 +1,449 @@ +# API Reference + +Complete reference for all Backlink Outreach API endpoints. All endpoints require Clerk authentication via `Depends(get_current_user)`. + +## Authentication + +All endpoints use Clerk authentication. Include the session token in the `Authorization` header: + +``` +Authorization: Bearer{subtitle}
+- Discover guest post opportunities, manage campaigns, and track outreach. -
- - {/* Tabs */} -+ AI-powered guest post outreach platform β discover opportunities, manage campaigns, compose emails, and track results. +
No campaigns yet. Create one to get started.
- )} - - {campaigns.map((c) => ( -Loading campaigns...
}Searching for opportunities using Exa + DuckDuckGo...
} - - {discoveredOpportunities.length > 0 && ( -- Status: {selectedCampaign.status} · {selectedCampaign.lead_count} leads + {campaigns.length === 0 && !isLoading && ( +
+ No campaigns yet. Create one above to get started.
- {selectedCampaign.leads.length === 0 && ( -No leads yet. Go to Discover tab to find opportunities.
- )} - {selectedCampaign.leads.map((lead) => ( -Loading...
} +Select a campaign from the Campaigns tab to view its leads.
- )} -+ Searching across Exa (neural) + DuckDuckGo... This may take 10β20 seconds. +
+ )} + {discoveredOpportunities.length > 0 && ( ++ Enter a keyword above and click Discover to find guest post opportunities. +
+ )} ++ {selectedCampaign.lead_count} leads · Status: {selectedCampaign.status} +
++ No leads yet. Go to the Discover tab to find and save opportunities. +
+ )} + + {/* Lead cards */} + {selectedCampaign.leads.map((lead) => ( +| {h} | + ))} +|||
|---|---|---|---|
| {a.subject} | +{renderStatusBadge(a.status)} | +{a.sender_email} | +{a.sent_at ? new Date(a.sent_at).toLocaleDateString() : '-'} | +
+ Select a campaign from the Campaigns tab to view its leads. +
+ )} +