diff options
Diffstat (limited to 'scripts/phenotypes_bulk_edit.py')
| -rw-r--r-- | scripts/phenotypes_bulk_edit.py | 265 |
1 files changed, 0 insertions, 265 deletions
diff --git a/scripts/phenotypes_bulk_edit.py b/scripts/phenotypes_bulk_edit.py deleted file mode 100644 index b647199..0000000 --- a/scripts/phenotypes_bulk_edit.py +++ /dev/null @@ -1,265 +0,0 @@ -import sys -import uuid -import logging -import argparse -from pathlib import Path -from typing import Iterator -from functools import reduce - -import requests -from MySQLdb.cursors import DictCursor - -from gn_libs import jobs, mysqldb, sqlite3 - -logging.basicConfig( - format="%(asctime)s — %(filename)s:%(lineno)s — %(levelname)s: %(message)s") -logger = logging.getLogger(__name__) - - -def check_ids(conn, ids: tuple[tuple[int, int], ...]) -> bool: - """Verify that all the `UniqueIdentifier` values are valid.""" - logger.info("Checking the 'UniqueIdentifier' values.") - with conn.cursor(cursorclass=DictCursor) as cursor: - paramstr = ",".join(["(%s, %s)"] * len(ids)) - cursor.execute( - "SELECT PhenotypeId AS phenotype_id, Id AS xref_id " - "FROM PublishXRef " - f"WHERE (PhenotypeId, Id) IN ({paramstr})", - tuple(item for row in ids for item in row)) - mysqldb.debug_query(cursor, logger) - found = tuple((row["phenotype_id"], row["xref_id"]) - for row in cursor.fetchall()) - - not_found = tuple(item for item in ids if item not in found) - if len(not_found) == 0: - logger.info("All 'UniqueIdentifier' are valid.") - return True - - for item in not_found: - logger.error(f"Invalid 'UniqueIdentifier' value: phId:%s::xrId:%s", item[0], item[1]) - - return False - - -def check_for_mandatory_fields(): - """Verify that mandatory fields have values.""" - pass - - -def __fetch_phenotypes__(conn, ids: tuple[int, ...]) -> tuple[dict, ...]: - """Fetch basic (non-numeric) phenotypes data from the database.""" - with conn.cursor(cursorclass=DictCursor) as cursor: - paramstr = ",".join(["%s"] * len(ids)) - cursor.execute(f"SELECT * FROM Phenotype WHERE Id IN ({paramstr}) " - "ORDER BY Id ASC", - ids) - return tuple(dict(row) for row in cursor.fetchall()) - - -def descriptions_differences(file_data, db_data) -> dict[str, str]: - """Compute differences in the descriptions.""" - logger.info("Computing differences in phenotype descriptions.") - assert len(file_data) == len(db_data), "The counts of phenotypes differ!" - description_columns = ("Pre_publication_description", - "Post_publication_description", - "Original_description", - "Pre_publication_abbreviation", - "Post_publication_abbreviation") - diff = tuple() - for file_row, db_row in zip(file_data, db_data): - assert file_row["phenotype_id"] == db_row["Id"] - inner_diff = { - key: file_row[key] - for key in description_columns - if not file_row[key] == db_row[key] - } - if bool(inner_diff): - diff = diff + ({ - "phenotype_id": file_row["phenotype_id"], - **inner_diff - },) - - return diff - - -def __fetch_publications__(conn, ids): - """Fetch publication from database by ID.""" - paramstr = ",".join(["(%s, %s)"] * len(ids)) - query = ( - "SELECT " - "pxr.PhenotypeId, pxr.Id AS xref_id, pxr.PublicationId, pub.PubMed_ID " - "FROM PublishXRef AS pxr INNER JOIN Publication AS pub " - "ON pxr.PublicationId=pub.Id " - f"WHERE (pxr.PhenotypeId, pxr.Id) IN ({paramstr})") - with conn.cursor(cursorclass=DictCursor) as cursor: - cursor.execute(query, tuple(item for row in ids for item in row)) - return tuple(dict(row) for row in cursor.fetchall()) - - -def __process_pubmed_publication_data__(text): - """Process the data from PubMed into usable data.""" - # Process with lxml - pass - - -def __fetch_new_pubmed_ids__(pubmed_ids): - """Retrieve data on new publications from NCBI.""" - # See whether we can retrieve multiple publications in one go - # Parse data and save to DB - # Return PublicationId(s) for new publication(s). - logger.info("Fetching publications data for the following PubMed IDs: %s", - ", ".join(pubmed_ids)) - - # Should we, perhaps, pass this in from a config variable? - uri = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi" - try: - response = request.get( - uri, - params={ - "db": "pubmed", - "retmode": "xml", - "id": ",".join(str(item) for item in pubmed_ids) - }) - - if response.status_code == 200: - return __process_pubmed_publication_data__(response.text) - - logger.error( - "Could not fetch the new publication from %s (status code: %s)", - uri, - response.status_code) - except requests.exceptions.ConnectionError: - logger.error("Could not find the domain %s", uri) - - return tuple() - - -def publications_differences(file_data, db_data, pubmed_ids): - """Compute differences in the publications.""" - logger.info("Computing differences in publications.") - db_pubmed_ids = reduce(lambda coll, curr: coll.union(set([curr["PubMed_ID"]])), - db_data, - set([None])) - new_pubmeds = __fetch_new_pubmed_ids__(tuple( - pubmed_ids.difference(db_pubmed_ids))) - pass - - -def compute_differences(conn, file_contents, pheno_ids, pheno_xref_ids, pubmed_ids) -> tuple[tuple[dict, ...], tuple[dict, ...], tuple[dict, ...]]: - """Compute differences between data in DB and edited data.""" - logger.info("Computing differences.") - # 1. Basic Phenotype data differences - # a. Descriptions differences - desc_diff = descriptions_differences(file_contents, __fetch_phenotypes__(conn, pheno_ids)) - logger.debug("DESCRIPTIONS DIFFERENCES: %s", desc_diff) - # b. Publications differences - # pub_diff = publications_differences(...) - # 2. Data differences - # data_diff = data_differences(...) - pass - - -def update_descriptions(): - """Update descriptions in the database""" - logger.info("Updating descriptions") - # Compute differences between db data and uploaded file - # Only run query for changed descriptions - pass - - -def link_publications(): - """Link phenotypes to relevant publications.""" - logger.info("Linking phenotypes to publications.") - # Create publication if PubMed_ID doesn't exist in db - pass - - -def update_values(): - """Update the phenotype values.""" - logger.info("Updating phenotypes values.") - # Compute differences between db data and uploaded file - # Only run query for changed data - pass - - -def parse_args(): - parser = argparse.ArgumentParser( - prog="Phenotypes Bulk-Edit Processor", - description="Process the bulk-edits to phenotype data and descriptions.") - parser.add_argument("db_uri", type=str, help="MariaDB/MySQL connection URL") - parser.add_argument( - "jobs_db_path", type=Path, help="Path to jobs' SQLite database.") - parser.add_argument("job_id", type=uuid.UUID, help="ID of the running job") - parser.add_argument( - "--log-level", - type=str, - help="Determines what is logged out.", - choices=("debug", "info", "warning", "error", "critical"), - default="info") - return parser.parse_args() - - -def read_file(filepath: Path) -> Iterator[str]: - """Read the file, one line at a time.""" - with filepath.open(mode="r", encoding="utf-8") as infile: - count = 0 - headers = None - for line in infile: - if line.startswith("#"): # ignore comments - continue; - - fields = line.strip().split("\t") - if count == 0: - headers = fields - count = count + 1 - continue - - _dict = dict(zip( - headers, - ((None if item.strip() == "" else item.strip()) - for item in fields))) - _pheno, _xref = _dict.pop("UniqueIdentifier").split("::") - _dict["phenotype_id"] = int(_pheno.split(":")[1]) - _dict["xref_id"] = int(_xref.split(":")[1]) - yield _dict - count = count + 1 - - -def run(conn, job): - """Process the data and update it.""" - file_contents = tuple(sorted(read_file(Path(job["metadata"]["edit-file"])), - key=lambda item: item["phenotype_id"])) - pheno_ids, pheno_xref_ids, pubmed_ids = reduce( - lambda coll, curr: ( - coll[0] + (curr["phenotype_id"],), - coll[1] + ((curr["phenotype_id"], curr["xref_id"]),), - coll[2].union(set([curr["PubMed_ID"]]))), - file_contents, - (tuple(), tuple(), set([None]))) - check_ids(conn, pheno_xref_ids) - check_for_mandatory_fields() - # stop running here if any errors are found. - compute_differences(conn, - file_contents, - pheno_ids, - pheno_xref_ids, - pubmed_ids) - update_descriptions() - link_publications() - update_values() - return 0 - - -def main(): - """Entry-point for this script.""" - args = parse_args() - logger.setLevel(args.log_level.upper()) - logger.debug("Arguments: %s", args) - - with (mysqldb.database_connection(args.db_uri) as conn, - sqlite3.connection(args.jobs_db_path) as jobs_conn): - return run(conn, jobs.job(jobs_conn, args.job_id)) - - -if __name__ == "__main__": - sys.exit(main()) |