Pipedream Integration

import { defineComponent } from "@pipedream/platform";

export default defineComponent({
  name: "Lead Generation Pipeline",
  description: "Automated lead discovery and CRM integration",
  version: "0.1.0",
  async run({ steps, $ }) {
    // Step 1: Search for companies
    const searchResults = await this.VENYM_SEARCH.search({
      query: "construction companies Seattle",
      maxResults: 20,
      extractContacts: true,
    });

    // Step 2: Filter valid leads
    const validLeads = searchResults.search_results.filter(result =>
      result.contact_info?.email &&
      !result.contact_info.email.includes('noreply')
    );

    // Step 3: Enrich with additional data
    const enrichedLeads = await Promise.all(
      validLeads.map(async (lead) => {
        const scrapeResult = await this.VENYM_SEARCH.scrape({
          url: lead.link,
          extractOptions: ["title", "text", "metadata"]
        });

        return {
          ...lead,
          company_size: this.extractCompanySize(scrapeResult.primary_content?.text),
          technology_stack: this.extractTechStack(scrapeResult.primary_content?.text),
          scraped_at: new Date().toISOString(),
        };
      })
    );

    // Step 4: Add to CRM
    const crmResults = await Promise.all(
      enrichedLeads.map(async (lead) => {
        return await this.hubspot.createContact({
          email: lead.contact_info.email,
          company: lead.title,
          website: lead.link,
          phone: lead.contact_info.phone,
          lead_source: "Venym Search Automation",
          company_size: lead.company_size,
        });
      })
    );

    // Step 5: Send notification
    await this.slack.sendMessage({
      channel: "#sales",
      text: `🎯 Found ${enrichedLeads.length} new qualified leads in construction industry`,
      attachments: [{
        color: "good",
        fields: [
          {
            title: "Companies Found",
            value: enrichedLeads.length.toString(),
            short: true
          },
          {
            title: "With Contact Info",
            value: enrichedLeads.filter(l => l.contact_info?.email).length.toString(),
            short: true
          }
        ]
      }]
    });

    $.export("$summary", `Processed ${enrichedLeads.length} leads successfully`);
    return {
      leads: enrichedLeads,
      crm_results: crmResults,
      processed_at: new Date().toISOString()
    };
  },

  extractCompanySize(text) {
    if (!text) return "Unknown";
    const sizeIndicators = {
      "startup": ["startup", "founded", "early stage"],
      "small": ["small business", "10-50 employees", "team of"],
      "medium": ["growing company", "50-200 employees", "established"],
      "large": ["enterprise", "500+ employees", "fortune", "publicly traded"]
    };

    for (const [size, indicators] of Object.entries(sizeIndicators)) {
      if (indicators.some(indicator => text.toLowerCase().includes(indicator))) {
        return size;
      }
    }
    return "Unknown";
  },

  extractTechStack(text) {
    if (!text) return [];
    const technologies = ["WordPress", "React", "Angular", "Vue", "Shopify", "Salesforce", "HubSpot"];
    return technologies.filter(tech =>
      text.toLowerCase().includes(tech.toLowerCase())
    );
  }
});

export default defineComponent({
  name: "Venym Search Webhook Trigger",
  description: "Trigger workflow when Venym Search finds new results",
  version: "0.1.0",
  type: "source",
  dedupe: "unique",
  props: {
    http: {
      type: "$.interface.http",
      customResponse: true,
    },
    VENYM_SEARCH: {
      type: "app",
      app: "VENYM_SEARCH",
    },
    query: {
      type: "string",
      label: "Search Query",
      description: "Query to monitor for new results",
    },
    checkInterval: {
      type: "integer",
      label: "Check Interval (minutes)",
      description: "How often to check for new results",
      default: 60,
      min: 5,
    },
  },
  hooks: {
    async activate() {
      // Store the last check timestamp
      this.db.set("lastCheck", Date.now());
    },
  },
  async run({ body, headers }) {
    const lastCheck = this.db.get("lastCheck") || 0;
    const now = Date.now();

    // Only run if enough time has passed
    if (now - lastCheck < this.checkInterval * 60 * 1000) {
      return;
    }

    try {
      const results = await this.VENYM_SEARCH.search({
        query: this.query,
        maxResults: 20,
      });

      // Filter results newer than last check
      const newResults = results.search_results.filter(result => {
        const resultDate = new Date(result.date || 0).getTime();
        return resultDate > lastCheck;
      });

      if (newResults.length > 0) {
        this.db.set("lastCheck", now);

        this.$emit({
          query: this.query,
          new_results: newResults,
          total_found: newResults.length,
          timestamp: new Date().toISOString(),
        }, {
          id: `${this.query}-${now}`,
          summary: `Found ${newResults.length} new results for "${this.query}"`,
          ts: now,
        });
      }
    } catch (error) {
      console.error("Venym Search trigger error:", error);
    }
  },
});