mirror of
https://github.com/Xevion/phototag.git
synced 2025-12-18 00:12:41 -06:00
exception handling -> printing, xmp identification and handling
This commit is contained in:
@@ -12,19 +12,20 @@ from google.cloud import vision
|
||||
from . import TEMP_PATH, INPUT_PATH, OUTPUT_PATH
|
||||
from .xmp import XMPParser
|
||||
|
||||
|
||||
class FileProcessor(object):
|
||||
def __init__(self, file_name : str):
|
||||
def __init__(self, file_name: str):
|
||||
self.file_name = file_name
|
||||
self.base, self.ext = os.path.splitext(self.file_name) # fileNAME and fileEXTENSIOn
|
||||
self.base, self.ext = os.path.splitext(
|
||||
self.file_name) # fileNAME and fileEXTENSIOn
|
||||
# Path to temporary file that will be optimized for upload to Google
|
||||
self.temp_file_path = os.path.join(TEMP_PATH, self.base + '.jpeg')
|
||||
|
||||
@property
|
||||
def hasXMP(self):
|
||||
return self.ext.lower() == 'xmp'
|
||||
|
||||
# Decide whether a XMP file is available
|
||||
self.xmp = os.path.join(INPUT_PATH, base + '.xmp')
|
||||
self.xmp = self.xmp if os.path.exists(self.xmp) else None
|
||||
|
||||
# Optimizes a file using JPEG thumbnailing and compression.
|
||||
def _optimize(self, file : str, size : tuple =(512, 512), quality=85, copy=None):
|
||||
def _optimize(self, file: str, size: tuple = (512, 512), quality : int = 85, copy : str = None):
|
||||
image = Image.open(file)
|
||||
image.thumbnail(size, resample=Image.ANTIALIAS)
|
||||
if copy:
|
||||
@@ -40,12 +41,13 @@ class FileProcessor(object):
|
||||
rgb.close()
|
||||
self._optimize(self.temp_file_path)
|
||||
else:
|
||||
self._optimize(os.path.join(INPUT_PATH, self.file_name), copy=self.temp_file_path)
|
||||
self._optimize(os.path.join(
|
||||
INPUT_PATH, self.file_name), copy=self.temp_file_path)
|
||||
|
||||
def run(self, client : vision.ImageAnnotatorClient):
|
||||
def run(self, client: vision.ImageAnnotatorClient):
|
||||
try:
|
||||
if self.hasXMP:
|
||||
self.optimize()
|
||||
|
||||
self.optimize()
|
||||
|
||||
# Open the image, read as bytes, convert to types Image
|
||||
image = Image.open(self.temp_file_path)
|
||||
@@ -53,15 +55,16 @@ class FileProcessor(object):
|
||||
image.save(bytesIO, format='jpeg')
|
||||
image.close()
|
||||
image = vision.types.Image(content=bytesIO.getvalue())
|
||||
|
||||
|
||||
# Performs label detection on the image file
|
||||
response = client.label_detection(image=image)
|
||||
labels = [label.description for label in response.label_annotations]
|
||||
print('\tLabels: {}'.format(', '.join(labels)))
|
||||
|
||||
|
||||
# XMP sidecar file specified, write to it using XML module
|
||||
if self.xmp_name:
|
||||
logging.info('\tWriting {} tags to output XMP.'.format(len(labels)))
|
||||
if self.hasXMP:
|
||||
logging.info(
|
||||
'\tWriting {} tags to output XMP.'.format(len(labels)))
|
||||
parser = XMPParser(os.path.join(INPUT_PATH, self.xmp_name))
|
||||
parser.add_keywords(labels)
|
||||
# Save the new XMP file
|
||||
@@ -71,8 +74,10 @@ class FileProcessor(object):
|
||||
os.remove(os.path.join(INPUT_PATH, self.xmp_name))
|
||||
# No XMP file is specified, using IPTC tagging
|
||||
else:
|
||||
logging.info('\tWriting {} tags to image IPTC'.format(len(labels)))
|
||||
info = iptcinfo3.IPTCInfo(os.path.join(INPUT_PATH, self.file_name))
|
||||
logging.info(
|
||||
'\tWriting {} tags to image IPTC'.format(len(labels)))
|
||||
info = iptcinfo3.IPTCInfo(
|
||||
os.path.join(INPUT_PATH, self.file_name))
|
||||
info['keywords'].extend(labels)
|
||||
info.save()
|
||||
# Remove the weird ghsot file created by this iptc read/writer.
|
||||
@@ -88,4 +93,4 @@ class FileProcessor(object):
|
||||
# Remove the temporary file (if it exists)
|
||||
def _cleanup(self):
|
||||
if os.path.exists(self.temp_file_path):
|
||||
os.remove(self.temp_file_path)
|
||||
os.remove(self.temp_file_path)
|
||||
|
||||
Reference in New Issue
Block a user