<s> from corehq . apps . sms . api import ( <EOL> MessageMetadata , <EOL> add_msg_tags , <EOL> send_sms_to_verified_number , <EOL> log_sms_exception , <EOL> ) <EOL> from corehq . apps . sms . messages import * <EOL> from corehq . apps . sms . util import format_message_list <EOL> from touchforms . formplayer . api import current_question <EOL> from corehq . apps . smsforms . app import ( <EOL> _get_responses , <EOL> _responses_to_text , <EOL> ) <EOL> from dateutil . parser import parse <EOL> from corehq . apps . smsforms . models import SQLXFormsSession <EOL> def form_session_handler ( v , text , msg ) : <EOL> """<STR_LIT>""" <EOL> multiple , session = get_single_open_session_or_close_multiple ( v . domain , v . owner_id ) <EOL> if multiple : <EOL> send_sms_to_verified_number ( v , get_message ( MSG_MULTIPLE_SESSIONS , v ) ) <EOL> return True <EOL> if session : <EOL> inbound_metadata = MessageMetadata ( <EOL> workflow = session . workflow , <EOL> reminder_id = session . reminder_id , <EOL> xforms_session_couch_id = session . _id , <EOL> ) <EOL> add_msg_tags ( msg , inbound_metadata ) <EOL> try : <EOL> answer_next_question ( v , text , msg , session ) <EOL> except Exception : <EOL> log_sms_exception ( msg ) <EOL> send_sms_to_verified_number ( v , get_message ( MSG_TOUCHFORMS_DOWN , v ) ) <EOL> return True <EOL> else : <EOL> return False <EOL> def get_single_open_session_or_close_multiple ( domain , contact_id ) : <EOL> """<STR_LIT>""" <EOL> sessions = SQLXFormsSession . get_all_open_sms_sessions ( domain , contact_id ) <EOL> count = sessions . count ( ) <EOL> if count > <NUM_LIT:1> : <EOL> for session in sessions : <EOL> session . end ( False ) <EOL> session . save ( ) <EOL> return ( True , None ) <EOL> session = sessions [ <NUM_LIT:0> ] if count == <NUM_LIT:1> else None <EOL> return ( False , session ) <EOL> def answer_next_question ( v , text , msg , session ) : <EOL> resp = current_question ( session . session_id ) <EOL> event = resp . event <EOL> valid , text , error_msg = validate_answer ( event , text , v ) <EOL> outbound_metadata = MessageMetadata ( <EOL> workflow = session . workflow , <EOL> reminder_id = session . reminder_id , <EOL> xforms_session_couch_id = session . _id , <EOL> ) <EOL> if valid : <EOL> responses = _get_responses ( v . domain , v . owner_id , text , <EOL> yield_responses = True ) <EOL> if has_invalid_response ( responses ) : <EOL> mark_as_invalid_response ( msg ) <EOL> text_responses = _responses_to_text ( responses ) <EOL> if len ( text_responses ) > <NUM_LIT:0> : <EOL> response_text = format_message_list ( text_responses ) <EOL> send_sms_to_verified_number ( v , response_text , <EOL> metadata = outbound_metadata ) <EOL> else : <EOL> mark_as_invalid_response ( msg ) <EOL> response_text = "<STR_LIT>" % ( error_msg , event . text_prompt ) <EOL> send_sms_to_verified_number ( v , response_text , <EOL> metadata = outbound_metadata ) <EOL> def validate_answer ( event , text , v ) : <EOL> text = text . strip ( ) <EOL> upper_text = text . upper ( ) <EOL> valid = False <EOL> error_msg = "<STR_LIT>" <EOL> if text == "<STR_LIT>" and event . _dict . get ( "<STR_LIT>" , False ) : <EOL> return ( False , text , get_message ( MSG_FIELD_REQUIRED , v ) ) <EOL> if event . datatype == "<STR_LIT>" : <EOL> choices = format_choices ( event . _dict [ "<STR_LIT>" ] ) <EOL> if upper_text in choices : <EOL> text = str ( choices [ upper_text ] ) <EOL> valid = True <EOL> else : <EOL> try : <EOL> answer = int ( text ) <EOL> if answer >= <NUM_LIT:1> and answer <= len ( event . _dict [ "<STR_LIT>" ] ) : <EOL> valid = True <EOL> else : <EOL> error_msg = get_message ( MSG_CHOICE_OUT_OF_RANGE , v ) <EOL> except ValueError : <EOL> error_msg = get_message ( MSG_INVALID_CHOICE , v ) <EOL> elif event . datatype == "<STR_LIT>" : <EOL> choices = format_choices ( event . _dict [ "<STR_LIT>" ] ) <EOL> max_index = len ( event . _dict [ "<STR_LIT>" ] ) <EOL> proposed_answers = text . split ( ) <EOL> final_answers = { } <EOL> try : <EOL> for answer in proposed_answers : <EOL> upper_answer = answer . upper ( ) <EOL> if upper_answer in choices : <EOL> final_answers [ str ( choices [ upper_answer ] ) ] = "<STR_LIT>" <EOL> else : <EOL> int_answer = int ( answer ) <EOL> assert int_answer >= <NUM_LIT:1> and int_answer <= max_index <EOL> final_answers [ str ( int_answer ) ] = "<STR_LIT>" <EOL> text = "<STR_LIT:U+0020>" . join ( final_answers . keys ( ) ) <EOL> valid = True <EOL> except Exception : <EOL> error_msg = get_message ( MSG_INVALID_CHOICE , v ) <EOL> elif event . datatype == "<STR_LIT:int>" : <EOL> try : <EOL> int ( text ) <EOL> valid = True <EOL> except ValueError : <EOL> error_msg = get_message ( MSG_INVALID_INT , v ) <EOL> elif event . datatype == "<STR_LIT:float>" : <EOL> try : <EOL> float ( text ) <EOL> valid = True <EOL> except ValueError : <EOL> error_msg = get_message ( MSG_INVALID_FLOAT , v ) <EOL> elif event . datatype == "<STR_LIT>" : <EOL> try : <EOL> long ( text ) <EOL> valid = True <EOL> except ValueError : <EOL> error_msg = get_message ( MSG_INVALID_LONG , v ) <EOL> elif event . datatype == "<STR_LIT:date>" : <EOL> try : <EOL> assert len ( text ) == <NUM_LIT:8> <EOL> int ( text ) <EOL> text = "<STR_LIT>" % ( text [ <NUM_LIT:0> : <NUM_LIT:4> ] , text [ <NUM_LIT:4> : <NUM_LIT:6> ] , text [ <NUM_LIT:6> : ] ) <EOL> parse ( text ) <EOL> valid = True <EOL> except Exception : <EOL> error_msg = get_message ( MSG_INVALID_DATE , v ) <EOL> elif event . datatype == "<STR_LIT:time>" : <EOL> try : <EOL> assert len ( text ) == <NUM_LIT:4> <EOL> hour = int ( text [ <NUM_LIT:0> : <NUM_LIT:2> ] ) <EOL> minute = int ( text [ <NUM_LIT:2> : ] ) <EOL> assert hour >= <NUM_LIT:0> and hour <= <NUM_LIT> <EOL> assert minute >= <NUM_LIT:0> and minute <= <NUM_LIT> <EOL> text = "<STR_LIT>" % ( hour , str ( minute ) . zfill ( <NUM_LIT:2> ) ) <EOL> valid = True <EOL> except Exception : <EOL> error_msg = get_message ( MSG_INVALID_TIME , v ) <EOL> else : <EOL> valid = True <EOL> return ( valid , text , error_msg ) <EOL> def format_choices ( choices_list ) : <EOL> choices = { } <EOL> for idx , choice in enumerate ( choices_list ) : <EOL> choices [ choice . strip ( ) . upper ( ) ] = idx + <NUM_LIT:1> <EOL> return choices <EOL> def has_invalid_response ( responses ) : <EOL> for r in responses : <EOL> if r . status == "<STR_LIT>" : <EOL> return True <EOL> return False <EOL> def mark_as_invalid_response ( msg ) : <EOL> msg . invalid_survey_response = True <EOL> msg . save ( ) </s>
