144 lines
4.1 KiB
Python
144 lines
4.1 KiB
Python
import os
|
|
import json
|
|
import requests
|
|
import glob
|
|
import exifread
|
|
|
|
from PIL import Image, ExifTags
|
|
from PIL.ExifTags import TAGS
|
|
from PIL.PngImagePlugin import PngImageFile, PngInfo
|
|
|
|
def load_config(file_path):
|
|
with open(file_path, "r") as config_file:
|
|
config_data = json.load(config_file)
|
|
return config_data
|
|
|
|
def getTags(file_path):
|
|
try:
|
|
file = open(file_path,'r')
|
|
fileContent = file.read()
|
|
file.close
|
|
except:
|
|
fileContent = ''
|
|
return fileContent
|
|
|
|
def getImages(dir):
|
|
result = glob.glob(dir + '/*.jpg')
|
|
return result
|
|
|
|
def getItemDescription(filename):
|
|
type = Image.open(filename)
|
|
|
|
exif_tags = open(filename, 'rb')
|
|
tags = exifread.process_file(exif_tags)
|
|
|
|
exif_array = []
|
|
|
|
if type.format != "PNG":
|
|
for i in tags:
|
|
compile = i, str(tags[i])
|
|
exif_array.append(compile)
|
|
|
|
if type.format == "PNG":
|
|
image = PngImageFile(filename)
|
|
metadata = PngInfo()
|
|
|
|
for i in image.text:
|
|
compile = i, str(image.text[i])
|
|
exif_array.append(compile)
|
|
|
|
Description = exif_array[0][1]
|
|
return Description
|
|
|
|
def mediaUpload(access_token, url, file):
|
|
api_url = url + "/api/v1/media"
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}"
|
|
}
|
|
|
|
try:
|
|
with open(file, "rb") as imageFile:
|
|
f = {"file": imageFile}
|
|
data = {
|
|
"file": (file, imageFile),
|
|
"description": "API TEST"
|
|
}
|
|
json_data = json.dumps(data)
|
|
|
|
response = requests.post(api_url, headers=headers, files=data)
|
|
response.raise_for_status() # Wirft eine HTTPError-Exception, wenn der Statuscode nicht erfolgreich ist
|
|
|
|
try:
|
|
data = response.json()
|
|
return(data)
|
|
except json.decoder.JSONDecodeError:
|
|
print("API response contains non-valid JSON data:")
|
|
print(response.text)
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"An error occurred during the API call: {e}")
|
|
|
|
def createNewPost(access_token, url, ImageID, ImageDescription):
|
|
api_url = url + "/api/v1/statuses"
|
|
ImageIDs = ImageID.split(",")
|
|
ImageIDsInt = list(map(int, ImageIDs))
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}"
|
|
}
|
|
|
|
data = {
|
|
"status": ImageDescription,
|
|
"media_ids": ImageIDsInt
|
|
}
|
|
|
|
try:
|
|
response = requests.post(api_url, headers=headers, json=data)
|
|
try:
|
|
responsedata = response.json()
|
|
return(responsedata)
|
|
except json.decoder.JSONDecodeError:
|
|
print("API response contains non-valid JSON data:")
|
|
print(response.text)
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"An error occurred during the API call: {e}")
|
|
|
|
def sendImages(Token, url, imagedir, uploadType, deleteFile):
|
|
files = getImages(imagedir)
|
|
tags = getTags("tags.txt")
|
|
tagDescription = ""
|
|
newFileID = ""
|
|
|
|
for f in files:
|
|
newFile = mediaUpload(Token, url, f)
|
|
description = getItemDescription(f)
|
|
if uploadType == 0:
|
|
tagDescription = description + ' ' + tags
|
|
newFileID = newFile.get("id")
|
|
createNewPost(Token, url, newFileID, tagDescription)
|
|
else:
|
|
if tagDescription == "":
|
|
tagDescription = description + ' ' + tags
|
|
if newFileID != "":
|
|
newFileID = newFileID + "," + newFile.get("id")
|
|
else:
|
|
newFileID = newFile.get("id")
|
|
if uploadType == 1:
|
|
createNewPost(Token, url, newFileID, tagDescription)
|
|
|
|
if deleteFile:
|
|
for f in files:
|
|
if os.path.exists(f):
|
|
os.remove(f)
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
config_file_path = "config.json"
|
|
config = load_config(config_file_path)
|
|
|
|
url = config.get("server_url")
|
|
accessToken = config.get("access_token")
|
|
imagedir = config.get("image_path")
|
|
uploadType = config.get("upload_type")
|
|
deleteFile = config.get("delete_file")
|
|
|
|
sendImages(accessToken, url, imagedir, uploadType, deleteFile) |