backup/warcprox
1
0
Fork 0
mirror of https://github.com/internetarchive/warcprox.git synced 2025-01-18 13:22:09 +01:00
Code Issues Projects Releases Activity

fixes for qa prototyping

Browse Source
This commit is contained in:
Barbara Miller 2023-06-27 11:47:55 -07:00
parent d9145eefb5
commit ef75164f8b
1 changed files with 15 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
warcprox/dedup.py
View File

@ -29,6 +29,7 @@ import warcprox
import sqlite3 import sqlite3
import doublethink import doublethink
import datetime import datetime
import psycopg2
import urllib3 import urllib3
from urllib3.exceptions import HTTPError from urllib3.exceptions import HTTPError
import collections import collections
@ -65,20 +66,17 @@ class LimitRevisitsPGMixin(object):
""" """
Limit revisits recorded to one per revisit_key Limit revisits recorded to one per revisit_key
""" """
def __init__(self, datasource, options=warcprox.Options()): def __init__(self, options=warcprox.Options()):
import psycopg2 self.datasource = "postgresql://archiveit@db.qa-archive-it.org/archiveit" # "postgresql://user@db_host/db_name"
from psycopg2 import extras # needed self.datatable = "crawl_revisits" # postgres table in db_name
self.datasource = datasource # "postgresql://user@db_host/db_name"
self.datatable = 'crawled_urls' # postgres table in db_name
def limit_revisits(self, recorded_url, hash_plus_url=None, revisit_key=None): def limit_revisits(self, recorded_url, hash_plus_url=None, revisit_key=None):
if not hash_plus_url: if not hash_plus_url:
hash_plus_url = b''.join( hash_plus_url = b"".join(
(warcprox.digest_str(recorded_url.payload_digest, (warcprox.digest_str(recorded_url.payload_digest,
self.options.base32), self.options.base32),
recorded_url.url) recorded_url.url)
) ).decode()
if not revisit_key: if not revisit_key:
# use ait-job-id if available # use ait-job-id if available
if ( if (
@ -90,7 +88,7 @@ class LimitRevisitsPGMixin(object):
else: else:
revisit_key = 'all' revisit_key = 'all'
query = f"SELECT exists(SELECT 1 FROM {self.datatable} WHERE hash_plus_url = {hash_plus_url} LIMIT 1;" query = "SELECT exists(SELECT 1 FROM crawl_revisits WHERE hash_plus_url = %s LIMIT 1);"
try: try:
conn = psycopg2.connect(self.datasource) conn = psycopg2.connect(self.datasource)
@ -99,19 +97,21 @@ class LimitRevisitsPGMixin(object):
return False return False
cur = conn.cursor() cur = conn.cursor()
try: try:
cur.execute(query) cur.execute(query, (hash_plus_url,))
except Exception as e: except Exception as e:
self.logger.warning("exception querying for %s in %s: %s", hash_plus_url, revisit_key, e) self.logger.warning("exception querying for %s in %s: %s", hash_plus_url, revisit_key, e)
return False return False
result = cur.fetchone() result = cur.fetchone()
if result[0]: if result[0]:
logging.info("result[0]: %s", result[0])
if result[0] and result[0] == True:
logging.info("skipping revisit for url %s and hash %s", recorded_url.url, hash) logging.info("skipping revisit for url %s and hash %s", recorded_url.url, hash)
return True return True
else: else:
query = "INSERT INTO {self.datable} VALUES(revisit_key, hash_plus_url);" query = "INSERT INTO crawl_revisits (crawl_id, hash_plus_url) VALUES (%s, %s);"
try: try:
cur.execute(query) cur.execute(query, (revisit_key, hash_plus_url))
except Exception as e: except Exception as e:
self.logger.warning("exception inserting %s in %s: %s", hash_plus_url, revisit_key, e) self.logger.warning("exception inserting %s in %s: %s", hash_plus_url, revisit_key, e)
@ -122,8 +122,6 @@ class LimitRecords(object):
Limit records to one per revisit_key, e.g., crawl id Limit records to one per revisit_key, e.g., crawl id
""" """
def __init__(self, datasource, options=warcprox.Options()): def __init__(self, datasource, options=warcprox.Options()):
import psycopg2
from psycopg2 import extras # needed
self.datasource = datasource # "postgresql://user@db_host/db_name" self.datasource = datasource # "postgresql://user@db_host/db_name"
# self.datatable = revisit_key # set in limit_revisits method # self.datatable = revisit_key # set in limit_revisits method
# verify partition table exists # verify partition table exists
@ -518,8 +516,7 @@ class BatchTroughLoader(warcprox.BaseBatchPostfetchProcessor, LimitRevisitsPGMix
def __init__(self, trough_dedup_db, options=warcprox.Options()): def __init__(self, trough_dedup_db, options=warcprox.Options()):
warcprox.BaseBatchPostfetchProcessor.__init__(self, options) warcprox.BaseBatchPostfetchProcessor.__init__(self, options)
LimitRevisitsPGMixin.__init__(self, datasource="", options) LimitRevisitsPGMixin.__init__(self, options)
# TODO: get datasource
self.trough_dedup_db = trough_dedup_db self.trough_dedup_db = trough_dedup_db
def _startup(self): def _startup(self):
@ -607,8 +604,7 @@ class BatchTroughLoader(warcprox.BaseBatchPostfetchProcessor, LimitRevisitsPGMix
for recorded_url in key_index[entry['digest_key']]: for recorded_url in key_index[entry['digest_key']]:
recorded_url.dedup_info = entry recorded_url.dedup_info = entry
if recorded_url.dedup_info: if recorded_url.dedup_info:
recorded_url.do_not_archive = \ recorded_url.do_not_archive = self.limit_revisits(recorded_url)
self.limit_revisits(recorded_url)
except Exception as e: except Exception as e:
# batch_lookup raised exception or something # batch_lookup raised exception or something
logging.warning( logging.warning(