forked from enviPath/enviPy
260 lines
9.0 KiB
Python
260 lines
9.0 KiB
Python
import base64
|
|
import logging
|
|
|
|
import requests
|
|
from django.conf import settings as s
|
|
from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseNotAllowed
|
|
from django.shortcuts import redirect
|
|
|
|
from bayer.models import PESCompound
|
|
from epdb.logic import PackageManager
|
|
from epdb.models import Pathway, Node, Group
|
|
from epdb.views import _anonymous_or_real, error
|
|
from utilities.decorators import package_permission_required
|
|
|
|
Package = s.GET_PACKAGE_MODEL()
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def has_secret_group(user):
|
|
"""
|
|
Determines if the specified user belongs to any secret group.
|
|
|
|
This function checks whether the given user is a member of any group
|
|
that is marked as secret.
|
|
|
|
Args:
|
|
user: The user for whom the check is performed.
|
|
|
|
Returns:
|
|
bool: True if the user belongs to at least one secret group,
|
|
False otherwise.
|
|
"""
|
|
return Group.objects.filter(secret=True, user_member=user).exists()
|
|
|
|
|
|
@package_permission_required()
|
|
def create_pes(request, package_uuid):
|
|
current_user = _anonymous_or_real(request)
|
|
current_package = PackageManager.get_package_by_id(current_user, package_uuid)
|
|
|
|
if request.method == "POST":
|
|
|
|
if current_package.classification_level == Package.Classification.INTERNAL:
|
|
return error(
|
|
request,
|
|
f'Creation of PESs for package {current_package.name} failed!',
|
|
"Creating PESs for internal packages is not allowed.",
|
|
)
|
|
|
|
compound_name = request.POST.get('compound-name')
|
|
compound_description = request.POST.get('compound-description')
|
|
pes_link = request.POST.get('pes-link')
|
|
|
|
if pes_link:
|
|
try:
|
|
pes_data = fetch_pes(request, pes_link, current_user)
|
|
except ValueError as e:
|
|
return error(
|
|
request,
|
|
"Could not fetch PES",
|
|
f"Could not fetch PES data for {pes_link}"
|
|
)
|
|
|
|
classification = pes_data.get("classificationLevel", "")
|
|
if "secret" == classification.lower():
|
|
|
|
if current_package.classification_level != Package.Classification.SECRET:
|
|
return error(
|
|
request,
|
|
"Classification Mismatch!",
|
|
"Cannot create secret PESs in non-secret packages."
|
|
)
|
|
|
|
if not current_package.data_pool or not current_package.data_pool.secret:
|
|
logger.info(f"The current package does not have a secret data pool.")
|
|
return error(
|
|
request,
|
|
"The current package does not have a secret data pool.",
|
|
"Cannot create secret PESs in package without a secret data pool."
|
|
)
|
|
|
|
pes = PESCompound.create(current_package, pes_data, compound_name, compound_description)
|
|
|
|
return redirect(pes.url)
|
|
else:
|
|
return error(
|
|
request,
|
|
"No PES link received",
|
|
"Please provide a PES link."
|
|
)
|
|
else:
|
|
return HttpResponseNotAllowed(["POST"])
|
|
|
|
|
|
@package_permission_required()
|
|
def create_pes_node(request, package_uuid, pathway_uuid):
|
|
current_user = _anonymous_or_real(request)
|
|
current_package = PackageManager.get_package_by_id(current_user, package_uuid)
|
|
current_pathway = Pathway.objects.get(package=current_package, uuid=pathway_uuid)
|
|
|
|
if request.method == "POST":
|
|
|
|
if current_package.classification_level == Package.Classification.INTERNAL:
|
|
return error(
|
|
request,
|
|
f'Creation of PESs for package {current_package.name} failed!',
|
|
"Creating PESs for internal packages is not allowed.",
|
|
)
|
|
|
|
compound_name = request.POST.get('compound-name')
|
|
compound_description = request.POST.get('compound-description')
|
|
pes_link = request.POST.get('pes-link')
|
|
|
|
if pes_link:
|
|
try:
|
|
pes_data = fetch_pes(request, pes_link, current_user)
|
|
except ValueError as e:
|
|
return error(
|
|
request,
|
|
"Could not fetch PES",
|
|
f"Could not fetch PES data for {pes_link}"
|
|
)
|
|
|
|
classification = pes_data.get("classificationLevel", "")
|
|
if "secret" == classification.lower():
|
|
|
|
if current_package.classification_level != Package.Classification.SECRET:
|
|
return error(
|
|
request,
|
|
"Classification Mismatch!",
|
|
"Cannot create secret PESs in non-secret packages."
|
|
)
|
|
|
|
if not current_package.data_pool or not current_package.data_pool.secret:
|
|
logger.info(f"The current package does not have a secret data pool.")
|
|
return error(
|
|
request,
|
|
"The current package does not have a secret data pool.",
|
|
"Cannot create secret PESs in package without a secret data pool."
|
|
)
|
|
|
|
pes = PESCompound.create(current_package, pes_data, compound_name, compound_description)
|
|
|
|
node_qs = Node.objects.filter(
|
|
pathway=current_pathway,
|
|
default_node_label=pes.default_structure
|
|
)
|
|
|
|
if node_qs.exists():
|
|
return redirect(current_pathway.url)
|
|
|
|
n = Node()
|
|
n.stereo_removed = False
|
|
n.pathway = current_pathway
|
|
n.depth = 0
|
|
|
|
n.default_node_label = pes.default_structure
|
|
n.save()
|
|
|
|
n.node_labels.add(pes.default_structure)
|
|
n.save()
|
|
|
|
return redirect(current_pathway.url)
|
|
|
|
else:
|
|
return error(
|
|
request,
|
|
"No PES link received",
|
|
"Please provide a PES link."
|
|
)
|
|
else:
|
|
return HttpResponseNotAllowed(["POST"])
|
|
|
|
|
|
def get_application_token(prod: bool) -> str:
|
|
scope = f"{s.PROD_PES_SCOPE if prod else s.NON_PROD_PES_SCOPE}/.default"
|
|
|
|
url = f"https://login.microsoftonline.com/{s.MS_TENANT_ID}/oauth2/v2.0/token"
|
|
data = {
|
|
"grant_type": "client_credentials",
|
|
"client_id": s.MS_ENTRA_CLIENT_ID,
|
|
"client_secret": s.MS_ENTRA_CLIENT_SECRET,
|
|
"scope": scope,
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, data=data)
|
|
response.raise_for_status()
|
|
return response.json()["access_token"]
|
|
except requests.exceptions.HTTPError as e:
|
|
logger.error(f"Could not fetch application token: {e}")
|
|
raise ValueError(f"Could not fetch application token!")
|
|
|
|
|
|
def fetch_pes(request, pes_url, user) -> dict:
|
|
|
|
for k, v in s.PES_API_MAPPING.items():
|
|
if pes_url.startswith(k):
|
|
|
|
prod = "cropkey-np" not in pes_url
|
|
|
|
pes_id = pes_url.split('/')[-1]
|
|
|
|
if pes_id == 'dummy':
|
|
import json
|
|
res_data = json.load(open(s.BASE_DIR / "fixtures/pes.json"))
|
|
res_data["pes_url"] = pes_url
|
|
return res_data
|
|
else:
|
|
headers = {
|
|
"accept": "*/*",
|
|
"authorization": "Bearer " + get_application_token(prod),
|
|
}
|
|
|
|
# Restrict request if user is not part of any secret group
|
|
if not has_secret_group(user):
|
|
headers["app-classification-level-restriction"] = "restrict-pes-secret-structure-access"
|
|
|
|
|
|
params = {"pes_reg_entity_corporate_id": pes_id}
|
|
|
|
res = requests.get(v, headers=headers, params=params, proxies=s.PROXIES or None)
|
|
|
|
try:
|
|
res.raise_for_status()
|
|
pes_data = res.json()
|
|
|
|
# Handle missing response
|
|
if "detail" in pes_data and "The following PES Reg Entities Corporate Ids could not be found" in pes_data["detail"]:
|
|
raise ValueError(f"PES with id {pes_id} not found")
|
|
|
|
# Ensure we have a entity
|
|
if len(pes_data) == 0:
|
|
raise ValueError(f"PES with id {pes_id} not found")
|
|
|
|
res_data = pes_data[0]
|
|
res_data["pes_url"] = pes_url
|
|
return res_data
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
raise ValueError(f"Error fetching PES with id {pes_id}: {e}")
|
|
|
|
raise ValueError(f"Unknown URL {pes_url}")
|
|
|
|
|
|
def visualize_pes(request):
|
|
pes_link = request.GET.get('pesLink')
|
|
|
|
if pes_link:
|
|
pes_data = fetch_pes(request, pes_link, request.user)
|
|
|
|
representations = pes_data.get('representations')
|
|
|
|
for rep in representations:
|
|
if rep.get('type') == 'color':
|
|
image_data = base64.b64decode(rep.get('base64').replace("data:image/png;base64,", ""))
|
|
return HttpResponse(image_data, content_type="image/png")
|