<s> from collections import namedtuple <EOL> import logging <EOL> from couchdbkit import ResourceNotFound <EOL> from django . db . models import Q <EOL> from casexml . apps . case . const import UNOWNED_EXTENSION_OWNER_ID , CASE_INDEX_EXTENSION <EOL> from casexml . apps . case . dbaccessors import get_extension_chain <EOL> from casexml . apps . case . signals import cases_received <EOL> from casexml . apps . case . util import validate_phone_datetime <EOL> from casexml . apps . phone . cleanliness import should_create_flags_on_submission <EOL> from casexml . apps . phone . models import OwnershipCleanlinessFlag <EOL> from corehq . toggles import LOOSE_SYNC_TOKEN_VALIDATION , EXTENSION_CASES_SYNC_ENABLED <EOL> from corehq . apps . users . util import SYSTEM_USER_ID <EOL> from corehq . form_processor . interfaces . processor import FormProcessorInterface <EOL> from couchforms . models import XFormInstance <EOL> from casexml . apps . case . exceptions import ( <EOL> NoDomainProvided , <EOL> ) <EOL> from django . conf import settings <EOL> from dimagi . utils . couch . database import iter_docs <EOL> from casexml . apps . case import const <EOL> from casexml . apps . case . models import CommCareCase <EOL> from casexml . apps . case . xml . parser import case_update_from_block <EOL> from dimagi . utils . logging import notify_exception <EOL> DirtinessFlag = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class CaseProcessingResult ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , domain , cases , dirtiness_flags , extensions_to_close = None ) : <EOL> self . domain = domain <EOL> self . cases = cases <EOL> self . dirtiness_flags = dirtiness_flags <EOL> if extensions_to_close is None : <EOL> extensions_to_close = set ( ) <EOL> self . extensions_to_close = extensions_to_close <EOL> def get_clean_owner_ids ( self ) : <EOL> dirty_flags = self . get_flags_to_save ( ) <EOL> return { c . owner_id for c in self . cases if c . owner_id and c . owner_id not in dirty_flags } <EOL> def set_cases ( self , cases ) : <EOL> self . cases = cases <EOL> def get_flags_to_save ( self ) : <EOL> return { f . owner_id : f . case_id for f in self . dirtiness_flags } <EOL> def close_extensions ( self ) : <EOL> from casexml . apps . case . cleanup import close_cases <EOL> extensions_to_close = list ( self . extensions_to_close ) <EOL> if extensions_to_close : <EOL> return close_cases ( list ( self . extensions_to_close ) , self . domain , SYSTEM_USER_ID ) <EOL> def commit_dirtiness_flags ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . domain : <EOL> flags_to_save = self . get_flags_to_save ( ) <EOL> if should_create_flags_on_submission ( self . domain ) : <EOL> assert settings . UNIT_TESTING <EOL> all_touched_ids = set ( flags_to_save . keys ( ) ) | self . get_clean_owner_ids ( ) <EOL> to_update = { f . owner_id : f for f in OwnershipCleanlinessFlag . objects . filter ( <EOL> domain = self . domain , <EOL> owner_id__in = list ( all_touched_ids ) , <EOL> ) } <EOL> for owner_id in all_touched_ids : <EOL> if owner_id not in to_update : <EOL> flag = OwnershipCleanlinessFlag ( domain = self . domain , owner_id = owner_id , is_clean = True ) <EOL> if owner_id in flags_to_save : <EOL> flag . is_clean = False <EOL> flag . hint = flags_to_save [ owner_id ] <EOL> flag . save ( ) <EOL> else : <EOL> flag = to_update [ owner_id ] <EOL> if owner_id in flags_to_save and ( flag . is_clean or not flag . hint ) : <EOL> flag . is_clean = False <EOL> flag . hint = flags_to_save [ owner_id ] <EOL> flag . save ( ) <EOL> else : <EOL> flags_to_update = OwnershipCleanlinessFlag . objects . filter ( <EOL> Q ( domain = self . domain ) , <EOL> Q ( owner_id__in = flags_to_save . keys ( ) ) , <EOL> Q ( is_clean = True ) | Q ( hint__isnull = True ) <EOL> ) <EOL> for flag in flags_to_update : <EOL> flag . is_clean = False <EOL> flag . hint = flags_to_save [ flag . owner_id ] <EOL> flag . save ( ) <EOL> def process_cases_with_casedb ( xforms , case_db , config = None ) : <EOL> config = config or CaseProcessingConfig ( ) <EOL> case_processing_result = _get_or_update_cases ( xforms , case_db ) <EOL> cases = case_processing_result . cases <EOL> xform = xforms [ <NUM_LIT:0> ] <EOL> try : <EOL> relevant_log = xform . get_sync_token ( ) <EOL> except ResourceNotFound : <EOL> if LOOSE_SYNC_TOKEN_VALIDATION . enabled ( xform . domain ) : <EOL> relevant_log = None <EOL> else : <EOL> raise <EOL> if relevant_log : <EOL> relevant_log . strict = config . strict_asserts <EOL> from casexml . apps . case . util import update_sync_log_with_checks <EOL> update_sync_log_with_checks ( relevant_log , xform , cases , case_db , <EOL> case_id_blacklist = config . case_id_blacklist ) <EOL> try : <EOL> cases_received . send ( sender = None , xform = xform , cases = cases ) <EOL> except Exception as e : <EOL> notify_exception ( <EOL> None , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( xform . form_id , e ) <EOL> ) <EOL> for case in cases : <EOL> case_db . post_process_case ( case , xform ) <EOL> case_db . mark_changed ( case ) <EOL> case_processing_result . set_cases ( cases ) <EOL> return case_processing_result <EOL> class CaseProcessingConfig ( object ) : <EOL> def __init__ ( self , strict_asserts = True , case_id_blacklist = None ) : <EOL> self . strict_asserts = strict_asserts <EOL> self . case_id_blacklist = case_id_blacklist if case_id_blacklist is not None else [ ] <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( <EOL> strict = self . strict_asserts , <EOL> ids = "<STR_LIT:U+002CU+0020>" . join ( self . case_id_blacklist ) <EOL> ) <EOL> def get_and_check_xform_domain ( xform ) : <EOL> try : <EOL> domain = xform . domain <EOL> except AttributeError : <EOL> domain = None <EOL> if not domain and settings . CASEXML_FORCE_DOMAIN_CHECK : <EOL> raise NoDomainProvided ( ) <EOL> return domain <EOL> def _get_or_update_cases ( xforms , case_db ) : <EOL> """<STR_LIT>""" <EOL> domain = getattr ( case_db , '<STR_LIT>' , None ) <EOL> touched_cases = FormProcessorInterface ( domain ) . get_cases_from_forms ( case_db , xforms ) <EOL> def _validate_indices ( case ) : <EOL> if case . indices : <EOL> for index in case . indices : <EOL> referenced_case = case_db . get ( index . referenced_id ) <EOL> if not referenced_case : <EOL> logging . error ( <EOL> "<STR_LIT>" , <EOL> case . case_id , <EOL> index . referenced_id , <EOL> ) <EOL> def _get_dirtiness_flags_for_outgoing_indices ( case , tree_owners = None ) : <EOL> """<STR_LIT>""" <EOL> if tree_owners is None : <EOL> tree_owners = set ( ) <EOL> extension_indices = [ index for index in case . indices if index . relationship == CASE_INDEX_EXTENSION ] <EOL> unowned_host_cases = [ ] <EOL> for index in extension_indices : <EOL> host_case = case_db . get ( index . referenced_id ) <EOL> if ( <EOL> host_case <EOL> and host_case . owner_id == UNOWNED_EXTENSION_OWNER_ID <EOL> and host_case not in unowned_host_cases <EOL> ) : <EOL> unowned_host_cases . append ( host_case ) <EOL> owner_ids = { case_db . get ( index . referenced_id ) . owner_id <EOL> for index in case . indices if case_db . get ( index . referenced_id ) } | tree_owners <EOL> potential_clean_owner_ids = owner_ids | set ( [ UNOWNED_EXTENSION_OWNER_ID ] ) <EOL> more_than_one_owner_touched = len ( owner_ids ) > <NUM_LIT:1> <EOL> touches_different_owner = len ( owner_ids ) == <NUM_LIT:1> and case . owner_id not in potential_clean_owner_ids <EOL> if ( more_than_one_owner_touched or touches_different_owner ) : <EOL> yield DirtinessFlag ( case . case_id , case . owner_id ) <EOL> if extension_indices : <EOL> for index in case . indices : <EOL> referenced_case = case_db . get ( index . referenced_id ) <EOL> yield DirtinessFlag ( referenced_case . case_id , referenced_case . owner_id ) <EOL> if case . owner_id != UNOWNED_EXTENSION_OWNER_ID : <EOL> tree_owners . add ( case . owner_id ) <EOL> for unowned_host_case in unowned_host_cases : <EOL> for dirtiness_flag in _get_dirtiness_flags_for_outgoing_indices ( unowned_host_case , <EOL> tree_owners = tree_owners ) : <EOL> yield dirtiness_flag <EOL> def _get_dirtiness_flags_for_child_cases ( cases ) : <EOL> child_cases = case_db . get_reverse_indexed_cases ( [ c . case_id for c in cases ] ) <EOL> case_owner_map = dict ( ( case . case_id , case . owner_id ) for case in cases ) <EOL> for child_case in child_cases : <EOL> for index in child_case . indices : <EOL> if ( index . referenced_id in case_owner_map <EOL> and child_case . owner_id != case_owner_map [ index . referenced_id ] ) : <EOL> yield DirtinessFlag ( child_case . case_id , child_case . owner_id ) <EOL> def _get_dirtiness_flags_for_reassigned_case ( case_metas ) : <EOL> for case_update_meta in case_metas : <EOL> if _is_change_of_ownership ( case_update_meta . previous_owner_id , case_update_meta . case . owner_id ) : <EOL> yield DirtinessFlag ( case_update_meta . case . case_id , case_update_meta . previous_owner_id ) <EOL> dirtiness_flags = [ ] <EOL> extensions_to_close = set ( ) <EOL> dirtiness_flags += list ( _get_dirtiness_flags_for_reassigned_case ( touched_cases . values ( ) ) ) <EOL> for case_update_meta in touched_cases . values ( ) : <EOL> _validate_indices ( case_update_meta . case ) <EOL> extensions_to_close = extensions_to_close | get_extensions_to_close ( case_update_meta . case , domain ) <EOL> dirtiness_flags += list ( _get_dirtiness_flags_for_outgoing_indices ( case_update_meta . case ) ) <EOL> dirtiness_flags += list ( _get_dirtiness_flags_for_child_cases ( [ meta . case for meta in touched_cases . values ( ) ] ) ) <EOL> return CaseProcessingResult ( <EOL> domain , <EOL> [ update . case for update in touched_cases . values ( ) ] , <EOL> dirtiness_flags , <EOL> extensions_to_close <EOL> ) <EOL> def _is_change_of_ownership ( previous_owner_id , next_owner_id ) : <EOL> return ( <EOL> previous_owner_id <EOL> and previous_owner_id != UNOWNED_EXTENSION_OWNER_ID <EOL> and previous_owner_id != next_owner_id <EOL> ) <EOL> def get_extensions_to_close ( case , domain ) : <EOL> outgoing_extension_indices = [ index . relationship for index in case . indices <EOL> if index . relationship == CASE_INDEX_EXTENSION ] <EOL> if not outgoing_extension_indices and case . closed and EXTENSION_CASES_SYNC_ENABLED . enabled ( domain ) : <EOL> return get_extension_chain ( [ case . case_id ] , domain ) <EOL> else : <EOL> return set ( ) <EOL> def is_device_report ( doc ) : <EOL> """<STR_LIT>""" <EOL> device_report_xmlns = "<STR_LIT>" <EOL> def _from_form_dict ( doc ) : <EOL> return isinstance ( doc , dict ) and "<STR_LIT>" in doc and doc [ "<STR_LIT>" ] == device_report_xmlns <EOL> def _from_xform_instance ( doc ) : <EOL> return getattr ( doc , '<STR_LIT>' , None ) == device_report_xmlns <EOL> return _from_form_dict ( doc ) or _from_xform_instance ( doc ) <EOL> def has_case_id ( case_block ) : <EOL> return const . CASE_TAG_ID in case_block or const . CASE_ATTR_ID in case_block <EOL> CaseBlockWithPath = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT:path>' ] ) <EOL> def extract_case_blocks ( doc , include_path = False ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( doc , XFormInstance ) : <EOL> form = doc . to_json ( ) [ '<STR_LIT>' ] <EOL> elif isinstance ( doc , dict ) : <EOL> form = doc <EOL> else : <EOL> form = doc . form_data <EOL> return [ struct if include_path else struct . caseblock for struct in _extract_case_blocks ( form ) ] <EOL> def _extract_case_blocks ( data , path = None ) : <EOL> """<STR_LIT>""" <EOL> path = path or [ ] <EOL> if isinstance ( data , list ) : <EOL> for item in data : <EOL> for case_block in _extract_case_blocks ( item , path = path ) : <EOL> yield case_block <EOL> elif isinstance ( data , dict ) and not is_device_report ( data ) : <EOL> for key , value in data . items ( ) : <EOL> new_path = path + [ key ] <EOL> if const . CASE_TAG == key : <EOL> if isinstance ( value , list ) : <EOL> case_blocks = value <EOL> else : <EOL> case_blocks = [ value ] <EOL> for case_block in case_blocks : <EOL> if has_case_id ( case_block ) : <EOL> validate_phone_datetime ( <EOL> case_block . get ( '<STR_LIT>' ) , none_ok = True ) <EOL> yield CaseBlockWithPath ( caseblock = case_block , path = path ) <EOL> else : <EOL> for case_block in _extract_case_blocks ( value , path = new_path ) : <EOL> yield case_block <EOL> def get_case_updates ( xform ) : <EOL> return [ case_update_from_block ( cb ) for cb in extract_case_blocks ( xform ) ] <EOL> def get_case_ids_from_form ( xform ) : <EOL> return set ( cu . id for cu in get_case_updates ( xform ) ) <EOL> def cases_referenced_by_xform ( xform ) : <EOL> """<STR_LIT>""" <EOL> from corehq . form_processor . backends . couch . dbaccessors import CaseAccessorCouch <EOL> from corehq . form_processor . interfaces . dbaccessors import CaseAccessors <EOL> case_ids = get_case_ids_from_form ( xform ) <EOL> domain = get_and_check_xform_domain ( xform ) <EOL> case_accessor = CaseAccessors ( domain ) <EOL> if domain is None : <EOL> assert case_accessor . db_accessor == CaseAccessorCouch <EOL> return case_accessor . get_cases ( list ( case_ids ) ) </s>
