aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorBryan Newbold <bnewbold@robocracy.org>2019-05-30 13:05:22 -0700
committerBryan Newbold <bnewbold@robocracy.org>2019-05-30 13:05:22 -0700
commit36bfdda186151972167c35cd0f642bc0c2ca1a54 (patch)
treed1de856018885c9460739d2e2c3e120541adf12b
parent393172a29e91bf1236e546b4a99f445dbbe849ef (diff)
downloadfatcat-36bfdda186151972167c35cd0f642bc0c2ca1a54.tar.gz
fatcat-36bfdda186151972167c35cd0f642bc0c2ca1a54.zip
fix and workaround container entities in release topic
-rw-r--r--python/fatcat_tools/workers/changelog.py4
-rw-r--r--python/fatcat_tools/workers/elasticsearch.py6
2 files changed, 8 insertions, 2 deletions
diff --git a/python/fatcat_tools/workers/changelog.py b/python/fatcat_tools/workers/changelog.py
index 5ef1c69c..a753bb49 100644
--- a/python/fatcat_tools/workers/changelog.py
+++ b/python/fatcat_tools/workers/changelog.py
@@ -75,8 +75,8 @@ class EntityUpdatesWorker(FatcatWorker):
def run(self):
changelog_topic = self.kafka.topics[self.consume_topic]
release_topic = self.kafka.topics[self.release_topic]
- file_topic = self.kafka.topics[self.release_topic]
- container_topic = self.kafka.topics[self.release_topic]
+ file_topic = self.kafka.topics[self.file_topic]
+ container_topic = self.kafka.topics[self.container_topic]
consumer = changelog_topic.get_balanced_consumer(
consumer_group=self.consumer_group,
diff --git a/python/fatcat_tools/workers/elasticsearch.py b/python/fatcat_tools/workers/elasticsearch.py
index fc96e3d4..a18ed244 100644
--- a/python/fatcat_tools/workers/elasticsearch.py
+++ b/python/fatcat_tools/workers/elasticsearch.py
@@ -43,6 +43,12 @@ class ElasticsearchReleaseWorker(FatcatWorker):
for msg in consumer:
json_str = msg.value.decode('utf-8')
+ # HACK: work around a bug where container entities got published to
+ # release_v03 topic
+ if self.elasticsearch_document_name == "release":
+ entity_dict = json.loads(json_str)
+ if entity_dict.get('name') and not entity_dict.get('title'):
+ continue
entity = entity_from_json(json_str, self.entity_type, api_client=ac)
#print(entity)
elasticsearch_endpoint = "{}/{}/{}/{}".format(