<s> import logging <EOL> import uuid <EOL> from sqlalchemy import sql <EOL> from acoustid import tables as schema , const <EOL> from acoustid . data . fingerprint import lookup_fingerprint , insert_fingerprint , inc_fingerprint_submission_count , FingerprintSearcher <EOL> from acoustid . data . musicbrainz import resolve_mbid_redirect <EOL> logger = logging . getLogger ( __name__ ) <EOL> def resolve_track_gid ( conn , gid ) : <EOL> query = sql . select ( [ schema . track . c . id , schema . track . c . new_id ] , <EOL> schema . track . c . gid == gid ) <EOL> row = conn . execute ( query ) . first ( ) <EOL> if row is None : <EOL> return None <EOL> track_id , new_track_id = row <EOL> if new_track_id is None : <EOL> return track_id <EOL> query = sql . select ( [ schema . track . c . id ] , <EOL> schema . track . c . id == new_track_id ) <EOL> return conn . execute ( query ) . scalar ( ) <EOL> def lookup_mbids ( conn , track_ids ) : <EOL> """<STR_LIT>""" <EOL> if not track_ids : <EOL> return { } <EOL> query = sql . select ( [ <EOL> schema . track_mbid . c . track_id , <EOL> schema . track_mbid . c . mbid , <EOL> schema . track_mbid . c . submission_count , <EOL> ] ) <EOL> query = query . where ( sql . and_ ( schema . track_mbid . c . track_id . in_ ( track_ids ) , schema . track_mbid . c . disabled == False ) ) <EOL> query = query . order_by ( schema . track_mbid . c . mbid ) <EOL> results = { } <EOL> for track_id , mbid , sources in conn . execute ( query ) : <EOL> results . setdefault ( track_id , [ ] ) . append ( ( mbid , sources ) ) <EOL> return results <EOL> def lookup_meta_ids ( conn , track_ids ) : <EOL> if not track_ids : <EOL> return { } <EOL> query = sql . select ( <EOL> [ schema . track_meta . c . track_id , schema . track_meta . c . meta_id ] , <EOL> sql . and_ ( schema . track_meta . c . track_id . in_ ( track_ids ) ) ) . order_by ( schema . track_meta . c . meta_id ) <EOL> results = { } <EOL> for track_id , meta_id in conn . execute ( query ) : <EOL> results . setdefault ( track_id , [ ] ) . append ( meta_id ) <EOL> return results <EOL> def lookup_tracks ( conn , mbids ) : <EOL> if not mbids : <EOL> return { } <EOL> query = sql . select ( <EOL> [ schema . track_mbid . c . track_id , schema . track . c . gid , schema . track_mbid . c . mbid ] , <EOL> sql . and_ ( schema . track_mbid . c . mbid . in_ ( mbids ) , schema . track_mbid . c . disabled == False ) , <EOL> from_obj = schema . track_mbid . join ( schema . track , schema . track_mbid . c . track_id == schema . track . c . id ) ) . order_by ( schema . track_mbid . c . track_id ) <EOL> results = { } <EOL> for track_id , track_gid , mbid in conn . execute ( query ) : <EOL> results . setdefault ( mbid , [ ] ) . append ( { '<STR_LIT:id>' : track_id , '<STR_LIT>' : track_gid } ) <EOL> return results <EOL> def merge_mbids ( conn , target_mbid , source_mbids ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" , '<STR_LIT:U+002CU+0020>' . join ( source_mbids ) , target_mbid ) <EOL> with conn . begin ( ) : <EOL> query = sql . select ( <EOL> [ <EOL> sql . func . min ( schema . track_mbid . c . id ) . label ( '<STR_LIT:id>' ) , <EOL> sql . func . array_agg ( schema . track_mbid . c . id ) . label ( '<STR_LIT>' ) , <EOL> schema . track_mbid . c . track_id , <EOL> sql . func . every ( schema . track_mbid . c . disabled ) . label ( '<STR_LIT>' ) , <EOL> sql . func . sum ( schema . track_mbid . c . submission_count ) . label ( '<STR_LIT:count>' ) , <EOL> ] , <EOL> schema . track_mbid . c . mbid . in_ ( source_mbids + [ target_mbid ] ) , <EOL> group_by = schema . track_mbid . c . track_id ) <EOL> rows = conn . execute ( query ) . fetchall ( ) <EOL> to_delete = set ( ) <EOL> to_update = [ ] <EOL> for row in rows : <EOL> old_ids = set ( row [ '<STR_LIT>' ] ) <EOL> old_ids . remove ( row [ '<STR_LIT:id>' ] ) <EOL> to_delete . update ( old_ids ) <EOL> to_update . append ( ( old_ids , row ) ) <EOL> if old_ids : <EOL> update_stmt = schema . track_mbid_source . update ( ) . where ( schema . track_mbid_source . c . track_mbid_id . in_ ( old_ids ) ) <EOL> conn . execute ( update_stmt . values ( track_mbid_id = row [ '<STR_LIT:id>' ] ) ) <EOL> update_stmt = schema . track_mbid_change . update ( ) . where ( schema . track_mbid_change . c . track_mbid_id . in_ ( old_ids ) ) <EOL> conn . execute ( update_stmt . values ( track_mbid_id = row [ '<STR_LIT:id>' ] ) ) <EOL> update_stmt = schema . track_mbid_flag . update ( ) . where ( schema . track_mbid_flag . c . track_mbid_id . in_ ( old_ids ) ) <EOL> conn . execute ( update_stmt . values ( track_mbid_id = row [ '<STR_LIT:id>' ] ) ) <EOL> if to_delete : <EOL> delete_stmt = schema . track_mbid . delete ( ) . where ( <EOL> schema . track_mbid . c . id . in_ ( to_delete ) ) <EOL> conn . execute ( delete_stmt ) <EOL> for old_ids , row in to_update : <EOL> update_stmt = schema . track_mbid . update ( ) . where ( schema . track_mbid . c . id == row [ '<STR_LIT:id>' ] ) <EOL> conn . execute ( update_stmt . values ( submission_count = row [ '<STR_LIT:count>' ] , <EOL> mbid = target_mbid , disabled = row [ '<STR_LIT>' ] ) ) <EOL> def merge_missing_mbids ( conn ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" ) <EOL> results = conn . execute ( """<STR_LIT>""" ) <EOL> merge = { } <EOL> for old_mbid , new_mbid in results : <EOL> merge . setdefault ( new_mbid , [ ] ) . append ( old_mbid ) <EOL> for new_mbid , old_mbids in merge . iteritems ( ) : <EOL> merge_mbids ( conn , new_mbid , old_mbids ) <EOL> def _merge_tracks_gids ( conn , name_with_id , target_id , source_ids ) : <EOL> name = name_with_id . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> tab = schema . metadata . tables [ '<STR_LIT>' % name ] <EOL> col = tab . columns [ name_with_id ] <EOL> tab_src = schema . metadata . tables [ '<STR_LIT>' % name ] <EOL> col_src = tab_src . columns [ '<STR_LIT>' % name ] <EOL> if name == '<STR_LIT>' : <EOL> tab_chg = schema . metadata . tables [ '<STR_LIT>' % name ] <EOL> col_chg = tab_chg . columns [ '<STR_LIT>' % name ] <EOL> tab_flag = schema . metadata . tables [ '<STR_LIT>' % name ] <EOL> col_flag = tab_flag . columns [ '<STR_LIT>' % name ] <EOL> columns = [ <EOL> sql . func . min ( tab . c . id ) . label ( '<STR_LIT:id>' ) , <EOL> sql . func . array_agg ( tab . c . id ) . label ( '<STR_LIT>' ) , <EOL> sql . func . sum ( tab . c . submission_count ) . label ( '<STR_LIT:count>' ) , <EOL> ] <EOL> if name == '<STR_LIT>' : <EOL> columns . append ( sql . func . every ( schema . track_mbid . c . disabled ) . label ( '<STR_LIT>' ) ) <EOL> query = sql . select ( columns , tab . c . track_id . in_ ( source_ids + [ target_id ] ) , group_by = col ) <EOL> rows = conn . execute ( query ) . fetchall ( ) <EOL> to_delete = set ( ) <EOL> to_update = [ ] <EOL> for row in rows : <EOL> old_ids = set ( row [ '<STR_LIT>' ] ) <EOL> old_ids . remove ( row [ '<STR_LIT:id>' ] ) <EOL> to_delete . update ( old_ids ) <EOL> to_update . append ( ( old_ids , row ) ) <EOL> if old_ids : <EOL> update_stmt = tab_src . update ( ) . where ( col_src . in_ ( old_ids ) ) <EOL> conn . execute ( update_stmt . values ( { col_src : row [ '<STR_LIT:id>' ] } ) ) <EOL> if name == '<STR_LIT>' : <EOL> update_stmt = tab_chg . update ( ) . where ( col_chg . in_ ( old_ids ) ) <EOL> conn . execute ( update_stmt . values ( { col_chg : row [ '<STR_LIT:id>' ] } ) ) <EOL> update_stmt = tab_flag . update ( ) . where ( col_flag . in_ ( old_ids ) ) <EOL> conn . execute ( update_stmt . values ( { col_flag : row [ '<STR_LIT:id>' ] } ) ) <EOL> if to_delete : <EOL> delete_stmt = tab . delete ( ) . where ( tab . c . id . in_ ( to_delete ) ) <EOL> conn . execute ( delete_stmt ) <EOL> for old_ids , row in to_update : <EOL> update_stmt = tab . update ( ) . where ( tab . c . id == row [ '<STR_LIT:id>' ] ) <EOL> if name == '<STR_LIT>' : <EOL> conn . execute ( update_stmt . values ( submission_count = row [ '<STR_LIT:count>' ] , track_id = target_id , disabled = row [ '<STR_LIT>' ] ) ) <EOL> else : <EOL> conn . execute ( update_stmt . values ( submission_count = row [ '<STR_LIT:count>' ] , track_id = target_id ) ) <EOL> def merge_tracks ( conn , target_id , source_ids ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" , '<STR_LIT:U+002CU+0020>' . join ( map ( str , source_ids ) ) , target_id ) <EOL> with conn . begin ( ) : <EOL> _merge_tracks_gids ( conn , '<STR_LIT>' , target_id , source_ids ) <EOL> _merge_tracks_gids ( conn , '<STR_LIT>' , target_id , source_ids ) <EOL> _merge_tracks_gids ( conn , '<STR_LIT>' , target_id , source_ids ) <EOL> _merge_tracks_gids ( conn , '<STR_LIT>' , target_id , source_ids ) <EOL> update_stmt = schema . fingerprint . update ( ) . where ( <EOL> schema . fingerprint . c . track_id . in_ ( source_ids ) ) <EOL> conn . execute ( update_stmt . values ( track_id = target_id ) ) <EOL> update_stmt = schema . track . update ( ) . where ( <EOL> sql . or_ ( schema . track . c . id . in_ ( source_ids ) , <EOL> schema . track . c . new_id . in_ ( source_ids ) ) ) <EOL> conn . execute ( update_stmt . values ( new_id = target_id ) ) <EOL> def insert_track ( conn ) : <EOL> """<STR_LIT>""" <EOL> insert_stmt = schema . track . insert ( ) . values ( { '<STR_LIT>' : str ( uuid . uuid4 ( ) ) } ) <EOL> id = conn . execute ( insert_stmt ) . inserted_primary_key [ <NUM_LIT:0> ] <EOL> logger . debug ( "<STR_LIT>" , id ) <EOL> return id <EOL> def _insert_gid ( conn , tab , tab_src , col , name , track_id , gid , submission_id = None , source_id = None ) : <EOL> cond = sql . and_ ( tab . c . track_id == track_id , col == gid ) <EOL> query = sql . select ( [ tab . c . id ] , cond ) <EOL> id = conn . execute ( query ) . scalar ( ) <EOL> if id is not None : <EOL> update_stmt = tab . update ( ) . where ( cond ) <EOL> values = { '<STR_LIT>' : sql . text ( '<STR_LIT>' ) } <EOL> conn . execute ( update_stmt . values ( ** values ) ) <EOL> else : <EOL> insert_stmt = tab . insert ( ) . values ( { <EOL> '<STR_LIT>' : track_id , name : gid , <EOL> '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> id = conn . execute ( insert_stmt ) . inserted_primary_key [ <NUM_LIT:0> ] <EOL> logger . debug ( "<STR_LIT>" , name . upper ( ) , gid , track_id ) <EOL> insert_stmt = tab_src . insert ( ) . values ( { <EOL> '<STR_LIT>' % name . replace ( '<STR_LIT>' , '<STR_LIT>' ) : id , <EOL> '<STR_LIT>' : submission_id , <EOL> '<STR_LIT>' : source_id , <EOL> } ) <EOL> conn . execute ( insert_stmt ) <EOL> return True <EOL> def insert_mbid ( conn , track_id , mbid , submission_id = None , source_id = None ) : <EOL> return _insert_gid ( conn , schema . track_mbid , schema . track_mbid_source , <EOL> schema . track_mbid . c . mbid , '<STR_LIT>' , track_id , mbid , submission_id , source_id ) <EOL> def insert_puid ( conn , track_id , puid , submission_id = None , source_id = None ) : <EOL> return _insert_gid ( conn , schema . track_puid , schema . track_puid_source , <EOL> schema . track_puid . c . puid , '<STR_LIT>' , track_id , puid , submission_id , source_id ) <EOL> def insert_track_foreignid ( conn , track_id , foreignid_id , submission_id = None , source_id = None ) : <EOL> return _insert_gid ( conn , schema . track_foreignid , schema . track_foreignid_source , <EOL> schema . track_foreignid . c . foreignid_id , '<STR_LIT>' , track_id , foreignid_id , <EOL> submission_id , source_id ) <EOL> def insert_track_meta ( conn , track_id , meta_id , submission_id = None , source_id = None ) : <EOL> return _insert_gid ( conn , schema . track_meta , schema . track_meta_source , <EOL> schema . track_meta . c . meta_id , '<STR_LIT>' , track_id , meta_id , submission_id , source_id ) <EOL> def calculate_fingerprint_similarity_matrix ( conn , track_ids ) : <EOL> fp1 = schema . fingerprint . alias ( '<STR_LIT>' ) <EOL> fp2 = schema . fingerprint . alias ( '<STR_LIT>' ) <EOL> src = fp1 . join ( fp2 , fp1 . c . id < fp2 . c . id ) <EOL> cond = sql . and_ ( fp1 . c . track_id . in_ ( track_ids ) , fp2 . c . track_id . in_ ( track_ids ) ) <EOL> query = sql . select ( [ <EOL> fp1 . c . id , fp2 . c . id , <EOL> sql . func . acoustid_compare2 ( fp1 . c . fingerprint , fp2 . c . fingerprint , const . TRACK_MAX_OFFSET ) , <EOL> ] , cond , from_obj = src ) . order_by ( fp1 . c . id , fp2 . c . id ) <EOL> result = { } <EOL> for fp1_id , fp2_id , score in conn . execute ( query ) : <EOL> result . setdefault ( fp1_id , { } ) [ fp2_id ] = score <EOL> result . setdefault ( fp2_id , { } ) [ fp1_id ] = score <EOL> result . setdefault ( fp1_id , { } ) [ fp1_id ] = <NUM_LIT:1.0> <EOL> result . setdefault ( fp2_id , { } ) [ fp2_id ] = <NUM_LIT:1.0> <EOL> return result <EOL> def can_merge_tracks ( conn , track_ids ) : <EOL> fp1 = schema . fingerprint . alias ( '<STR_LIT>' ) <EOL> fp2 = schema . fingerprint . alias ( '<STR_LIT>' ) <EOL> join_cond = sql . and_ ( fp1 . c . id < fp2 . c . id , fp1 . c . track_id < fp2 . c . track_id ) <EOL> src = fp1 . join ( fp2 , join_cond ) <EOL> cond = sql . and_ ( fp1 . c . track_id . in_ ( track_ids ) , fp2 . c . track_id . in_ ( track_ids ) ) <EOL> query = sql . select ( [ <EOL> fp1 . c . track_id , fp2 . c . track_id , <EOL> sql . func . max ( sql . func . abs ( fp1 . c . length - fp2 . c . length ) ) , <EOL> sql . func . min ( sql . func . acoustid_compare2 ( fp1 . c . fingerprint , fp2 . c . fingerprint , const . TRACK_MAX_OFFSET ) ) , <EOL> ] , cond , from_obj = src ) . group_by ( fp1 . c . track_id , fp2 . c . track_id ) . order_by ( fp1 . c . track_id , fp2 . c . track_id ) <EOL> rows = conn . execute ( query ) <EOL> merges = { } <EOL> for fp1_id , fp2_id , length_diff , score in rows : <EOL> if score < const . TRACK_GROUP_MERGE_THRESHOLD : <EOL> continue <EOL> if length_diff > const . FINGERPRINT_MAX_LENGTH_DIFF : <EOL> continue <EOL> group = fp1_id <EOL> if group in merges : <EOL> group = merges [ group ] <EOL> merges [ fp2_id ] = group <EOL> result = [ ] <EOL> for group in set ( merges . values ( ) ) : <EOL> result . append ( set ( [ group ] + [ i for i in merges if merges [ i ] == group ] ) ) <EOL> return result <EOL> def can_add_fp_to_track ( conn , track_id , fingerprint , length ) : <EOL> cond = schema . fingerprint . c . track_id == track_id <EOL> query = sql . select ( [ <EOL> sql . func . acoustid_compare2 ( schema . fingerprint . c . fingerprint , fingerprint , const . TRACK_MAX_OFFSET ) , <EOL> schema . fingerprint . c . length , <EOL> ] , cond , from_obj = schema . fingerprint ) <EOL> for fp_score , fp_length in conn . execute ( query ) : <EOL> if fp_score < const . TRACK_GROUP_MERGE_THRESHOLD : <EOL> return False <EOL> if abs ( fp_length - length ) > const . FINGERPRINT_MAX_LENGTH_DIFF : <EOL> return False <EOL> return True <EOL> def find_track_duplicates ( conn , fingerprint , index = None ) : <EOL> with conn . begin ( ) : <EOL> searcher = FingerprintSearcher ( conn , index ) <EOL> searcher . min_score = const . TRACK_MERGE_THRESHOLD <EOL> matches = searcher . search ( fingerprint [ '<STR_LIT>' ] , fingerprint [ '<STR_LIT>' ] ) <EOL> if not matches : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> return <EOL> logged = False <EOL> match = matches [ <NUM_LIT:0> ] <EOL> all_track_ids = set ( ) <EOL> possible_track_ids = set ( ) <EOL> for m in matches : <EOL> if m [ '<STR_LIT>' ] in all_track_ids : <EOL> continue <EOL> all_track_ids . add ( m [ '<STR_LIT>' ] ) <EOL> if can_add_fp_to_track ( conn , m [ '<STR_LIT>' ] , fingerprint [ '<STR_LIT>' ] , fingerprint [ '<STR_LIT>' ] ) : <EOL> if m [ '<STR_LIT:id>' ] != fingerprint [ '<STR_LIT:id>' ] : <EOL> if not logged : <EOL> logger . debug ( "<STR_LIT>" , fingerprint [ '<STR_LIT:id>' ] ) <EOL> logged = True <EOL> logger . debug ( "<STR_LIT>" , m [ '<STR_LIT:id>' ] , m [ '<STR_LIT>' ] , m [ '<STR_LIT>' ] * <NUM_LIT:100> ) <EOL> possible_track_ids . add ( m [ '<STR_LIT>' ] ) <EOL> if len ( possible_track_ids ) > <NUM_LIT:1> : <EOL> for group in can_merge_tracks ( conn , possible_track_ids ) : <EOL> if len ( group ) > <NUM_LIT:1> : <EOL> target_track_id = min ( group ) <EOL> group . remove ( target_track_id ) <EOL> merge_tracks ( conn , target_track_id , list ( group ) ) <EOL> break <EOL> conn . execute ( "<STR_LIT>" , fingerprint [ '<STR_LIT:id>' ] ) <EOL> def find_duplicates ( conn , limit = <NUM_LIT:50> , index = None ) : <EOL> query = "<STR_LIT>" <EOL> for fingerprint in conn . execute ( query ) : <EOL> find_track_duplicates ( conn , fingerprint , index = index ) </s>
