File Coverage

blib/lib/Net/Async/Firecrawl.pm

Criterion	Covered	Total	%
statement	180	201	89.5
branch	41	68	60.2
condition	18	37	48.6
subroutine	42	46	91.3
pod	11	11	100.0
total	292	363	80.4

package Net::Async::Firecrawl;
# ABSTRACT: IO::Async Firecrawl v2 client with flow helpers
use strict;
use warnings;
use parent 'IO::Async::Notifier';

use Carp qw( croak );
use WWW::Firecrawl ();
use WWW::Firecrawl::Error ();
use Net::Async::HTTP ();
use Future ();
use Future::Utils qw( repeat );

our $VERSION = '0.001';

sub _init {
  my ( $self, $args ) = @_;
  $self->SUPER::_init($args);
  $self->{firecrawl} ||= WWW::Firecrawl->new(
    ( exists $args->{base_url}    ? ( base_url    => delete $args->{base_url} )    : () ),
    ( exists $args->{api_key}     ? ( api_key     => delete $args->{api_key} )     : () ),
    ( exists $args->{api_version} ? ( api_version => delete $args->{api_version} ) : () ),
  );
  $self->{poll_interval} = exists $args->{poll_interval} ? delete $args->{poll_interval} : 3;
  $self->{http} = delete $args->{http};
  $self->{firecrawl} = delete $args->{firecrawl} if exists $args->{firecrawl};
  $self->{delay_sub} = delete $args->{delay_sub};
  return;
}

sub configure_unknown {
  my ( $self, %args ) = @_;
  for my $k (qw( base_url api_key api_version poll_interval firecrawl http delay_sub )) {
    delete $args{$k};
  }
  return unless %args;
  croak "Unknown configuration keys: ".join(',', sort keys %args);
}

sub firecrawl     { $_[0]->{firecrawl} }
sub poll_interval { @_ > 1 ? ($_[0]->{poll_interval} = $_[1]) : $_[0]->{poll_interval} }

sub http {
  my ( $self ) = @_;
  return $self->{http} if $self->{http};
  my $http = Net::Async::HTTP->new(
    user_agent => $self->firecrawl->user_agent_string,
    max_connections_per_host => 4,
  );
  $self->add_child($http);
  return $self->{http} = $http;
}

sub _on_added_to_loop {
  my ( $self, $loop ) = @_;
  $self->SUPER::_on_added_to_loop($loop) if $self->can('SUPER::_on_added_to_loop');
  # Lazy-build http so it's parented properly
  $self->http;
}

#----------------------------------------------------------------------
# Generic request dispatch
#----------------------------------------------------------------------

sub do_request {
  my ( $self, $request ) = @_;
  croak "do_request requires HTTP::Request" unless $self->firecrawl->is_request($request);
  return $self->_do_request_with_retry($request, 1);
}

sub _delay_future {
  my ( $self, $seconds ) = @_;
  return $self->{delay_sub}->($seconds) if $self->{delay_sub};
  return $self->loop->delay_future( after => $seconds );
}

sub _do_request_with_retry {
  my ( $self, $request, $attempt ) = @_;
  my $fc = $self->firecrawl;
  my $max = $fc->max_attempts;
  return $self->http->do_request( request => $request )->then(sub {
    my ( $response ) = @_;
    my ( $err, $retryable ) = $fc->_classify_response( $response, $attempt );
    return Future->done($response) unless $err;
    if ( $retryable && $attempt < $max ) {
      my $delay = $fc->_retry_delay( $response, $attempt );
      if ( my $cb = $fc->on_retry ) {
        $cb->( $attempt, $delay, $err );
      }
      return $self->_delay_future($delay)->then(sub {
        $self->_do_request_with_retry( $request, $attempt + 1 );
      });
    }
    return Future->fail($err, 'firecrawl', $attempt);
  });
}

# Build a Future-returning wrapper named $name around a WWW::Firecrawl
# request builder + response parser pair.
sub _install_wrapper {
  my ( $class, $name, $opts ) = @_;
  $opts ||= {};
  my $builder = $opts->{builder} || "${name}_request";
  my $parser  = $opts->{parser}  || "parse_${name}_response";

  no strict 'refs';
  *{"${class}::${name}"} = sub {
    my ( $self, @args ) = @_;
    my $fc = $self->firecrawl;
    my $req = $fc->$builder(@args);
    return $self->do_request($req)->then(sub {
      my $response = $_[0];
      my $data = eval { $fc->$parser($response) };
      if ( my $e = $@ ) {
        my $err = ref $e && $e->isa('WWW::Firecrawl::Error')
          ? $e
          : WWW::Firecrawl::Error->new( type => 'api', message => "$e", response => $response );
        return Future->fail($err, 'firecrawl');
      }
      return Future->done($data);
    });
  };
}

# Declarative endpoint list: one Future-returning method per endpoint.
my @ENDPOINTS = qw(
  scrape
  crawl
  crawl_status
  crawl_cancel
  crawl_errors
  crawl_active
  crawl_params_preview
  map
  search
  batch_scrape
  batch_scrape_status
  batch_scrape_cancel
  batch_scrape_errors
  extract
  extract_status
  agent
  agent_status
  agent_cancel
  browser_create
  browser_list
  browser_delete
  browser_execute
  scrape_execute
  scrape_browser_stop
  credit_usage
  credit_usage_historical
  token_usage
  token_usage_historical
  queue_status
  activity
);

__PACKAGE__->_install_wrapper($_) for @ENDPOINTS;

# Pagination-follow helpers — same parser as their base endpoint.
__PACKAGE__->_install_wrapper('crawl_status_next', {
  builder => 'crawl_status_next_request',
  parser  => 'parse_crawl_status_response',
});
__PACKAGE__->_install_wrapper('batch_scrape_status_next', {
  builder => 'batch_scrape_status_next_request',
  parser  => 'parse_batch_scrape_status_response',
});

#----------------------------------------------------------------------
# Flow helpers
#----------------------------------------------------------------------

sub _poll_until_done {
  my ( $self, %args ) = @_;
  my $status_cb = $args{status};
  my $interval  = $args{interval} || $self->poll_interval;
  my $loop      = $self->loop or croak "not added to a loop yet";

  return repeat {
    $status_cb->()->then(sub {
      my ( $status ) = @_;
      my $st = $status->{status} // '';
      if ( $st eq 'failed' || $st eq 'cancelled' ) {
        my $msg = "Firecrawl job $st";
        $msg .= ': ' . $status->{error} if defined $status->{error};
        return Future->fail(
          WWW::Firecrawl::Error->new(
            type => 'job',
            message => $msg,
            data => $status,
          ),
          'firecrawl',
        );
      }
      return Future->done($status) if $st eq 'completed';
      return $self->_delay_future($interval)
        ->then(sub { Future->done($status) });
    });
  } until => sub {
    my $f = $_[0];
    return 1 if $f->is_failed;
    my $s = $f->get;
    return ($s->{status} // '') eq 'completed';
  };
}

sub _collect_pages {
  my ( $self, $first_status, $next_method ) = @_;
  my @pages = @{ $first_status->{data} || [] };
  my $next = $first_status->{next};
  my $last_status = $first_status;
  return Future->done({ %$first_status, data => \@pages }) unless $next;
  my $current = $next;
  my $loop_f = repeat {
    my $url = $current;
    $self->$next_method($url)->on_done(sub {
      my ( $s ) = @_;
      $last_status = $s;
      push @pages, @{ $s->{data} || [] };
      $current = $s->{next};
    });
  } while => sub { defined $current };
  return $loop_f->then(sub {
    Future->done({
      %$first_status,
      data   => \@pages,
      status => $last_status->{status},
    });
  });
}

sub _collect_crawl_pages { $_[0]->_collect_pages($_[1], 'crawl_status_next') }
sub _collect_batch_pages { $_[0]->_collect_pages($_[1], 'batch_scrape_status_next') }

# Apply is_failure classification to a collected crawl/batch result,
# producing { data, failed, raw_data, stats } from raw `data`.
sub _split_pages {
  my ( $self, $result ) = @_;
  my $fc = $self->firecrawl;
  my @raw = @{ $result->{data} || [] };
  my @ok;
  my @failed;
  for my $page (@raw) {
    if ( $fc->is_failure->($page) ) {
      my $meta = $page->{metadata} || {};
      push @failed, {
        url        => $meta->{sourceURL} // $meta->{url},
        statusCode => $meta->{statusCode},
        error      => $fc->scrape_error($page),
        page       => $page,
      };
    }
    else {
      push @ok, $page;
    }
  }
  return {
    %$result,
    data     => \@ok,
    failed   => \@failed,
    raw_data => \@raw,
    stats    => {
      ok     => scalar @ok,
      failed => scalar @failed,
      total  => scalar @raw,
    },
  };
}


sub crawl_and_collect {
  my ( $self, %args ) = @_;
  my $interval = delete $args{poll_interval};
  $self->crawl(%args)->then(sub {
    my ( $job ) = @_;
    my $id = $job->{id} or return Future->fail(
      WWW::Firecrawl::Error->new( type => 'api', message => "crawl returned no id" ),
      'firecrawl',
    );
    $self->_poll_until_done(
      status   => sub { $self->crawl_status($id) },
      interval => $interval,
    )
    ->then(sub { $self->_collect_crawl_pages($_[0]) })
    ->then(sub { Future->done( $self->_split_pages($_[0]) ) });
  });
}

sub batch_scrape_and_wait {
  my ( $self, %args ) = @_;
  my $interval = delete $args{poll_interval};
  $self->batch_scrape(%args)->then(sub {
    my ( $job ) = @_;
    my $id = $job->{id} or return Future->fail(
      WWW::Firecrawl::Error->new( type => 'api', message => "batch_scrape returned no id" ),
      'firecrawl',
    );
    $self->_poll_until_done(
      status   => sub { $self->batch_scrape_status($id) },
      interval => $interval,
    )
    ->then(sub { $self->_collect_batch_pages($_[0]) })
    ->then(sub { Future->done( $self->_split_pages($_[0]) ) });
  });
}

# Start extract → poll until done → return the final payload.
sub extract_and_wait {
  my ( $self, %args ) = @_;
  my $interval = delete $args{poll_interval};
  $self->extract(%args)->then(sub {
    my ( $job ) = @_;
    my $id = $job->{id} or return Future->fail("extract returned no id");
    $self->_poll_until_done(
      status   => sub { $self->extract_status($id) },
      interval => $interval,
    );
  });
}

# Start agent → poll until done.
sub agent_and_wait {
  my ( $self, %args ) = @_;
  my $interval = delete $args{poll_interval};
  $self->agent(%args)->then(sub {
    my ( $job ) = @_;
    my $id = $job->{id} or return Future->fail("agent returned no id");
    $self->_poll_until_done(
      status   => sub { $self->agent_status($id) },
      interval => $interval,
    );
  });
}

sub scrape_many {
  my ( $self, $urls, %common ) = @_;
  croak "scrape_many: first arg must be arrayref of URLs" unless ref $urls eq 'ARRAY';
  my $fc = $self->firecrawl;
  my @futures = map {
    my $url = $_;
    $self->scrape( url => $url, %common )->then(
      sub {
        my $data = $_[0];
        if ( $fc->is_failure->($data) ) {
          my $err = WWW::Firecrawl::Error->new(
            type => 'page',
            message => 'Firecrawl scrape failed: ' . ($fc->scrape_error($data) // 'unknown'),
            data => $data,
            status_code => $fc->scrape_status($data),
            url => $url,
          );
          return Future->done({ url => $url, failed => { url => $url, error => $err } });
        }
        return Future->done({ url => $url, ok => { url => $url, data => $data } });
      },
      sub {
        my ( $err ) = @_;
        my $e = ref $err && $err->isa('WWW::Firecrawl::Error')
          ? $err
          : WWW::Firecrawl::Error->new( type => 'api', message => "$err", url => $url );
        return Future->done({ url => $url, failed => { url => $url, error => $e } });
      },
    );
  } @$urls;
  return Future->wait_all(@futures)->then(sub {
    my @resolved = map { $_->get } @_;
    my @ok     = map { $_->{ok}     } grep { exists $_->{ok}     } @resolved;
    my @failed = map { $_->{failed} } grep { exists $_->{failed} } @resolved;
    Future->done({
      ok => \@ok,
      failed => \@failed,
      stats => { ok => scalar @ok, failed => scalar @failed, total => scalar @$urls },
    });
  });
}

sub retry_failed_pages {
  my ( $self, $result, %scrape_opts ) = @_;
  my @urls = map { $_->{url} } @{ $result->{failed} || [] };
  return $self->scrape_many( \@urls, %scrape_opts );
}

1;

__END__