mirror of https://github.com/papermerge/mglib
6 changed files with 168 additions and 6 deletions
@ -0,0 +1,40 @@ |
|||
import logging |
|||
from pmworker import wrapper |
|||
|
|||
|
|||
logger = logging.getLogger(__name__) |
|||
|
|||
|
|||
class Mime(wrapper.Wrapper): |
|||
def __init__(self, filepath): |
|||
super().__init__(exec_name="file") |
|||
self.filepath = filepath |
|||
|
|||
def get_cmd(self): |
|||
cmd = super().get_cmd() |
|||
|
|||
cmd.extend(['--mime-type']) |
|||
cmd.extend(['-b']) |
|||
cmd.extend([self.filepath]) |
|||
|
|||
return cmd |
|||
|
|||
def is_tiff(self): |
|||
return self.guess() == 'image/tiff' |
|||
|
|||
def is_pdf(self): |
|||
return self.guess() == 'application/pdf' |
|||
|
|||
def is_image(self): |
|||
""" |
|||
Returns true if MIME type is one of following: |
|||
* image/png |
|||
* image/jpg |
|||
""" |
|||
return self.guess() in ('image/png', 'image/jpg') |
|||
|
|||
def guess(self): |
|||
cmd = self.get_cmd() |
|||
complete = self.run(cmd) |
|||
|
|||
return complete.stdout.strip() |
@ -0,0 +1,32 @@ |
|||
import os |
|||
import logging |
|||
|
|||
from mglib.runcmd import run |
|||
|
|||
logger = logging.getLogger(__name__) |
|||
|
|||
|
|||
def convert_tiff2pdf(doc_url): |
|||
|
|||
logger.debug(f"convert_tiff2pdf for {doc_url}") |
|||
# basename is filename + ext (no path) |
|||
|
|||
basename = os.path.basename(doc_url) |
|||
base_root, base_ext = os.path.splitext(basename) |
|||
root, ext = os.path.splitext(doc_url) |
|||
new_doc_url = f"{root}.pdf" |
|||
|
|||
logger.debug( |
|||
f"tiff2pdf source={doc_url} dest={new_doc_url}" |
|||
) |
|||
|
|||
cmd = ( |
|||
"convert", |
|||
doc_url, |
|||
new_doc_url, |
|||
) |
|||
|
|||
run(cmd) |
|||
|
|||
# returns new filename |
|||
return f"{base_root}.pdf" |
@ -0,0 +1,49 @@ |
|||
import logging |
|||
import subprocess |
|||
|
|||
logger = logging.getLogger(__name__) |
|||
|
|||
|
|||
class Wrapper: |
|||
|
|||
def __init__(self, exec_name, check=True, dry_run=False): |
|||
self.exec_name = exec_name |
|||
self.check = check, |
|||
# usefull for debugging purpose |
|||
self.dry_run = dry_run |
|||
|
|||
def get_cmd(self): |
|||
cmd = [] |
|||
|
|||
if self.exec_name: |
|||
cmd.extend([self.exec_name]) |
|||
|
|||
return cmd |
|||
|
|||
def call_no_args(self): |
|||
|
|||
cmd = self.get_cmd() |
|||
self.run(cmd) |
|||
|
|||
def run(self, cmd): |
|||
|
|||
command_to_run = ' '.join(cmd) |
|||
|
|||
if (self.dry_run): |
|||
logger.debug(f"Dry run: {command_to_run}") |
|||
|
|||
logger.debug(f"subprocess: {command_to_run}") |
|||
|
|||
ret = subprocess.run( |
|||
cmd, |
|||
stdout=subprocess.PIPE, |
|||
stderr=subprocess.PIPE, |
|||
encoding="utf-8" |
|||
) |
|||
if ret.returncode != 0: |
|||
logger.error(( |
|||
f"returncode={ret.returncode}" |
|||
f" stdout={ret.stdout}" |
|||
f" stderr={ret.stderr}" |
|||
)) |
|||
return ret |
Loading…
Reference in new issue