aboutsummaryrefslogtreecommitdiffstats
path: root/python/sandcrawler_worker.py
diff options
context:
space:
mode:
Diffstat (limited to 'python/sandcrawler_worker.py')
-rwxr-xr-xpython/sandcrawler_worker.py20
1 files changed, 15 insertions, 5 deletions
diff --git a/python/sandcrawler_worker.py b/python/sandcrawler_worker.py
index f314218..895a5b9 100755
--- a/python/sandcrawler_worker.py
+++ b/python/sandcrawler_worker.py
@@ -30,11 +30,10 @@ def run_grobid_extract(args):
def run_grobid_persist(args):
consume_topic = "sandcrawler-{}.grobid-output-pg".format(args.env)
- raise NotImplementedError
- #worker = GrobidPersistWorker()
- #pusher = KafkaJsonPusher(worker=worker, kafka_hosts=args.kafka_hosts,
- # consume_topic=consume_topic, group="grobid-persist")
- #pusher.run()
+ worker = PersistGrobidWorker()
+ pusher = KafkaJsonPusher(worker=worker, kafka_hosts=args.kafka_hosts,
+ consume_topic=consume_topic, group="grobid-persist")
+ pusher.run()
def run_ingest_file(args):
consume_topic = "sandcrawler-{}.ingest-file-requests".format(args.env)
@@ -46,6 +45,13 @@ def run_ingest_file(args):
consume_topic=consume_topic, group="ingest-file", batch_size=1)
pusher.run()
+def run_ingest_file_persist(args):
+ consume_topic = "sandcrawler-{}.ingest-file-results".format(args.env)
+ worker = PersistIngestFileResultWorker()
+ pusher = KafkaJsonPusher(worker=worker, kafka_hosts=args.kafka_hosts,
+ consume_topic=consume_topic, group="ingest-persist")
+ pusher.run()
+
def main():
parser = argparse.ArgumentParser(
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
@@ -72,6 +78,10 @@ def main():
help="daemon that consumes requests from Kafka, ingests, pushes results to Kafka")
sub_ingest_file.set_defaults(func=run_ingest_file)
+ sub_ingest_file_persist = subparsers.add_parser('ingest-file-persist',
+ help="daemon that consumes ingest-file output from Kafka and pushes to postgres")
+ sub_ingest_file_persist.set_defaults(func=run_ingest_file_persist)
+
args = parser.parse_args()
if not args.__dict__.get("func"):
print("tell me what to do!")