Added automatic detection of image file type.
Create a new windw if necessary. Added support for AVIF format.
This commit is contained in:
parent
c9decd19f1
commit
91df24ca4f
2 changed files with 74 additions and 33 deletions
|
@ -10,3 +10,8 @@ python -m pip install requests
|
|||
|
||||
python -m pip install pillow
|
||||
|
||||
python -m pip install pillow-avif-plugin
|
||||
|
||||
python -m pip install magika==0.6.1rc1
|
||||
|
||||
|
||||
|
|
102
aecpic.py
102
aecpic.py
|
@ -6,6 +6,7 @@ get a picture at some url in use curl, then open new anvil window and set the pi
|
|||
author: moli
|
||||
'''
|
||||
|
||||
import sys
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
@ -13,54 +14,70 @@ import requests
|
|||
import json
|
||||
import mimetypes
|
||||
from PIL import Image
|
||||
import pillow_avif
|
||||
import argparse
|
||||
from magika import Magika
|
||||
|
||||
def getSuffix(url):
|
||||
|
||||
if 'ANVIL_API_PORT' not in os.environ:
|
||||
print('not in Anvil.')
|
||||
sys.exit(1)
|
||||
|
||||
host = 'localhost'
|
||||
ANVIL_API_PORT = os.environ['ANVIL_API_PORT']
|
||||
ANVIL_API_SESS = os.environ['ANVIL_API_SESS']
|
||||
ANVIL_WIN_ID = os.environ['ANVIL_WIN_ID']
|
||||
|
||||
def getSuffix(byts):
|
||||
'''
|
||||
get the suffix in the url.
|
||||
get the filetype in the byts, and change sufifix
|
||||
|
||||
>>> import apcpic
|
||||
>>> apcpic.getSuffix('https://moli-green.xyz/img.webp')
|
||||
'.webp'
|
||||
'''
|
||||
mimetype = mimetypes.guess_type(url)
|
||||
match mimetype:
|
||||
case 'image/png', encoding:
|
||||
m = Magika()
|
||||
mr = m.identify_bytes(byts)
|
||||
filetype = str(mr.output.label)
|
||||
|
||||
match filetype:
|
||||
case 'png':
|
||||
return '.png'
|
||||
case 'image/jpeg', encoding:
|
||||
case 'jpeg':
|
||||
return '.jpeg'
|
||||
case 'image/webp', encoding:
|
||||
case 'webp':
|
||||
return '.webp'
|
||||
case 'image/avif', encoding:
|
||||
case 'mp4':
|
||||
return '.avif'
|
||||
case 'image/gif', encoding:
|
||||
case 'gif':
|
||||
return '.gif'
|
||||
case _:
|
||||
raise RuntimeError('this image type not supported.')
|
||||
raise RuntimeError(f'this image type not supported. {filetype}')
|
||||
|
||||
def download(url):
|
||||
'''
|
||||
download the iamge file to the temporary file.
|
||||
|
||||
>>> import apcpic
|
||||
>>> r = apcpic.download('https://moli-green.xyz/img.webp')
|
||||
>>> if r != '':
|
||||
... pass
|
||||
>>> import aecpic
|
||||
>>> n, t = aecpic.download('https://moli-green.xyz/img.webp')
|
||||
>>> assert n != None and t == 'webp'
|
||||
'''
|
||||
suffix = getSuffix(url)
|
||||
filename = ''
|
||||
|
||||
cr = subprocess.run(['curl', url], capture_output = True)
|
||||
if cr.returncode != 0:
|
||||
raise RuntimeError('Downlad failed.')
|
||||
|
||||
suffix = getSuffix(cr.stdout)
|
||||
filename = None
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode='wb',
|
||||
delete=False,
|
||||
suffix=suffix) as f:
|
||||
r = subprocess.run(['curl', url], capture_output = True)
|
||||
f.write(r.stdout)
|
||||
f.write(cr.stdout)
|
||||
filename = f.name
|
||||
return filename
|
||||
|
||||
def getPNG(url):
|
||||
'''
|
||||
get PNG file path.
|
||||
Download the file from the URL, convert and save to PNG format.
|
||||
|
||||
>>> import apcpic
|
||||
>>> r = apcpic.getPNG('https://moli-green.xyz/img.webp')
|
||||
|
@ -79,6 +96,32 @@ def getPNG(url):
|
|||
img.save(png_filename)
|
||||
return png_filename
|
||||
|
||||
def get_anvil_window():
|
||||
'''
|
||||
Get or create Anvil's window.
|
||||
'''
|
||||
|
||||
global ANVIL_WIN_ID
|
||||
|
||||
if ANVIL_WIN_ID != '':
|
||||
return ANVIL_WIN_ID
|
||||
|
||||
h = {
|
||||
'Anvil-Sess': ANVIL_API_SESS,
|
||||
'Accept': 'application/json',
|
||||
}
|
||||
|
||||
s = f'http://{host}:{int(ANVIL_API_PORT)}/wins'
|
||||
|
||||
r = requests.post(s, headers=h)
|
||||
|
||||
if r.status_code != requests.codes.ok:
|
||||
raise RuntimeError(f'status code: {r.status_code}')
|
||||
|
||||
ANVIL_WIN_ID = r.json()['Id']
|
||||
|
||||
return ANVIL_WIN_ID
|
||||
|
||||
def anvil_Pic(host='localhost', url=None):
|
||||
'''
|
||||
Pic command send to Anvil.
|
||||
|
@ -86,20 +129,12 @@ def anvil_Pic(host='localhost', url=None):
|
|||
if not url:
|
||||
raise RuntimeError('not have url.')
|
||||
|
||||
if 'ANVIL_API_PORT' not in os.environ:
|
||||
raise RuntimeError('not in Anvil')
|
||||
|
||||
filename = getPNG(url)
|
||||
|
||||
ANVIL_API_PORT = os.environ['ANVIL_API_PORT']
|
||||
ANVIL_API_SESS = os.environ['ANVIL_API_SESS']
|
||||
ANVIL_WIN_ID = os.environ['ANVIL_WIN_ID']
|
||||
|
||||
if ANVIL_WIN_ID == '':
|
||||
raise RuntimeError('not on window.')
|
||||
anvil_win_id = get_anvil_window()
|
||||
|
||||
h = {
|
||||
'Anvil-Sess': ANVIL_API_SESS,
|
||||
'Anvil-Sess': ANVIL_API_SESS,
|
||||
'Accept': 'application/json',
|
||||
}
|
||||
|
||||
|
@ -122,5 +157,6 @@ if __name__ == '__main__':
|
|||
p.add_argument('--host', default='localhost', help='Hostname')
|
||||
p.add_argument('url', help='Image URL')
|
||||
arg = p.parse_args()
|
||||
anvil_Pic(arg.host, arg.url)
|
||||
host = arg.host
|
||||
anvil_Pic(host, arg.url)
|
||||
|
||||
|
|
Loading…
Add table
Reference in a new issue