MetaPhase-Consulting/State-TalentMAP-API

View on GitHub
talentmap_api/fsbid/services/bid.py

Summary

Maintainability
A
3 hrs
Test Coverage
D
65%
import logging
from copy import deepcopy
from functools import partial
from urllib.parse import urlencode, quote

from django.conf import settings
from django.utils.encoding import smart_str

import maya
import pydash
import jwt

from talentmap_api.bidding.models import BidHandshakeCycle

from talentmap_api.common.common_helpers import ensure_date
from talentmap_api.fsbid.services.positions import fsbid_to_talentmap_pos

from talentmap_api.bidding.models import Bid
from talentmap_api.fsbid.requests import requests

import talentmap_api.bidding.services.bidhandshake as bh_services
import talentmap_api.fsbid.services.available_positions as ap_services

API_ROOT = settings.WS_ROOT_API_URL
BIDS_V2_ROOT = settings.BIDS_API_V2_URL

logger = logging.getLogger(__name__)

def user_bids(employee_id, jwt_token, position_id=None, query={}):
    '''
    Get bids for a user on a position or all if no position
    '''
    from talentmap_api.fsbid.services.common import sort_bids
    url = f"{API_ROOT}/v1/bids/?perdet_seq_num={employee_id}"
    ordering_query = query.get("ordering", None)
    bids = requests.get(url, headers={'JWTAuthorization': jwt_token, 'Content-Type': 'application/json'}).json()
    filteredBids = {}
    # Filter out any bids with a status of "D" (deleted)
    filteredBids['Data'] = [b for b in list(pydash.get(bids, 'Data') or []) if smart_str(b["bs_cd"]) != 'D']
    if position_id:
        mappedBids = [fsbid_bid_to_talentmap_bid(bid, jwt_token) for bid in filteredBids.get('Data', []) if bid.get('cp_id') == int(position_id)]
    else:
        mappedBids = map(lambda b: fsbid_bid_to_talentmap_bid(b, jwt_token), filteredBids.get('Data', []))
    mappedBids = sort_bids(bidlist=mappedBids, ordering_query=ordering_query)
    return map_bids_to_disable_handshake_if_accepted(mappedBids)

def get_user_bids_csv(employee_id, jwt_token, position_id=None, query={}):
    '''
    Export bids for a user to CSV
    '''
    from talentmap_api.fsbid.services.common import get_bids_csv
    data = user_bids(employee_id, jwt_token, position_id, query)

    response = get_bids_csv(list(data), "bids", jwt_token)

    return response


def bid_on_position(employeeId, cyclePositionId, jwt_token):
    '''
    Adds a bid on a position
    '''
    ad_id = jwt.decode(jwt_token, verify=False).get('unique_name')
    url = f"{API_ROOT}/v1/bids/?cp_id={cyclePositionId}&perdet_seq_num={employeeId}&ad_id={ad_id}"
    response = requests.post(url, data={}, headers={'JWTAuthorization': jwt_token, 'Content-Type': 'application/json'})
    response.raise_for_status()
    return response


def submit_bid_on_position(employeeId, cyclePositionId, jwt_token):
    '''
    Submits a bid on a position
    '''
    ad_id = jwt.decode(jwt_token, verify=False).get('unique_name')
    url = f"{API_ROOT}/v1/bids/?cp_id={cyclePositionId}&perdet_seq_num={employeeId}&ad_id={ad_id}"
    response = requests.put(url, data={}, headers={'JWTAuthorization': jwt_token, 'Content-Type': 'application/json'})
    response.raise_for_status()
    return response


def register_bid_on_position(employeeId, cyclePositionId, jwt_token):
    '''
    Registers a bid on a position
    '''
    ad_id = jwt.decode(jwt_token, verify=False).get('unique_name')
    url = f"{API_ROOT}/v1/bids/handshake/?cp_id={cyclePositionId}&perdet_seq_num={employeeId}&ad_id={ad_id}&hs_cd=HS"
    response = requests.patch(url, data={}, headers={'JWTAuthorization': jwt_token, 'Content-Type': 'application/json'})
    response.raise_for_status()
    return response


def unregister_bid_on_position(employeeId, cyclePositionId, jwt_token):
    '''
    Unregisters a bid on a position
    '''
    ad_id = jwt.decode(jwt_token, verify=False).get('unique_name')
    url = f"{API_ROOT}/v1/bids/handshake/?cp_id={cyclePositionId}&perdet_seq_num={employeeId}&ad_id={ad_id}"
    response = requests.patch(url, data={}, headers={'JWTAuthorization': jwt_token, 'Content-Type': 'application/json'})
    response.raise_for_status()
    return response


def remove_bid(employeeId, cyclePositionId, jwt_token):
    '''
    Removes a bid from the users bid list
    '''
    ad_id = jwt.decode(jwt_token, verify=False).get('unique_name')
    url = f"{API_ROOT}/v1/bids?cp_id={cyclePositionId}&perdet_seq_num={employeeId}&ad_id={ad_id}"
    return requests.delete(url, headers={'JWTAuthorization': jwt_token, 'Content-Type': 'application/json'})


def map_bids_to_disable_handshake_if_accepted(bids):
    # Accepting a handshake should be disabled if another handshake within the same bid cycle has already been accepted

    bidsClone = deepcopy(list(bids))

    # get the cp_id and related bid cycle id of accepted handshakes
    hasAcceptedHandshakeIds = pydash.chain(bidsClone).filter_(
        lambda x: pydash.get(x, 'handshake.bidder_hs_code') == 'handshake_accepted' and pydash.get(x, 'handshake.hs_status_code') != 'handshake_revoked'
    ).map(
        lambda x: {'cp_id': pydash.get(x, 'position_info.id'), 'cycle_id': pydash.get(x, 'position_info.bidcycle.id')}
    ).value()

    bidsClone = pydash.map_(bidsClone, lambda x: {
        **x,
        'accept_handshake_disabled': True if pydash.find(hasAcceptedHandshakeIds, lambda y:
                                                         y['cycle_id'] == pydash.get(x, 'position_info.bidcycle.id'))
                                     and not pydash.find(hasAcceptedHandshakeIds, lambda y: y['cp_id'] == pydash.get(x, 'position_info.id'))
                                     else False,
    })
    return bidsClone


def get_bid_status(statusCode, handshakeCode, assignmentCreateDate, panelMeetingStatus, handshakeAllowed):
    '''
    Map the FSBid status code and handshake code to a TalentMap status
        statusCode - W → Draft

        statusCode - A → Submitted

        handShakeCode Y, statusCode A → Handshake Accepted

        handshakeAllowed Y, statusCode A → Handshake Needs Registered

        statusCode - P → Paneled

        statusCode - D → Deleted

        statusCode - C → Closed

        statusCode - U → Unavailable
    '''
    if assignmentCreateDate is not None:
        return Bid.Status.approved
    if statusCode == 'C':
        return Bid.Status.closed
    if statusCode == 'U':
        return Bid.Status.closed
    if statusCode == 'P':
        return Bid.Status.in_panel
    if panelMeetingStatus is not None:
        return Bid.Status.in_panel
    if statusCode == 'W':
        return Bid.Status.draft
    if statusCode == 'A':
        if handshakeCode == 'Y':
            return Bid.Status.handshake_accepted
        # display register handshake (FSBid will only return this for CDOs)
        if handshakeAllowed == 'Y':
            return Bid.Status.handshake_needs_registered
        # null will just let the bidder know that it is submitted.
        # Similarly, handshakeAllowed == 'N' will let a CDO know that it is still pending review.
        else:
            return Bid.Status.submitted


def can_delete_bid(bidStatus, cycleStatus):
    '''
    Draft bids and submitted bids in an active cycle can be deleted
    '''
    return bidStatus == Bid.Status.draft or (bidStatus == Bid.Status.submitted and cycleStatus == 'A')


def fsbid_bid_to_talentmap_bid(data, jwt_token):
    bidStatus = get_bid_status(
        data.get('bs_cd'),
        data.get('ubw_hndshk_offrd_flg'),
        data.get('assignment_date'),
        data.get('panel_meeting_status'),
        data.get('handshake_allowed_ind')
    )
    canDelete = True if data.get('delete_ind', 'Y') == 'Y' else False
    cpId = int(data.get('cp_id'))
    perdet = str(int(float(data.get('perdet_seq_num'))))
    positionInfo = ap_services.get_all_position(str(cpId), jwt_token) or {}
    cycle = pydash.get(positionInfo, 'bidcycle.id')

    showHandshakeData = True
    handshakeCycle = BidHandshakeCycle.objects.filter(cycle_id=cycle)
    if handshakeCycle:
        handshakeCycle = handshakeCycle.first()
        handshake_allowed_date = handshakeCycle.handshake_allowed_date
        if handshake_allowed_date and handshake_allowed_date > maya.now().datetime():
            showHandshakeData = False

    data = {
        "id": f"{perdet}_{cpId}",
        "emp_id": data.get('perdet_seq_num'),
        "user": "",
        "waivers": [],
        "can_delete": canDelete,
        "status": bidStatus,
        "panel_status": data.get('panel_meeting_status', ''),
        "draft_date": ensure_date(data.get('ubw_create_dt'), utc_offset=-5),
        "submitted_date": ensure_date(data.get('ubw_submit_dt'), utc_offset=-5),
        "handshake_accepted_date": ensure_date(data.get("ubw_hndshk_offrd_dt"), utc_offset=-5),
        "in_panel_date": ensure_date(data.get('panel_meeting_date'), utc_offset=-5),
        "scheduled_panel_date": ensure_date(data.get('panel_meeting_date'), utc_offset=-5),
        "approved_date": ensure_date(data.get('assignment_date'), utc_offset=-5),
        "declined_date": "",
        "closed_date": "",
        "is_priority": False,
        "panel_reschedule_count": 0,
        "create_date": ensure_date(data.get('ubw_create_dt'), utc_offset=-5),
        "update_date": "",
        "reviewer": "",
        "cdo_bid": data.get('cdo_bid') == 'Y',
        "position_info": {
            "id": cpId, # even if we can't return positionInfo, we can at least return id
            **positionInfo,
        },
    }

    if showHandshakeData:
        data["handshake"] = {
            **bh_services.get_bidder_handshake_data(cpId, perdet, True),
        }

    return data

def get_bids(query, jwt_token, pk):
    '''
    Get bids
    '''
    from talentmap_api.fsbid.services.common import send_get_request

    args = {
        "uri": "",
        "query": query,
        "query_mapping_function": partial(convert_bids_query, pk),
        "jwt_token": jwt_token,
        "mapping_function": fsbid_to_talentmap_bids,
        "count_function": None,
        "base_url": "api/v1/bidding/",
        "api_root": BIDS_V2_ROOT,
    }

    bids = send_get_request(
        **args
    )

    return bids


def convert_bids_query(pk, query):
    '''
    Converts TalentMap query into FSBid query
    '''
    from talentmap_api.fsbid.services.common import convert_to_fsbid_ql

    values = {
        "rp.pageNum": int(query.get("page", 1)),
        "rp.pageRows": int(query.get("limit", 1000)),
        "rp.filter": convert_to_fsbid_ql([*(query.get("filters") or []), {'col': 'ubwperdetseqnum', 'val': pk}]),
    }

    valuesToReturn = pydash.omit_by(values, lambda o: o is None or o == [])

    return urlencode(valuesToReturn, doseq=True, quote_via=quote)


def fsbid_to_talentmap_bids(data):
    pos = data.get('position', [])

    return {
        'hs_code': data.get('ubwhscode'),
        'cp_id': data.get('ubwcpid'),
        'pos_seq_num': data.get('cpposseqnum'),
        'pos_num': data.get('posnumtext'),
        'pos_org_short_desc': data.get('posorgshortdesc'),
        'pos_title': data.get('postitledesc'),
        'perdet': data.get('perdet_seq_num'),
        'pos': fsbid_to_talentmap_pos(pos[0] if len(pos) else {}),
    }