cic-cli/cic/contract/components/attachment.py

83 lines
2.5 KiB
Python
Raw Normal View History

2021-10-09 19:56:29 +02:00
# standard imports
import logging
import os
2021-10-09 19:56:29 +02:00
# local imports
from cic.contract.base import Data, data_dir
2021-10-09 19:56:29 +02:00
logg = logging.getLogger(__name__)
class Attachment(Data):
2021-11-29 08:02:12 +01:00
"""Processes, serialized and publishes all attachments found in the "attachments" subdirectory of the settings directory.
2021-10-09 19:56:29 +02:00
2021-11-29 08:02:12 +01:00
:param path: Path to settings directory
:type path: str
:param writer: Writer interface receiving the output of the processor
:type writer: cic.writers.OutputWriter
2021-11-29 08:02:12 +01:00
"""
def __init__(self, path=".", writer=None, interactive=False):
2021-10-09 19:56:29 +02:00
super(Attachment, self).__init__()
2021-10-10 21:30:18 +02:00
self.contents = {}
2021-10-09 19:56:29 +02:00
self.path = path
self.writer = writer
self.attachment_path = os.path.join(self.path, "attachments")
self.start()
if interactive:
input(
f"Please add attachment files to '{os.path.abspath(os.path.join(self.path,'attachments'))}' and then press ENTER to continue"
)
self.load()
2021-10-09 19:56:29 +02:00
def load(self):
"""Loads attachment data from settings."""
2021-10-09 19:56:29 +02:00
for s in os.listdir(self.attachment_path):
fp = os.path.realpath(os.path.join(self.attachment_path, s))
with open(fp, "rb") as f:
r = f.read()
2021-10-09 19:56:29 +02:00
2021-10-10 21:30:18 +02:00
z = self.hash(r).hex()
self.contents[z] = fp
2021-10-09 19:56:29 +02:00
logg.debug(f"loaded attachment file {fp} digest {z}")
2021-10-09 19:56:29 +02:00
def start(self):
"""Initialize attachment settings from template."""
2021-10-09 19:56:29 +02:00
super(Attachment, self).start()
2022-04-26 20:25:02 +02:00
os.makedirs(self.attachment_path, exist_ok=True)
2021-10-09 20:37:54 +02:00
2021-10-10 21:30:18 +02:00
def get(self, k):
2021-11-29 08:02:12 +01:00
"""Get a single attachment by the sha256 hash of the content.
:param k: Content hash
:type k: str (hex)
"""
2021-10-10 21:30:18 +02:00
return self.contents[k]
def asdict(self):
"""Output attachment state to dict"""
2021-10-10 21:30:18 +02:00
return self.contents
2021-10-21 15:11:05 +02:00
def process(self, token_address=None, token_symbol=None, writer=None):
2021-11-29 08:02:12 +01:00
"""Serialize and publish attachments.
See cic.processor.Processor.process
2021-11-29 08:02:12 +01:00
"""
if writer == None:
writer = self.writer
for key, value in self.contents.items():
fp = os.path.join(self.attachment_path, value)
with open(fp, "rb") as f:
data = f.read()
logg.debug(f"writing attachment {key}")
writer.write(key, data)
2021-10-11 19:02:42 +02:00
2021-10-09 20:37:54 +02:00
def __str__(self):
s = ""
for key, value in self.contents.items():
s += f"{key} = {value}\n" # self.digests[i].hex(), self.contents[i])
2021-10-09 20:37:54 +02:00
return s