Technologies cleanup automation (#52)

* cleanup drafts

* aggregated query

* crawl pipeline patch

* staging tech assert

* simpler category rewrite

* typos

* comments

* alias

* Update definitions/output/crawl/pages.js

Co-authored-by: Barry Pollard <barrypollard@google.com>

* Update definitions/output/crawl/pages.js

Co-authored-by: Barry Pollard <barrypollard@google.com>

* Update definitions/output/crawl/pages.js

Co-authored-by: Barry Pollard <barrypollard@google.com>

* lint

* Update definitions/declarations/httparchive.js

Co-authored-by: Barry Pollard <barrypollard@google.com>

---------

Co-authored-by: Barry Pollard <barrypollard@google.com>

Author: max-ostapenko

definitions/declarations/httparchive.js

@@ -1,17 +1,42 @@
-const stagingTables = ['pages', 'requests', 'parsed_css']
-for (const table of stagingTables) {
+// Staging tables source: https://github.com/HTTPArchive/crawl/blob/main/crawl.py
+['pages', 'requests', 'parsed_css'].forEach(table =>
   declare({
     schema: 'crawl_staging',
     name: table
   })
-}
+)
 
-declare({
-  schema: 'wappalyzer',
-  name: 'technologies'
-})
+// See https://github.com/HTTPArchive/dataform/issues/43
+assert('corrupted_technology_values')
+  .tags(['crawl_complete'])
+  .query(ctx => `
+SELECT
+  date,
+  client,
+  tech,
+  COUNT(DISTINCT page) AS cnt_pages,
+  ARRAY_AGG(DISTINCT page LIMIT 3) AS sample_pages
+FROM ${ctx.ref('crawl_staging', 'pages')} AS pages
+LEFT JOIN pages.technologies AS tech
+LEFT JOIN tech.categories AS category
+WHERE
+  date = '${constants.currentMonth}' AND
+  (
+    tech.technology NOT IN (SELECT DISTINCT name FROM wappalyzer.technologies)
+    OR category NOT IN (SELECT DISTINCT name FROM wappalyzer.categories)
+    OR ARRAY_LENGTH(tech.categories) = 0
+  )
+GROUP BY
+  date,
+  client,
+  tech
+ORDER BY cnt_pages DESC
+`);
 
-declare({
-  schema: 'wappalyzer',
-  name: 'categories'
-})
+// Wappalyzer tables source: https://github.com/HTTPArchive/wappalyzer/blob/main/.github/workflows/upload.yml
+['technologies', 'categories'].forEach(table =>
+  declare({
+    schema: 'wappalyzer',
+    name: table
+  })
+)

definitions/output/crawl/pages.js

@@ -52,23 +52,97 @@ publish('pages', {
 DELETE FROM ${ctx.self()}
 WHERE date = '${constants.currentMonth}' AND
   client = 'desktop';
-`).query(ctx => `
+
+INSERT INTO ${ctx.self()}
 SELECT
   *
 FROM ${ctx.ref('crawl_staging', 'pages')}
 WHERE date = '${constants.currentMonth}' AND
   client = 'desktop'
-  ${constants.devRankFilter}
-`).postOps(ctx => `
+  ${constants.devRankFilter};
+
 DELETE FROM ${ctx.self()}
 WHERE date = '${constants.currentMonth}' AND
   client = 'mobile';
-
-INSERT INTO ${ctx.self()}
+`).query(ctx => `
 SELECT
   *
 FROM ${ctx.ref('crawl_staging', 'pages')}
 WHERE date = '${constants.currentMonth}' AND
   client = 'mobile'
   ${constants.devRankFilter}
+`).postOps(ctx => `
+CREATE TEMP TABLE technologies_cleaned AS (
+  WITH wappalyzer AS (
+    SELECT DISTINCT
+      name AS technology,
+      categories
+    FROM ${ctx.ref('wappalyzer', 'technologies')}
+  ),
+
+  pages AS (
+    SELECT
+      client,
+      page,
+      tech.technology,
+      tech.categories,
+      tech.info
+    FROM ${ctx.self()} AS pages
+    LEFT JOIN pages.technologies AS tech
+    WHERE date = '${constants.currentMonth}' ${constants.devRankFilter}
+  ),
+
+  -- Identify impacted pages
+  impacted_pages AS (
+    SELECT DISTINCT
+      client,
+      page
+    FROM pages
+    LEFT JOIN pages.categories AS category
+    WHERE
+      -- Technology is corrupted
+      technology NOT IN (SELECT DISTINCT technology FROM wappalyzer) OR
+      -- Technology's category is corrupted
+      CONCAT(technology, category) NOT IN (
+        SELECT DISTINCT
+          CONCAT(technology, category)
+        FROM wappalyzer
+        LEFT JOIN wappalyzer.categories AS category
+      )
+  ),
+
+  -- Keep valid technologies and use correct categories
+  reconstructed_technologies AS (
+    SELECT
+      client,
+      page,
+      ARRAY_AGG(STRUCT(
+        pages.technology,
+        wappalyzer.categories,
+        pages.info
+      )) AS technologies
+    FROM pages
+    INNER JOIN impacted_pages
+    USING (client, page)
+    INNER JOIN wappalyzer
+    ON pages.technology = wappalyzer.technology
+    GROUP BY
+      client,
+      page
+  )
+
+  SELECT
+    client,
+    page,
+    technologies
+  FROM reconstructed_technologies
+);
+
+-- Update the crawl.pages table with the cleaned and restored technologies
+UPDATE ${ctx.self()} AS pages
+SET technologies = technologies_cleaned.technologies
+FROM technologies_cleaned
+WHERE pages.date = '${constants.currentMonth}' AND
+  pages.client = technologies_cleaned.client AND
+  pages.page = technologies_cleaned.page;
 `)