unpacker.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115
  1. import os
  2. import json
  3. import socket
  4. import sys
  5. import time
  6. import tempfile
  7. from threading import Thread
  8. from subprocess import check_output
  9. from bookworm import s3
  10. from bookworm.logger import log, setup_logger
  11. from bookworm.constants import UNPACKABLE_EXTENSIONS, REDIS
  12. import redis
  13. def should_unpack(fname):
  14. fname = fname.lower()
  15. return fname.endswith('rar') or fname.endswith('zip')
  16. def archive_contents(fd):
  17. to_extract = {}
  18. contents = check_output(['lsar', '-j', fd.name]).decode('utf-8')
  19. contents = json.loads(contents)
  20. log.debug(contents['lsarContents'])
  21. for f in contents['lsarContents']:
  22. fname = f['XADFileName']
  23. if any([extension in fname.lower() for extension in UNPACKABLE_EXTENSIONS]):
  24. log.info('Extracting %s from the archive', fname)
  25. to_extract[f['XADIndex']] = fname
  26. return to_extract
  27. def convert_to_mobi(orig_fname) -> bytes:
  28. with tempfile.NamedTemporaryFile(suffix='.mobi') as fd:
  29. log.info('Converting %s to mobi at %s', orig_fname, fd.name)
  30. check_output(['ebook-convert', orig_fname, fd.name, '--output-profile=kindle_pw'])
  31. log.info('Done converting')
  32. buff = open(fd.name, 'rb').read()
  33. return buff
  34. def store_file(s3client, fname, file_contents, meta):
  35. log.info('Puttin in s3 under bucket %s with key %s', meta['processed_file_bucket'], fname)
  36. s3client.put_object(Body=file_contents, Bucket=meta['processed_file_bucket'], Key=fname)
  37. log.info('Put in s3 under bucket %s with key %s', meta['processed_file_bucket'], fname)
  38. def delete_raw_file(s3client, s3key, meta):
  39. log.info('Deleting %s from %s', s3key, meta['raw_file_bucket'])
  40. s3client.delete_object(Bucket=meta['raw_file_bucket'], Key=s3key)
  41. def unpack_and_convert(job_key, s3key, s3client, redis, meta):
  42. redis.hset(job_key, REDIS.STEP_KEY, 'UNPACKING')
  43. unpacked_files = unpack(s3key, s3client, redis, meta)
  44. redis.hset(job_key, REDIS.STEP_KEY, 'UNPACK_DONE')
  45. log.info('Done unpacking job %s', job_key)
  46. for fname, data in unpacked_files:
  47. if not fname.lower().endswith('mobi'):
  48. redis.hset(job_key, REDIS.STEP_KEY, 'CONVERTING')
  49. redis.hset(job_key, REDIS.STATE_KEY, fname)
  50. log.info('Asked to store %s, need to convert first', fname)
  51. fname_no_ext, ext = os.path.splitext(fname)
  52. with tempfile.NamedTemporaryFile(suffix=ext) as original:
  53. original.write(data)
  54. original.flush()
  55. converted_data = convert_to_mobi(original.name)
  56. data = converted_data
  57. fname = fname_no_ext + '.mobi'
  58. store_file(s3client, fname, data, meta)
  59. delete_raw_file(s3client, s3key, meta)
  60. redis.delete(job_key)
  61. def unpack(s3key, s3client, redis, meta):
  62. log.info('Got a request to unpack %s', s3key)
  63. data = s3client.get_object(Key=s3key, Bucket=meta['raw_file_bucket'])
  64. with tempfile.NamedTemporaryFile() as fd:
  65. raw_file_contents = data['Body'].read()
  66. fd.write(raw_file_contents)
  67. fd.flush()
  68. if not should_unpack(s3key):
  69. log.info("Not unpacking %s", s3key)
  70. return [(s3key, raw_file_contents)]
  71. to_extract = archive_contents(fd)
  72. if not to_extract:
  73. log.error("Could not find any valid file")
  74. return []
  75. ret = []
  76. for index, fname in to_extract.items():
  77. log.info('Processing %s %s', index, fname)
  78. file_contents = check_output(['unar', '-o', '-', '-i', fd.name, str(index)])
  79. log.info('Got %d bytes', len(file_contents))
  80. ret.append((fname, file_contents))
  81. return ret
  82. def main():
  83. r = redis.StrictRedis(host='localhost', port=6379)
  84. setup_logger()
  85. s3client = s3.client()
  86. while True:
  87. log.info('Waiting for message on %s', REDIS.Q_UNPACK_FILE)
  88. topic, message = r.blpop(REDIS.Q_UNPACK_FILE)
  89. log.info('got message: %s', message)
  90. params = json.loads(message.decode('utf-8'))
  91. log.info('params for unpacker: %s', params)
  92. params['s3client'] = s3client
  93. params['redis'] = r
  94. t = Thread(target=unpack_and_convert, kwargs=params)
  95. t.daemon = True
  96. t.start()
  97. main()