<commit_before>#!/opt/perl

use lib qw(/opt/infoservant.com/docroot/info_servant/lib);

use SiteCode::Modern;

use autodie;

use File::Slurp;
use File::Path;
use Sys::Syslog;

use SiteCode::DBX;
use XML::Feed;

use Mojo::Util;

use Mojo::UserAgent;

use SiteCode::Feed;
use SiteCode::Feeds;
use DateTime;

my $data_dir = "/opt/infoservant.com/data/feed_files";
my $html_dir = "/opt/infoservant.com/data/html_files";
Mojo::Util::spurt($$, "/tmp/feeder.pl.pid");

while (1) {
    my $dbx = SiteCode::DBX->new();

    my $feeds = $dbx->array(qq(
        select id, name as url from feed
    ));

    foreach my $feed (@{ $feeds }) {
        eval {
            $dbx->do("SELECT id FROM feed WHERE id = ? FOR UPDATE NOWAIT", undef, $$feed{id});
        };
        if ($@) {
            next;
        }

        my $url_dir = Mojo::Util::url_escape($$feed{url});

        my $the_dir = "$data_dir/$url_dir";

        my $obj = SiteCode::Feed->new(id => $$feed{id});

        if (!-d $the_dir) {
            mkdir($the_dir);
        }

        my $last_check = $obj->key("last_check") || 0;
        my $diff = time() - $last_check;
        if (900 > $diff) {
            $dbx->dbh->rollback();
            next;
        }
        $obj->key("last_check", time());
        $dbx->dbh->commit;

        my $ua = Mojo::UserAgent->new;
        $ua->max_redirects(10);

        info('ua->head :: %s', $$feed{url});

        my $tx = $ua->head($$feed{url});
        if (my $res = $tx->success) { 
            my $last_modified = $tx->res->headers->last_modified || "";
            my $prev_modified = $obj->key("last_modified") || "";

            if ($last_modified eq $prev_modified) {
                next;
            }

            $obj->key("last_modified", $last_modified);
        }
        else {
            my ($err, $code) = $tx->error;
            my $string =  $code ? "$code response: $err" : "Connection error: $err";

            info("head error(%s) :: %s", $$feed{url}, $string);

            system("/bin/touch", $the_dir);

            next;
        }

        info('ua->get :: %s', $$feed{url});

        $tx = $ua->get($$feed{url});
        if (my $res = $tx->success) { 
            Mojo::Util::spurt($res->body, "$the_dir/the.feed");

            info("spurt(%s) :: %s", "$the_dir/the.feed", $$feed{url});
        } else {
            my ($err, $code) = $tx->error;
            my $string =  $code ? "$code response: $err" : "Connection error: $err";

            info("get error(%s) :: %s", $$feed{url}, $string);
        }

        system("/bin/touch", $the_dir);

        my $parse;
        eval {
            $parse = XML::Feed->parse("$the_dir/the.feed");
            $obj->key("title", $parse->title());
            $obj->key("base", $parse->base());
            $obj->key("link", $parse->link());
        };
        if ($@) {
            info("parse error(%s) :: %s", $$feed{url}, $@);
        }
        my $entries = $parse ? [$parse->entries()] : [];

        my $feed_path = "$html_dir/$$feed{id}";
        if (@{ $entries}) {
            info("DELETE FROM entry (%s) :: %s", $$feed{url}, scalar(@{ $entries }) || 0);

            eval {
                $dbx->do("DELETE FROM entry WHERE feed_name = ?", undef, $$feed{url});
            };
            if ($@) {
                info("DELETE FROM entry error(%s) :: %s", $$feed{url}, $@);
            }
            eval {
                if (-d $feed_path && $feed_path =~ m#^/opt/infoservant.com/data/html_files/\d+$#) {
                    info("remove_tree (%s) :: %s", $$feed{url}, $feed_path);
                    File::Path::remove_tree($feed_path);
                }
                info("mkdir (%s) :: %s", $$feed{url}, $feed_path);
                mkdir($feed_path) or die("mkdir :: $feed_path: $!");
            };
            if ($@) {
                info("Feed path error: %s :: %s :: %s :: %s", $$feed{url}, $feed_path, $@, $!);
            }
            else {
                $dbx->dbh->commit;
                foreach my $entry (@{ $entries }) {
                    eval {
                        my $exists = $dbx->col("SELECT id FROM entry WHERE feed_name = ? and entry_id = ?", undef, $$feed{url}, $entry->id());
                        unless ($exists) {
                            $dbx->do(
                                "INSERT INTO entry (feed_name, feed_title, issued, title, entry_id, link) VALUES (?, ?, ?, ?, ?, ?)", 
                                undef, 
                                $$feed{url},
                                $parse->title() || "",
                                $entry->issued() || DateTime->from_epoch(epoch => time() - 604800),
                                $entry->title() || "", 
                                $entry->id(), 
                                $entry->link(),
                            );
                            my $id = $dbx->last_insert_id(undef,undef,undef,undef,{sequence=>'entry_id_seq'});
                            my $html_file = "$feed_path/$id.html";
                            # Mojo::Util::spurt(utf8::encode($entry->content->body()), $html_file);
                            File::Slurp::write_file($html_file, {binmode => ':utf8', atomic => 1}, $entry->content->body() || "No content found.");
                            $dbx->dbh->commit;
                        }
                    };
                    if ($@) {
                        my $err = $@;
                        info("INSERT INTO entry error(%s) :: %s", $$feed{url}, $err);
                        $dbx->dbh->rollback;
                    }
                }

                info("INSERT INTO entry :: %s :: %s", $$feed{url}, scalar(@{ $entries }));
            }
        }

        $dbx->dbh->commit;
    }

    info("COMPLETE full scan");

    mem_usage();

    sleep(10);
}

sub info {
    openlog("feeder", 'cons,pid', 'user');<commit_msg>Keep entries for a while.<commit_after>#!/opt/perl

use lib qw(/opt/infoservant.com/docroot/info_servant/lib);

use SiteCode::Modern;

use autodie;

use File::Slurp;
use File::Path;
use Sys::Syslog;

use SiteCode::DBX;
use XML::Feed;

use Mojo::Util;

use Mojo::UserAgent;

use SiteCode::Feed;
use SiteCode::Feeds;
use DateTime;

my $data_dir = "/opt/infoservant.com/data/feed_files";
my $html_dir = "/opt/infoservant.com/data/html_files";
Mojo::Util::spurt($$, "/tmp/feeder.pl.pid");

while (1) {
    my $dbx = SiteCode::DBX->new();

    my $feeds = $dbx->array(qq(
        select id, name as url from feed
    ));

    foreach my $feed (@{ $feeds }) {
        eval {
            $dbx->do("SELECT id FROM feed WHERE id = ? FOR UPDATE NOWAIT", undef, $$feed{id});
        };
        if ($@) {
            next;
        }

        my $url_dir = Mojo::Util::url_escape($$feed{url});

        my $the_dir = "$data_dir/$url_dir";

        my $obj = SiteCode::Feed->new(id => $$feed{id});

        if (!-d $the_dir) {
            mkdir($the_dir);
        }

        my $last_check = $obj->key("last_check") || 0;
        my $diff = time() - $last_check;
        if (900 > $diff) {
            $dbx->dbh->rollback();
            next;
        }
        $obj->key("last_check", time());
        $dbx->dbh->commit;

        my $ua = Mojo::UserAgent->new;
        $ua->max_redirects(10);

        info('ua->head :: %s', $$feed{url});

        my $tx = $ua->head($$feed{url});
        if (my $res = $tx->success) { 
            my $last_modified = $tx->res->headers->last_modified || "";
            my $prev_modified = $obj->key("last_modified") || "";

            if ($last_modified eq $prev_modified) {
                next;
            }

            $obj->key("last_modified", $last_modified);
        }
        else {
            my ($err, $code) = $tx->error;
            my $string =  $code ? "$code response: $err" : "Connection error: $err";

            info("head error(%s) :: %s", $$feed{url}, $string);

            system("/bin/touch", $the_dir);

            next;
        }

        info('ua->get :: %s', $$feed{url});

        $tx = $ua->get($$feed{url});
        if (my $res = $tx->success) { 
            Mojo::Util::spurt($res->body, "$the_dir/the.feed");

            info("spurt(%s) :: %s", "$the_dir/the.feed", $$feed{url});
        } else {
            my ($err, $code) = $tx->error;
            my $string =  $code ? "$code response: $err" : "Connection error: $err";

            info("get error(%s) :: %s", $$feed{url}, $string);
        }

        system("/bin/touch", $the_dir);

        my $parse;
        eval {
            $parse = XML::Feed->parse("$the_dir/the.feed");
            $obj->key("title", $parse->title());
            $obj->key("base", $parse->base());
            $obj->key("link", $parse->link());
        };
        if ($@) {
            info("parse error(%s) :: %s", $$feed{url}, $@);
        }
        my $entries = $parse ? [$parse->entries()] : [];

        my $feed_path = "$html_dir/$$feed{id}";
        if (!-d $feed_path) {
            mkdir($feed_path);
        }
        if (@{ $entries}) {
                foreach my $entry (@{ $entries }) {
                    eval {
                        my $exists = $dbx->col("SELECT id FROM entry WHERE feed_name = ? and entry_id = ?", undef, $$feed{url}, $entry->id());
                        unless ($exists) {
                            $dbx->do(
                                "INSERT INTO entry (feed_name, feed_title, issued, title, entry_id, link) VALUES (?, ?, ?, ?, ?, ?)", 
                                undef, 
                                $$feed{url},
                                $parse->title() || "",
                                $entry->issued() || DateTime->from_epoch(epoch => time() - 604800),
                                $entry->title() || "", 
                                $entry->id(), 
                                $entry->link(),
                            );
                            my $id = $dbx->last_insert_id(undef,undef,undef,undef,{sequence=>'entry_id_seq'});
                            my $html_file = "$feed_path/$id.html";
                            # Mojo::Util::spurt(utf8::encode($entry->content->body()), $html_file);
                            File::Slurp::write_file($html_file, {binmode => ':utf8', atomic => 1}, $entry->content->body() || "No content found.");
                            $dbx->dbh->commit;
                        }
                    };
                    if ($@) {
                        my $err = $@;
                        info("INSERT INTO entry error(%s) :: %s", $$feed{url}, $err);
                        $dbx->dbh->rollback;
                    }
                }

                info("INSERT INTO entry :: %s :: %s", $$feed{url}, scalar(@{ $entries }));
            # }
        }

        $dbx->dbh->commit;
    }

    info("COMPLETE full scan");

    mem_usage();

    sleep(10);
}

sub info {
    openlog("feeder", 'cons,pid', 'user');
{'fraction_non_alphanumeric': 0.19113877700476017, 'fraction_numerical': 0.003387037715122666, 'mean_word_length': 1.5, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 126, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '51695', 'n_tokens_mistral': 3527, 'n_tokens_neox': 3345, 'n_words': 660}