generated from pavel.muhortov/template-bash
add Worpress() class
This commit is contained in:
parent
699492f77f
commit
0af51a29b6
|
|
@ -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.
|
||||
"""
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user