<s> import hashlib <EOL> import datetime <EOL> from corehq . form_processor . interfaces . dbaccessors import FormAccessors <EOL> from corehq . form_processor . interfaces . processor import FormProcessorInterface <EOL> from corehq . form_processor . models import Attachment <EOL> from corehq . form_processor . utils import convert_xform_to_json , adjust_datetimes <EOL> from couchforms import XMLSyntaxError <EOL> from couchforms . exceptions import DuplicateError <EOL> from dimagi . utils . couch import LockManager , ReleaseOnError <EOL> class MultiLockManager ( list ) : <EOL> def __enter__ ( self ) : <EOL> return [ lock_manager . __enter__ ( ) for lock_manager in self ] <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> for lock_manager in self : <EOL> lock_manager . __exit__ ( exc_type , exc_val , exc_tb ) <EOL> class FormProcessingResult ( object ) : <EOL> def __init__ ( self , submitted_form , existing_duplicate = None ) : <EOL> self . submitted_form = submitted_form <EOL> self . existing_duplicate = existing_duplicate <EOL> if submitted_form . is_duplicate : <EOL> assert existing_duplicate is None <EOL> if existing_duplicate : <EOL> assert existing_duplicate . is_deprecated <EOL> self . interface = FormProcessorInterface ( self . submitted_form . domain ) <EOL> def _get_form_lock ( self , form_id ) : <EOL> return self . interface . acquire_lock_for_xform ( form_id ) <EOL> def get_locked_forms ( self ) : <EOL> if self . existing_duplicate : <EOL> new_id = self . existing_duplicate . form_id <EOL> old_id = self . existing_duplicate . orig_id <EOL> return MultiLockManager ( [ <EOL> LockManager ( self . submitted_form , self . _get_form_lock ( new_id ) ) , <EOL> LockManager ( self . existing_duplicate , self . _get_form_lock ( old_id ) ) , <EOL> ] ) <EOL> else : <EOL> return MultiLockManager ( [ <EOL> LockManager ( self . submitted_form , self . _get_form_lock ( self . submitted_form . form_id ) ) <EOL> ] ) <EOL> class LockedFormProcessingResult ( FormProcessingResult ) : <EOL> def __init__ ( self , submitted_form ) : <EOL> super ( LockedFormProcessingResult , self ) . __init__ ( submitted_form ) <EOL> assert submitted_form . is_normal <EOL> self . lock = self . _get_form_lock ( submitted_form . form_id ) <EOL> def get_locked_forms ( self ) : <EOL> return MultiLockManager ( [ LockManager ( self . submitted_form , self . lock ) ] ) <EOL> def process_xform_xml ( domain , instance , attachments = None ) : <EOL> """<STR_LIT>""" <EOL> attachments = attachments or { } <EOL> try : <EOL> return _create_new_xform ( domain , instance , attachments = attachments ) <EOL> except XMLSyntaxError as e : <EOL> return _get_submission_error ( domain , instance , e ) <EOL> except DuplicateError as e : <EOL> return _handle_id_conflict ( instance , e . xform , domain ) <EOL> def _create_new_xform ( domain , instance_xml , attachments = None ) : <EOL> """<STR_LIT>""" <EOL> from corehq . form_processor . interfaces . processor import FormProcessorInterface <EOL> interface = FormProcessorInterface ( domain ) <EOL> assert attachments is not None <EOL> form_data = convert_xform_to_json ( instance_xml ) <EOL> adjust_datetimes ( form_data ) <EOL> xform = interface . new_xform ( form_data ) <EOL> xform . domain = domain <EOL> attachments = map ( <EOL> lambda a : Attachment ( name = a [ <NUM_LIT:0> ] , raw_content = a [ <NUM_LIT:1> ] , content_type = a [ <NUM_LIT:1> ] . content_type ) , <EOL> attachments . items ( ) <EOL> ) <EOL> attachments . append ( Attachment ( name = '<STR_LIT>' , raw_content = instance_xml , content_type = '<STR_LIT>' ) ) <EOL> interface . store_attachments ( xform , attachments ) <EOL> result = LockedFormProcessingResult ( xform ) <EOL> with ReleaseOnError ( result . lock ) : <EOL> if interface . is_duplicate ( xform . form_id ) : <EOL> raise DuplicateError ( xform ) <EOL> return result <EOL> def _get_submission_error ( domain , instance , error ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> message = unicode ( error ) <EOL> except UnicodeDecodeError : <EOL> message = unicode ( str ( error ) , encoding = '<STR_LIT:utf-8>' ) <EOL> xform = FormProcessorInterface ( domain ) . submission_error_form_instance ( instance , message ) <EOL> return FormProcessingResult ( xform ) <EOL> def _handle_id_conflict ( instance , xform , domain ) : <EOL> """<STR_LIT>""" <EOL> assert domain <EOL> conflict_id = xform . form_id <EOL> interface = FormProcessorInterface ( domain ) <EOL> if interface . is_duplicate ( conflict_id , domain ) : <EOL> return _handle_duplicate ( xform , instance ) <EOL> else : <EOL> xform = interface . assign_new_id ( xform ) <EOL> return FormProcessingResult ( xform ) <EOL> def _handle_duplicate ( new_doc , instance ) : <EOL> """<STR_LIT>""" <EOL> interface = FormProcessorInterface ( new_doc . domain ) <EOL> conflict_id = new_doc . form_id <EOL> existing_doc = FormAccessors ( new_doc . domain ) . get_with_attachments ( conflict_id ) <EOL> existing_md5 = existing_doc . xml_md5 ( ) <EOL> new_md5 = new_doc . xml_md5 ( ) <EOL> if existing_md5 != new_md5 : <EOL> existing_doc , new_doc = apply_deprecation ( existing_doc , new_doc , interface ) <EOL> return FormProcessingResult ( new_doc , existing_doc ) <EOL> else : <EOL> duplicate = interface . deduplicate_xform ( new_doc ) <EOL> return FormProcessingResult ( duplicate ) <EOL> def apply_deprecation ( existing_xform , new_xform , interface = None ) : <EOL> interface = interface or FormProcessorInterface ( existing_xform . domain ) <EOL> new_xform . form_id = existing_xform . form_id <EOL> existing_xform = interface . assign_new_id ( existing_xform ) <EOL> existing_xform . orig_id = new_xform . form_id <EOL> new_xform . received_on = existing_xform . received_on <EOL> new_xform . deprecated_form_id = existing_xform . form_id <EOL> new_xform . edited_on = datetime . datetime . utcnow ( ) <EOL> existing_xform . edited_on = new_xform . edited_on <EOL> return interface . apply_deprecation ( existing_xform , new_xform ) </s>
