readme: documentar warcificator

readme.md

@@ -8,7 +8,7 @@ scrapeo "masivo" de precios y datos en supermercados argentinos
 
   (no hace falta correrlos porque ya hay listas armadas en [data/](./data/))
 
-- se usa wget (potencialmente reemplazado por algo custom en el futuro) que genera un archivo [WARC](https://iipc.github.io/warc-specifications/specifications/warc-format/warc-1.0/) con todas las paginas de productos
+- [warcificator](./warcificator/) descarga las paginas de productos y genera un archivo [WARC](https://iipc.github.io/warc-specifications/specifications/warc-format/warc-1.0/) con ellas
 - el [scraper](./scraper/) procesa estos WARCs, extrayendo varios datos y guardandolos en una base de datos SQLite (definida en [db-datos](./db-datos/schema.ts))
 - el [sitio](./sitio/) renderiza páginas a partir de la base de datos y hace gráficos lindos
 

scraper/Containerfile

@@ -8,12 +8,27 @@ RUN bun install --frozen-lockfile \
     && bun build scraper/cli.ts --target=bun --outfile=/tmp/cli.build.js \
     && rm -rf node_modules/
 
+# https://dev.to/deciduously/use-multi-stage-docker-builds-for-statically-linked-rust-binaries-3jgd
+FROM docker.io/rust:1.74 AS warcificator-builder
+WORKDIR /usr/src/
+RUN rustup target add x86_64-unknown-linux-musl
+RUN apt-get update && apt-get install -y musl-tools musl-dev
+
+RUN USER=root cargo new warcificator
+WORKDIR /usr/src/warcificator
+COPY ./warcificator/Cargo.toml ./warcificator/Cargo.lock ./
+RUN cargo build --release
+
+COPY ./warcificator/src ./src
+RUN cargo install --target x86_64-unknown-linux-musl --path .
+
 FROM base
 RUN apk add --no-cache wget zstd tini
 RUN printf "#!/bin/sh\nexec bun /bin/scraper auto\n" > /etc/periodic/daily/scraper \
     && chmod +x /etc/periodic/daily/scraper
 
 COPY --from=builder /tmp/cli.build.js /bin/scraper
+COPY --from=warcificator-builder /usr/local/cargo/bin/warcificator /bin/
 COPY --from=builder /usr/src/app/db-datos/drizzle /bin/drizzle
 COPY --from=builder /usr/src/app/data /listas
 WORKDIR /app

scraper/auto.ts

@@ -22,9 +22,6 @@ const supermercados: Supermercado[] = [
   Supermercado.Dia,
 ];
 
-// hacemos una cola para la compresión para no sobrecargar la CPU
-const compressionQueue = new PQueue({ concurrency: 1 });
-
 // hacemos una cola para el scrapeo para no tener varios writers a la BD y no sobrecargar la CPU
 const scrapQueue = new PQueue({ concurrency: 1 });
 
@@ -77,7 +74,7 @@ class Auto {
   }
 
   async downloadList(supermercado: Supermercado) {
-    const ctxPath = await mkdtemp(join(tmpdir(), "preciazo-scraper-wget-"));
+    const ctxPath = await mkdtemp(join(tmpdir(), "preciazo-scraper-download-"));
 
     let listPath: string;
     {
@@ -117,15 +114,7 @@ class Auto {
     )}.warc.zst`;
     const zstdWarcPath = join(ctxPath, zstdWarcName);
     const subproc = Bun.spawn({
-      cmd: [
-        "wget",
-        "--no-verbose",
-        "--tries=3",
-        "--delete-after",
-        "--input-file",
-        listPath,
-        `--warc-file=temp`,
-      ],
+      cmd: ["warcificator", listPath, zstdWarcPath],
       stderr: "ignore",
       stdout: "ignore",
       cwd: ctxPath,
@@ -133,18 +122,9 @@ class Auto {
     const t0 = performance.now();
     await subproc.exited;
     this.inform(
-      `[wget] ${zstdWarcName} tardó ${formatMs(performance.now() - t0)}`
+      `[downloader] ${zstdWarcName} tardó ${formatMs(performance.now() - t0)}`
     );
 
-    const gzippedWarcPath = join(ctxPath, "temp.warc.gz");
-    if (!(await fileExists(gzippedWarcPath))) {
-      const err = this.report(`no encontré el ${gzippedWarcPath}`);
-      throw err;
-    }
-
-    await compressionQueue.add(() =>
-      this.recompress(gzippedWarcPath, zstdWarcPath)
-    );
     if (!(await fileExists(zstdWarcPath))) {
       const err = this.report(`no encontré el ${zstdWarcPath}`);
       throw err;
@@ -190,49 +170,6 @@ class Auto {
     }
   }
 
-  /**
-   * toma un archivo gzippeado y lo recomprime con zstd.
-   * borra el archivo original.
-   */
-  recompress(inputPath: string, outputPath: string) {
-    // XXX: por alguna razón no funciona en Bun 1.0.20
-    // const decompressor = Bun.spawn({
-    //   cmd: ["gzip", "-dc", inputPath],
-    //   stderr: "inherit",
-    // });
-    // const compressor = Bun.spawn({
-    //   cmd: ["zstd", "-T0", "-15", "--long", "-o", outputPath],
-    //   stdin: decompressor.stdout,
-    //   // stderr: "inherit",
-    // });
-    // const errorCode = await compressor.exited;
-    // if (errorCode !== 0) {
-    //   const err = report(`zstd threw error code ${errorCode}`);
-    //   throw err;
-    // }
-
-    return new Promise((resolve, reject) => {
-      const decompressor = spawn("gzip", ["-dc", inputPath], {
-        stdio: [null, "pipe", null],
-      });
-      const compressor = spawn(
-        "zstd",
-        ["-T0", "-15", "--long", "-o", outputPath],
-        {
-          stdio: ["pipe", null, null],
-        }
-      );
-      decompressor.stdout.pipe(compressor.stdin);
-      compressor.on("close", (code) => {
-        if (code !== 0) {
-          const err = this.report(`zstd threw error code ${code}`);
-          reject(err);
-        }
-        resolve(void 0);
-      });
-    });
-  }
-
   async uploadToBucket({
     fileName,
     file,
@@ -278,7 +215,6 @@ class Auto {
     await fetch(url);
   }
 }
-// await recompress("sqlite.db.gz", "sqlite.db.zst");
 
 // no se llama exists porque bun tiene un bug en el que usa fs.exists por mas que exista una funcion llamada exists
 async function fileExists(path: string) {

warcificator/src/main.rs

@@ -1,16 +1,24 @@
 use async_channel::{Receiver, Sender};
-use std::{env::args, fs, io::stdout, net::SocketAddr};
+use std::{
+    env::args,
+    fs,
+    net::SocketAddr,
+    process::{Command, Stdio},
+};
+use tokio::io::{stderr, AsyncWriteExt};
 use warc::{RecordBuilder, WarcHeader, WarcWriter};
 
 struct FullExchange {
-    socket_addr: SocketAddr,
+    socket_addr: Option<SocketAddr>,
     request: http::Request<&'static str>,
     response: http::Response<Vec<u8>>,
 }
 
 #[tokio::main]
 async fn main() {
-    let links_list_path = args().skip(1).next().unwrap();
+    let mut args = args().skip(1);
+    let links_list_path = args.next().unwrap();
+    let output_zstd_path = args.next().unwrap();
     let links_str = fs::read_to_string(links_list_path).unwrap();
     let links = links_str
         .split("\n")
@@ -18,6 +26,7 @@ async fn main() {
         .filter(|s| s.len() > 0)
         .map(|s| s.to_owned())
         .collect::<Vec<_>>();
+
     let handle = {
         let (sender, receiver) = async_channel::bounded::<String>(1);
         let (res_sender, res_receiver) = async_channel::unbounded::<FullExchange>();
@@ -29,7 +38,7 @@ async fn main() {
             handles.push(tokio::spawn(worker(rx, tx)));
         }
 
-        let warc_writer_handle = tokio::spawn(warc_writer(res_receiver));
+        let warc_writer_handle = tokio::spawn(warc_writer(res_receiver, output_zstd_path));
 
         for link in links {
             sender.send_blocking(link).unwrap();
@@ -48,6 +57,22 @@ async fn main() {
 async fn worker(rx: Receiver<String>, tx: Sender<FullExchange>) {
     let client = reqwest::ClientBuilder::default().build().unwrap();
     while let Ok(url) = rx.recv().await {
+        let res = fetch(&client, url.clone()).await;
+        match res {
+            Ok(ex) => {
+                tx.send(ex).await.unwrap();
+            }
+            Err(err) => {
+                stderr()
+                    .write_all(format!("Failed to fetch {}: {:#?}", url.as_str(), err).as_bytes())
+                    .await
+                    .unwrap();
+            }
+        }
+    }
+}
+
+async fn fetch(client: &reqwest::Client, url: String) -> Result<FullExchange, reqwest::Error> {
     let request = client.get(url).build().unwrap();
     let mut http_request_builder = http::Request::builder()
         .method(request.method())
@@ -55,9 +80,9 @@ async fn worker(rx: Receiver<String>, tx: Sender<FullExchange>) {
     for (key, val) in request.headers() {
         http_request_builder = http_request_builder.header(key, val);
     }
-        let response = client.execute(request).await.unwrap();
+    let response = client.execute(request).await?;
 
-        let ip_address = response.remote_addr().unwrap();
+    let ip_address = response.remote_addr();
 
     let http_request = {
         http_request_builder
@@ -73,65 +98,75 @@ async fn worker(rx: Receiver<String>, tx: Sender<FullExchange>) {
         for (key, val) in response.headers() {
             http_response_builder = http_response_builder.header(key, val);
         }
-            let body = response.bytes().await.unwrap();
+        let body = response.bytes().await?;
         http_response_builder.body(body.to_vec()).unwrap()
     };
 
-        tx.send(FullExchange {
+    Ok(FullExchange {
         socket_addr: ip_address,
         request: http_request,
         response: http_response,
     })
-        .await
-        .unwrap();
-    }
 }
 
-async fn warc_writer(rx: Receiver<FullExchange>) {
-    let mut writer = WarcWriter::new(stdout());
-    let warc_fields = format!("software: preciazo-warcificator/0.0.0\nformat: WARC file version 1.0\nconformsTo: http://www.archive.org/documents/WarcFileFormat-1.0.html");
+async fn warc_writer(rx: Receiver<FullExchange>, output_zstd_path: String) {
+    let zstd_proc = Command::new("zstd")
+        .args(&["-T0", "-15", "--long", "-o", &output_zstd_path])
+        .stdin(Stdio::piped())
+        .stderr(Stdio::null())
+        .stdout(Stdio::null())
+        .spawn()
+        .unwrap();
+
+    let mut writer = WarcWriter::new(zstd_proc.stdin.unwrap());
     writer
         .write(
             &RecordBuilder::default()
                 .version("1.0".to_owned())
                 .warc_type(warc::RecordType::WarcInfo)
                 .header(WarcHeader::ContentType, "application/warc-fields")
-                .body(warc_fields.into())
+                .body(format!("software: preciazo-warcificator/0.0.0\nformat: WARC file version 1.0\nconformsTo: http://www.archive.org/documents/WarcFileFormat-1.0.html").into())
                 .build()
                 .unwrap(),
         )
         .unwrap();
     while let Ok(res) = rx.recv().await {
         let uri = res.request.uri().to_string();
-        writer
-            .write(
-                &RecordBuilder::default()
+        let req_record = {
+            let mut builder = RecordBuilder::default()
                 .version("1.0".to_owned())
                 .warc_type(warc::RecordType::Request)
                 .header(WarcHeader::TargetURI, uri.clone())
-                    .header(WarcHeader::IPAddress, res.socket_addr.ip().to_string())
                 .header(WarcHeader::ContentType, "application/http;msgtype=request")
                 .header(
                     WarcHeader::Unknown("X-Warcificator-Lying".to_string()),
                     "the request contains other headers not included here",
-                    )
+                );
+            if let Some(addr) = res.socket_addr {
+                builder = builder.header(WarcHeader::IPAddress, addr.ip().to_string());
+            }
+            builder
                 .body(format_http11_request(res.request).into_bytes())
                 .build()
-                    .unwrap(),
-            )
-            .unwrap();
+                .unwrap()
+        };
+        writer.write(&req_record).unwrap();
         writer
-            .write(
-                &RecordBuilder::default()
+            .write(&{
+                let mut builder = RecordBuilder::default()
                     .version("1.0".to_owned())
                     .warc_type(warc::RecordType::Response)
                     .header(WarcHeader::TargetURI, uri)
-                    .header(WarcHeader::IPAddress, res.socket_addr.ip().to_string())
-                    .header(WarcHeader::ContentType, "application/http;msgtype=response")
+                    .header(WarcHeader::ConcurrentTo, req_record.warc_id())
+                    .header(WarcHeader::ContentType, "application/http;msgtype=response");
+                if let Some(addr) = res.socket_addr {
+                    builder = builder.header(WarcHeader::IPAddress, addr.ip().to_string());
+                }
+                builder
                     .body(format_http11_response(res.response))
                     .build()
-                    .unwrap(),
-            )
+                    .unwrap()
+            })
             .unwrap();
     }
 }