Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def on_hit(self, hit, tweet_count):
# Cycle tweet id files
if tweet_count % self.max_per_file == 0:
if self.file:
self.file.close()
self.file = open(
os.path.join(self.dataset_path, 'tweets-{}.csv'.format(str(self.file_count).zfill(3))), 'w')
self.sheet = csv.writer(self.file)
self.sheet.writerow(json2csv.get_headings())
self.file_count += 1
# Write to tweet file
self.sheet.writerow(json2csv.get_row(json.loads(hit.tweet), excel=True))
def get_headings(extra_headings=None):
fields = json2csv.get_headings()
if extra_headings:
fields.extend(extra_headings)
return fields
def _header_row(self):
return twarc.json2csv.get_headings()
parser.error("csv output not available for %s" % command)
elif args.format in ("csv", "csv-excel"):
csv_writer = csv.writer(fh)
csv_writer.writerow(get_headings())
line_count = 0
file_count = 0
for thing in things:
# rotate the files if necessary
if args.output and args.split and line_count % args.split == 0:
file_count += 1
fh = codecs.open(numbered_filepath(args.output, file_count), 'wb', 'utf8')
if csv_writer:
csv_writer = csv.writer(fh)
csv_writer.writerow(get_headings())
line_count += 1
# ready to output
kind_of = type(thing)
if kind_of == str_type:
# user or tweet IDs
print(thing, file=fh)
log.info("archived %s" % thing)
elif 'id_str' in thing:
# tweets and users
if (args.format == "json"):
print(json.dumps(thing), file=fh)
elif (args.format == "csv"):
csv_writer.writerow(get_row(thing))