Cleanup ISOData command

Cache downloaded bytes in memory and dont store a BytesIO instance,
instead create one fresh on zip file access.
This commit is contained in:
Kovid Goyal 2024-06-28 10:12:17 +05:30
parent 4804084665
commit c721338b43
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C

View File

@ -8,51 +8,49 @@ import shutil
import time import time
import zipfile import zipfile
from contextlib import suppress from contextlib import suppress
from functools import lru_cache
from io import BytesIO from io import BytesIO
from setup import Command, download_securely from setup import Command, download_securely
@lru_cache(2)
def iso_codes_data():
URL = 'https://salsa.debian.org/iso-codes-team/iso-codes/-/archive/main/iso-codes-main.zip'
return download_securely(URL)
class ISOData(Command): class ISOData(Command):
description = 'Get ISO codes name localization data' description = 'Get ISO codes name localization data'
URL = 'https://salsa.debian.org/iso-codes-team/iso-codes/-/archive/main/iso-codes-main.zip' top_level_filename = 'iso-codes-main'
_zip_data = None
def add_options(self, parser): def add_options(self, parser):
with suppress(optparse.OptionConflictError): # ignore if option already added with suppress(optparse.OptionConflictError): # ignore if option already added
parser.add_option('--path-to-isocodes', help='Path to previously downloaded iso-codes-main.zip') parser.add_option('--path-to-isocodes', help='Path to previously downloaded iso-codes-main.zip')
def run(self, opts): def run(self, opts):
if self._zip_data is None: if self._zip_data is None and opts.path_to_isocodes:
if opts.path_to_isocodes:
with open(opts.path_to_isocodes, 'rb') as f: with open(opts.path_to_isocodes, 'rb') as f:
self._zip_data = BytesIO(f.read()) self._zip_data = f.read()
# get top level directory # get top level directory
top = {item.split('/')[0] for item in zipfile.ZipFile(self.zip_data).namelist()} top = {item.split('/')[0] for item in zipfile.ZipFile(self.zip_data).namelist()}
assert len(top) == 1 assert len(top) == 1
self.top_level = top.pop() self.top_level_filename = top.pop()
else:
self._zip_data = BytesIO(download_securely(self.URL))
def __init__(self):
super().__init__()
self._zip_data = None
self.top_level = 'iso-codes-main'
@property @property
def zip_data(self): def zip_data(self):
return self._zip_data return self._zip_data or iso_codes_data()
def db_data(self, name: str) -> bytes: def db_data(self, name: str) -> bytes:
with zipfile.ZipFile(self.zip_data) as zf: with zipfile.ZipFile(BytesIO(self.zip_data)) as zf:
with zf.open(f'{self.top_level}/data/{name}') as f: with zf.open(f'{self.top_level_filename}/data/{name}') as f:
return f.read() return f.read()
def extract_po_files(self, name: str, output_dir: str) -> None: def extract_po_files(self, name: str, output_dir: str) -> None:
name = name.split('.', 1)[0] name = name.split('.', 1)[0]
pat = f'{self.top_level}/{name}/*.po' pat = f'{self.top_level_filename}/{name}/*.po'
if self.zip_data is None: with zipfile.ZipFile(BytesIO(self.zip_data)) as zf:
self._zip_data = BytesIO(download_securely(self.URL))
with zipfile.ZipFile(self.zip_data) as zf:
for name in fnmatch.filter(zf.namelist(), pat): for name in fnmatch.filter(zf.namelist(), pat):
dest = os.path.join(output_dir, name.split('/')[-1]) dest = os.path.join(output_dir, name.split('/')[-1])
zi = zf.getinfo(name) zi = zf.getinfo(name)