<commit_before># See file COPYING distributed with the xnnppx package for the copyright 
# and license.

"""XNAT Python workflow module"""

import sys
import os
import traceback
import datetime
import urllib2
import smtplib
import email.mime.text
import xml.dom.minidom
import suds.client
import suds.xsd.doctor
import suds.transport.http

# the following variables are set by XnatPythonLauncher:
#     arguments
#     from_email
#     log_file
#     mail_host
#     parameters
#     workflow_info

class WorkflowNotFoundError(Exception):
    """workflow not found"""

class _WorkflowInfo:

    """class for mirroring workflow information (XML) in XNAT"""

    def _call(self, 
              jws, 
              operation, 
              inputs, 
              authenticated=False, 
              fix_import=False):
        """perform a SOAP call"""
        url = '%s/axis/%s' % (self._base_url, jws)
        if authenticated:
            t = suds.transport.http.HttpAuthenticated(username=self._username, 
                                                      password=self._password)
        else:
            t = suds.transport.http.HttpTransport()
        if self._cookiejar is not None:
            t.cookiejar = self._cookiejar
        if fix_import:
            xsd_url = 'http://schemas.xmlsoap.org/soap/encoding/'
            imp = suds.xsd.doctor.Import(xsd_url)
            doctor = suds.xsd.doctor.ImportDoctor(imp)
            client = suds.client.Client('%s?wsdl' % url, 
                                        transport=t, 
                                        doctor=doctor)
        else:
            client = suds.client.Client('%s?wsdl' % url, transport=t)
        typed_inputs = []
        for (dtype, val) in inputs:
            ti = client.factory.create(dtype)
            ti.value = val
            typed_inputs.append(ti)
        # the WSDL returns the local IP address in the URLs; these need 
        # to be corrected if XNAT is behind a proxy
        client.set_options(location=url)
        f = getattr(client.service, operation)
        result = f(*typed_inputs)
        if self._cookiejar is None:
            self._cookiejar = t.cookiejar
        return result

    def __init__(self, base_url, username, password, workflow_id):
        self._base_url = base_url
        self._username = username
        self._password = password
        self._cookiejar = None
        res = self._call('CreateServiceSession.jws', 
                         'execute', 
                         (), 
                         authenticated=True)
        self._session = str(res)
        args = (('ns1:string', self._session), 
                ('ns1:string', 'wrk:workflowData.ID'), 
                ('ns1:string', '='), 
                ('ns1:string', workflow_id), 
                ('ns1:string', 'wrk:workflowData'))
        workflow_ids = self._call('GetIdentifiers.jws', 'search', args)
        self._doc = None
        for w_id in workflow_ids:
            url = '%s/app/template/XMLSearch.vm/id/%s/data_type/wrk:workflowData' % (self._base_url, str(w_id))
            r = urllib2.Request(url)
            self._cookiejar.add_cookie_header(r)
            data = urllib2.urlopen(r).read()
            doc = xml.dom.minidom.parseString(data)
            workflow_node = doc.getElementsByTagName('wrk:Workflow')[0]
            status = workflow_node.getAttribute('status').lower()
            if status in ('queued', 'awaiting action', 'hold'):
                self._doc = doc
                break
        if self._doc is None:
            raise WorkflowNotFoundError
        return
<commit_msg>added the workflow ID as an attribute to _WorkflowInfo objects<commit_after># See file COPYING distributed with the xnnppx package for the copyright 
# and license.

"""XNAT Python workflow module"""

import sys
import os
import traceback
import datetime
import urllib2
import smtplib
import email.mime.text
import xml.dom.minidom
import suds.client
import suds.xsd.doctor
import suds.transport.http

# the following variables are set by XnatPythonLauncher:
#     arguments
#     from_email
#     log_file
#     mail_host
#     parameters
#     workflow_info

class WorkflowNotFoundError(Exception):
    """workflow not found"""

class _WorkflowInfo:

    """class for mirroring workflow information (XML) in XNAT"""

    def _call(self, 
              jws, 
              operation, 
              inputs, 
              authenticated=False, 
              fix_import=False):
        """perform a SOAP call"""
        url = '%s/axis/%s' % (self._base_url, jws)
        if authenticated:
            t = suds.transport.http.HttpAuthenticated(username=self._username, 
                                                      password=self._password)
        else:
            t = suds.transport.http.HttpTransport()
        if self._cookiejar is not None:
            t.cookiejar = self._cookiejar
        if fix_import:
            xsd_url = 'http://schemas.xmlsoap.org/soap/encoding/'
            imp = suds.xsd.doctor.Import(xsd_url)
            doctor = suds.xsd.doctor.ImportDoctor(imp)
            client = suds.client.Client('%s?wsdl' % url, 
                                        transport=t, 
                                        doctor=doctor)
        else:
            client = suds.client.Client('%s?wsdl' % url, transport=t)
        typed_inputs = []
        for (dtype, val) in inputs:
            ti = client.factory.create(dtype)
            ti.value = val
            typed_inputs.append(ti)
        # the WSDL returns the local IP address in the URLs; these need 
        # to be corrected if XNAT is behind a proxy
        client.set_options(location=url)
        f = getattr(client.service, operation)
        result = f(*typed_inputs)
        if self._cookiejar is None:
            self._cookiejar = t.cookiejar
        return result

    def __init__(self, base_url, username, password, workflow_id):
        self._base_url = base_url
        self._username = username
        self._password = password
        self._cookiejar = None
        res = self._call('CreateServiceSession.jws', 
                         'execute', 
                         (), 
                         authenticated=True)
        self._session = str(res)
        args = (('ns1:string', self._session), 
                ('ns1:string', 'wrk:workflowData.ID'), 
                ('ns1:string', '='), 
                ('ns1:string', workflow_id), 
                ('ns1:string', 'wrk:workflowData'))
        workflow_ids = self._call('GetIdentifiers.jws', 'search', args)
        self._doc = None
        for w_id in workflow_ids:
            url = '%s/app/template/XMLSearch.vm/id/%s/data_type/wrk:workflowData' % (self._base_url, str(w_id))
            r = urllib2.Request(url)
            self._cookiejar.add_cookie_header(r)
            data = urllib2.urlopen(r).read()
            doc = xml.dom.minidom.parseString(data)
            workflow_node = doc.getElementsByTagName('wrk:Workflow')[0]
            status = workflow_node.getAttribute('status').lower()
            if status in ('queued', 'awaiting action', 'hold'):
                self._doc = doc
                self.id = w_id
                break
        if self._doc is None:
            raise WorkflowNotFoundError
        return

{'fraction_non_alphanumeric': 0.09846827133479212, 'fraction_numerical': 0.002461706783369803, 'mean_word_length': 1.8191981495759444, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '553396', 'n_tokens_mistral': 2132, 'n_tokens_neox': 2011, 'n_words': 479}