From 2f3605540d5d8cf2dc77e33649445fcd8a34b9cd Mon Sep 17 00:00:00 2001 From: Adam Miller Date: Thu, 6 Jun 2024 14:21:43 -0700 Subject: [PATCH] feat: Add logging to crawl.log for metadata records created by ExtractorYoutubeDL --- .../modules/extractor/ExtractorYoutubeDL.java | 69 ++++++++++++++++++- .../modules/warc/BaseWARCRecordBuilder.java | 4 ++ .../modules/warc/WARCRecordBuilder.java | 2 + .../writer/WARCWriterChainProcessor.java | 3 + 4 files changed, 76 insertions(+), 2 deletions(-) diff --git a/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java b/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java index 8bcdda8b..af9ecd0b 100644 --- a/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java +++ b/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java @@ -41,10 +41,12 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Level; import java.util.logging.Logger; import org.apache.commons.httpclient.URIException; +import org.archive.crawler.framework.CrawlController; import org.archive.crawler.frontier.AMQPUrlReceiver; import org.archive.crawler.reporting.CrawlerLoggerModule; import org.archive.format.warc.WARCConstants.WARCRecordType; @@ -53,10 +55,9 @@ import org.archive.modules.CoreAttributeConstants; import org.archive.modules.CrawlURI; import org.archive.modules.warc.BaseWARCRecordBuilder; import org.archive.modules.warc.WARCRecordBuilder; -import org.archive.net.UURI; -import org.archive.net.UURIFactory; import org.archive.util.ArchiveUtils; import org.archive.util.MimetypeUtils; +import org.archive.util.Recorder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.Lifecycle; @@ -110,6 +111,7 @@ public class ExtractorYoutubeDL extends Extractor protected static final String YDL_CONTAINING_PAGE_DIGEST = "ydl-containing-page-digest"; protected static final String YDL_CONTAINING_PAGE_TIMESTAMP = "ydl-containing-page-timestamp"; protected static final String YDL_CONTAINING_PAGE_URI = "ydl-containing-page-uri"; + protected static final String YDL_JSON_FILE_DIGEST = "ydl-json-file-digest"; protected static final int MAX_VIDEOS_PER_PAGE = 1000; protected static final int NICE_MOD = 10; @@ -118,6 +120,7 @@ public class ExtractorYoutubeDL extends Extractor protected HashMap seedsYDLd = new HashMap(); protected transient Logger ydlLogger = null; + private static final AtomicLong nextRecorderId = new AtomicLong(); // unnamed toethread-local temporary file protected transient ThreadLocal tempfile = new ThreadLocal() { @@ -178,6 +181,12 @@ public class ExtractorYoutubeDL extends Extractor this.crawlerLoggerModule = crawlerLoggerModule; } + @Autowired + protected CrawlController controller; + public void setCrawlController(CrawlController controller) { + this.controller = controller; + } + @Override public void start() { if (!isRunning) { @@ -389,6 +398,12 @@ public class ExtractorYoutubeDL extends Extractor } } + /** Dummy output stream to swallow bytes without storing anything. */ + public class NullOutputStream extends OutputStream { + @Override + public void write(int b) throws IOException {} + } + /** * Streams through yt-dlp json output. Sticks video urls in * results.videoUrls, web page urls in @@ -598,6 +613,22 @@ public class ExtractorYoutubeDL extends Extractor recordInfo.setMimetype("application/vnd.youtube-dl_formats+json;charset=utf-8"); recordInfo.setEnforceLength(true); + //Use the recorder object to calculate the content digest and store it on the curi. + //Must be calculated now, before the warc writer closes the file stream. + //We don't need an extra copy, so just write to NullOutputStream. + String recorderBaseName = "ExtractorYoutubeDL-" + nextRecorderId.getAndIncrement(); + Recorder recorder = new Recorder(new File(controller.getScratchDir().getFile(), recorderBaseName), + controller.getRecorderOutBufferBytes(), controller.getRecorderInBufferBytes()); + recorder.getRecordedInput().setDigest("sha1"); + getLocalTempFile().seek(0); + recorder.inputWrap(Channels.newInputStream(getLocalTempFile().getChannel())); + recorder.getRecordedInput().startDigest(); + recorder.outputWrap(new NullOutputStream()); + recorder.getRecordedInput().readFully(); + curi.getData().put(YDL_JSON_FILE_DIGEST,recorder.getRecordedInput().getDigestValue()); + recorder.getRecordedOutput().close(); + //Leave InputStream open for warc writer to handle, but close our NullOutputStream + getLocalTempFile().seek(0); InputStream inputStream = Channels.newInputStream(getLocalTempFile().getChannel()); recordInfo.setContentStream(inputStream); @@ -608,6 +639,40 @@ public class ExtractorYoutubeDL extends Extractor return recordInfo; } + /** + * Because we are writing an additional WARC Metadata Record for the json video info, there is no CrawlURI for that + * record, and thus nothing ever goes through the frontier to be logged to the crawl.log. To log this capture we + * Create a CrawlURI pseudoCuri object and assign the appropriate values and then call to the logger. + * + * @param recordInfo WARCRecordInfo object that was just written + * @param curi CrawlURI that generated the WARCRecordInfo Object + */ + @Override + public void postWrite(WARCRecordInfo recordInfo, CrawlURI curi) { + CrawlURI pseudoCuri = null; + try { + pseudoCuri = curi.createCrawlURI(recordInfo.getUrl(), LinkContext.EMBED_MISC, Hop.INFERRED); + + pseudoCuri.getAnnotations().add("youtube-dl:"); + pseudoCuri.setThreadNumber(curi.getThreadNumber()); + pseudoCuri.setContentSize(recordInfo.getContentLength()); + pseudoCuri.setContentType(recordInfo.getMimetype()); + pseudoCuri.addExtraInfo("warcFilename", recordInfo.getWARCFilename()); + pseudoCuri.addExtraInfo("warcFileOffset", recordInfo.getWARCFileOffset()); + pseudoCuri.setFetchStatus(204); + pseudoCuri.setContentDigest("sha1",(byte[])curi.getData().get(YDL_JSON_FILE_DIGEST)); + pseudoCuri.addExtraInfo("contentSize", recordInfo.getContentLength()); + + Object array[] = {pseudoCuri}; + this.controller.getLoggerModule().getUriProcessing().log(Level.INFO, + curi.getUURI().toString(), array); + } catch (URIException e) { + logger.log(Level.WARNING, "Exception while parsing UURI for youtube-dl metadata record " + recordInfo.getUrl(), e); + } catch (IOException e) { + logger.log(Level.WARNING, "Exception while generating digest for youtube-dl metadata record " + recordInfo.getUrl(), e); + } + } + public static void main(String[] args) throws IOException { /* File t = File.createTempFile("ydl", ".json"); diff --git a/modules/src/main/java/org/archive/modules/warc/BaseWARCRecordBuilder.java b/modules/src/main/java/org/archive/modules/warc/BaseWARCRecordBuilder.java index 20dbf556..ff306dbc 100644 --- a/modules/src/main/java/org/archive/modules/warc/BaseWARCRecordBuilder.java +++ b/modules/src/main/java/org/archive/modules/warc/BaseWARCRecordBuilder.java @@ -7,6 +7,7 @@ import java.net.URI; import java.net.URISyntaxException; import java.util.UUID; +import org.archive.io.warc.WARCRecordInfo; import org.archive.modules.CrawlURI; import org.archive.modules.net.CrawlHost; import org.archive.modules.net.ServerCache; @@ -63,4 +64,7 @@ public abstract class BaseWARCRecordBuilder implements WARCRecordBuilder { return h.getIP().getHostAddress(); } + public void postWrite(WARCRecordInfo recordInfo, CrawlURI curi) { + return; + } } diff --git a/modules/src/main/java/org/archive/modules/warc/WARCRecordBuilder.java b/modules/src/main/java/org/archive/modules/warc/WARCRecordBuilder.java index 2b5d2eba..017d867a 100644 --- a/modules/src/main/java/org/archive/modules/warc/WARCRecordBuilder.java +++ b/modules/src/main/java/org/archive/modules/warc/WARCRecordBuilder.java @@ -43,4 +43,6 @@ public interface WARCRecordBuilder { WARCRecordInfo buildRecord(CrawlURI curi, URI concurrentTo) throws IOException; + void postWrite(WARCRecordInfo warcRecordInfo, CrawlURI curi); + } \ No newline at end of file diff --git a/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java b/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java index 9b86057d..7403fc0d 100644 --- a/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java @@ -163,7 +163,9 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements if (recordBuilder.shouldBuildRecord(curi)) { WARCRecordInfo record = recordBuilder.buildRecord(curi, concurrentTo); if (record != null) { + record.setWARCFileOffset(writer.getPosition()); writer.writeRecord(record); + record.setWARCFilename(writer.getFilenameWithoutOccupiedSuffix()); InputStream is = null; try { is = record.getContentStream(); @@ -178,6 +180,7 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements if (concurrentTo == null) { concurrentTo = record.getRecordId(); } + recordBuilder.postWrite(record, curi); } } }