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

more updates qa prototyping

Browse Source
This commit is contained in:
Barbara Miller 2023-06-28 11:43:53 -07:00
parent ef75164f8b
commit 702afbd098
1 changed files with 21 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
warcprox/dedup.py
View File

@ -62,21 +62,21 @@ class DedupableMixin(object):
else: else:
return recorded_url.response_recorder.payload_size() > self.min_binary_size return recorded_url.response_recorder.payload_size() > self.min_binary_size
class LimitRevisitsPGMixin(object): class LimitRevisitsPGMixin():
""" """
Limit revisits recorded to one per revisit_key Limit revisits recorded to one per revisit_key
""" """
def __init__(self, options=warcprox.Options()): def __init__(self):
self.datasource = "postgresql://archiveit@db.qa-archive-it.org/archiveit" # "postgresql://user@db_host/db_name" self.datasource = "postgresql://archiveit@db.qa-archive-it.org/archiveit" # "postgresql://user@db_host/db_name"
self.datatable = "crawl_revisits" # postgres table in db_name self.datatable = "crawl_revisits" # 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):
# tracks revisits, returns True when we've seen revisit before, else False
if not hash_plus_url: if not hash_plus_url:
hash_plus_url = b"".join( digest = warcprox.digest_str(recorded_url.payload_digest,
(warcprox.digest_str(recorded_url.payload_digest, self.options.base32)
self.options.base32), digest = digest[5:] if digest.startswith(b'sha1:') else digest
recorded_url.url) hash_plus_url = b"".join([digest, recorded_url.url]).decode()
).decode()
if not revisit_key: if not revisit_key:
# use ait-job-id if available # use ait-job-id if available
if ( if (
@ -86,9 +86,9 @@ class LimitRevisitsPGMixin(object):
): ):
revisit_key = recorded_url.warcprox_meta["metadata"]["ait-job-id"] revisit_key = recorded_url.warcprox_meta["metadata"]["ait-job-id"]
else: else:
revisit_key = 'all' revisit_key = '__unspecified__'
query = "SELECT exists(SELECT 1 FROM crawl_revisits WHERE hash_plus_url = %s LIMIT 1);" query = "SELECT exists(SELECT 1 FROM crawl_revisits WHERE hash_plus_url = %s and crawl_id = %s LIMIT 1);"
try: try:
conn = psycopg2.connect(self.datasource) conn = psycopg2.connect(self.datasource)
@ -97,23 +97,22 @@ class LimitRevisitsPGMixin(object):
return False return False
cur = conn.cursor() cur = conn.cursor()
try: try:
cur.execute(query, (hash_plus_url,)) cur.execute(query, (hash_plus_url, revisit_key))
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", digest, revisit_key, e)
return False return False
result = cur.fetchone() result = cur.fetchone()
if result[0]:
logging.info("result[0]: %s", result[0])
if result[0] and result[0] == True: if result and result == (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, digest)
return True return True
else:
query = "INSERT INTO crawl_revisits (crawl_id, hash_plus_url) VALUES (%s, %s);" query = "INSERT INTO crawl_revisits (crawl_id, hash_plus_url) VALUES (%s, %s);"
try: try:
cur.execute(query, (revisit_key, hash_plus_url)) cur.execute(query, (revisit_key, hash_plus_url))
except Exception as e: conn.commit()
self.logger.warning("exception inserting %s in %s: %s", hash_plus_url, revisit_key, e) except Exception as e:
self.logger.warning("exception inserting %s in %s for %s: %s", digest, revisit_key, recorded_url.url, e)
return False return False
@ -516,7 +515,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, options) LimitRevisitsPGMixin.__init__(self)
self.trough_dedup_db = trough_dedup_db self.trough_dedup_db = trough_dedup_db
def _startup(self): def _startup(self):