diff --git a/cctv-scheduler.py b/cctv-scheduler.py index 383820b..39f0e13 100644 --- a/cctv-scheduler.py +++ b/cctv-scheduler.py @@ -5,6 +5,7 @@ import base64 import json import logging +import re import urllib.request from argparse import ArgumentParser from datetime import datetime @@ -1043,6 +1044,272 @@ class Sensor(Connect): return self.__temperature(nodename=self._node) +class Wordpress(Connect): + """Set of methods (functions) for Wordpress API. + Reference: https://developer.wordpress.org/rest-api/reference/ + + Args: + Connect (_type_): class with 'http' method. + """ + def __init__( + self, + hostname: str, + username: str, + password: str + ): + """Object constructor. + + Args: + hostname (str, optional): www.wordpress.site. + username (str, optional): wordpress username. + password (str, optional): wordpress passwrod. + """ + if Do.args_valid(locals(), self.__init__.__annotations__): + self._host = hostname + self._user = username + self._pass = password + self.api_event = 'https://' + self._host + '/wp-json/tribe/events/v1/events' + self.api_media = 'https://' + self._host + '/wp-json/wp/v2/media' + self.api_pages = 'https://' + self._host + '/wp-json/wp/v2/pages' + self.url_files = 'https://' + self._host + '/wp-content/uploads' + + def event_create( + self, + title: str, + slug: str, + date_start: str, + date_end: str, + date_publish: str = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S"), + all_day: bool = True, + description: str = None + ) -> dict: + """Create event by 'wp-json' and 'The Events Calendar'. + + Args: + title (str, optional): event title. + slug (str, optional): event slug. + date_start (str, optional): '2022-12-31T23:59:59' format. + date_end (str, optional): '2022-12-31T23:59:59' format. + date_publish (_type_, optional): '2022-12-31T23:59:59' format. Defaults to now. + all_day (bool, optional): all day event duration flag. Defaults to True. + description (str, optional): event body. Defaults to None. + + Raises: + ValueError: date formate is not 2022-12-31T23:59:59. + ValueError: description can't be empty. + + Returns: + dict: {'success':bool,'result':'http/url/to/event'}. + """ + if Do.args_valid(locals(), self.event_create.__annotations__): + pattern = "^([0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2})+$" + if ( + not re.fullmatch(pattern, date_start) or + not re.fullmatch(pattern, date_end) or + not re.fullmatch(pattern, date_publish) + ): + raise ValueError("date formate is not 2022-12-31T23:59:59") + if description == '': + raise ValueError("description can't be empty") + + event_json = { + "title": title, + "slug": slug, + "date": date_publish, + "start_date": date_start, + "end_date": date_end, + "all_day": str(all_day), + "description": description + } + response = self.http( + url=self.api_event, + method='POST', + username=self._user, + password=self._pass, + authtype='basic', + contenttype='application/json; charset=UTF-8', + contentdata=json.dumps(event_json, indent=2) + ) + logging.debug( + msg="" + + "\n" + "event API response: " + + "\n" + json.dumps(json.loads(response['result']), indent=2) + ) + if response['success']: + for key, val in json.loads(response['result']).items(): + if key == "url": + logging.info('event created: %s', val) + return {"success": True, "result": val} + else: + logging.warning("event didn't create") + return {"success": False, "result": "ERROR"} + + def media_search( + self, + media_name: (str, type(None)) = None, + media_type: (str, type(None)) = None + ) -> dict: + """Search uploaded media by 'wp-json'. + + Args: + media_name (str, type, optional): results matching a string. Defaults to None. + media_type (str, type, optional): application,image,video,audio,text. Defaults to None. + + Returns: + dict: {'success':bool,'result':['list/of/link/to/media']} + """ + if Do.args_valid(locals(), self.media_search.__annotations__): + url = self.api_media + '?per_page=100' + if media_name: + url = url + '&search=' + media_name + if (media_type == 'application' or + media_type == 'image' or + media_type == 'video' or + media_type == 'audio' or + media_type == 'text' + ): + url = url + '&media_type=' + media_type + + media_list = [] + response = self.http(url=url, method='GET') + logging.debug( + msg="" + + "\n" + "media API response: " + + "\n" + json.dumps(json.loads(response['result']), indent=2) + ) + if response['success']: + for media in json.loads(response['result']): + media_list.append(media['guid']['rendered']) + return {"success": True, "result": media_list} + else: + logging.warning("media didn't list") + return {"success": False, "result": "ERROR"} + + def media_upload( + self, + mediafile: str, + mediatype: str, + aliasfile: str = '' + ) -> dict: + """Upload media by 'wp-json'. + + Args: + mediafile (str, optional): path to file. + mediatype (str, optional): 'image/jpeg', 'video/mp4', etc. + aliasfile (str, optional): uploaded media name. Defaults to original file. + + Raises: + FileExistsError: mediafile is not exist. + + Returns: + dict: {'success':bool,'result':'http/url/to/media'}. + """ + if Do.args_valid(locals(), self.media_upload.__annotations__): + if not path.exists(mediafile): + raise FileExistsError(mediafile + " is not exist") + else: + with open(mediafile, mode='rb') as file: + mediadata = file.read() + if aliasfile == '': + aliasfile = path.basename(mediafile) + + response = self.http( + url=self.api_media, + method='POST', + username=self._user, + password=self._pass, + authtype='basic', + contenttype=mediatype, + contentdata=mediadata, + headers={ + "Accept": "application/json", + 'Content-Disposition': 'attachment; filename=' + aliasfile, + 'Cache-Control': 'no-cache' + } + ) + logging.debug( + msg="" + + "\n" + "media API response: " + + "\n" + json.dumps(json.loads(response['result']), indent=2) + ) + if response['success']: + for key, val in json.loads(response['result']).items(): + if key == "source_url": + logging.info('media uploaded: %s', val) + return {"success": True, "result": val} + else: + logging.warning("media didn't upload") + return {"success": False, "result": "ERROR"} + + def pages_read( + self, + page_id: int + ) -> dict: + """Read page by 'wp-json'. + + Args: + page_id (int): unique identifier for the page. + + Returns: + dict: {'success':bool,'result':'page data'} + """ + if Do.args_valid(locals(), self.pages_read.__annotations__): + page_link = self.api_pages + '/' + str(page_id) + response = self.http(url=page_link) + if response['success']: + logging.debug( + msg="" + + "\n" + "wp page API response: " + + "\n" + json.dumps(json.loads(response['result']), indent=2) + ) + return {"success": True, "result": response['result']} + else: + logging.warning("wp page didn't read") + return {"success": False, "result": "ERROR"} + + def pages_update( + self, + page_id: int, + content: str + ) -> dict: + """Update page by 'wp-json'. + + Args: + page_id (int): unique identifier for the page. + content (str): the content for the page. + + Returns: + dict: {'success':bool,'result':'http/url/to/page'} + """ + if Do.args_valid(locals(), self.pages_update.__annotations__): + page_link = self.api_pages + '/' + str(page_id) + page_json = { + "content": content + } + response = self.http( + url=page_link, + method='POST', + username=self._user, + password=self._pass, + authtype='basic', + contenttype='application/json; charset=UTF-8', + contentdata=json.dumps(page_json) + ) + logging.debug( + msg="" + + "\n" + "wp page API response: " + + "\n" + json.dumps(json.loads(response['result']), indent=2) + ) + if response['success']: + for key, val in json.loads(response['result']).items(): + if key == "link": + logging.info(msg="wp page " + str(page_id) + " updated: " + val) + return {"success": True, "result": val} + else: + logging.warning("wp page didn't update") + return {"success": False, "result": "ERROR"} + + class Sequence: """Sequence handling. """