File Coverage

blib/lib/JSON/Lines.pm

Criterion	Covered	Total	%
statement	244	273	89.3
branch	87	130	66.9
condition	26	39	66.6
subroutine	29	31	93.5
pod	21	21	100.0
total	407	494	82.3

package JSON::Lines;
use 5.006; use strict; use warnings; our $VERSION = '1.11';
use Cpanel::JSON::XS; use base 'Import::Export';
our ($JSON, $LINES, $STRING, %EX);
BEGIN {
	$JSON = Cpanel::JSON::XS->new;
	$STRING = qr{ " (?> (?: [^"\\]++ | \\. )*+ ) " }x;
	$LINES = qr{ ( [\[\{] (?> (?: (?> [^\[\]\{\}"]++ ) | $STRING | (??{ $LINES }) )*+ ) [\]\}] ) }x;
	%EX = (
		jsonl => [qw/all/]
	);
}
sub jsonl {
	my %args = (scalar @_ == 1 ? %{$_[0]} : @_);
	my $self = __PACKAGE__->new(%args);
	return $args{file}
		? $self->encode_file($args{file}, $args{data})
		: $self->encode($args{data})
	if ($args{encode});
	return $args{file}
		? $self->decode_file($args{file})
		: $self->decode($args{data})
	if ($args{decode});
}
sub new {
	my ($pkg, %args) = (shift, scalar @_ == 1 ? %{$_[0]} : @_);
	my $self = bless { headers => [], _buffer => '' }, $pkg;
	exists $args{$_} && $JSON->$_($args{$_}) for qw/pretty canonical utf8/;
	$self->{$_} = $args{$_} for qw/parse_headers error_cb success_cb/;
	$self;
}
sub pretty {
	my ($self, $val) = @_;
	return $JSON->get_indent if @_ == 1;
	$JSON->pretty($val);
	return $self;
}
sub canonical {
	my ($self, $val) = @_;
	return $JSON->get_canonical if @_ == 1;
	$JSON->canonical($val);
	return $self;
}
sub utf8 {
	my ($self, $val) = @_;
	return $JSON->get_utf8 if @_ == 1;
	$JSON->utf8($val);
	return $self;
}
sub parse_headers {
	my ($self, $val) = @_;
	return $self->{parse_headers} if @_ == 1;
	$self->{parse_headers} = $val;
	return $self;
}
sub error_cb {
	my ($self, $val) = @_;
	return $self->{error_cb} if @_ == 1;
	$self->{error_cb} = $val;
	return $self;
}
sub success_cb {
	my ($self, $val) = @_;
	return $self->{success_cb} if @_ == 1;
	$self->{success_cb} = $val;
	return $self;
}
sub encode {
	my ($self, @data) = (shift, scalar @_ == 1 ? @{$_[0]} : @_);
	@data = $self->_parse_headers(@data)
		if ($self->{parse_headers});
	my $stream;
	for (@data) {
		my $json = eval { $JSON->encode($_) };
		if ($@) {
			if ($self->{error_cb}) {
				$self->{error_cb}->('encode', $@, $_);
			} else {
				die $@;
			}
		} else {
			$self->{success_cb}->('encode', $json, $_) if $self->{success_cb};
			$stream .= $json . ($json =~ m/\n$/ ? "" : "\n");
		}
	}
	return $self->{stream} = $stream;
}
sub encode_file {
	my ($self, $file) = (shift, shift);
	open my $fh, '>', $file;
	print $fh $self->encode(@_);
	close $fh;
	return $file;
}
sub decode {
	my ($self, $string) = @_;
	if (defined $self->{_buffer} && length $self->{_buffer}) {
		$string = $self->{_buffer} . $string;
		$self->{_buffer} = '';
	}
	my @lines;
	my $pos = 0;
	my $len = length($string);
	while ($pos < $len) {
		if ($string =~ m/\G[\s\n]+/gc) {
			$pos = pos($string);
			next;
		}
		my $remaining = substr($string, $pos);
		my ($obj, $chars_consumed, $regex_matched) = $self->_decode_one($remaining);
		if (defined $obj) {
			push @lines, $obj;
			$pos += $chars_consumed;
		} elsif ($regex_matched) {
			$pos += $chars_consumed if $chars_consumed > 0;
			$pos++ if $chars_consumed == 0;
		} else {
			if ($remaining =~ /^[\[\{]/) {
				$self->{_buffer} = $remaining;
				last;
			}
			if ($string =~ m/\G[^\[\{]+/gc) {
				$pos = pos($string);
			} else {
				$pos++;
			}
		}
		pos($string) = $pos;
	}
	@lines = $self->_deparse_headers(@lines)
		if ($self->{parse_headers});
	return wantarray ? @lines : \@lines;
}
sub remaining {
	my ($self) = @_;
	return $self->{_buffer} // '';
}
sub clear_buffer {
	my ($self) = @_;
	$self->{_buffer} = '';
	return $self;
}
sub decode_file {
	my ($self, $file) = (shift, shift);
	open my $fh, '<', $file;
	my $content = do { local $/; <$fh> };
	close $fh;
	return $self->decode($content);
}
sub add_line {
	my ($self, $line, $fh) = @_;
	if (defined $fh) {
		print $fh $self->encode([$line]);
	} else {
		my $stream = $self->{stream};
		my $add = $self->encode([$line]);
		$self->{stream} =  $stream .  $add;
		$self->{stream};
	}
}
sub clear_stream {
	$_[0]->{stream} = '';
}
sub get_lines {
	my ($self, $fh, $lines) = @_;
	my @lines;
	for (1 .. $lines) {
		my $line = $self->get_line($fh);
		push @lines, $line;
		last if eof($fh);
	}
	return wantarray ? @lines : \@lines;
}
sub get_line {
	my ($self, $fh) = @_;
	$self->{_line_buffer} //= [];
	if (@{$self->{_line_buffer}}) {
		return shift @{$self->{_line_buffer}};
	}
	my $line = '';
	$line .= <$fh> while ($line !~ m/^$LINES/ && !eof($fh));
	return undef if $line eq '' && eof($fh);
	my @objects = $self->decode($line);
	return undef unless @objects;
	my $first = shift @objects;
	push @{$self->{_line_buffer}}, @objects;
	return $first;
}
sub get_line_at {
	my ($self, $fh, $index, $seek) = @_;
	my $fh_id = fileno($fh);
	my ($target_line, $target_offset) = $index =~ /:/
		? split(/:/, $index)
		: ($index, 0);
	$self->{_line_pos} //= {};
	$self->{_parsed_lines} //= {};
	if ($seek) {
		seek $fh, 0, 0;
		$self->{_line_pos}{$fh_id} = 0;
		$self->{_parsed_lines}{$fh_id} = {};
	}
	$self->{_line_pos}{$fh_id} //= 0;
	$self->{_parsed_lines}{$fh_id} //= {};
	if ($target_line < $self->{_line_pos}{$fh_id}) {
		seek $fh, 0, 0;
		$self->{_line_pos}{$fh_id} = 0;
		$self->{_parsed_lines}{$fh_id} = {};
	}
	while ($self->{_line_pos}{$fh_id} < $target_line && !eof($fh)) {
		<$fh>;
		$self->{_line_pos}{$fh_id}++;
	}
	return undef if eof($fh);
	if (exists $self->{_parsed_lines}{$fh_id}{$target_line}) {
		my $line_objects = $self->{_parsed_lines}{$fh_id}{$target_line};
		return undef unless $line_objects && @$line_objects > $target_offset;
		return $line_objects->[$target_offset];
	}
	my $line = '';
	my $start_line = $self->{_line_pos}{$fh_id};
	while ($line !~ m/^$LINES/ && !eof($fh)) {
		$line .= <$fh>;
		$self->{_line_pos}{$fh_id}++;
	}
	return undef unless $line =~ m/^$LINES/;
	my @objects = $self->decode($line);
	$self->{_parsed_lines}{$fh_id}{$start_line} = \@objects;
	return undef unless @objects > $target_offset;
	return $objects[$target_offset];
}
sub get_subset {
	my ($self, $fh, $offset, $length, $out_file) = @_;
	my $in_fh = 1;
	if (ref $fh ne 'GLOB') {
		my $file = $fh;
		open my $ffh, '<', $file or die "cannot open file: $!";
		$fh = $ffh;
		$in_fh = 0;
	}
	seek $fh, 0, 0;
	my @all_objects;
	my $line = '';
	while (!eof($fh)) {
		do { $line .= <$fh> } while ($line !~ m/^$LINES/ && !eof($fh));
		if ($line =~ m/^$LINES/) {
			push @all_objects, $self->decode($line);
			$line = "";
		}
	}
	if (!$in_fh) {
		close $fh;
	}
	my $end = $length;
	$end = $#all_objects if $end > $#all_objects;
	my @subset = @all_objects[$offset .. $end];
	if ($out_file) {
		open my $cfh, '>', $out_file or die "cannot open file for writing: $!";
		print $cfh $self->encode(\@subset);
		close $cfh;
		return 1;
	}
	return \@subset;
}
sub group_lines {
	my ($self, $fh, $key) = @_;
	die "Key must be provided for grouping" unless defined $key;
	seek $fh, 0, 0;
	my (%groups, $line_num, $line);
	($line_num, $line) = (0, '');
	while (!eof($fh)) {
		my $start_line = $line_num;
		while ($line !~ m/^$LINES/ && !eof($fh)) {
			$line .= <$fh>;
			$line_num++;
		}
		last unless $line =~ m/^$LINES/;
		my @objects = $self->decode($line);
		$line = '';
		my $obj_offset = 0;
		for my $data (@objects) {
			my $value = ref $key eq 'CODE'
				? do { local $_ = $data; $key->($data) }
				: $data->{$key};
			my $index = @objects > 1 ? "$start_line:$obj_offset" : $start_line;
			push @{ $groups{$value} }, $index;
			$obj_offset++;
		}
	}
	return \%groups;
}
sub _decode_one {
	my ($self, $string) = @_;
	return (undef, 0, 0) unless $string =~ /^[\[\{]/;
	if ($string =~ m/^($LINES)/) {
		my $json_str = $1;
		my $chars_consumed = length($json_str);
		my $struct = eval { $JSON->decode($json_str) };
		if ($@) {
			return (undef, $chars_consumed, 1);
		}
		$self->{success_cb}->('decode', $struct, $json_str) if $self->{success_cb};
		return ($struct, $chars_consumed, 1);
	}
	return (undef, 0, 0);
}
sub _decode_line {
	my ($self, $line) = @_;
	my $struct = eval { $JSON->decode($line) };
	if ($@) {
		if ($self->{error_cb}) {
			return $self->{error_cb}->('decode', $@, $line);
		} else {
			die $@;
		}
	}
	return $self->{success_cb}->('decode', $struct, $line) if $self->{success_cb};
	return $struct;
}
sub _parse_headers {
	my ($self, @data) = @_;
	my @headers = @{ $self->{headers} };
	unless (@headers) {
		if (ref $data[0] eq 'ARRAY') {
			@headers = @{ shift @data };
		} else {
			my %key_map;
			for (@data) {
				%key_map = (%key_map, %{$_});
			}
			@headers = sort keys %key_map;
		}
		$self->{headers} = \@headers;
	}
	my @body;
	for my $d (@data) {
		push @body, (ref $d eq 'ARRAY')
			? $d
			: [
				map {
					$d->{$_}
				} @headers
			];
	}
	return (
		\@headers,
		@body
	);
}
sub _deparse_headers {
	my ($self, @data) = @_;
	return @data unless ref $data[0] eq 'ARRAY';
	my @headers = @{ shift @data };
	my @body;
	for my $d (@data) {
		my $i = 0;
		push @body, (ref $d eq 'HASH')
			? $d
			: {
				map {
					$_ => $d->[$i++]
				} @headers
			};
	}
 	return @body;
}
1;

__END__