"""
sheepuri.py
URI method handler classes for the sheep branch of the URI tree.
Copyright PR Hardman 2009 - 2023.
This program 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.
This program 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 this program. If not, see <https://www.gnu.org/licenses/>.
"""
import os
import json
import cherrypy
from cherrypy.lib.static import serve_file
from ppdb import const
from ppdb.handlers import uribase
from ppdb.lib import (util, perslib, sheeplib, shpupd, flocklib, transfer, memlib, fbklib,
insplib, userlib)
xhrc = const.JSConsts()
#~ DEBUG = True
DEBUG = False
class SheepBase(uribase.UriBase):
""" Provides generic methods for the 'sheep' branch of the URI tree.
Instance variables are shared by all threads. Therefore method
parameters which vary by thread may not use instance variables.
This class should not be instantiated itself, but must be subclassed.
"""
def __init__(self, config, path_conf=None):
""" Default constructor. """
super().__init__(config, path_conf)
self.prog_limit = config.get('progeny-limit', 3)
self.sheep_sort_params = list(sheeplib.SHEEP_SORT_TRANSLATE.keys())
@cherrypy.expose
class Sheep(SheepBase):
""" Sheep root class for the Database. """
def GET(self, rid=None, start=0, step=20, sortby='regno', sortdir='asc'):
"""
- If the looged in user is not an 'officer' and is a 'breeder' then the sheep
displayed in thes form will be limited to those bred by, or registered by the
the breeder or currently owned by the breeder.
- A GET with no rid or an unresolvable rid wil return the sheep data page
displaying the most recently registered sheep for the Database.
- A GET that resolves to a single sheep wil return the sheep's data page.
- A GET that resolves to multiple sheep will return the Lookup Sheep page.
- If the summary page is returned then:
- If specified, 'sortby' must be in self.sheep_sort_params
- If specified, 'sortdir' must be 'asc' (ascending) or 'desc' (descending).
"""
if DEBUG:
print(f"rid: {rid}, sortby: {sortby}, sortdir: {sortdir}, "
"accept: {cherrypy.request.headers.get('Accept', '')}")
dlg_msg = ""
show_fndtn = bool('admin' in cherrypy.request.loginroles or
'regsec' in cherrypy.request.loginroles)
if not rid:
rid = sheeplib.last_sheep()
# 'result' is a message or else a list of registration numbers matching 'rid'.
result = sheeplib.find_sheep(rid, sortdir)
if not isinstance(result, list):
# Some sort of error or unexpected case.
dlg_msg = result
result = [sheeplib.last_sheep()]
if len(result) > 1:
dlg_msg = (f"More than one sheep found matching '{rid}'.<br>"
"Only the first sheep is shown. "
"Use the search facility on this page to see all the results.")
# Display a data entry form
regn_no = result[0]
page_data = sheeplib.get_details_page_data(regn_no)
page_data['markings'] = sheeplib.get_recognised_markings(separators=True)
page_data['colours'] = [[row["colour_name"], row["description"]]
for row in sheeplib.get_recognised_colours()]
page_info = self.html_appconf_items()
page_info.update({'requestpath': cherrypy.request.path_info,
'rid': regn_no,
'hdr_nav': ''})
# Insert the sheep data into the page_info dict
page_info.update(page_data)
if page_data['registering_flock'] == page_data['originating_flock']:
page_info['flock_no'] = page_data['registering_flock']
page_info['flock_name'] = page_data['regn_flock_name']
page_info['flock_text'] = "Born in flock"
else:
page_info['flock_no'] = page_data['originating_flock']
page_info['flock_name'] = page_data['org_flock_name']
page_info['flock_text'] = "Registered in flock"
page_info['pagetitle'] = f"{page_info['public_no']} {page_info['full_name']}"
page_info['titlebar'] = page_info['pagetitle']
# Disable tag changes if the transfer state is dead or unknown, or if the owning
# flock has no prefixes
if page_data['xfer_state'] == const.XFER_STATE_ALIVE:
if len(page_info['prefixes']) > 0:
page_info['allow_retag'] = ''
page_info['no_prefixes'] = 'false'
else:
page_info['allow_retag'] = 'disabled'
page_info['no_prefixes'] = 'true'
else: # Dead or in unknown flock
page_info['allow_retag'] = 'disabled'
page_info['no_prefixes'] = 'false'
page_info['sire_no_fld'] = f"{page_info['sire_no']}, {page_info['sire_ear_tag']}"
page_info['dam_no_fld'] = f"{page_info['dam_no']}, {page_info['dam_ear_tag']}"
page_info['rid'] = rid if rid else ''
page_info["dlg_msg"] = dlg_msg
template = const.JINJA_ENV.get_template('templates/sheepdata.tmpl')
return template.render(page_info)
@cherrypy.tools.json_out()
def POST(self):
""" Create a new sheep from the request body.
The body must contain the following key:value pairs:
- regn_flock The registering flock number.
- org_flock The originating flock number.
- tag_prefix The ear tag prefix.
- tag_from Indicates the origin of the tag: 'regflk' or 'orgflk'.
- sex Either 'M' or 'F'.
- tag_no The ear tag individual number. Up to 7 characters.
- text_dob The date of birth as a text string in D(D)/M(M)/(YY)YY format.
- sheep_name The sheep's individual name. Up to 32 characters.
- litter_size A single numeric character or 'X'.
- colour The sheep's colour and/or markings. Up to 48 characters.
- horns A single numeric character or 'X' or 'S'.
- breeders_temp_mark A free-form string of up to 15 characters.
- sire_no The registration number of the sheep's sire.
- sire_xfer_date An ISO date string if xfer_sire is present, otherwise empty.
- dam_no The registration number of the sheep's dam.
- dam_xfer_date An ISO date string if xfer_dam is present, otherwise empty.
- cttee_appr The Committee Approval status - 'none', 'raret' or 'reins'
The following key:value pairs may also be present:
- xfer_sire If present must be 'xfer'.
- xfer_dam If present must be 'xfer'.
- ignore_warnings If present must be 'ignore'.
Any other key:value pairs wil be ignored.
The 'rid' must be 'None' and will be assigned by the program.
The registration number of the new sheep will be returned in the response
the registration is successful.
"""
req_body = self.check_request(None, ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
try:
regn_no = shpupd.insert_new_sheep(req_body)
except (shpupd.ValidationError, util.PPDError, util.PPDWarning) as err:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": str(err)}
if regn_no:
cherrypy.response.status = 201
return {"rcode": xhrc.XHR_CREATED, "data": regn_no}
# This should have raised an exception!
cherrypy.response.status = 500
return {"rcode": xhrc.XHR_ERROR, "data": "Uncaught exception"}
@cherrypy.tools.json_out()
def PUT(self):
""" Update an sheep using the data in the request body.
The body must contain the following key:value pairs:
- regn_no The registration number of the sheep being updated.
The body may contain any or all of the following key:value pairs:
- tag_prefix The ear tag prefix.
- tag_no The ear tag individual number. Up to 7 characters.
- tag_from Must be present if tag_prefix and/or tag_no are present.
Indicates the origin of the tag: 'currflkpfx' or 'owningflk'.
- sex Either 'M' or 'F'.
- text_dob The date of birth as a text string in D(D)/M(M)/(YY)YY format.
- sheep_name The sheep's individual name. Up to 32 characters.
- litter_size A single numeric character or 'X'.
- colour The sheep's colour and/or markings. Up to 48 characters.
- horns A single numeric character or 'X' or 'S'.
- breeders_temp_mark A free-form string of up to 15 characters.
- sire_no The registration number of the sheep's sire.
- sire_xfer_date An ISO date string if xfer_sire is present, otherwise empty.
- dam_no The registration number of the sheep's dam.
- dam_xfer_date An ISO date string if xfer_dam is present, otherwise empty.
The following key:value pairs may also be present:
- xfer_sire If present must be 'xfer'.
- xfer_dam If present must be 'xfer'.
- ignore_warnings If present must be 'ignore'.
Any other key:value pairs will be ignored.
"""
req_body = self.check_request('regn_no', ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
if not sheeplib.exists_sheep(req_body['regn_no']):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": "'regn_no' item is invalid"}
try:
result = shpupd.update_sheep(req_body)
except shpupd.ValidationError as err:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": str(err)}
if result == 'OK':
cherrypy.response.status = 200
return {"rcode": xhrc.XHR_UPDATED, "data": "Updated"}
# This should have raised an exception!
cherrypy.response.status = 500
return {"rcode": xhrc.XHR_ERROR, "data": "Uncaught exception"}
@cherrypy.expose
class AISire(SheepBase):
""" Sheep known to have semen taken for AI """
def GET(self, rid=None, param=None):
""" Return the AI Sires page """
page_info = self.html_appconf_items()
rows = util.rows2dicts(sheeplib.get_ai_sires())
for row in rows:
row['sire_name'] = sheeplib.make_fullname(row)
page_info['sires'] = rows
page_info['pagetitle'] = "AI Sires"
page_info['titlebar'] = page_info['pagetitle']
page_info['resource'] = "sheep/aisire"
template = const.JINJA_ENV.get_template('templates/aisires.tmpl')
return template.render(page_info)
@cherrypy.tools.json_out()
def PUT(self):
""" Data from the AI Sires page. req_body may only include columns changes """
req_body = self.check_request("regn_no", ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
try:
regn_no = shpupd.update_ai_sire(req_body)
except (shpupd.ValidationError, util.PPDError, util.PPDWarning) as err:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": str(err)}
cherrypy.response.status = 200
return {"rcode": xhrc.XHR_UPDATED, "data": regn_no}
@cherrypy.tools.json_out()
def POST(self):
""" Data from the ai sire dialog """
req_body = self.check_request("regn_no", ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
try:
regn_no = shpupd.add_ai_sire(req_body)
except (shpupd.ValidationError, util.PPDError, util.PPDWarning) as err:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": str(err)}
cherrypy.response.status = 201
return {"rcode": xhrc.XHR_CREATED, "data": regn_no}
@cherrypy.expose
class Lookup(SheepBase):
""" The database sheep Lookup page. This page is mounted at sheep/lookup """
def GET(self):
""" Return the Sheep Lookup page. """
fndtn = False
brdr_flocks = []
loginroles = cherrypy.request.loginroles
logingroups = cherrypy.request.logingroups
userpers = userlib.get_user_login(cherrypy.request.login)['person_id']
if ('breeder' in loginroles and not 'officer'in logingroups):
brdr_flocks = [[row[0], row[1]]
for row in flocklib.get_persons_flocks(userpers)]
page_info = self.html_appconf_items()
page_info['pagetitle'] = "Sheep Lookup"
page_info['titlebar'] = page_info['pagetitle']
page_info.update({'resource': "sheep/lookup",
'rid': '',
'loginroles': ','.join(loginroles),
'logingroups': ','.join(logingroups),
'flocks': brdr_flocks,
'srchby': 'name',
})
cherrypy.response.status = 200
template = const.JINJA_ENV.get_template('templates/sheeplookup.tmpl')
return template.render(page_info)
@cherrypy.expose
class Data(SheepBase):
""" Miscellaneous sheep data items """
@cherrypy.tools.json_out()
def GET(self, vpath=None, param=None):
""" Return the data items as specified by 'vpath' and 'param'"""
cherrypy.response.headers['content-type'] = 'application/json'
loginroles = cherrypy.request.loginroles
logingroups = cherrypy.request.logingroups
if vpath == 'data':
# Return a list of dicts of data formatted for the results dialog
if not param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": "'rid' is required"}
loginroles = cherrypy.request.loginroles
show_fndtn = bool('admin' in loginroles or 'regsec' in loginroles)
result = sheeplib.find_sheep(param)
if not isinstance(result, list):
# Some sort of error or unexpected case.
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": result}
if len(result) > 300:
return {"rcode": xhrc.XHR_WARN, "data": f"Too many ({len(result)}) matches "
f"found for '{param}'. Try a more specific search."}
sheep_data = sheeplib.get_data_for_list(result, 'regno', 'asc')
list_data = sheeplib.make_dialog_data(sheep_data, show_fndtn)
return {"rcode": xhrc.XHR_NOERR, "data": list_data}
if vpath == 'colour':
# 'param' if present is the sort parameter
colours = [[row["colour_name"], row["description"], row['name_lc']]
for row in sheeplib.get_recognised_colours()]
return {"rcode": xhrc.XHR_NOERR,"data": colours}
if vpath == 'marking':
# param is the sort parameter if specified
markings = [[row["marking_name"], row['description'], row['category'],
row['name_lc']] for row in sheeplib.get_recognised_markings()]
return {"rcode": xhrc.XHR_NOERR,"data": markings}
if vpath == 'owner':
# 'param' is the colon separated registration number and optional date
if not ':' in param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "parms must be separated by a colon"}
parms = param.split(':')
regn_no = parms[0]
iso_date = ""
if len(parms) == 2:
iso_date = parms[1]
if iso_date:
flock_info = transfer.get_flock_info(regn_no, iso_date)
owner = flocklib.get_flock_owner(flock_info['flock_no'], iso_date)
else:
flock_info = transfer.get_flock_info(regn_no)
owner = flocklib.get_flock_owner(flock_info['flock_no'])
if flock_info['flock_no'] == flocklib.FLOCK_DEAD:
return {"rcode": xhrc.XHR_WARN,
"data": "This sheep is recorded as dead and cannot be inspected"}
if flock_info['flock_no'] == flocklib.FLOCK_UNKNOWN:
return {"rcode": xhrc.XHR_WARN,
"data": "The owner of this sheep is unknown and it cannot be inspected"}
owner_name = perslib.make_person_string(owner)
owner_inf = {"flock": f"{flock_info['flock_name']} - {flock_info['flock_no']}",
"flock_no": flock_info['flock_no'], "owner_name": owner_name,
"owner_id": owner['person_id'], "active": owner['active']}
return {"rcode": xhrc.XHR_NOERR, "data": owner_inf}
if vpath == 'fndtn':
# Get the foundation sheep.
if not ':' in param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "parms must be separated by a colon"}
parms = param.split(':')
if len(parms) != 2:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Invalid parameter for {vpath}"}
if parms[0] not in ('F', 'M'):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Invalid parameter for {vpath}"}
regn_nos = sheeplib.get_foundation_sheep(parms[0], parms[1])
list_data = sheeplib.get_data_for_list(regn_nos)
dlg_data = []
for sheep in list_data:
dlg_data.append(sheeplib.format_data(sheep))
return {"rcode": xhrc.XHR_NOERR, "data": dlg_data}
if vpath == 'parent_xfer':
# Get the sire's transfer status.
# Params is: 'sire_id':dam-id:'iso_dob':'birth_flock'
if not ':' in param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "params must be separated by a colon"}
args = param.split(":")
sire_status = ""
dam_status = ""
if args[0]:
if not sheeplib.exists_sheep(args[0]):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Invalid sire_id {args[0]}"}
sire_data = sheeplib.get_pedigree(args[0], 1)[1]
if isinstance(sire_data, str):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": sire_data}
sire_status = transfer.get_parent_transfer_status(sire_data, args[2], args[3])
if args[1]:
if not sheeplib.exists_sheep(args[1]):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Invalid dam_id {args[1]}"}
dam_data = sheeplib.get_pedigree(args[1], 1)[1]
if isinstance(dam_data, str):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": dam_data}
dam_status = transfer.get_parent_transfer_status(dam_data, args[2], args[3])
return {"rcode": xhrc.XHR_NOERR,
"data": {"sire_status": sire_status, "dam_status": dam_status}}
if vpath == 'breeder':
# Return the breeder person id of the sheep identified by colon separated
# registration numbers in 'param'
if not ':' in param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "params muts be separated by a colon"}
args = param.split(":")
owners = []
for arg in args:
owners.append(sheeplib.get_sheep_breeder_id(arg))
return {"rcode": xhrc.XHR_NOERR, "data": owners}
if vpath == 'datahist':
# Return a sheep's history from the audit_history table
# args are : tablename:regn_no:[attr0:[attr1]]
if not ':' in param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "params muts be separated by a colon"}
args = param.split(":")
rows = sheeplib.get_sheep_history(args[0], args[1], args[2], args[3])
drows = sheeplib.make_data_history_display_rows(rows)
return {"rcode": xhrc.XHR_NOERR, "data": drows}
if vpath == 'taghist':
# Return a sheep's ear tag from the audit_history table
# args are : regn_no:[attr0:[attr1]]
if not ':' in param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "params muts be separated by a colon"}
args = param.split(":")
rows = sheeplib.get_sheep_history('ear_tag', args[0], args[1], args[2])
drows = sheeplib.make_tag_history_display_rows(rows)
return {"rcode": xhrc.XHR_NOERR, "data": drows}
if vpath == 'lookup':
# Return data on sheep identified by 'args' - 'rid' and 'srchby'
cherrypy.response.status = 400
if not ':' in param:
return {"rcode": xhrc.XHR_ERROR,
"data": "params muts be separated by a colon"}
rid, srchby = param.split(":")
if not rid and srchby != 'fndtn':
return {"rcode": xhrc.XHR_ERROR, "data": "A search argument is required"}
if srchby == "name":
regn_nos = sheeplib.find_sheep(rid)
if not isinstance(regn_nos, list):
return {"rcode": xhrc.XHR_ERROR, "data": regn_nos}
elif srchby in ("regflk", "orgflk", "nowin", "everin", "movedto"):
flocks = flocklib.find_flock(rid)
if not flocks:
return {"rcode": xhrc.XHR_ERROR, "data": f"No flock found matching '{rid}'"}
if len(flocks) > 1:
return {"rcode": xhrc.XHR_ERROR,
"data": f"More than one flock found matching '{rid}'"}
flock = flocks[0][0]
# Check the search is allowed
userpers = userlib.get_user_login(cherrypy.request.login)['person_id']
if ('breeder' in loginroles and not 'officer'in logingroups):
brdr_flocks = [[row[0], row[1]]
for row in flocklib.get_persons_flocks(userpers)]
if DEBUG:
print(f"Breeder flocks: {brdr_flocks}")
found = False
if srchby not in ("regflk", "orgflk"):
if 'breeder' in loginroles and not 'officer' in logingroups:
found = True
brdr_flocks = [[row[0], row[1]]
for row in flocklib.get_persons_flocks(userpers)]
for brdr_flock in brdr_flocks:
if brdr_flock[0] == flock:
found = True
break
if not found:
return {"rcode": xhrc.XHR_ERROR,
"data": "The flock specified must be "
"owned by the logged in breeder"}
if srchby == "regflk":
regn_nos = transfer.get_sheep_registered_in_flock(flock)
elif srchby == "orgflk":
regn_nos = transfer.get_sheep_originating_in_flock(flock)
elif srchby == "nowin":
regn_nos = transfer.get_sheep_now_in_flock(flock)
elif srchby == "everin":
regn_nos = transfer.get_sheep_ever_in_flock(flock)
elif srchby == "movedto":
regn_nos = transfer.get_sheep_moved_into_flock(flock)
elif srchby == "tagno":
if len(rid) > const.MAX_INDIV_NO_LEN:
return {"rcode": xhrc.XHR_ERROR,
"data": "Individual tag numbers are less than "
f"{const.MAX_INDIV_NO_LEN} characters long"}
regn_nos = sheeplib.get_sheep_with_tag(rid)
else: # Foundation sheep
regn_nos = sheeplib.get_foundation_sheep(None, None)
if not regn_nos:
return {"rcode": xhrc.XHR_ERROR, "data": f"Nothing found matching '{rid}'"}
if len(regn_nos) > 300:
return {"rcode": xhrc.XHR_WARN,"data": (f"Too many ({len(regn_nos)}) matches "
f"found for '{rid}'. Try a more specific search.")}
sheep_data = sheeplib.get_data_for_list(regn_nos, 'regno', 'asc')
show_fndtn = bool('admin' in loginroles or 'regsec' in loginroles)
list_data = sheeplib.make_dialog_data(sheep_data, show_fndtn)
cherrypy.response.status = 200
return {"rcode": xhrc.XHR_NOERR, "data": list_data}
if vpath == 'aisire':
# Return the ai sire's data
if sheeplib.exists_sheep(param):
# Return sire data
if not sheeplib.is_ai_sire(param):
return {"rcode": xhrc.XHR_NOERR, "data": {"ai_sire": False}}
data = util.row2dict(sheeplib.get_ai_sires(param)[0])
data['ai_sire'] = True
data['owner_name'] = perslib.make_person_string(data, False)
data['sire_name'] = sheeplib.make_fullname(data)
return {"rcode": xhrc.XHR_NOERR, "data": data}
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Sheep {param} not found"}
if vpath == 'aiprog':
# Return the ai sire's progeny
if sheeplib.exists_sheep(param):
# Return sire's progeny data
if not sheeplib.is_ai_sire(param):
return {"rcode": xhrc.XHR_NOERR, "data": {"ai_sire": False}}
rows = sheeplib.get_aisire_progeny(param)
data = sheeplib.make_ai_progeny_data(rows)
return {"rcode": xhrc.XHR_NOERR, "data": data}
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Sheep {param} not found"}
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Bad vpath: {vpath}"}
@cherrypy.expose
class EarTag(SheepBase):
""" Sheep ear tags URLs """
def GET(self, rid=None, ltype='full'):
""" GET ear tag data """
if ltype not in ('full', 'prefix', 'indiv'):
raise cherrypy.HTTPError(400, f"Invalid format: {type}")
# tag and pfx may have been passed in rid
if rid:
if ltype == 'full' and rid.find(':') != -1:
tag, pfx = rid.split(':')
elif ltype == 'prefix':
tag = None
pfx = rid
else:
tag = rid
pfx = None
# All queries must specifiy 'rid'
return "Tag queries not yet available!"
return "Tag queries not yet available!"
@cherrypy.tools.json_out()
def PUT(self):
""" Swap ear tags between sheep. Prefixes must be the same """
req_body = self.check_request("regn_no1", ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
try:
shpupd.swap_tags(req_body['regn_no1'], req_body['regn_no2'], req_body['req_by'])
except shpupd.ValidationError as e:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": f"{e}"}
cherrypy.response.status = 200
return {"rcode": xhrc.XHR_UPDATED, "data": "Tags swapped"}
@cherrypy.expose
class Nav(SheepBase):
""" Class and method to return the sheep number as JSON after applying the navigation
request. This URL is assumed to be called by an AJAX call from the client.
"""
@cherrypy.tools.json_out()
def GET(self, rid=None, oldno=None):
""" Return the sheep (by regn no) located in reponse to the navigation request """
cherrypy.response.headers['content-type'] = 'application/json'
new_sheep = None
if rid in ('first', 'last'):
# oldno is optional
if oldno and not sheeplib.exists_sheep(oldno):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": f"Invalid 'oldno': {oldno}"}
if rid == 'first':
new_sheep = sheeplib.first_sheep(oldno)
elif rid == 'last':
new_sheep = sheeplib.last_sheep(oldno)
else:
if not oldno:
# An oldno must be specified
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": "No 'oldno' specified"}
if not sheeplib.exists_sheep(oldno):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": f"Invalid 'oldno': {oldno}"}
if rid == 'fprev':
new_sheep = sheeplib.prev_sheep(oldno, recs=10)
elif rid == 'prev':
new_sheep = sheeplib.prev_sheep(oldno)
elif rid == 'next':
new_sheep = sheeplib.next_sheep(oldno)
elif rid == 'fnext':
new_sheep = sheeplib.next_sheep(oldno, recs=10)
else:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": f"Invalid 'rid': {rid}"}
cherrypy.response.status = 200
return {"rcode": xhrc.XHR_NOERR, "data": new_sheep}
@cherrypy.expose
class New(SheepBase):
""" Sheep Registration flock session class """
def GET(self):
""" GET the registration form - return an empty form to start a registration
'session'. The session is maintained by Javascript on the returned page.
"""
if 'breeder' in cherrypy.request.loginroles:
member = perslib.get_person_member(cherrypy.request.userpers)
if not memlib.is_membership_current(member['member_no']):
raise cherrypy.HTTPError('403 Forbidden',
'Your membership must be paid up to register sheep')
page_info = self.html_appconf_items()
page_info['markings'] = sheeplib.get_recognised_markings(separators=True)
page_info['colours'] = [[row["colour_name"], row["description"]]
for row in sheeplib.get_recognised_colours()]
page_info['pagetitle'] = "Register Sheep"
page_info['titlebar'] = page_info['pagetitle']
page_info['regn_page'] = 'true'
page_info['transfer'] = {'transfer_text': "",}
page_info.update({'sex': "F", 'tag_no': "", 'allow_retag': "", 'text_dob': "",
'sort_dob': "", 'sheep_name': "", 'litter_size': "0",
'colour': "Not specified",
'horns': "0", 'breeders_temp_mark': "", 'sire_no_fld': "",
'sire_full_name': "", 'sire_colour': "", 'dam_no_fld': "",
'dam_full_name': "", 'dam_colour': "",})
page_info['register_code'] = ''
template = const.JINJA_ENV.get_template('templates/sheepnew.tmpl')
return template.render(page_info)
@cherrypy.tools.json_out()
def POST(self):
""" Data from the 'New Registering Flock Session'.dialog
The POST data contains flock statistics and a flag if a flock book
is to be ordered.
"""
req_body = self.check_request(None, ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
# Check the args
if not 'flock_no' in req_body or not 'fbk_order' in req_body:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": "Missing parameters in request"}
flockno = req_body['flock_no']
fbk_order = req_body['fbk_order']
try:
flock_stats = {'f_pure': int(req_body['f_pure']),
'f_cross': int(req_body['f_cross']),
'f_other': int(req_body['f_other']),
'males': int(req_body['males']),
'm_other': int(req_body['m_other'])}
except KeyError as err:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": "Missing parameter in request"}
except ValueError as err:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": f"{err}: Stats fields can only contain numbers"}
flklist = flocklib.find_flock(flockno)
if len(flklist) != 1 or flklist[0][0] != flockno:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_WARN, "data": f"Nothing found matching {flockno}"}
if sum(flock_stats.values()) > 0:
curr_yr = util.isodate_now()[:4]
if not fbklib.get_flock_stats(flockno, curr_yr):
# Write the flock stats
fbklib.write_flock_stats(flock_stats, flockno, curr_yr)
if fbk_order:
owner = flocklib.get_flock_owner(flockno)['person_id']
member_info = perslib.get_person_member(owner)
fbklib.add_fbk_order(member_info['member_no'])
prefixes = flocklib.get_flock_prefixes(flockno)
if not prefixes:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "The Registration Secretary must assign a tag prefix before "
"sheep can be registered in this flock"}
cherrypy.response.status = 201
return {"rcode": xhrc.XHR_NOERR,
"data": {"flock_no": flockno, "flock_name": flklist[0][1],
"prefixes": prefixes}}
@cherrypy.expose
class Certificates(SheepBase):
""" Pedigree and Registration Certificates
The GET methods return one or more HTML pages in multiple new tabs.
The user's browser must be set to allow popups from PPDB or only one of the set of
tabs will be opened.
Chrome/Chromium can be used headless to convert the HTML into PDF - this is the
same as using the browser's 'Save to PDF' option in the Ctrl+P page.
The command line is:
chromium --headless --quiet --print-to-pdf="<output path and filename" url
"""
def GET(self, vpath=None, param=None):
""" GET one or more certificates or the printed certificates data
if vpath is None then the Sheep Certificates page is returned.
All other vpaths return a JSON object
vpath is one of 'pcqueue', 'regcert', 'pcbyid', 'pcbyflk', 'byflock' and 'pcprint'
param is a colon separated list of parameters applicable to the vpath
filtval is only applicable to the 'pcqueue' and 'pcprint' vpaths
"""
if DEBUG:
print(f"vpath: {vpath}, 'param': {param}")
def get_pedcerts_data(regn_nos):
# Compose data for pegigree certificates
pc_data = []
for regn_no in regn_nos:
# Generate the data for the pedigree certificate
ped_dict = sheeplib.get_pedigree(regn_no, 4)
page_data = sheeplib.ped_data_for_ped_cert(ped_dict)
page_data[1].update(sheeplib.pedcert_subject_data(ped_dict[1]))
# Copy subject data to the pc_info dict and set the watermark if appropriate
pc_info = {'regd_on': page_data[1]['pc_regd_on'],
'watermark_line1': '',
'watermark_line2': '',
'regn_by': page_data[1]['pc_regd_by'],
'public_no': page_data[1]['public_no'],
'full_name': page_data[1]['full_name'],
'full_sex': page_data[1]['full_sex'],
'colour': page_data[1]['colour'],
'ear_tag': page_data[1]['ear_tag'],
'comp_dob': page_data[1]['comp_dob'],
'register_code': ped_dict[1]['register_code'],
'approved': page_data[1]['approved'],
'genotype': page_data[1]['genotype']
}
# Get the breeder or registering person name name
if ped_dict[1]['breeder_person_id'] == 0:
pers_data = perslib.get_person_data(ped_dict[1]['regn_person_id'])
else:
pers_data = perslib.get_person_data(ped_dict[1]['breeder_person_id'])
pc_info['regn_pers_name'] = perslib.make_person_string(pers_data, False)
if pc_info['register_code'] in ('A', 'B'):
pc_info['watermark_line1'] = const.PC_WM1
pc_info['watermark_line2'] = const.PC_WM2
# The ped cert has a 3-column x 31-row grid of data
# cell_rows is a tuple of tuples of into pc_info
# The pedigree data is sparse so need to check that the ancestor is
# in page_data. Patch in missing page_data
for i in range(16):
if i not in page_data:
page_data[i] = {'full_name': ' ', 'colour': ' ',
'id_cell': ' ', 'ear_no': ' '}
anc_rows = ((' ', ' ', page_data[8]['full_name']),
(' ', ' ', page_data[8]['colour']),
(' ', page_data[4]['full_name'], page_data[8]['id_cell']),
(' ', page_data[4]['colour'], ' '),
(' ', page_data[4]['id_cell'], page_data[9]['full_name']),
('Sire', page_data[4]['ear_no'], page_data[9]['colour']),
(page_data[2]['full_name'], ' ', page_data[9]['id_cell']),
(page_data[2]['colour'], ' ', ' '),
(page_data[2]['id_cell'], ' ', page_data[10]['full_name']),
(page_data[2]['ear_no'], ' ', page_data[10]['colour']),
(' ', page_data[5]['full_name'], page_data[10]['id_cell']),
(' ', page_data[5]['colour'], ' '),
(' ', page_data[5]['id_cell'], page_data[11]['full_name']),
(' ', page_data[5]['ear_no'], page_data[11]['colour']),
(' ', ' ', page_data[11]['id_cell']),
(' ', ' ', ' '),
(' ', ' ', page_data[12]['full_name']),
(' ', ' ', page_data[12]['colour']),
(' ', page_data[5]['full_name'], page_data[12]['id_cell']),
(' ', page_data[5]['colour'], ' '),
(' ', page_data[5]['id_cell'], page_data[13]['full_name']),
('Dam', page_data[5]['ear_no'], page_data[13]['colour']),
(page_data[3]['full_name'], ' ', page_data[13]['id_cell']),
(page_data[3]['colour'], ' ', ' '),
(page_data[3]['id_cell'], ' ', page_data[14]['full_name']),
(page_data[3]['ear_no'], ' ', page_data[14]['colour']),
(' ', page_data[6]['full_name'], page_data[14]['id_cell']),
(' ', page_data[6]['colour'], ' '),
(' ', page_data[6]['id_cell'], page_data[15]['full_name']),
(' ', page_data[6]['ear_no'], page_data[15]['colour']),
(' ', ' ', page_data[15]['id_cell']))
pc_info['anc_rows'] = anc_rows
pc_info['issue_date'] = util.isodate_now()
pc_data.append(pc_info)
return pc_data
if not vpath:
# Return the sheepcerts page. This is the only vpath not to return JSON
page_info = self.html_appconf_items()
page_info['sb_regno'] = '\u25BC'
page_info['pedcerts'] = sheeplib.get_pedcert_queue()
page_info['pc_counts'] = sheeplib.get_pedcert_counts()
page_info['titlebar'] = "Pedigree and Registration Certificates"
template = const.JINJA_ENV.get_template('templates/sheepcerts.tmpl')
return template.render(page_info)
if vpath == 'file':
# param is the full path of the file to download
result = serve_file(param, "application/x-download", "attachment")
os.unlink(param)
return result
if vpath == 'pcqueue':
# Return the pedcert queue, 'param' is the column to sort by, filtval is the
# filter value
if not 'json' in cherrypy.request.headers.get("Accept", ''):
raise cherrypy.HTTPError(400, 'Invalid content-type')
cherrypy.response.headers['content-type'] = 'application/json'
try:
sortby, sortdir = param.split(':')
except ValueError as err:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid 'param': {err}"})
if sortby not in ('flockno', 'reqby', 'regno', 'reqbrdr', 'reqpers', 'reason',
'added'):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid 'sortby' param: {sortby}"})
if sortdir not in ('asc', 'desc'):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid 'sortdir' param: {sortdir}"})
pedcerts = util.rows2dicts(sheeplib.get_pedcert_queue(sortby, sortdir))
cherrypy.response.status = 200
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": pedcerts})
if vpath == 'pcprint':
# Return the printed pedigree certificates data,
# 'param' is the colon separated filter parameters
# This vpath only for JSON requests
if not 'json' in cherrypy.request.headers.get("Accept", ''):
raise cherrypy.HTTPError(400, 'Invalid content-type')
cherrypy.response.headers['content-type'] = 'application/json'
try:
filtby, filtval, sortby, sortdir = param.split(':')
except ValueError as err:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid 'param': {err}"})
if sortby not in ('flockno', 'reqbrdr', 'reqpers', 'regno', 'reason', 'added',
'printed'):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid 'sortby' param: {sortby}"})
if sortdir not in ('asc', 'desc'):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid 'sortdir' param: {sortdir}"})
if not filtval:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": "'filtval' is required"})
if filtby == 'regno':
result = sheeplib.find_sheep(filtval)
if not result:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Sheep {filtval} does not exist"})
if len(result) > 1:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"{filtval} refers to multiple sheep"})
filtval = result[0]
elif filtby in 'flockno':
filtval = filtval.rjust(4, '0')
if not flocklib.exists_flock(filtval):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Flock {filtval} does not exist"})
elif filtby == 'reqbrdr':
if filtval not in ('regsec', 'admin'):
filtval = filtval.rjust(4, '0')
if not memlib.exists_member(filtval):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Member {filtval} does not exist"})
elif filtby == 'reqpers':
try:
persid = int(filtval)
except: # pylint: disable=bare-except
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"The person id {filtval} must be an integer"})
if not perslib.exists_person(persid):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Person {filtval} does not exist"})
elif filtval in ('added', 'printed'):
try:
filtval = util.is_valid_year(filtval)
except util.PPDWarning as err:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR, "data": str(err)})
except Exception as err:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Error checking year {err}"})
pedcerts = sheeplib.get_printed_pedcerts(filtby, filtval, sortby, sortdir)
if pedcerts:
data = util.rows2dicts(pedcerts)
else:
data = f"No pedcerts found matching {filtval}."
cherrypy.response.status = 200
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": data})
if vpath == 'regcert':
# Validate the the regcert request data and either return the validation status
# for a JSON request or return a Registration Certificate.
# param is a flock_no:fkb_year string
# This vpath only for JSON requests
if not 'json' in cherrypy.request.headers.get("Accept", ''):
raise cherrypy.HTTPError(400, 'Invalid content-type')
cherrypy.response.headers['content-type'] = 'application/json'
try:
parms = param.split(':')
except ValueError as err:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid parameter: {err}"})
if parms[0] not in ('file', 'page'):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid output parameter{parms[0]}"})
if not flocklib.exists_flock(parms[1]):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"No flock found matching '{parms[1]}'"})
fbk_year = parms[2]
if not fbk_year:
fbk_year = util.isodate_now()[:4]
raw_data = sheeplib.get_data_for_regcert(parms[1], fbk_year)
if not raw_data:
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"No sheep were registered by flock {parms[1]} "
f"in {fbk_year}"})
# Use the data collected above
rc_data = fbklib.make_flockbook_data(raw_data)
page_info = self.html_appconf_items()
page_info['issue_date'] = util.isodate_now()
page_info['fbk_year'] = fbk_year
page_info["rc_data"] = rc_data
if len(rc_data) == 1:
page_info['titlebar'] = f"Registration Certificate - Flock {parms[1]}"
else:
page_info['titlebar'] = "Registration Certificates"
template = const.JINJA_ENV.get_template('templates/regcert.tmpl')
html_text = template.render(page_info)
if parms[0] == 'page':
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": html_text})
# Write the html to a file and return the filepath
fname = f"regcerts-by-flock-{util.isots_now()}.html"
filepath = os.path.join(self.export_path, fname)
with open(filepath, mode='w', newline='\r\n') as fh:
fh.write(html_text)
cherrypy.response.status = 200
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": filepath})
if vpath == "pcbyflk":
# 'param' is a string of colon separated flock numbers
# Compose and return a page with the queued pedigree certificates for
# the flocks in 'param'. If the first parameter is 'files' return a file with
# the pedcerts, otherwise return an HTML page.
#
# This vpath only for JSON requests
if not 'json' in cherrypy.request.headers.get("Accept", ''):
raise cherrypy.HTTPError(400, 'Invalid content-type')
cherrypy.response.headers['content-type'] = 'application/json'
parms = param.split(':')
for flock_no in parms:
if not flocklib.exists_flock(flock_no ):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"No flock found matching '{flock_no}'"})
page_info = self.html_appconf_items()
page_info['issue_date'] = util.isodate_now()
# Get the pedcert data
regn_nos = sheeplib.get_pedcert_regnos_by_flock(parms[1:])
page_info["pc_data"] = get_pedcerts_data(regn_nos)
page_info['titlebar'] = ("Queued Pedigree Certificates for "
f"{len(parms[1:])} Flocks")
template = const.JINJA_ENV.get_template('templates/pedcert.tmpl')
html_text = template.render(page_info)
# Write the html to a file and return the filepath
fname = f"pedcerts-by-flock-{util.isots_now()}.html"
filepath = os.path.join(self.export_path, fname)
with open(filepath, mode='w', newline='\r\n') as fh:
fh.write(html_text)
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": filepath})
if vpath == 'pcbyid':
# Return one or more pedigree certificates as an HTML page or file.
# param is a colon separated list of pc_queue rec_ids
#
# This vpath only for JSON requests
if not 'json' in cherrypy.request.headers.get("Accept", ''):
raise cherrypy.HTTPError(400, 'Invalid content-type')
cherrypy.response.headers['content-type'] = 'application/json'
if not param:
return "Must supply destination and sheep registration number(s) or name(s)"
parms = param.split(':')
if parms[0] not in ('file', 'page'):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR,
"data": f"Invalid output parameter{parms[0]}"})
if len(parms) == 3 and parms[1] == '0':
# parmst[2] is a registration number for a new pedigree certificate
regn_nos = sheeplib.find_sheep(parms[2])
else:
idlist = [int(id) for id in parms[1:]]
# get the regn nos
regn_nos = sheeplib.get_pedcert_regnos_from_recids(idlist)
if not isinstance(regn_nos, list):
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR, "data": regn_nos})
# Build the pedigree certificates page. This is multi-certificate page.
pc_data = get_pedcerts_data(regn_nos)
page_info = self.html_appconf_items()
if len(pc_data) == 1:
page_info['titlebar'] = (f"Pedigree Certificate - {pc_data[0]['public_no']} "
f"{pc_data[0]['full_name']}")
else:
page_info['titlebar'] = "Pedigree Certificates"
page_info['issue_date'] = util.isodate_now()
page_info["pc_data"] = pc_data
template = const.JINJA_ENV.get_template('templates/pedcert.tmpl')
html_text = template.render(page_info)
if parms[0] == 'page':
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": html_text})
# Write the html to a file and return the filepath
fname = f"pedcerts-by-regnno-{util.isots_now()}.html"
filepath = os.path.join(self.export_path, fname)
with open(filepath, mode='w', newline='\r\n') as fh:
fh.write(html_text)
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": filepath})
cherrypy.response.status = 400
return self.json_out({"rcode": xhrc.XHR_ERROR, "data": f"Invalid vpath: {vpath}"})
@cherrypy.tools.json_out()
def POST(self):
""" Add a sheep to the Deferred Pedigree Certificates table """
req_body = self.check_request('regn_no', ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
resp = sheeplib.find_sheep(req_body['regn_no'])
if not isinstance(resp, list):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": resp}
req_body['regn_no'] = resp[0]
req_body['req_person'] = cherrypy.request.userpers
req_body['req_breeder'] = None
regn_no = sheeplib.add_pedcert_to_queue(req_body)
cherrypy.response.status = 201
return {"rcode": xhrc.XHR_NOERR, "data": regn_no}
@cherrypy.tools.json_out()
def DELETE(self):
""" Delete a sheep from the Deferred Pedigree Certificate table """
req_body = self.check_request('cert_ids', ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
if not req_body['cert_ids']:
return {"rcode": xhrc.XHR_ERROR, "data": "the cert_ids list is required"}
return {"rcode": xhrc.XHR_NOERR,
"data": sheeplib.delete_pedcerts_from_queue(
[int(id) for id in req_body['cert_ids']])}
@cherrypy.expose
class Transfer(SheepBase):
""" Sheep transfers.
This class supports the GET, POST and PUT methods.
Additional path elements/query args are: 'rid' and 'hist'.
- 'rid' is a sheep registration number, or a name which resolves to a
single sheep
- Return an HTML form allowing the sheep's
transfer history to be updated is returned.
"""
def GET(self, rid=None):
""" GET a sheep's transfer history. """
if not rid:
rid = sheeplib.last_sheep()
result = sheeplib.find_sheep(rid)
if not isinstance(result, list):
# Some sort of error or unexpected case.
regn_no = sheeplib.last_sheep()
else:
regn_no = result[0]
sheep_data = sheeplib.get_sheep_basic_data(regn_no)
page_data = {'pagetitle': "Transfer History of "
f"{sheeplib.make_public_regn_no(sheep_data)} "
f"{sheeplib.make_fullname(sheep_data)}"}
page_data['titlebar'] = page_data['pagetitle']
page_data['curr_flock'] = transfer.get_flock_info(regn_no)
page_data['history'] = util.rows2dicts(transfer.get_in_flock_history(regn_no))
changes = []
raw_chngs = transfer.get_transfer_changes(regn_no)
for row in raw_chngs:
if row['command'] != 'INSERT':
chng = {'changed': row['audit_ts'], 'user': row['app_user'],
'data': json.loads(row['old_row'])}
chng['data']['flock_name'] = flocklib.exists_flock(
chng['data']['flock_no'])[1]
changes.append(chng)
page_data['changes'] = changes
page_data['rid'] = rid if rid else ''
if 'json' in cherrypy.request.headers.get("Accept", ''):
cherrypy.response.headers['content-type'] = 'application/json'
return self.json_out({"rcode": xhrc.XHR_NOERR, "data": page_data})
page_info = self.html_appconf_items()
page_info.update({'requestpath': cherrypy.request.path_info,
'regn_no': regn_no})
page_info.update(page_data)
template = const.JINJA_ENV.get_template('templates/transferhist.tmpl')
return template.render(page_info)
@cherrypy.tools.json_out()
def PUT(self):
""" Change the date of a sheep's transfer. The new date must not be on or before the
previous record, nor, if there is one, on or after the next record.
"""
req_body = self.check_request('regn_no', ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
# Check the transfer date
try:
req_body['new_text_date'], req_body['new_date'] = util.check_text_date(
req_body['new_text_date'])
except util.PPDError as err:
return {'rcode': xhrc.XHR_ERROR, "data": f"{err}"}
except util.PPDWarning as err:
return {'rcode': xhrc.XHR_WARN, "data": f"{err}"}
# Check that the sheep exists and that the transfer date is after the DoB
sheep_data = sheeplib.exists_sheep(req_body['regn_no'])
if not sheep_data:
return {"rcode": xhrc.XHR_ERROR,
"data": f"Sheep {req_body['regn_no']} does not exist"}
hist = util.rows2dicts(transfer.get_in_flock_history(req_body["regn_no"]))
this_ix = -1
for ix, xfer in enumerate(hist):
if (str(xfer["xfer_id"]) == req_body["xfer_id"] and
xfer["regn_no"] == req_body['regn_no'] and
xfer["transfer_date"] == req_body['old_transfer_date']):
this_ix = ix
break
if this_ix == -1:
return {"rcode": xhrc.XHR_WARN,
"data": "The transfer you want to change does not exist"}
if hist[this_ix]["xfer_id"] == 0:
return {"rcode": xhrc.XHR_WARN,
"data": "A 'Birth' or 'Registration' record cannot be changed here"}
if req_body["new_date"] <= hist[this_ix - 1]["transfer_date"]:
return {"rcode": xhrc.XHR_WARN,
"data": "The new date must be greater than the previous transfer date"}
if (this_ix + 1 != len(hist) and
req_body["new__date"] >= hist[this_ix + 1]["transfer_date"]):
return {"rcode": xhrc.XHR_WARN,
"data": "The new date must be less than the next recorded transfer date"}
# Change the date
if transfer.update_transfer_date(req_body):
return {"rcode": xhrc.XHR_UPDATED,
"data": f"Transfer date for Sheep {req_body['regn_no']} updated"}
return {"rcode": xhrc.XHR_ERROR,
"data": "Failed to update the transfer date "
f"of sheep {req_body['regn_no']}"}
@cherrypy.tools.json_out()
def POST(self):
""" Transfer a sheep to a flock.
Check that the proposed transfer date does not lead to any conflicts with
existing data.
A sheep's death is recorded by a transfer to the 'Dead flock'
- Multiple transfers on the same date are not allowed.
- A transfer to any flock must be > date of birth. A transfer to any flock may
pre-date the date of registration.
- A transfer to any flock other than the 'Dead' flock must pre-date the date of
transfer to the 'Dead' flock.
- A transfer to the 'Dead' flock must post-date all other transfers, including
registration.
- If the sheep is already reportted dead then the sheep cannot be transferred.
- Check the transfer date is after the most recent live transfer
- Check the flock exists
- Insert a new transfer record
"""
req_body = self.check_request('regn_no', ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
# Check the transfer date
try:
req_body['text_transfer_date'], req_body['transfer_date'] = util.check_text_date(
req_body['text_transfer_date'])
except util.PPDError as err:
return {'rcode': xhrc.XHR_ERROR, "data": f"{err}"}
except util.PPDWarning as err:
return {'rcode': xhrc.XHR_WARN, "data": f"{err}"}
# Check that the sheep exists and that the transfer date is after the DoB
sheep_data = sheeplib.exists_sheep(req_body['regn_no'])
if not sheep_data:
return {"rcode": xhrc.XHR_ERROR,
"data": f"Sheep {req_body['regn_no']} does not exist"}
if req_body['transfer_date'] <= sheep_data['sort_dob']:
return {"rcode": xhrc.XHR_WARN,
"data": "Transfer date is on or before the sheep was born"}
# Check that there are no existing transfers on the proposed transefr date
if transfer.exists_transfer(req_body['regn_no'], req_body['transfer_date']):
return {"rcode": xhrc.XHR_WARN,
"data": f"A transfer on {req_body['text_transfer_date']} already exists."}
# Check that the 'to' flock exists end get the flock_no
flock = flocklib.exists_flock(req_body['new_flock'])
if not flock:
return {"rcode": xhrc.XHR_ERROR,
"data": f"Flock {req_body['new_flock']} does not exist"}
flock_no = flock['flock_no']
# Check whether the sheep is reported dead or presumed dead
curr_flock = transfer.get_flock_info(req_body['regn_no'])
if curr_flock['flock_no'] == flocklib.FLOCK_DEAD:
if flock_no == flocklib.FLOCK_DEAD:
return {"rcode": xhrc.XHR_WARN,
"data": "Sheep is already recorded as dead!"}
if req_body['transfer_date'] >= curr_flock['transfer_date']:
if curr_flock['transfer_reason'] == 'Reported':
return {"rcode": xhrc.XHR_WARN,
"data": "Sheep has been reported Dead before the new transfer date "
"and cannot be transferred"}
elif curr_flock['flock_no'] == flocklib.FLOCK_UNKNOWN:
# The last transfer is to 'Location Unknown' - delete that tansfer
transfer.delete_location_unknown(req_body['regn_no'])
# The users must confirm any transfer is before the last 'live' transfer
last_live = transfer.get_last_live_xfer(req_body['regn_no'])
if req_body['transfer_date'] < last_live['transfer_date']:
if not 'inter' in req_body:
return {"rcode": xhrc.XHR_ASK,
"data": "The new transfer precedes the most recent transfer.<br>Are you "
f"sure the new transfer date of {req_body['transfer_date']} is correct?"}
# Transfer the sheep
return transfer.transfer_sheep(flock_no, req_body)
@cherrypy.tools.json_out()
def DELETE(self):
""" This method will delete an existing transfer record to correct an error """
req_body = self.check_request('regn_no', ("regsec",))
if not isinstance(req_body, dict):
# Some sort of error or unexpected case.
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
if not 'action' in req_body:
return {"rcode": xhrc.XHR_ERROR, "data": "Missing attribute 'action'"}
if req_body['action'] == xhrc.XHR_CANCEL:
if transfer.delete_transfer(req_body['xfer_id'], req_body['regn_no']):
return {"rcode": xhrc.XHR_DELETED,
"data": f"Sheep {req_body['regn_no']} transfer to "
f"flock {req_body['flock_no']} on {req_body['xfer_date']} "
"has been cancelled"}
return {"rcode": xhrc.XHR_ERROR,
"data": f"Failed to cancel transfer of sheep {req_body['regn_no']} to "
"flock {req_body['flock_no']} on {req_body['xfer_date']}"}
return {"rcode": xhrc.XHR_ERROR, "data": req_body}
@cherrypy.expose
class Validators(SheepBase):
""" This class provides data validation URLS for the sheep data input forms.
Unless there is an error in the vpath or parameters all validators return HTTP
status 200
"""
def get_parent_data(self, parent_id):
""" Check the parent id (must be a registration number) and get its data """
result = sheeplib.exists_sheep(parent_id)
if not result:
# 'result' is an error message
return f"Registration number '{parent_id}' not found"
return sheeplib.get_pedigree(result[0], 1)[1]
@cherrypy.tools.json_out()
def GET(self, vpath=None, param=None):
""" Validators for sheep date of birth, ear tag, colour and parent (sire or dam).
The validator is specified by the vpath parameter.
All data, including error responses other than 500, is returned as a
JSON object.
1) vpath = dob. The string passed in 'param' is validated and the validated
text string, the corresponding ISO date and any messages are returned.
2) vpath = tag. The tag prefix/tag no colon separated pair in 'param' is
parsed and validated. The validated individual number or messages are
returned. If the tag is already used return the tag_id and data for the
using sheep.
3) vpath = sire. The data (sire id, iso dob, birth flock) in 'param' is
validated and sire data, age check and transfer status information is
returned.
4) vpath = dam. The data (dam id, iso dob, birth flock) in 'param' is
validated and dam data, age check and transfer status information is
returned.
5) vpath = breeding. validates the rcode, asib and colour genetics. For the
colour genetics the sire colour, dam colour and subject colour are checked to
confirm that when interpreted as a genotype they match a possible breeding.
6) vpath = rcode. The data (sire id, dam id,registering flock, progeny_sex)
in 'param' is examined and a putative register code for the sheep is returned.
7) vpath = sibs. The data (subj_id, sire id, dam id, birth flock, iso dob,
text dob, litter size) are checked to ensure that any uterine siblings
were all born on the same date (within a 147 day range), have the same
sire, were born in the same flock, and have the same declared litter size.
If the subject is a new sheep then subj_no is empty.
8) vpath = colgen. The sire colour, dam colour and subject colour are checked to
confirm that when interpreted as a genotype they match a possible breeding.
9) vpath = name. The name field is is checked to make sure it does not contain
sheep's flock name,and to verify that the name is unique within the flock.
10) vpath = horns. The horns field is is checked to make sure it is a valid value.
"""
if DEBUG:
print(f"vpath: {vpath}, param: {param}")
cherrypy.response.headers['content-type'] = 'application/json'
if not param:
# Return a 400 with a message
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": "Missing parameter"}
param = util.unquote(param)
if vpath == 'dob':
# Validate the date passed in param
result = {'ok_date': None, 'iso_date': None, 'message': ''}
try:
result['ok_date'], result['iso_date'] = util.check_text_date(param)
except util.PPDWarning as err:
result['message'] = err.msg
return {"rcode": xhrc.XHR_ERROR, "data": result}
return {"rcode": xhrc.XHR_NOERR, "data": result}
if vpath == 'tag':
# Validate an ear tag individual number in conjunction with the prefix.
parms = param.split(':')
if len(parms) == 3:
subj = parms[0]
pfx = parms[1]
tag = parms[2]
# Request to validate a tag
result = {"rcode": xhrc.XHR_NOERR, "data": {}}
try:
tag_data = shpupd.check_ear_tag(tag, pfx)
result["data"] = tag_data["indiv_no"]
except shpupd.ValidationError as err:
result["data"] = str(err)
result["rcode"] = xhrc.XHR_INVALID
return result
except shpupd.NotKnownError as err:
result["data"] = str(err)
result["rcode"] = xhrc.XHR_NOTKNOWN
return result
dup_list = shpupd.is_eartag_used(tag_data)
if dup_list:
if len(dup_list) == 1:
if dup_list[0]['regn_no'] != subj:
if not dup_list[0]['regn_no']:
# Tag is known but sheep is un-registered
# Get the data from the inspection record
insp_rec = insplib.get_insp_by_tagid(dup_list[0]['tag_id'])
result['data'] = dup_list[0]
result['data']['ur_dob'] = insp_rec['ur_dob']
result['data']['ur_sex'] = insp_rec['ur_sex']
result['data']['ur_name'] = insp_rec['ur_name']
result['data']['ur_flock'] = insp_rec['ur_flock']
result['data']['insp_result'] = insp_rec['insp_result']
else:
ped_dicts = sheeplib.get_pedigree(dup_list[0]['regn_no'], 1)
result["data"] = sheeplib.format_short_data(ped_dicts[1])
result["data"]["tag_id"] = dup_list[0]['tag_id']
result["rcode"] = xhrc.XHR_USED
else:
result["data"] = f"{len(dup_list)} duplicate tags found"
result["rcode"] = xhrc.XHR_ERROR
else:
cherrypy.response.status = 400
result = {"rcode": xhrc.XHR_ERROR,
"data": "Must supply colon separated subject no, prefix and tag"}
return result
if vpath in ('sire', 'dam'):
# Params is: 'parentid':'iso_dob':'birth_flock'.
args = param.split(':')
if len(args) != 3:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "Must supply 3 colon separated parameters"}
parent_data = self.get_parent_data(args[0])
if isinstance(parent_data, str):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": parent_data}
# Check the sex
if ((vpath == 'sire' and parent_data['sex'] != 'M') or
(vpath == 'dam' and parent_data['sex'] != 'F')):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": f"Parent {parent_data['regn_no']} is the wrong sex"}
rtn_data = {"parent": sheeplib.format_short_data(parent_data)}
# Get the transfer status
rtn_data["status"] = transfer.get_parent_transfer_status(
parent_data, args[1], args[2])
# Perform the age checks
if rtn_data["status"]["status"] != "Dead":
age_msg = shpupd.check_parent_dates(parent_data, args[1])
rtn_data["age_check"] = age_msg
if vpath == 'sire':
rtn_data['status']['ai_sire'] = sheeplib.is_ai_sire(parent_data['regn_no'],
args[1])
#~ else:
#~ rtn_data['status']['ai_sire'] = False
return {"rcode": xhrc.XHR_NOERR, "data": rtn_data}
if vpath == 'breeding':
# Parameters are: 'sireid':'damid':'regn_flock':'progeny_sex':'prefix':'tag_no'
# :'cttee_appr'[:'subj_no':'birth_flock':'iso_dob':'litter_size':'colour']
# cttee_appr may be '', 'reinstate' or 'raretrait'
args = param.split(':')
if len(args) != 7 and len(args) != 12:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "Must supply 7 or 12 colon separated "
f"parameters ({len(args)} supplied)"}
sire_data = self.get_parent_data(args[0])
if not isinstance(sire_data, dict):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": sire_data}
dam_data = self.get_parent_data(args[1])
if not isinstance(dam_data, dict):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": dam_data}
# Check the register code. The tag indiv_no may be empty in which case set
# 'fail'
rc_args = {'regn_flock': args[2], 'sex': args[3], 'prefix': args[4],
'tag_no': args[5], 'cttee_appr': args[6]}
if len(args) == 12:
sibs_args = {'subj_no': args[7], 'birth_flock': args[8], 'iso_dob': args[9],
'litter_size': args[10]}
rtn_data = {'rcodes': shpupd.check_register(sire_data, dam_data, rc_args)}
if len(args) == 12:
rtn_data["sibs"] = shpupd.check_siblings(sire_data, dam_data, sibs_args)
rtn_data["colgen"] = shpupd.check_colour_genetics(sire_data, dam_data,
args[11])
return {"rcode": xhrc.XHR_NOERR, "data": rtn_data}
if vpath == 'sibs':
# Parameters are:
# 'sireid':'damid':'subj_no':'birth_flock':'iso_dob':'litter_size'
args = param.split(':')
if len(args) != 6:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "Must supply 6 colon separated parameters"}
sire_data = self.get_parent_data(args[0])
if not isinstance(sire_data, dict):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": sire_data}
dam_data = self.get_parent_data(args[1])
if not isinstance(dam_data, dict):
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": dam_data}
sibs_args = {'subj_no': args[2], 'birth_flock': args[3], 'iso_dob': args[4],
'litter_size': args[5]}
rtn_data = shpupd.check_siblings(sire_data, dam_data, sibs_args)
return {"rcode": xhrc.XHR_NOERR, "data": rtn_data}
if vpath == "colgen":
# Parameters are: 'sireid':'damid':'colour'
args = param.split(':')
if len(args) != 3:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "Must supply 3 colon separated "
f"parameters ({len(args)} supplied)"}
sire_data = sheeplib.get_sheep_basic_data(args[0])
dam_data = sheeplib.get_sheep_basic_data(args[1])
rtn_data = shpupd.check_colour_genetics(sire_data, dam_data, args[2])
return {"rcode": xhrc.XHR_NOERR, "data": rtn_data}
if vpath == "name":
args = param.split(':')
if len(args) != 2:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "Must supply 2 colon separated "
f"parameters ({len(args)} supplied)"}
rtn_data = shpupd.check_name(args[0], args[1])
return {"rcode": xhrc.XHR_NOERR, "data": rtn_data}
if vpath == "horns":
args = param.split(':')
if len(args) != 2:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": "Must supply 2 colon separated "
f"parameters ({len(args)} supplied)"}
try:
horns = shpupd.check_horns(args[0], args[1])
except shpupd.ValidationError as err:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR, "data": str(err)}
return {"rcode": xhrc.XHR_NOERR, "data": horns}
if vpath == 'aisire':
# Validate and return an ai sire's owner
if not ':' in param:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": "Invalid parameter"}
args = param.split(':')
if args[0] == 'person_id':
# param should be an owner person id or a name
pers_data = perslib.get_person_data(args[1])
if not pers_data:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Person {args[1]} not found"}
pers_name = perslib.make_person_string(pers_data, False)
return {"rcode": xhrc.XHR_NOERR,
"data": [{"person_id": args[1], 'person_name': pers_name}]}
if args[0] == 'flock_no':
owners = util.rows2dicts(flocklib.get_flock_owners(args[1]))
if not owners:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Flock {args[1]} not found"}
for owner in owners:
owner['person_name'] = perslib.make_person_string(owner, False)
owner['person_id'] = owner['owner_id']
return {"rcode": xhrc.XHR_NOERR, "data": owners}
if args[0] == 'name':
if args[1] in ('rbst', 'RBST', 'sss', 'SSS'):
owners = [{"person_id": None, "person_name": param.upper()}]
else:
owners = util.rows2dicts(perslib.find_person(args[1], 'persname', False))
if not owners:
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,
"data": f"No member found matching {args[1]}"}
for owner in owners:
owner['person_name'] = perslib.make_person_string(owner, False)
# Get the flock list for the persmult_dialog
flock_list = []
flocks = util.rows2dicts(flocklib.get_persons_flocks(
owner['person_id']))
flock_list = []
for flock in flocks:
flock_list.append(f"{flock['flock_no']}-{flock['flock_name']}")
owner['flocks'] = ", ".join(flock_list)
return {"rcode": xhrc.XHR_NOERR, "data": owners}
cherrypy.response.status = 400
return {"rcode": xhrc.XHR_ERROR,"data": f"Bad vpath: {vpath}"}