<s> from youtrack import YouTrackException , Issue <EOL> import youtrack <EOL> from youtrack . connection import Connection <EOL> from youtrack . importHelper import create_custom_field <EOL> import itertools <EOL> __author__ = '<STR_LIT:user>' <EOL> NAME = u'<STR_LIT:name>' <EOL> TYPE = u'<STR_LIT:type>' <EOL> POLICY = u'<STR_LIT>' <EOL> AUTO_ATTACHED = u'<STR_LIT>' <EOL> NUMBER_IN_PROJECT = u'<STR_LIT>' <EOL> class YouTrackImporter ( object ) : <EOL> def __init__ ( self , source , target , import_config ) : <EOL> self . _source = source <EOL> self . _target = target <EOL> self . _import_config = import_config <EOL> def do_import ( self , projects , new_projects_owner_login = u'<STR_LIT:root>' ) : <EOL> project_ids = projects . keys ( ) <EOL> self . _create_auto_attached_fields ( ) <EOL> self . _create_custom_fields ( project_ids ) <EOL> for project_id , project_name in projects . items ( ) : <EOL> self . _create_project ( project_id , project_name , new_projects_owner_login ) <EOL> self . _attach_fields_to_project ( project_id ) <EOL> self . _add_value_to_fields_in_project ( project_id ) <EOL> self . _import_issues ( project_id ) <EOL> self . _import_tags ( project_ids ) <EOL> self . _import_issue_links ( project_ids ) <EOL> def _create_auto_attached_fields ( self ) : <EOL> predefined_fields = self . _import_config . get_predefined_fields ( ) <EOL> for field in predefined_fields : <EOL> self . _create_field ( field [ NAME ] , field [ TYPE ] , field . get ( POLICY ) ) <EOL> def _create_field ( self , field_name , field_type , attach_bundle_policy , auto_attached = True ) : <EOL> if field_name in youtrack . EXISTING_FIELDS : <EOL> return <EOL> create_custom_field ( self . _target , field_type , field_name , auto_attached , bundle_policy = attach_bundle_policy ) <EOL> def _create_custom_fields ( self , project_ids ) : <EOL> custom_fields = self . _get_custom_fields_for_projects ( project_ids ) <EOL> for yt_field in custom_fields : <EOL> if ( yt_field is not None ) and ( yt_field [ TYPE ] is not None ) : <EOL> self . _create_field ( yt_field [ NAME ] , yt_field [ TYPE ] , yt_field . get ( POLICY ) , yt_field [ AUTO_ATTACHED ] ) <EOL> def _get_custom_fields_for_projects ( self , project_ids ) : <EOL> raise NotImplementedError ( ) <EOL> def _create_project ( self , project_id , project_name , project_lead_login ) : <EOL> try : <EOL> self . _target . getProject ( project_id ) <EOL> except YouTrackException : <EOL> self . _target . createProjectDetailed ( project_id , project_name , u'<STR_LIT>' , project_lead_login ) <EOL> def _attach_fields_to_project ( self , project_id ) : <EOL> for yt_field in self . _get_custom_fields_for_projects ( [ project_id ] ) : <EOL> if not yt_field [ u'<STR_LIT>' ] : <EOL> field_name = yt_field [ NAME ] <EOL> try : <EOL> self . _target . createProjectCustomFieldDetailed ( project_id , field_name , u'<STR_LIT>' + field_name ) <EOL> except YouTrackException : <EOL> print ( u'<STR_LIT>' % field_name ) <EOL> def _import_issues ( self , project_id ) : <EOL> limit = <NUM_LIT:100> <EOL> all_issues = self . _get_issues ( project_id ) <EOL> while True : <EOL> issues = list ( itertools . islice ( all_issues , None , limit ) ) <EOL> if not len ( issues ) : <EOL> break <EOL> self . _target . importIssues ( project_id , project_id + u'<STR_LIT>' , <EOL> [ self . _to_yt_issue ( issue , project_id ) for issue in issues ] ) <EOL> for issue in issues : <EOL> issue_id = self . _get_issue_id ( issue ) <EOL> issue_attachments = self . _get_attachments ( issue ) <EOL> yt_issue_id = u'<STR_LIT>' % ( project_id , issue_id ) <EOL> self . _import_attachments ( yt_issue_id , issue_attachments ) <EOL> def _import_tags ( self , project_ids ) : <EOL> limit = <NUM_LIT:100> <EOL> existing_tags = set ( [ ] ) <EOL> for project_id in project_ids : <EOL> all_tags = self . _get_issue_tags ( project_id ) <EOL> while True : <EOL> l = list ( itertools . islice ( all_tags , None , limit ) ) <EOL> if not len ( l ) : <EOL> break <EOL> issue_tags = zip ( * l ) [ <NUM_LIT:1> ] <EOL> existing_tags |= set ( [ item for tags in issue_tags for item in tags ] ) <EOL> self . _do_import_tags ( project_ids , existing_tags ) <EOL> def _is_prefix_of_any_other_tag ( self , tag , other_tags ) : <EOL> for t in other_tags : <EOL> if t . startswith ( tag ) and ( t != tag ) : <EOL> return True <EOL> return False <EOL> def _do_import_tags ( self , project_ids , collected_tags ) : <EOL> tags_to_import_now = set ( [ ] ) <EOL> tags_to_import_after = set ( [ ] ) <EOL> for tag in collected_tags : <EOL> if self . _is_prefix_of_any_other_tag ( tag , collected_tags ) : <EOL> tags_to_import_after . add ( tag ) <EOL> else : <EOL> tags_to_import_now . add ( tag ) <EOL> for project_id in project_ids : <EOL> for ( issue_id , tags ) in self . _get_issue_tags ( project_id ) : <EOL> yt_issue_id = u'<STR_LIT>' % ( project_id , issue_id ) <EOL> for tag in tags : <EOL> if tag in tags_to_import_now : <EOL> try : <EOL> self . _target . executeCommand ( yt_issue_id , u'<STR_LIT>' + tag ) <EOL> except YouTrackException : <EOL> print ( u'<STR_LIT>' % yt_issue_id ) <EOL> if len ( tags_to_import_after ) : <EOL> self . _do_import_tags ( project_ids , tags_to_import_after ) <EOL> def _import_issue_links ( self , project_ids ) : <EOL> limit = <NUM_LIT:100> <EOL> for project_id in project_ids : <EOL> after = <NUM_LIT:0> <EOL> while True : <EOL> links = self . _get_issue_links ( project_id , after , limit ) <EOL> if not len ( links ) : <EOL> break <EOL> self . _target . importLinks ( links ) <EOL> after += limit <EOL> def process_field ( self , key , project_id , result , value ) : <EOL> if value is None : <EOL> return <EOL> if isinstance ( value , list ) and not len ( value ) : <EOL> return <EOL> if ( isinstance ( value , str ) or isinstance ( value , unicode ) ) and not len ( value ) : <EOL> return <EOL> field_name = self . _get_field_name ( key , project_id ) <EOL> if field_name is None or field_name == NUMBER_IN_PROJECT : <EOL> return <EOL> field_type = self . _get_field_type ( field_name ) <EOL> if ( field_type is None ) and ( field_name not in youtrack . EXISTING_FIELDS ) : <EOL> return <EOL> value = self . get_field_value ( field_name , field_type , value ) <EOL> if isinstance ( value , list ) : <EOL> for v in value : <EOL> self . _add_value_to_field ( project_id , field_name , field_type , v ) <EOL> else : <EOL> self . _add_value_to_field ( project_id , field_name , field_type , value ) <EOL> if ( field_type is not None ) and field_type . startswith ( u'<STR_LIT:user>' ) : <EOL> if isinstance ( value , list ) : <EOL> value = [ v . login for v in value ] <EOL> else : <EOL> value = value . login <EOL> if not isinstance ( value , list ) : <EOL> value = str ( value ) <EOL> result [ field_name ] = value <EOL> def _to_yt_issue ( self , issue , project_id ) : <EOL> result = Issue ( ) <EOL> result . comments = [ self . _to_yt_comment ( comment ) for comment in self . _get_comments ( issue ) ] <EOL> result . numberInProject = self . _get_issue_id ( issue ) <EOL> for ( key , value ) in issue . items ( ) : <EOL> self . process_field ( key , project_id , result , value ) <EOL> return result <EOL> def _get_field_name ( self , field_name , project_id ) : <EOL> field_name = self . _import_config . get_field_name ( field_name ) <EOL> if field_name in youtrack . EXISTING_FIELDS : <EOL> return field_name <EOL> try : <EOL> self . _target . getProjectCustomField ( project_id , field_name ) <EOL> return field_name <EOL> except YouTrackException : <EOL> return None <EOL> def _get_field_type ( self , field_name ) : <EOL> if field_name in youtrack . EXISTING_FIELD_TYPES : <EOL> return youtrack . EXISTING_FIELD_TYPES [ field_name ] <EOL> try : <EOL> return self . _target . getCustomField ( field_name ) . type <EOL> except YouTrackException : <EOL> return None <EOL> def _import_user ( self , user ) : <EOL> self . _target . importUsers ( [ user ] ) <EOL> for group in user . getGroups ( ) : <EOL> try : <EOL> self . _target . createGroup ( group ) <EOL> except YouTrackException : <EOL> pass <EOL> self . _target . setUserGroup ( user . login , group . name ) <EOL> def _add_value_to_field ( self , project_id , field_name , field_type , value ) : <EOL> if ( field_type is not None ) and field_type . startswith ( u'<STR_LIT:user>' ) : <EOL> self . _import_user ( value ) <EOL> value = value . login <EOL> if field_name in youtrack . EXISTING_FIELDS : <EOL> return <EOL> custom_field = self . _target . getProjectCustomField ( project_id , field_name ) <EOL> if hasattr ( custom_field , u'<STR_LIT>' ) : <EOL> bundle = self . _target . getBundle ( field_type , custom_field . bundle ) <EOL> try : <EOL> self . _target . addValueToBundle ( bundle , value ) <EOL> except YouTrackException : <EOL> pass <EOL> def get_field_value ( self , field_name , field_type , value ) : <EOL> if value is None : <EOL> return None <EOL> values_map = self . _import_config . get_value_mapping ( field_name ) <EOL> if isinstance ( value , list ) : <EOL> return [ self . get_field_value ( field_name , field_type , v ) for v in value ] <EOL> if field_type . startswith ( u'<STR_LIT:user>' ) : <EOL> return self . _to_yt_user ( value ) <EOL> if field_type . lower ( ) == u"<STR_LIT:date>" : <EOL> return self . to_unix_date ( value ) <EOL> if isinstance ( value , basestring ) : <EOL> return values_map . get ( value , value ) <EOL> if isinstance ( value , int ) : <EOL> return values_map . get ( value , str ( value ) ) <EOL> def to_unix_date ( self , date ) : <EOL> return date <EOL> def _add_value_to_fields_in_project ( self , project_id ) : <EOL> for field in self . _get_fields_with_values ( project_id ) : <EOL> field_name = self . _get_field_name ( field [ NAME ] , project_id ) <EOL> pcf = self . _target . getProjectCustomField ( project_id , field_name ) <EOL> if hasattr ( pcf , u'<STR_LIT>' ) : <EOL> field_type = pcf . type [ <NUM_LIT:0> : - <NUM_LIT:3> ] <EOL> bundle = self . _target . getBundle ( field_type , pcf . bundle ) <EOL> yt_values = [ v for v in [ field [ u'<STR_LIT>' ] ( value , bundle , <EOL> lambda name , value_name : self . _import_config . get_field_value ( name , field_type , value_name ) ) for <EOL> value in <EOL> field [ u'<STR_LIT>' ] ] if len ( v ) ] <EOL> for value in yt_values : <EOL> self . _target . addValueToBundle ( bundle , value ) <EOL> def _get_issue_id ( self , issue ) : <EOL> return str ( issue [ self . _import_config . get_key_for_field_name ( NUMBER_IN_PROJECT ) ] ) <EOL> def _get_fields_with_values ( self , project_id ) : <EOL> return [ ] <EOL> def _to_yt_comment ( self , comment ) : <EOL> raise NotImplementedError <EOL> def _get_attachments ( self , issue_id ) : <EOL> return [ ] <EOL> def _get_issues ( self , project_id ) : <EOL> raise NotImplementedError <EOL> def _import_attachments ( self , issue_id , issue_attachments ) : <EOL> for attach in issue_attachments : <EOL> self . _target . createAttachmentFromAttachment ( issue_id , attach ) <EOL> def _get_comments ( self , issue ) : <EOL> raise NotImplementedError <EOL> def _get_issue_tags ( self , project_id ) : <EOL> key = self . _import_config . get_key_for_field_name ( u'<STR_LIT>' ) <EOL> return ( ( self . _get_issue_id ( issue ) , issue [ key ] ) for issue in self . _get_issues ( project_id ) if ( key in issue ) and len ( issue [ key ] ) ) <EOL> def _get_issue_links ( self , project_id , after , limit ) : <EOL> return [ ] <EOL> def _to_yt_user ( self , value ) : <EOL> raise NotImplementedError <EOL> class YouTrackImportConfig ( object ) : <EOL> def __init__ ( self , name_mapping , type_mapping , value_mapping = None , link_type_mapping = None ) : <EOL> self . _name_mapping = name_mapping <EOL> self . _type_mapping = type_mapping <EOL> self . _value_mapping = value_mapping if value_mapping is not None else { } <EOL> self . _link_type_mapping = link_type_mapping if link_type_mapping is not None else { } <EOL> def _get_default_auto_attached ( self ) : <EOL> return True <EOL> def _get_default_bundle_policy ( self ) : <EOL> return <NUM_LIT:0> <EOL> def get_field_name ( self , field_name ) : <EOL> return field_name if field_name not in self . _name_mapping else self . _name_mapping [ field_name ] <EOL> def get_predefined_fields ( self ) : <EOL> return [ ] <EOL> def get_link_type ( self , type ) : <EOL> return self . _link_type_mapping [ type ] if type in self . _link_type_mapping else type <EOL> def get_key_for_field_name ( self , field_name ) : <EOL> for ( key , value ) in self . _name_mapping . items ( ) : <EOL> if value == field_name : <EOL> return key <EOL> return field_name <EOL> def get_value_mapping ( self , field_name ) : <EOL> return self . _value_mapping [ field_name ] if field_name in self . _value_mapping else { } </s>
