#!/usr/bin/env python3
# == Rapid Develop Macros in LibreOffice ==
# ~ https://git.cuates.net/elmau/easymacro
# ~ easymacro is free software: you can redistribute it and/or modify
# ~ it under the terms of the GNU General Public License as published by
# ~ the Free Software Foundation, either version 3 of the License, or
# ~ (at your option) any later version.
# ~ easymacro is distributed in the hope that it will be useful,
# ~ but WITHOUT ANY WARRANTY; without even the implied warranty of
# ~ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# ~ GNU General Public License for more details.
# ~ You should have received a copy of the GNU General Public License
# ~ along with easymacro. If not, see .
import datetime
import getpass
import hashlib
import json
import logging
import os
import platform
import shlex
import shutil
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import time
import traceback
from functools import wraps
from pathlib import Path
from pprint import pprint
from string import Template
from typing import Any, Union
from socket import timeout
from urllib import parse
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
import uno
from com.sun.star.awt import MessageBoxButtons as MSG_BUTTONS
from com.sun.star.awt.MessageBoxResults import YES
from com.sun.star.beans import PropertyValue, NamedValue
from com.sun.star.ui.dialogs import TemplateDescription
# Global variables
OS = platform.system()
DESKTOP = os.environ.get('DESKTOP_SESSION', '')
PC = platform.node()
USER = getpass.getuser()
IS_WIN = OS == 'Windows'
IS_MAC = OS == 'Darwin'
LOG_FORMAT = '%(asctime)s - %(levelname)s - %(message)s'
LOG_DATE = '%d/%m/%Y %H:%M:%S'
if IS_WIN:
logging.addLevelName(logging.ERROR, 'ERROR')
logging.addLevelName(logging.DEBUG, 'DEBUG')
logging.addLevelName(logging.INFO, 'INFO')
else:
logging.addLevelName(logging.ERROR, '\033[1;41mERROR\033[1;0m')
logging.addLevelName(logging.DEBUG, '\x1b[33mDEBUG\033[1;0m')
logging.addLevelName(logging.INFO, '\x1b[32mINFO\033[1;0m')
logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT, datefmt=LOG_DATE)
log = logging.getLogger(__name__)
_info_debug = f"Python: {sys.version}\n\n{platform.platform()}\n\n" + '\n'.join(sys.path)
TIMEOUT = 10
SALT = b'00a1bfb05353bb3fd8e7aa7fe5efdccc'
_EVENTS = {}
PYTHON = 'python'
if IS_WIN:
PYTHON = 'python.exe'
FILES = {
'CONFIG': 'zaz-{}.json',
}
DIRS = {}
MESSAGES = {
'es': {
'OK': 'Aceptar',
'Cancel': 'Cancelar',
'Select path': 'Seleccionar ruta',
'Select directory': 'Seleccionar directorio',
'Select file': 'Seleccionar archivo',
'Incorrect user or password': 'Nombre de usuario o contraseña inválidos',
'Allow less secure apps in GMail': 'Activa: Permitir aplicaciones menos segura en GMail',
}
}
CTX = uno.getComponentContext()
SM = CTX.getServiceManager()
# UNO Enum
class MessageBoxType():
"""Class for import enum
`See Api MessageBoxType `_
"""
from com.sun.star.awt.MessageBoxType \
import MESSAGEBOX, INFOBOX, WARNINGBOX, ERRORBOX, QUERYBOX
MBT = MessageBoxType
def create_instance(name: str, with_context: bool=False, argument: Any=None) -> Any:
"""Create a service instance
:param name: Name of service
:type name: str
:param with_context: If used context
:type with_context: bool
:param argument: If needed some argument
:type argument: Any
:return: PyUno instance
:rtype: PyUno Object
"""
if with_context:
instance = SM.createInstanceWithContext(name, CTX)
elif argument:
instance = SM.createInstanceWithArguments(name, (argument,))
else:
instance = SM.createInstance(name)
return instance
def get_app_config(node_name: str, key: str='') -> Any:
"""Get any key from any node from LibreOffice configuration.
:param node_name: Name of node
:type name: str
:param key: Name of key
:type key: str
:return: Any value
:rtype: Any
`See Api ConfigurationProvider `_
"""
name = 'com.sun.star.configuration.ConfigurationProvider'
service = 'com.sun.star.configuration.ConfigurationAccess'
cp = create_instance(name, True)
node = PropertyValue(Name='nodepath', Value=node_name)
value = ''
try:
value = cp.createInstanceWithArguments(service, (node,))
if value and value.hasByName(key):
value = value.getPropertyValue(key)
except Exception as e:
error(e)
value = ''
return value
# Get info LibO
NAME = TITLE = get_app_config('org.openoffice.Setup/Product', 'ooName')
VERSION = get_app_config('org.openoffice.Setup/Product','ooSetupVersion')
LANGUAGE = get_app_config('org.openoffice.Setup/L10N/', 'ooLocale')
LANG = LANGUAGE.split('-')[0]
INFO_DEBUG = f"{NAME} v{VERSION} {LANGUAGE}\n\n{_info_debug}"
# Get start date from Calc configuration
node = '/org.openoffice.Office.Calc/Calculate/Other/Date'
year = get_app_config(node, 'YY')
month = get_app_config(node, 'MM')
day = get_app_config(node, 'DD')
DATE_OFFSET = datetime.date(year, month, day).toordinal()
def _(msg):
if LANG == 'en':
return msg
if not LANG in MESSAGES:
return msg
return MESSAGES[LANG][msg]
def set_app_config(node_name: str, key: str, new_value: Any) -> Any:
"""Update value for key in node name.
:param node_name: Name of node
:type name: str
:param key: Name of key
:type key: str
:return: True if update sucesfully
:rtype: bool
`See Api ConfigurationUpdateAccess `_
"""
result = True
current_value = ''
name = 'com.sun.star.configuration.ConfigurationProvider'
service = 'com.sun.star.configuration.ConfigurationUpdateAccess'
cp = create_instance(name, True)
node = PropertyValue(Name='nodepath', Value=node_name)
update = cp.createInstanceWithArguments(service, (node,))
try:
current_value = update.getPropertyValue(key)
update.setPropertyValue(key, new_value)
update.commitChanges()
except Exception as e:
error(e)
if update.hasByName(key) and current_value:
update.setPropertyValue(key, current_value)
update.commitChanges()
result = False
return result
def debug(*messages) -> None:
"""Show messages debug
:param messages: List of messages to debug
:type messages: list[Any]
"""
data = [str(m) for m in messages]
log.debug('\t'.join(data))
return
def error(message: Any) -> None:
"""Show message error
:param message: The message error
:type message: Any
"""
log.error(message)
return
def info(*messages) -> None:
"""Show messages info
:param messages: List of messages to debug
:type messages: list[Any]
"""
data = [str(m) for m in messages]
log.info('\t'.join(data))
return
def save_log(path: str, data: Any) -> None:
"""Save data in file, data append to end and automatic add current time.
:param path: Path to save log
:type path: str
:param data: Data to save in file log
:type data: Any
"""
with open(path, 'a') as f:
f.write(f'{str(now())[:19]} - ')
pprint(data, stream=f)
return
def _set_app_command(command: str, disable: bool):
NEW_NODE_NAME = f'zaz_disable_command_{command.lower()}'
name = 'com.sun.star.configuration.ConfigurationProvider'
service = 'com.sun.star.configuration.ConfigurationUpdateAccess'
node_name = '/org.openoffice.Office.Commands/Execute/Disabled'
cp = create_instance(name, True)
node = PropertyValue(Name='nodepath', Value=node_name)
update = cp.createInstanceWithArguments(service, (node,))
result = True
try:
if disable:
new_node = update.createInstanceWithArguments(())
new_node.setPropertyValue('Command', command)
update.insertByName(NEW_NODE_NAME, new_node)
else:
update.removeByName(NEW_NODE_NAME)
update.commitChanges()
except Exception as e:
result = False
return result
class commands():
"""Class for disable and enable commands
:param command: UNO Command for disable or enable
:type command: str
`See DispatchCommands `_
"""
def __init__(self, command):
self._command = command
def disable(self):
"""Disable command
:return: True if correctly disable, False if not.
:rtype: bool
"""
return _set_app_command(self._command, True)
def enabled(self):
"""Enable command
:return: True if correctly enable, False if not.
:rtype: bool
"""
return _set_app_command(self._command, False)
def mri(obj: Any) -> None:
"""Inspect object with MRI Extension
:param obj: Any pyUno object
:type obj: Any
`See MRI `_
"""
mri = create_instance('mytools.Mri')
if m is None:
msg = 'Extension MRI not found'
error(msg)
return
if hasattr(obj, 'obj'):
obj = obj.obj
mri.inspect(obj)
return
def catch_exception(f):
"""Catch exception for any function
:param f: Any Python function
:type f: Function instance
"""
@wraps(f)
def func(*args, **kwargs):
try:
return f(*args, **kwargs)
except Exception as e:
name = f.__name__
if IS_WIN:
msgbox(traceback.format_exc())
log.error(name, exc_info=True)
return func
def msgbox(message: Any, title: str=TITLE, buttons=MSG_BUTTONS.BUTTONS_OK, \
type_message_box=MessageBoxType.INFOBOX) -> int:
"""Create message box
:param message: Any type message, all is converted to string.
:type message: Any
:param title: The title for message box
:type title: str
:param buttons: A combination of `com::sun::star::awt::MessageBoxButtons `_
:type buttons: long
:param type_message_box: The `message box type `_
:type type_message_box: enum
:return: `MessageBoxResult `_
:rtype: int
`See Api XMessageBoxFactory `_
"""
toolkit = create_instance('com.sun.star.awt.Toolkit')
parent = toolkit.getDesktopWindow()
box = toolkit.createMessageBox(parent, type_message_box, buttons, title, str(message))
return box.execute()
def question(message: str, title: str=TITLE) -> bool:
"""Create message box question, show buttons YES and NO
:param message: Message question
:type message: str
:param title: The title for message box
:type title: str
:return: True if user click YES and False if click NO
:rtype: bool
"""
result = msgbox(message, title, MSG_BUTTONS.BUTTONS_YES_NO, MessageBoxType.QUERYBOX)
return result == YES
def warning(message: Any, title: str=TITLE) -> int:
"""Create message box with icon warning
:param message: Any type message, all is converted to string.
:type message: Any
:param title: The title for message box
:type title: str
:return: MessageBoxResult
:rtype: int
"""
return msgbox(message, title, type_message_box=MessageBoxType.WARNINGBOX)
def errorbox(message: Any, title: str=TITLE) -> int:
"""Create message box with icon error
:param message: Any type message, all is converted to string.
:type message: Any
:param title: The title for message box
:type title: str
:return: MessageBoxResult
:rtype: int
"""
return msgbox(message, title, type_message_box=MessageBoxType.ERRORBOX)
def sleep(seconds: int):
"""Sleep
"""
time.sleep(seconds)
return
def run_in_thread(fn):
"""Run any function in thread
:param fn: Any Python function (macro)
:type fn: Function instance
"""
def run(*k, **kw):
t = threading.Thread(target=fn, args=k, kwargs=kw)
t.start()
return t
return run
def dict_to_property(values: dict, uno_any: bool=False):
"""Convert dictionary to array of PropertyValue
:param values: Dictionary of values
:type values: dict
:param uno_any: If return like array uno.Any
:type uno_any: bool
:return: Tuple of PropertyValue or array uno.Any
:rtype: tuples or uno.Any
"""
ps = tuple([PropertyValue(Name=n, Value=v) for n, v in values.items()])
if uno_any:
ps = uno.Any('[]com.sun.star.beans.PropertyValue', ps)
return ps
def _property_to_dict(values):
d = {v.Name: v.Value for v in values}
return d
def data_to_dict(data) -> dict:
"""Convert tuples, list, PropertyValue, NamedValue to dictionary
:param data: Dictionary of values
:type data: array of tuples, list, PropertyValue or NamedValue
:return: Dictionary
:rtype: dict
"""
d = {}
if not isinstance(data, (tuple, list)):
return d
if isinstance(data[0], (tuple, list)):
d = {r[0]: r[1] for r in data}
elif isinstance(data[0], (PropertyValue, NamedValue)):
d = _property_to_dict(data)
return d
def render(template, data):
s = Template(template)
return s.safe_substitute(**data)
# Classes
class _classproperty:
def __init__(self, method=None):
self.fget = method
def __get__(self, instance, cls=None):
return self.fget(cls)
def getter(self, method):
self.fget = method
return self
class Dates(object):
"""Class for datetimes
"""
_start = None
@_classproperty
def now(cls):
"""Current local date and time
:return: Return the current local date and time
:rtype: datetime
"""
return datetime.datetime.now().replace(microsecond=0)
@_classproperty
def today(cls):
"""Current local date
:return: Return the current local date
:rtype: date
"""
return datetime.date.today()
@_classproperty
def time(cls):
"""Current local time
:return: Return the current local time
:rtype: datetime.time
"""
t = cls.now.time().replace(microsecond=0)
return t
@_classproperty
def epoch(cls):
"""Get unix time
:return: Return unix time
:rtype: int
`See Unix Time `_
"""
n = cls.now
e = int(time.mktime(n.timetuple()))
return e
@classmethod
def date(cls, year: int, month: int, day: int):
"""Get date from year, month, day
:param year: Year of date
:type year: int
:param month: Month of date
:type month: int
:param day: Day of day
:type day: int
:return: Return the date
:rtype: date
`See Python date `_
"""
d = datetime.date(year, month, day)
return d
@classmethod
def str_to_date(cls, str_date: str, template: str, to_calc: bool=False):
"""Get date from string
:param str_date: Date in string
:type str_date: str
:param template: Formato of date string
:type template: str
:param to_calc: If date is for used in Calc cell
:type to_calc: bool
:return: Return date or int if used in Calc
:rtype: date or int
`See Python strptime `_
"""
d = datetime.datetime.strptime(str_date, template).date()
if to_calc:
d = d.toordinal() - DATE_OFFSET
return d
@classmethod
def calc_to_date(cls, value: float):
"""Get date from calc value
:param value: Float value from cell
:type value: float
:return: Return the current local date
:rtype: date
`See Python fromordinal `_
"""
d = datetime.date.fromordinal(int(value) + DATE_OFFSET)
return d
@classmethod
def start(cls):
"""Start counter
"""
cls._start = cls.now
info('Start: ', cls._start)
return
@classmethod
def end(cls, get_seconds: bool=True):
"""End counter
:param get_seconds: If return value in total seconds
:type get_seconds: bool
:return: Return the timedelta or total seconds
:rtype: timedelta or int
"""
e = cls.now
td = e - cls._start
result = str(td)
if get_seconds:
result = td.total_seconds()
info('End: ', e)
return result
class Json(object):
"""Class for json data
"""
@classmethod
def dumps(cls, data: Any) -> str:
"""Dumps
:param data: Any data
:type data: Any
:return: Return string json
:rtype: str
"""
return json.dumps(data, indent=4, sort_keys=True)
@classmethod
def loads(cls, data: str) -> Any:
"""Loads
:param data: String data
:type data: str
:return: Return any object
:rtype: Any
"""
return json.loads(data)
class Macro(object):
"""Class for call macro
`See Scripting Framework `_
"""
@classmethod
def call(cls, args: dict, in_thread: bool=False):
"""Call any macro
:param args: Dictionary with macro location
:type args: dict
:param in_thread: If execute in thread
:type in_thread: bool
:return: Return None or result of call macro
:rtype: Any
"""
result = None
if in_thread:
t = threading.Thread(target=cls._call, args=(args,))
t.start()
else:
result = cls._call(args)
return result
@classmethod
def _get_url_script(cls, args: dict):
library = args['library']
name = args['name']
language = args.get('language', 'Python')
location = args.get('location', 'user')
module = args.get('module', '.')
if language == 'Python':
module = '.py$'
elif language == 'Basic':
module = f".{module}."
if location == 'user':
location = 'application'
url = 'vnd.sun.star.script'
url = f'{url}:{library}{module}{name}?language={language}&location={location}'
return url
@classmethod
def _call(cls, args: dict):
url = cls._get_url_script(args)
args = args.get('args', ())
service = 'com.sun.star.script.provider.MasterScriptProviderFactory'
factory = create_instance(service)
script = factory.createScriptProvider('').getScript(url)
result = script.invoke(args, None, None)[0]
return result
class Shell(object):
"""Class for subprocess
`See Subprocess `_
"""
@classmethod
def run(cls, command, capture=False, split=False):
"""Execute commands
:param command: Command to run
:type command: str
:param capture: If capture result of command
:type capture: bool
:param split: Some commands need split.
:type split: bool
:return: Result of command
:rtype: Any
"""
if split:
cmd = shlex.split(command)
result = subprocess.run(cmd, capture_output=capture, text=True, shell=IS_WIN)
if capture:
result = result.stdout
else:
result = result.returncode
else:
if capture:
result = subprocess.check_output(command, shell=True).decode()
else:
result = subprocess.Popen(command)
return result
@classmethod
def popen(cls, command):
"""Execute commands and return line by line
:param command: Command to run
:type command: str
:return: Result of command
:rtype: Any
"""
try:
proc = subprocess.Popen(shlex.split(command), shell=IS_WIN,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
for line in proc.stdout:
yield line.decode().rstrip()
except Exception as e:
error(e)
yield (e.errno, e.strerror)
class Timer(object):
"""Class for timer thread"""
class TimerThread(threading.Thread):
def __init__(self, event, seconds, macro):
threading.Thread.__init__(self)
self._event = event
self._seconds = seconds
self._macro = macro
def run(self):
while not self._event.wait(self._seconds):
Macro.call(self._macro)
info('\tTimer stopped... ')
return
@classmethod
def exists(cls, name):
"""Validate in timer **name** exists
:param name: Timer name, it must be unique
:type name: str
:return: True if exists timer name
:rtype: bool
"""
global _EVENTS
return name in _EVENTS
@classmethod
def start(cls, name: str, seconds: float, macro: dict):
"""Start timer **name** every **seconds** and execute **macro**
:param name: Timer name, it must be unique
:type name: str
:param seconds: Seconds for wait
:type seconds: float
:param macro: Macro for execute
:type macro: dict
"""
global _EVENTS
_EVENTS[name] = threading.Event()
info(f"Timer '{name}' started, execute macro: '{macro['name']}'")
thread = cls.TimerThread(_EVENTS[name], seconds, macro)
thread.start()
return
@classmethod
def stop(cls, name: str):
"""Stop timer **name**
:param name: Timer name
:type name: str
"""
global _EVENTS
_EVENTS[name].set()
del _EVENTS[name]
return
@classmethod
def once(cls, name: str, seconds: float, macro: dict):
"""Start timer **name** only once in **seconds** and execute **macro**
:param name: Timer name, it must be unique
:type name: str
:param seconds: Seconds for wait before execute macro
:type seconds: float
:param macro: Macro for execute
:type macro: dict
"""
global _EVENTS
_EVENTS[name] = threading.Timer(seconds, Macro.call, (macro,))
_EVENTS[name].start()
info(f'Event: "{name}", started... execute in {seconds} seconds')
return
@classmethod
def cancel(cls, name: str):
"""Cancel timer **name** only once events.
:param name: Timer name, it must be unique
:type name: str
"""
global _EVENTS
if name in _EVENTS:
try:
_EVENTS[name].cancel()
del _EVENTS[name]
info(f'Cancel event: "{name}", ok...')
except Exception as e:
error(e)
else:
debug(f'Cancel event: "{name}", not exists...')
return
class Hash(object):
"""Class for hash
"""
@classmethod
def digest(cls, method: str, data: str, in_hex: bool=True):
"""Get digest from data with method
:param method: Digest method: md5, sha1, sha256, sha512, etc...
:type method: str
:param data: Data for get digest
:type data: str
:param in_hex: If True, get digest in hexadecimal, if False, get bytes
:type in_hex: bool
:return: bytes or hex digest
:rtype: bytes or str
"""
result = ''
obj = getattr(hashlib, method)(data.encode())
if in_hex:
result = obj.hexdigest()
else:
result = obj.digest()
return result
class Paths(object):
"""Class for paths
"""
FILE_PICKER = 'com.sun.star.ui.dialogs.FilePicker'
FOLDER_PICKER = 'com.sun.star.ui.dialogs.FolderPicker'
REMOTE_FILE_PICKER = 'com.sun.star.ui.dialogs.RemoteFilePicker'
OFFICE_FILE_PICKER = 'com.sun.star.ui.dialogs.OfficeFilePicker'
def __init__(self, path=''):
if path.startswith('file://'):
path = str(Path(uno.fileUrlToSystemPath(path)).resolve())
self._path = Path(path)
@property
def path(self):
"""Get base path"""
return str(self._path.parent)
@property
def file_name(self):
"""Get file name"""
return self._path.name
@property
def name(self):
"""Get name"""
return self._path.stem
@property
def ext(self):
"""Get extension"""
return self._path.suffix[1:]
@property
def size(self):
"""Get size"""
return self._path.stat().st_size
@property
def url(self):
"""Get like URL"""
return self._path.as_uri()
@property
def info(self):
"""Get all info like tuple"""
i = (self.path, self.file_name, self.name, self.ext, self.size, self.url)
return i
@property
def dict(self):
"""Get all info like dict"""
data = {
'path': self.path,
'file_name': self.file_name,
'name': self.name,
'ext': self.ext,
'size': self.size,
'url': self.url,
}
return data
@_classproperty
def home(self):
"""Get user home"""
return str(Path.home())
@_classproperty
def documents(self):
"""Get user save documents"""
return self.config()
@_classproperty
def user_profile(self):
"""Get path user profile"""
path = self.config('UserConfig')
path = str(Path(path).parent)
return path
@_classproperty
def user_config(self):
"""Get path config in user profile"""
path = self.config('UserConfig')
return path
@_classproperty
def python(self):
"""Get path executable python"""
if IS_WIN:
path = self.join(self.config('Module'), PYTHON)
elif IS_MAC:
path = self.join(self.config('Module'), '..', 'Resources', PYTHON)
else:
path = sys.executable
return path
@classmethod
def to_system(cls, path:str) -> str:
"""Convert paths in URL to system
:param path: Path to convert
:type path: str
:return: Path system format
:rtype: str
"""
if path.startswith('file://'):
path = str(Path(uno.fileUrlToSystemPath(path)).resolve())
return path
@classmethod
def to_url(cls, path: str) -> str:
"""Convert paths in format system to URL
:param path: Path to convert
:type path: str
:return: Path in URL
:rtype: str
"""
if not path.startswith('file://'):
path = Path(path).as_uri()
return path
@classmethod
def config(cls, name: str='Work') -> Union[str, list]:
"""Return path from config
:param name: Name in service PathSettings, default get path documents
:type name: str
:return: Path in config, if exists.
:rtype: str or list
`See Api XPathSettings `_
"""
path = create_instance('com.sun.star.util.PathSettings')
path = cls.to_system(getattr(path, name)).split(';')
if len(path) == 1:
path = path[0]
return path
@classmethod
def join(cls, *paths: str) -> str:
"""Join paths
:param paths: Paths to join
:type paths: list
:return: New path with joins
:rtype: str
"""
path = str(Path(paths[0]).joinpath(*paths[1:]))
return path
@classmethod
def exists(cls, path: str) -> bool:
"""If exists path
:param path: Path for validate
:type path: str
:return: True if path exists, False if not.
:rtype: bool
"""
path = cls.to_system(path)
result = Path(path).exists()
return result
@classmethod
def exists_app(cls, name_app: str) -> bool:
"""If exists app in system
:param name_app: Name of application
:type name_app: str
:return: True if app exists, False if not.
:rtype: bool
"""
result = bool(shutil.which(name_app))
return result
@classmethod
def temp_file(self):
"""Make temporary file"""
return tempfile.NamedTemporaryFile(mode='w')
@classmethod
def temp_dir(self):
"""Make temporary directory"""
return tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
@classmethod
def get(cls, init_dir: str='', filters: str='') -> str:
"""Get path for save
:param init_dir: Initial default path
:type init_dir: str
:param filters: Filter for show type files: 'xml' or 'txt,xml'
:type filters: str
:return: Selected path
:rtype: str
`See API `_
"""
if not init_dir:
init_dir = cls.documents
init_dir = cls.to_url(init_dir)
file_picker = create_instance(cls.FILE_PICKER)
file_picker.setTitle(_('Select path'))
file_picker.setDisplayDirectory(init_dir)
file_picker.initialize((TemplateDescription.FILEOPEN_SIMPLE,))
if filters:
for f in filters.split(','):
file_picker.appendFilter(f.upper(), f'*.{f.lower()}')
path = ''
if file_picker.execute():
path = cls.to_system(file_picker.getSelectedFiles()[0])
return path
@classmethod
def get_dir(cls, init_dir: str='') -> str:
"""Get path dir
:param init_dir: Initial default path
:type init_dir: str
:param filters: Filter for show type files: 'xml' or 'txt,xml'
:type filters: str
:return: Selected path
:rtype: str
"""
folder_picker = create_instance(cls.FOLDER_PICKER)
if not init_dir:
init_dir = cls.documents
init_dir = cls.to_url(init_dir)
folder_picker.setTitle(_('Select directory'))
folder_picker.setDisplayDirectory(init_dir)
path = ''
if folder_picker.execute():
path = cls.to_system(folder_picker.getDirectory())
return path
@classmethod
def get_file(cls, init_dir: str='', filters: str='', multiple: bool=False):
"""
Get path file
init_folder: folder default open
filters: 'xml' or 'xml,txt'
multiple: True for multiple selected
"""
if not init_dir:
init_dir = cls.documents
init_dir = cls.to_url(init_dir)
file_picker = create_instance(cls.FILE_PICKER)
file_picker.setTitle(_('Select file'))
file_picker.setDisplayDirectory(init_dir)
file_picker.setMultiSelectionMode(multiple)
if filters:
filters = [(f.upper(), f'*.{f.lower()}') for f in filters.split(',')]
file_picker.setCurrentFilter(filters[0][0])
for f in filters:
file_picker.appendFilter(f[0], f[1])
path = ''
if file_picker.execute():
files = file_picker.getSelectedFiles()
path = [cls.to_system(f) for f in files]
if not multiple:
path = path[0]
return path
# ~ Save/read data
@classmethod
def save(cls, path: str, data: str, encoding: str='utf-8') -> bool:
"""Save data in path with encoding
:param path: Path to file save
:type path: str
:param data: Data to save
:type data: str
:param encoding: Encoding for save data, default utf-8
:type encoding: str
:return: True, if save corrrectly
:rtype: bool
"""
result = bool(Path(path).write_text(data, encoding=encoding))
return result
@classmethod
def save_bin(cls, path: str, data: bytes) -> bool:
"""Save binary data in path
:param path: Path to file save
:type path: str
:param data: Data to save
:type data: bytes
:return: True, if save corrrectly
:rtype: bool
"""
result = bool(Path(path).write_bytes(data))
return result
@classmethod
def read(cls, path: str, get_lines: bool=False, encoding: str='utf-8') -> Union[str, list]:
"""Read data in path
:param path: Path to file read
:type path: str
:param get_lines: If read file line by line
:type get_lines: bool
:return: File content
:rtype: str or list
"""
if get_lines:
with Path(path).open(encoding=encoding) as f:
data = f.readlines()
else:
data = Path(path).read_text(encoding=encoding)
return data
@classmethod
def read_bin(cls, path: str) -> bytes:
"""Read binary data in path
:param path: Path to file read
:type path: str
:return: File content
:rtype: bytes
"""
data = Path(path).read_bytes()
return data
# ~ Import/export data
@classmethod
def from_json(cls, path: str) -> Any:
"""Read path file and load json data
:param path: Path to file
:type path: str
:return: Any data
:rtype: Any
"""
data = json.loads(cls.read(path))
return data
@classmethod
def to_json(cls, path: str, data: str):
"""Save data in path file like json
:param path: Path to file
:type path: str
:return: True if save correctly
:rtype: bool
"""
data = json.dumps(data, indent=4, ensure_ascii=False, sort_keys=True)
return cls.save(path, data)
class Config(object):
"""Class for set and get configurations
"""
@classmethod
def set(cls, prefix: str, value: Any, key: str='') -> bool:
"""Save data config in user config like json
:param prefix: Unique prefix for this data
:type prefix: str
:param value: Value for save
:type value: Any
:param key: Key for value
:type key: str
:return: True if save correctly
:rtype: bool
"""
name_file = FILES['CONFIG'].format(prefix)
path = Paths.join(Paths.user_config, name_file)
data = value
if key:
data = cls.get(prefix)
data[key] = value
result = Paths.to_json(path, data)
return result
@classmethod
def get(cls, prefix: str, key: str='', default: Any={}) -> Any:
"""Get data config from user config like json
:param prefix: Unique prefix for this data
:type prefix: str
:param key: Key for value
:type key: str
:param default: Get if not exists key
:type default: Any
:return: data
:rtype: Any
"""
data = {}
name_file = FILES['CONFIG'].format(prefix)
path = Paths.join(Paths.user_config, name_file)
if not Paths.exists(path):
return data
data = Paths.from_json(path)
if key:
data = data.get(key, default)
return data
class Url(object):
"""Class for simple url open
"""
@classmethod
def _open(cls, url: str, data: Any=None, headers: dict={}, verify: bool=True, \
json: bool=False, timeout: int=TIMEOUT, method: str='GET') -> tuple:
"""URL Open"""
debug(url)
result = None
context = None
rheaders = {}
err = ''
if verify:
if not data is None and isinstance(data, str):
data = data.encode()
else:
context = ssl._create_unverified_context()
try:
req = Request(url, data=data, headers=headers, method=method)
response = urlopen(req, timeout=timeout, context=context)
except HTTPError as e:
error(e)
err = str(e)
except URLError as e:
error(e.reason)
err = str(e.reason)
# ToDo
# ~ except timeout:
# ~ err = 'timeout'
# ~ error(err)
else:
rheaders = dict(response.info())
result = response.read().decode()
if json:
result = Json.loads(result)
return result, rheaders, err
@classmethod
def get(cls, url: str, data: Any=None, headers: dict={}, verify: bool=True, \
json: bool=False, timeout: int=TIMEOUT) -> tuple:
"""Method GET
:param url: Url to open
:type url: str
:return: result, headers and error
:rtype: tuple
"""
return cls._open(url, data, headers, verify, json, timeout)
# ToDo
@classmethod
def _post(cls, url: str, data: Any=None, headers: dict={}, verify: bool=True, \
json: bool=False, timeout: int=TIMEOUT) -> tuple:
"""Method POST
"""
data = parse.urlencode(data).encode('ascii')
return cls._open(url, data, headers, verify, json, timeout, 'POST')
class Color(object):
"""Class for colors
`See Web Colors `_
"""
COLORS = {
'aliceblue': 15792383,
'antiquewhite': 16444375,
'aqua': 65535,
'aquamarine': 8388564,
'azure': 15794175,
'beige': 16119260,
'bisque': 16770244,
'black': 0,
'blanchedalmond': 16772045,
'blue': 255,
'blueviolet': 9055202,
'brown': 10824234,
'burlywood': 14596231,
'cadetblue': 6266528,
'chartreuse': 8388352,
'chocolate': 13789470,
'coral': 16744272,
'cornflowerblue': 6591981,
'cornsilk': 16775388,
'crimson': 14423100,
'cyan': 65535,
'darkblue': 139,
'darkcyan': 35723,
'darkgoldenrod': 12092939,
'darkgray': 11119017,
'darkgreen': 25600,
'darkgrey': 11119017,
'darkkhaki': 12433259,
'darkmagenta': 9109643,
'darkolivegreen': 5597999,
'darkorange': 16747520,
'darkorchid': 10040012,
'darkred': 9109504,
'darksalmon': 15308410,
'darkseagreen': 9419919,
'darkslateblue': 4734347,
'darkslategray': 3100495,
'darkslategrey': 3100495,
'darkturquoise': 52945,
'darkviolet': 9699539,
'deeppink': 16716947,
'deepskyblue': 49151,
'dimgray': 6908265,
'dimgrey': 6908265,
'dodgerblue': 2003199,
'firebrick': 11674146,
'floralwhite': 16775920,
'forestgreen': 2263842,
'fuchsia': 16711935,
'gainsboro': 14474460,
'ghostwhite': 16316671,
'gold': 16766720,
'goldenrod': 14329120,
'gray': 8421504,
'grey': 8421504,
'green': 32768,
'greenyellow': 11403055,
'honeydew': 15794160,
'hotpink': 16738740,
'indianred': 13458524,
'indigo': 4915330,
'ivory': 16777200,
'khaki': 15787660,
'lavender': 15132410,
'lavenderblush': 16773365,
'lawngreen': 8190976,
'lemonchiffon': 16775885,
'lightblue': 11393254,
'lightcoral': 15761536,
'lightcyan': 14745599,
'lightgoldenrodyellow': 16448210,
'lightgray': 13882323,
'lightgreen': 9498256,
'lightgrey': 13882323,
'lightpink': 16758465,
'lightsalmon': 16752762,
'lightseagreen': 2142890,
'lightskyblue': 8900346,
'lightslategray': 7833753,
'lightslategrey': 7833753,
'lightsteelblue': 11584734,
'lightyellow': 16777184,
'lime': 65280,
'limegreen': 3329330,
'linen': 16445670,
'magenta': 16711935,
'maroon': 8388608,
'mediumaquamarine': 6737322,
'mediumblue': 205,
'mediumorchid': 12211667,
'mediumpurple': 9662683,
'mediumseagreen': 3978097,
'mediumslateblue': 8087790,
'mediumspringgreen': 64154,
'mediumturquoise': 4772300,
'mediumvioletred': 13047173,
'midnightblue': 1644912,
'mintcream': 16121850,
'mistyrose': 16770273,
'moccasin': 16770229,
'navajowhite': 16768685,
'navy': 128,
'oldlace': 16643558,
'olive': 8421376,
'olivedrab': 7048739,
'orange': 16753920,
'orangered': 16729344,
'orchid': 14315734,
'palegoldenrod': 15657130,
'palegreen': 10025880,
'paleturquoise': 11529966,
'palevioletred': 14381203,
'papayawhip': 16773077,
'peachpuff': 16767673,
'peru': 13468991,
'pink': 16761035,
'plum': 14524637,
'powderblue': 11591910,
'purple': 8388736,
'red': 16711680,
'rosybrown': 12357519,
'royalblue': 4286945,
'saddlebrown': 9127187,
'salmon': 16416882,
'sandybrown': 16032864,
'seagreen': 3050327,
'seashell': 16774638,
'sienna': 10506797,
'silver': 12632256,
'skyblue': 8900331,
'slateblue': 6970061,
'slategray': 7372944,
'slategrey': 7372944,
'snow': 16775930,
'springgreen': 65407,
'steelblue': 4620980,
'tan': 13808780,
'teal': 32896,
'thistle': 14204888,
'tomato': 16737095,
'turquoise': 4251856,
'violet': 15631086,
'wheat': 16113331,
'white': 16777215,
'whitesmoke': 16119285,
'yellow': 16776960,
'yellowgreen': 10145074,
}
def _get_color(self, index):
if isinstance(index, tuple):
color = (index[0] << 16) + (index[1] << 8) + index[2]
else:
if index[0] == '#':
r, g, b = bytes.fromhex(index[1:])
color = (r << 16) + (g << 8) + b
else:
color = self.COLORS.get(index.lower(), -1)
return color
def __call__(self, index):
return self._get_color(index)
def __getitem__(self, index):
return self._get_color(index)
COLOR_ON_FOCUS = Color()('LightYellow')
def __getattr__(name):
classes = {
'dates': Dates,
'json': Json,
'macro': Macro,
'shell': Shell,
'timer': Timer,
'hash': Hash,
'path': Paths,
'config': Config,
'url': Url,
'color': Color(),
}
if name in classes:
return classes[name]
raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
class LOServer(object):
"""Started LibeOffice like server
"""
HOST = 'localhost'
PORT = '8100'
ARG = f'socket,host={HOST},port={PORT};urp;StarOffice.ComponentContext'
CMD = ['soffice',
'-env:SingleAppInstance=false',
'-env:UserInstallation=file:///tmp/LO_Process8100',
'--headless', '--norestore', '--invisible',
f'--accept={ARG}']
def __init__(self):
self._server = None
self._ctx = None
self._sm = None
self._start_server()
self._init_values()
def _init_values(self):
global CTX
global SM
if not self.is_running:
return
ctx = uno.getComponentContext()
service = 'com.sun.star.bridge.UnoUrlResolver'
resolver = ctx.ServiceManager.createInstanceWithContext(service, ctx)
self._ctx = resolver.resolve('uno:{}'.format(self.ARG))
self._sm = self._ctx.getServiceManager()
CTX = self._ctx
SM = self._sm
return
@property
def is_running(self):
try:
s = socket.create_connection((self.HOST, self.PORT), 5.0)
s.close()
debug('LibreOffice is running...')
return True
except ConnectionRefusedError:
return False
def _start_server(self):
if self.is_running:
return
for i in range(3):
self._server = subprocess.Popen(self.CMD,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
time.sleep(3)
if self.is_running:
break
return
def stop(self):
"""Stop server
"""
if self._server is None:
print('Search pgrep soffice')
else:
self._server.terminate()
debug('LibreOffice is stop...')
return
def _create_instance(self, name, with_context=True):
if with_context:
instance = self._sm.createInstanceWithContext(name, self._ctx)
else:
instance = self._sm.createInstance(name)
return instance