<s> from __future__ import absolute_import <EOL> from zerver . models import get_client <EOL> from zerver . lib . actions import check_send_message <EOL> from zerver . lib . response import json_success , json_error <EOL> from zerver . decorator import api_key_only_webhook_view <EOL> from defusedxml . ElementTree import fromstring as xml_fromstring <EOL> import logging <EOL> import re <EOL> import ujson <EOL> def api_pivotal_webhook_v3 ( request , user_profile , stream ) : <EOL> payload = xml_fromstring ( request . body ) <EOL> def get_text ( attrs ) : <EOL> start = payload <EOL> try : <EOL> for attr in attrs : <EOL> start = start . find ( attr ) <EOL> return start . text <EOL> except AttributeError : <EOL> return "<STR_LIT>" <EOL> event_type = payload . find ( '<STR_LIT>' ) . text <EOL> description = payload . find ( '<STR_LIT:description>' ) . text <EOL> project_id = payload . find ( '<STR_LIT>' ) . text <EOL> story_id = get_text ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:id>' ] ) <EOL> url = "<STR_LIT>" % ( project_id , story_id ) <EOL> name_re = re . compile ( r'<STR_LIT>' ) <EOL> match = name_re . match ( description ) <EOL> if match and len ( match . groups ( ) ) : <EOL> name = match . group ( <NUM_LIT:1> ) <EOL> else : <EOL> name = "<STR_LIT>" <EOL> more_info = "<STR_LIT>" % ( url , ) <EOL> if event_type == '<STR_LIT>' : <EOL> subject = name <EOL> content = description + more_info <EOL> elif event_type == '<STR_LIT>' : <EOL> subject = "<STR_LIT>" <EOL> content = description + more_info <EOL> elif event_type == '<STR_LIT>' : <EOL> issue_desc = get_text ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:description>' ] ) <EOL> issue_type = get_text ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> issue_status = get_text ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> estimate = get_text ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> if estimate != '<STR_LIT>' : <EOL> estimate = "<STR_LIT>" % ( estimate , ) <EOL> subject = name <EOL> content = "<STR_LIT>" % ( description , <EOL> issue_status , <EOL> issue_type , <EOL> estimate , <EOL> issue_desc , <EOL> more_info ) <EOL> return subject , content <EOL> def api_pivotal_webhook_v5 ( request , user_profile , stream ) : <EOL> payload = ujson . loads ( request . body ) <EOL> event_type = payload [ "<STR_LIT>" ] <EOL> project_name = payload [ "<STR_LIT>" ] [ "<STR_LIT:name>" ] <EOL> project_id = payload [ "<STR_LIT>" ] [ "<STR_LIT:id>" ] <EOL> primary_resources = payload [ "<STR_LIT>" ] [ <NUM_LIT:0> ] <EOL> story_url = primary_resources [ "<STR_LIT:url>" ] <EOL> story_type = primary_resources [ "<STR_LIT>" ] <EOL> story_id = primary_resources [ "<STR_LIT:id>" ] <EOL> story_name = primary_resources [ "<STR_LIT:name>" ] <EOL> performed_by = payload . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT>" ) <EOL> story_info = "<STR_LIT>" % ( project_name , project_id , story_name , story_url ) <EOL> changes = payload . get ( "<STR_LIT>" , [ ] ) <EOL> content = "<STR_LIT>" <EOL> subject = "<STR_LIT>" % ( story_id , story_name ) <EOL> def extract_comment ( change ) : <EOL> if change . get ( "<STR_LIT>" ) == "<STR_LIT>" : <EOL> return change . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:text>" , None ) <EOL> return None <EOL> if event_type == "<STR_LIT>" : <EOL> content += "<STR_LIT>" % ( performed_by , story_info ) <EOL> for change in changes : <EOL> old_values = change . get ( "<STR_LIT>" , { } ) <EOL> new_values = change [ "<STR_LIT>" ] <EOL> if "<STR_LIT>" in old_values and "<STR_LIT>" in new_values : <EOL> content += "<STR_LIT>" % ( old_values [ "<STR_LIT>" ] , new_values [ "<STR_LIT>" ] ) <EOL> if "<STR_LIT>" in old_values and "<STR_LIT>" in new_values : <EOL> old_estimate = old_values . get ( "<STR_LIT>" , None ) <EOL> if old_estimate is None : <EOL> estimate = "<STR_LIT>" <EOL> else : <EOL> estimate = "<STR_LIT>" % ( old_estimate , ) <EOL> new_estimate = new_values [ "<STR_LIT>" ] if new_values [ "<STR_LIT>" ] is not None else "<STR_LIT:0>" <EOL> content += "<STR_LIT>" % ( estimate , new_estimate ) <EOL> if "<STR_LIT>" in old_values and "<STR_LIT>" in new_values : <EOL> content += "<STR_LIT>" % ( old_values [ "<STR_LIT>" ] , new_values [ "<STR_LIT>" ] ) <EOL> comment = extract_comment ( change ) <EOL> if comment is not None : <EOL> content += "<STR_LIT>" % ( comment , ) <EOL> elif event_type == "<STR_LIT>" : <EOL> for change in changes : <EOL> comment = extract_comment ( change ) <EOL> if comment is not None : <EOL> content += "<STR_LIT>" % ( performed_by , story_info , comment ) <EOL> elif event_type == "<STR_LIT>" : <EOL> content += "<STR_LIT>" % ( performed_by , story_type , story_info ) <EOL> for change in changes : <EOL> new_values = change . get ( "<STR_LIT>" , { } ) <EOL> if "<STR_LIT>" in new_values : <EOL> content += "<STR_LIT>" % ( new_values [ "<STR_LIT>" ] , ) <EOL> if "<STR_LIT:description>" in new_values : <EOL> content += "<STR_LIT>" % ( new_values [ "<STR_LIT:description>" ] , ) <EOL> elif event_type == "<STR_LIT>" : <EOL> content = "<STR_LIT>" % ( performed_by , story_info ) <EOL> for change in changes : <EOL> old_values = change . get ( "<STR_LIT>" , { } ) <EOL> new_values = change [ "<STR_LIT>" ] <EOL> if "<STR_LIT>" in old_values and "<STR_LIT>" in new_values : <EOL> content += "<STR_LIT>" % ( old_values [ "<STR_LIT>" ] , new_values [ "<STR_LIT>" ] ) <EOL> elif event_type in [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] : <EOL> pass <EOL> else : <EOL> logging . warning ( "<STR_LIT>" % ( event_type , ) ) <EOL> return subject , content <EOL> @ api_key_only_webhook_view <EOL> def api_pivotal_webhook ( request , user_profile ) : <EOL> try : <EOL> stream = request . GET [ '<STR_LIT>' ] <EOL> except ( AttributeError , KeyError ) : <EOL> return json_error ( "<STR_LIT>" ) <EOL> subject = content = None <EOL> try : <EOL> subject , content = api_pivotal_webhook_v3 ( request , user_profile , stream ) <EOL> except AttributeError : <EOL> return json_error ( "<STR_LIT>" ) <EOL> except : <EOL> try : <EOL> subject , content = api_pivotal_webhook_v5 ( request , user_profile , stream ) <EOL> except AttributeError : <EOL> return json_error ( "<STR_LIT>" ) <EOL> if subject is None or content is None : <EOL> return json_error ( "<STR_LIT>" ) <EOL> check_send_message ( user_profile , get_client ( "<STR_LIT>" ) , "<STR_LIT>" , <EOL> [ stream ] , subject , content ) <EOL> return json_success ( ) </s>
