tkpull.cc

#include <fmt/format.h>
#include <fmt/printf.h>
#include <fmt/ranges.h>
#include <mutex>
#include <iostream>
#include "sqlwriter.hh"
#include <atomic>

#include "httplib.h"
#include <set>
#include "support.hh"

using namespace std;

void ifExistsThenRename(const std::string& fname)
{
  struct stat sb;
  if(stat(fname.c_str(), &sb) < 0)
    return;
  
  string newname = fmt::sprintf("%s.%d", fname, sb.st_mtime);
  if(rename(fname.c_str(), newname.c_str()) == 0) {
    fmt::print("Already had a file for {}, renamed to {}\n",
		fname, newname);
  }
}

void storeDocument(const std::string& id, const std::string& content, const string& prefix)
{
  string fname=makePathForId(id, prefix, "", true);
  ifExistsThenRename(fname);
  FILE* t = fopen((fname+".tmp").c_str(), "w");
  if(!t)
    throw runtime_error("Unable to open file "+fname+": "+string(strerror(errno)));
  
  shared_ptr<FILE> fp(t, fclose);
  if(fwrite(content.c_str(), 1, content.size(), fp.get()) != content.size()) {
    unlink(fname.c_str());
    throw runtime_error("Partial write to file "+fname);
  }
  fp.reset();
  
  if(rename((fname+".tmp").c_str(), fname.c_str()) < 0) {
    int e = errno;
    unlink((fname+".tmp").c_str());
    throw runtime_error("Unable to rename saved file "+fname+".tmp - "+strerror(e));
  }
}

struct ThrottleDB
{
  ThrottleDB() : d_sqlw("meta.sqlite3")
  {
    d_sqlw.query("create table if not exists throttle (thing TEXT, reason TEXT, timestamp INT) STRICT");
    time_t now = time(0);
    d_sqlw.query("delete from throttle where timestamp < ?", {now - s_retention});
  }

  void report(const std::string& thing, const std::string& reason="")
  {
    d_sqlw.addValue({{"thing", thing}, {"reason", reason}, {"timestamp", time(nullptr)}}, "throttle");
  }

  bool shouldThrottle(const std::string& thing, int limitSeconds, int limit)
  {
    time_t lim = time(nullptr) - limitSeconds;
    auto res = d_sqlw.queryT("select count(1) as c from throttle where thing=? and timestamp > ?",
		 {thing, lim});
    if(res.empty())
      return false;
    //    cout<<"Asked for '"<<thing<<"', limitSeconds "<<limitSeconds<<" limit "<<limit<<" lim "<<lim<<", count: "<<std::get<int64_t>(res[0]["c"])<<endl;
    
    return std::get<int64_t>(res[0]["c"]) >= limit;
  }
  
  SQLiteWriter d_sqlw;
  constexpr static unsigned int s_retention = 7*86400;
};

int main(int argc, char** argv)
{
  SQLiteWriter sqlw("tk.sqlite3", SQLWFlag::ReadOnly);

  int sizlim = 250000000;
  string limit="2007-01-01";
  auto wantDocs = sqlw.queryT("select id,enclosure,contentLength from Document where datum > ? and contentLength < ?", {limit, sizlim});

  auto alleVerslagen = sqlw.queryT("select Verslag.id as id, vergadering.id as vergaderingid,enclosure,contentLength,datum from Verslag,Vergadering where Verslag.vergaderingId=Vergadering.id and datum > ? order by datum desc, verslag.updated desc", {limit});

  auto wantPhotos = sqlw.queryT("select Persoon.id as id, enclosure, contentLength from Persoon where contentLength > 0");
  
  set<string> seenvergadering;
  decltype(alleVerslagen) wantVerslagen, todelete;
  for(auto& v: alleVerslagen) {
    string vid = get<string>(v["vergaderingid"]);
    if(seenvergadering.count(vid)) {
      todelete.push_back(v);
      continue;
    }
    wantVerslagen.push_back(v);
    seenvergadering.insert(vid);
  }
  fmt::print("We desire {} documents and {} photos and {} verslagen, and found {} older verslagen\n", wantDocs.size(), wantPhotos.size(), wantVerslagen.size(), todelete.size());

  int unlinked=0;
  for(auto& td : todelete) {
    string verslagid=get<string>(td["id"]);
    string fname=makePathForId(verslagid);
    int rc = unlink(fname.c_str());
    if(!rc)
      unlinked++;
    else {
      if(errno != ENOENT)
	fmt::print("Error removing file {}: {}\n", fname, strerror(errno));
    }
  }
  fmt::print("{} niet-nieuwste versies van verslagen gewist\n", unlinked);

  struct RetStore
  {
    string id;
    string enclosure;
    int64_t contentLength;
    bool operator<(const RetStore& rhs) const
    {
      return id < rhs.id;
    }
  };

  ThrottleDB tdb;
  
  map<string, decltype(&wantDocs)> work = {
    {"docs", &wantDocs},
    {"verslagen", &wantVerslagen},
    {"photos", &wantPhotos}};
  for(auto& [name, store] : work) {
    int present=0;
    int toolarge=0, retrieved=0;
    int error=0;
    cout<<"Starting from store '"<<name<<"', got "<<store->size()<<" docs to go"<<endl;
    set<RetStore> toRetrieve;
    string prefix = (store == &wantPhotos) ? "photos" : "docs";
    for(auto& d : *store) {
      if(isPresentRightSize(get<string>(d["id"]), get<int64_t>(d["contentLength"]), prefix)  )
	present++;
      else {
	auto contentLength = get_if<int64_t>(&d["contentLength"]);

	toRetrieve.insert({get<string>(d["id"]), get<string>(d["enclosure"]),
	    contentLength ? *contentLength : 0});
	
	size_t siz;
	if(isPresentNonEmpty(get<string>(d["id"]), prefix, "", &siz)) {
	  fmt::print("Re-retrieving {} {}, has wrong size on disk {}, should be {}\n",
		     name, get<string>(d["id"]), siz, get<int64_t>(d["contentLength"]));
	}
      }
    }
    fmt::print("We have {} files to retrieve, {} are already present\n", toRetrieve.size(), present);

    for(const auto& need : toRetrieve) {
      if(!need.contentLength || need.contentLength > sizlim) {
	toolarge++;
	fmt::print("Skipping {}, too large for server ({}) or unknown\n",
		   need.id, need.contentLength);
	continue;
      }

      if(tdb.shouldThrottle(need.enclosure, 86400, 2)) {
	fmt::print("Not retrieving {}, throttled\n", need.enclosure);
	continue;
      }
      httplib::Client cli("https://gegevensmagazijn.tweedekamer.nl");
      cli.set_connection_timeout(10, 0); 
      cli.set_read_timeout(10, 0); 
      cli.set_write_timeout(10, 0); 
      
      fmt::print("Retrieving from {} (expect {} bytes).. ", need.enclosure, need.contentLength);
      cout.flush();
      auto res = cli.Get(need.enclosure);
      
      if(!res) {
	auto err = res.error();
	fmt::print("Oops retrieving from {} -> {}\n", need.enclosure, httplib::to_string(err));
	error++;
	continue;
      }
      if(res->status != 200) {
	fmt::print("Wrong status code {} for url {}, not storing\n",
		   res->status, need.enclosure);
	error++;
	continue;
      }

      fmt::print("Got {} bytes\n", res->body.size());
      if(res->body.size() == (unsigned int)need.contentLength) {
	storeDocument(need.id, res->body, prefix);
	retrieved++;
      }
      else {
	tdb.report(need.enclosure, "wrong size");
	fmt::print("Unexpected size received, not storing document\n");
      }
    }
    fmt::print("Retrieved {} documents, {} were too large, {} errors\n", retrieved, toolarge, error);
  }
}