ci_mapping/run_pipeline.py
def parse_mag(self):
        """Parse MAG responses to PostgreSQL."""
        # Connect to postgresql
        s = self._create_session()

        # Read MAG responses
        data = []
        for filename in glob.iglob("".join([self.external_data, "*.pickle"])):
            with open(filename, "rb") as h:
                data.extend(pickle.load(h))

        # Collect IDs from tables to ensure we're not inserting duplicates
        paper_ids = {id_[0] for id_ in s.query(Paper.id)}
        author_ids = {id_[0] for id_ in s.query(Author.id)}
        fos_ids = {id_[0] for id_ in s.query(FieldOfStudy.id)}
        aff_ids = {id_[0] for id_ in s.query(Affiliation.id)}

        # Remove duplicates and keep only papers that are not already in the mag_papers table.
        data = [
            d for d in unique_dicts_by_value(data, "Id") if d["Id"] not in paper_ids
        ]
        logger.info(f"Number of unique  papers not existing in DB: {len(data)}")

        papers = [parse_papers(response) for response in data]
        logger.info(f"Completed parsing papers: {len(papers)}")

        journals = [
            parse_journal(response, response["Id"])
            for response in data
            if "J" in response.keys()
        ]
        logger.info(f"Completed parsing journals: {len(journals)}")

        conferences = [
            parse_conference(response, response["Id"])
            for response in data
            if "C" in response.keys()
        ]
        logger.info(f"Completed parsing conferences: {len(conferences)}")

        # Parse author information
        items = [parse_authors(response, response["Id"]) for response in data]
        authors = [
            d
            for d in unique_dicts_by_value(
                flatten_lists([item[0] for item in items]), "id"
            )
            if d["id"] not in author_ids
        ]

        paper_with_authors = unique_dicts(flatten_lists([item[1] for item in items]))
        logger.info(f"Completed parsing authors: {len(authors)}")
        logger.info(f"Completed parsing papers_with_authors: {len(paper_with_authors)}")

        # Parse Fields of Study
        items = [
            parse_fos(response, response["Id"])
            for response in data
            if "F" in response.keys()
        ]
        paper_with_fos = unique_dicts(flatten_lists([item[0] for item in items]))
        fields_of_study = [
            d
            for d in unique_dicts(flatten_lists([item[1] for item in items]))
            if d["id"] not in fos_ids
        ]
        logger.info(f"Completed parsing fields_of_study: {len(fields_of_study)}")
        logger.info(f"Completed parsing paper_with_fos: {len(paper_with_fos)}")

        # Parse affiliations
        items = [parse_affiliations(response, response["Id"]) for response in data]
        affiliations = [
            d
            for d in unique_dicts(flatten_lists([item[0] for item in items]))
            if d["id"] not in aff_ids
        ]
        paper_author_aff = unique_dicts(flatten_lists([item[1] for item in items]))
        logger.info(f"Completed parsing affiliations: {len(affiliations)}")
        logger.info(f"Completed parsing author_with_aff: {len(paper_author_aff)}")

        logger.info("Parsing completed!")

        # Insert dicts into postgresql
        s.bulk_insert_mappings(Paper, papers)
        s.bulk_insert_mappings(Journal, journals)
        s.bulk_insert_mappings(Conference, conferences)
        s.bulk_insert_mappings(Author, authors)
        s.bulk_insert_mappings(PaperAuthor, paper_with_authors)
        s.bulk_insert_mappings(FieldOfStudy, fields_of_study)
        s.bulk_insert_mappings(PaperFieldsOfStudy, paper_with_fos)
        s.bulk_insert_mappings(Affiliation, affiliations)
        s.bulk_insert_mappings(AuthorAffiliation, paper_author_aff)
        s.commit()
        logger.info("Committed to DB!")

        self.next(self.collect_fields_of_study_level)
Parse MAG responses to PostgreSQL.
Parse MAG responses to PostgreSQL.
Parse MAG responses to PostgreSQL .
def parse_mag(self):
        s = self._create_session()
        data = []
        for filename in glob.iglob("".join([self.external_data, "*.pickle"])):
            with open(filename, "rb") as h:
                data.extend(pickle.load(h))
        paper_ids = {id_[0] for id_ in s.query(Paper.id)}
        author_ids = {id_[0] for id_ in s.query(Author.id)}
        fos_ids = {id_[0] for id_ in s.query(FieldOfStudy.id)}
        aff_ids = {id_[0] for id_ in s.query(Affiliation.id)}
        data = [
            d for d in unique_dicts_by_value(data, "Id") if d["Id"] not in paper_ids
        ]
        logger.info(f"Number of unique  papers not existing in DB: {len(data)}")
        papers = [parse_papers(response) for response in data]
        logger.info(f"Completed parsing papers: {len(papers)}")
        journals = [
            parse_journal(response, response["Id"])
            for response in data
            if "J" in response.keys()
        ]
        logger.info(f"Completed parsing journals: {len(journals)}")
        conferences = [
            parse_conference(response, response["Id"])
            for response in data
            if "C" in response.keys()
        ]
        logger.info(f"Completed parsing conferences: {len(conferences)}")
        items = [parse_authors(response, response["Id"]) for response in data]
        authors = [
            d
            for d in unique_dicts_by_value(
                flatten_lists([item[0] for item in items]), "id"
            )
            if d["id"] not in author_ids
        ]
        paper_with_authors = unique_dicts(flatten_lists([item[1] for item in items]))
        logger.info(f"Completed parsing authors: {len(authors)}")
        logger.info(f"Completed parsing papers_with_authors: {len(paper_with_authors)}")
        items = [
            parse_fos(response, response["Id"])
            for response in data
            if "F" in response.keys()
        ]
        paper_with_fos = unique_dicts(flatten_lists([item[0] for item in items]))
        fields_of_study = [
            d
            for d in unique_dicts(flatten_lists([item[1] for item in items]))
            if d["id"] not in fos_ids
        ]
        logger.info(f"Completed parsing fields_of_study: {len(fields_of_study)}")
        logger.info(f"Completed parsing paper_with_fos: {len(paper_with_fos)}")
        items = [parse_affiliations(response, response["Id"]) for response in data]
        affiliations = [
            d
            for d in unique_dicts(flatten_lists([item[0] for item in items]))
            if d["id"] not in aff_ids
        ]
        paper_author_aff = unique_dicts(flatten_lists([item[1] for item in items]))
        logger.info(f"Completed parsing affiliations: {len(affiliations)}")
        logger.info(f"Completed parsing author_with_aff: {len(paper_author_aff)}")
        logger.info("Parsing completed!")
        s.bulk_insert_mappings(Paper, papers)
        s.bulk_insert_mappings(Journal, journals)
        s.bulk_insert_mappings(Conference, conferences)
        s.bulk_insert_mappings(Author, authors)
        s.bulk_insert_mappings(PaperAuthor, paper_with_authors)
        s.bulk_insert_mappings(FieldOfStudy, fields_of_study)
        s.bulk_insert_mappings(PaperFieldsOfStudy, paper_with_fos)
        s.bulk_insert_mappings(Affiliation, affiliations)
        s.bulk_insert_mappings(AuthorAffiliation, paper_author_aff)
        s.commit()
        logger.info("Committed to DB!")
        self.next(self.collect_fields_of_study_level)
def parse_mag ( self ) : s = self . _create_session ( ) data = [ ] for filename in glob . iglob ( "" . join ( [ self . external_data , "*.pickle" ] ) ) : with open ( filename , "rb" ) as h : data . extend ( pickle . load ( h ) ) paper_ids = { id_ [ 0 ] for id_ in s . query ( Paper . id ) } author_ids = { id_ [ 0 ] for id_ in s . query ( Author . id ) } fos_ids = { id_ [ 0 ] for id_ in s . query ( FieldOfStudy . id ) } aff_ids = { id_ [ 0 ] for id_ in s . query ( Affiliation . id ) } data = [ d for d in unique_dicts_by_value ( data , "Id" ) if d [ "Id" ] not in paper_ids ] logger . info ( f"Number of unique  papers not existing in DB: {len(data)}" ) papers = [ parse_papers ( response ) for response in data ] logger . info ( f"Completed parsing papers: {len(papers)}" ) journals = [ parse_journal ( response , response [ "Id" ] ) for response in data if "J" in response . keys ( ) ] logger . info ( f"Completed parsing journals: {len(journals)}" ) conferences = [ parse_conference ( response , response [ "Id" ] ) for response in data if "C" in response . keys ( ) ] logger . info ( f"Completed parsing conferences: {len(conferences)}" ) items = [ parse_authors ( response , response [ "Id" ] ) for response in data ] authors = [ d for d in unique_dicts_by_value ( flatten_lists ( [ item [ 0 ] for item in items ] ) , "id" ) if d [ "id" ] not in author_ids ] paper_with_authors = unique_dicts ( flatten_lists ( [ item [ 1 ] for item in items ] ) ) logger . info ( f"Completed parsing authors: {len(authors)}" ) logger . info ( f"Completed parsing papers_with_authors: {len(paper_with_authors)}" ) items = [ parse_fos ( response , response [ "Id" ] ) for response in data if "F" in response . keys ( ) ] paper_with_fos = unique_dicts ( flatten_lists ( [ item [ 0 ] for item in items ] ) ) fields_of_study = [ d for d in unique_dicts ( flatten_lists ( [ item [ 1 ] for item in items ] ) ) if d [ "id" ] not in fos_ids ] logger . info ( f"Completed parsing fields_of_study: {len(fields_of_study)}" ) logger . info ( f"Completed parsing paper_with_fos: {len(paper_with_fos)}" ) items = [ parse_affiliations ( response , response [ "Id" ] ) for response in data ] affiliations = [ d for d in unique_dicts ( flatten_lists ( [ item [ 0 ] for item in items ] ) ) if d [ "id" ] not in aff_ids ] paper_author_aff = unique_dicts ( flatten_lists ( [ item [ 1 ] for item in items ] ) ) logger . info ( f"Completed parsing affiliations: {len(affiliations)}" ) logger . info ( f"Completed parsing author_with_aff: {len(paper_author_aff)}" ) logger . info ( "Parsing completed!" ) s . bulk_insert_mappings ( Paper , papers ) s . bulk_insert_mappings ( Journal , journals ) s . bulk_insert_mappings ( Conference , conferences ) s . bulk_insert_mappings ( Author , authors ) s . bulk_insert_mappings ( PaperAuthor , paper_with_authors ) s . bulk_insert_mappings ( FieldOfStudy , fields_of_study ) s . bulk_insert_mappings ( PaperFieldsOfStudy , paper_with_fos ) s . bulk_insert_mappings ( Affiliation , affiliations ) s . bulk_insert_mappings ( AuthorAffiliation , paper_author_aff ) s . commit ( ) logger . info ( "Committed to DB!" ) self . next ( self . collect_fields_of_study_level )
Parse MAG responses to PostgreSQL.
Parse MAG responses to PostgreSQL .
"""Parse MAG responses to PostgreSQL.""" # Connect to postgresql # Read MAG responses # Collect IDs from tables to ensure we're not inserting duplicates # Remove duplicates and keep only papers that are not already in the mag_papers table. # Parse author information # Parse Fields of Study # Parse affiliations # Insert dicts into postgresql
self