Merge pull request #92 from dvklopfenstein/dev

Fix dict key potential error; replaced removing files/dirs using os.system for greater safety

Author: dvklopfenstein

CHANGELOG.md

@@ -7,6 +7,7 @@
 ## Summary
 
 * [**Unreleased**](#unreleased)
+* [**Release 2026-04-14 v0.3.2**]dir key err fixed; Replaced removing files/dirs using system
 * [**Release 2025-12-15 v0.3.1**]Removed links to cite which went roguer (Gambling) than it was before (creepy use of your data). 
 * [**Release 2025-12-01 v0.3.0**](#release-2025-12-01-v030) Updated for new NIH iCite format which may no longer supply keys such as `author`
 * [**Release 2025-09-06 v0.2.0**](#release-2025-09-06-v020) Updated for new NIH iCite format which may no longer have fields: cited_by_clin cited_by references provisional

pmidcite/__init__.py

@@ -2,6 +2,6 @@
 
 __copyright__ = 'Copyright (C) 2019-present, DV Klopfenstein, PhD. All rights reserved'
 __author__ = 'DV Klopfenstein, PhD'
-__version__ = '0.3.1'
+__version__ = '0.3.2'
 
 # Copyright (C) 2019-present, DV Klopfenstein, PhD. All rights reserved

pmidcite/eutils/cmds/pubmed.py

@@ -4,10 +4,10 @@
 __copyright__ = "Copyright (C) 2016-present DV Klopfenstein, PhD. All rights reserved."
 __license__ = "AGPL-3.0"
 
-from os import system
-from os import getcwd
-from os.path import exists
 import sys
+import os
+import os.path as op
+from pathlib import Path
 from pmidcite.eutils.cmds.base import EntrezUtilities
 from pmidcite.eutils.cmds.esearch import ESearch
 
@@ -31,8 +31,7 @@ def dnld_query_pmids(self, query, num_ids_p_epost=10):
 
     def _dnld_wr_all(self, fout_pubmed, efetch_idxs, efetch_params):
         """Download and write all PMIDs PubMed text entries into one file"""
-        if exists(fout_pubmed):
-            system(f'rm {fout_pubmed}')
+        Path(fout_pubmed).unlink(missing_ok=True)
         if not efetch_idxs:
             return
         for desc, start, pmids_exp, querykey in efetch_idxs:
@@ -50,7 +49,7 @@ def _dnld_query(self, query):
 
     def dnld_wr1_per_pmid(self, pmids, force_download, dir_pubmed_txt, pmid2name=None):
         """Download and write one PubMed text file entry per PMID"""
-        if not exists(dir_pubmed_txt):
+        if not op.exists(dir_pubmed_txt):
             dir_pubmed_txt = self._err_exists(dir_pubmed_txt)
         pmid_nt_list = self.get_pmid_nt_list(pmids, force_download, dir_pubmed_txt, pmid2name)
         efetch_idxs, efetch_params = self.epost_ids(pmids, 'pubmed', 10, 1, **self.medline_text)
@@ -59,8 +58,8 @@ def dnld_wr1_per_pmid(self, pmids, force_download, dir_pubmed_txt, pmid2name=Non
     @staticmethod
     def _err_exists(dir_pubmed_txt):
         """Print warning message if dir not exist, return current working directory"""
-        cwd = getcwd()
-        print(f'**WARNING: DIR({dir_pubmed_txt}) NOT EXIST RETURNING CWD({getcwd()})')
+        cwd = os.getcwd()
+        print(f'**WARNING: DIR({dir_pubmed_txt}) NOT EXIST RETURNING CWD({cwd})')
         return cwd
 
     def dnld_texts(self, efetch_idxs, efetch_params):

pmidcite/icite/entry.py

@@ -268,7 +268,7 @@ def get_aart_translation(self):
         lst.append('H' if dct['human'] != 0.0 else '.')
         lst.append('A' if dct['animal'] != 0.0 else '.')
         lst.append('M' if dct['molecular_cellular'] != 0.0 else '.')
-        lst.append('C' if dct['is_clinical'] else '.')
+        lst.append('C' if dct.get('is_clinical') else '.')
         lst.append('c' if dct.get('cited_by_clin') else '.')
         return ''.join(lst)
 

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "pmidcite"
 description="Turbocharge a PubMed literature search using citation data from the NIH"
-version = "0.3.1"
+version = "0.3.2"
 license = "AGPL-3.0-or-later"
 authors = [
   {name = 'DV Klopfenstein, PhD', email = 'dvklopfenstein@protonmail.com'},

tests/icite.py

@@ -4,43 +4,40 @@
 __copyright__ = "Copyright (C) 2019-present, DV Klopfenstein, PhD. All rights reserved."
 __author__ = "DV Klopfenstein, PhD"
 
+import os
+import os.path as op
+import shutil
 from os import system
-from os import mkdir
-from os import environ
-from os.path import join
-from os.path import dirname
-from os.path import abspath
-from os.path import exists
-from os.path import getmtime
+from pathlib import Path
 from glob import glob
 from sys import stdout
 from pmidcite.icite.pmid_dnlder import NIHiCiteDownloader
 
-DIR_TEST = dirname(abspath(__file__))
-DIR_TESTDATA = join(DIR_TEST, "data")
-DIR_ICITE = abspath(join(DIR_TEST, "./icite"))
-DIR_REPO = abspath(join(DIR_TEST, ".."))
-DL = environ.get('DLCYG')
+DIR_TEST = op.dirname(op.abspath(__file__))
+DIR_TESTDATA = op.join(DIR_TEST, "data")
+DIR_ICITE = op.abspath(op.join(DIR_TEST, "./icite"))
+DIR_REPO = op.abspath(op.join(DIR_TEST, ".."))
+DL = os.environ.get('DLCYG')
 
 
 def get_dnld_files(glob_pattern):
     """Get the filenames of downloaded files matching the researcher's glob pattern"""
-    return glob(join(DL, glob_pattern))
+    return glob(op.join(DL, glob_pattern))
 
 def get_filename_test(basename):
     """Get the full filename of a test file, basename"""
-    return join(DIR_TEST, basename)
+    return op.join(DIR_TEST, basename)
 
 def get_filename_testdata(basename):
     """Get the full filename of a test file, basename"""
-    return join(DIR_TESTDATA, basename)
+    return op.join(DIR_TESTDATA, basename)
 
 def dir_icite_clobber(prt=stdout):
     """Create an empty dir, ./src/tests/icite, removing old contents if necessary"""
-    cmd = 'rm -rf {ICITE}; mkdir {ICITE}'.format(ICITE=DIR_ICITE)
-    system(cmd)
+    shutil.rmtree(DIR_ICITE, ignore_errors=True)
+    Path(DIR_ICITE).mkdir(parents=True, exist_ok=True)
     if prt:
-        prt.write(f"{cmd}\n")
+        print('rm -rf', DIR_ICITE)
 
 def dir_icite_wc_l(prt=stdout):
     """Print count of p{PMID}.py files in dir ./src/tests/icite"""
@@ -53,10 +50,10 @@ def dir_icite_wc_l(prt=stdout):
 def mk_dir(dir_name, rmdir=False):
     """Get the directory where data downloaded from NIH-OCC are stored"""
     if rmdir:
-        system(f'rm -rf {dir_name}')
-    if not exists(dir_name):
-        mkdir(dir_name)
-        print(f'**CREATED DIR: {abspath(dir_name)}')
+        shutil.rmtree(dir_name, ignore_errors=True)
+    if not op.exists(dir_name):
+        os.mkdir(dir_name)
+        print(f'**CREATED DIR: {op.abspath(dir_name)}')
     return dir_name
 
 
@@ -65,12 +62,12 @@ class ICiteTester:
 
     def __init__(self):
         self.dir_icite = mk_dir(DIR_ICITE)
-        self.icite_files = join(self.dir_icite, '*.py')
+        self.icite_files = op.join(self.dir_icite, '*.py')
 
     def rm_icitefiles(self):
         """Remove downloaded NIH-OCC iCite files"""
-        if list(glob(self.icite_files)):
-            system(f'rm {self.icite_files}')
+        for fname in list(glob(self.icite_files)):
+            os.remove(fname)
         assert not list(glob(self.icite_files)), 'BAD INITIAL CLEAN UP'
 
     def get_paper(self, pmid, force_download=False, do_prt=True):
@@ -86,7 +83,7 @@ def get_paper(self, pmid, force_download=False, do_prt=True):
 
     def get_f2mtime(self, min_files):
         """Get mofification times of globbed files"""
-        f2mtime = {fin:getmtime(fin) for fin in glob(self.icite_files)}
+        f2mtime = {fin:op.getmtime(fin) for fin in glob(self.icite_files)}
         assert len(f2mtime) >= min_files, \
             f'iCite FILES NOT DOWNLOADED {len(f2mtime)=} < min_files({min_files})'
         return f2mtime

tests/test_cfg_icite.py

@@ -1,38 +1,36 @@
 #!/usr/bin/env python3
 """Test cfg file documentation matches cfg init"""
 
-from os import system
 from os import stat
-from os.path import join
-from os.path import exists
-from os.path import basename
-from os.path import relpath
+import os.path as op
+from pathlib import Path
 from sys import argv
 import filecmp
 from pmidcite.cfg import Cfg
 from pmidcite.cfgini import wr_rcfile
 from tests.icite import DIR_REPO
+from tests.utils import cat
 
 
 # pylint: disable=line-too-long
 def test_cfg_desecriptive():
     """Test default cfg file that contains detailed comments"""
-    file_cfg = join(DIR_REPO, 'test_icite.cfg')
+    file_cfg = op.join(DIR_REPO, 'test_icite.cfg')
 
     # Remove test configuration file and test that it does not exist
-    system(f'rm -f {file_cfg}')
-    assert not exists(file_cfg)
+    Path(file_cfg).unlink(missing_ok=True)
+    assert not op.exists(file_cfg)
 
     # Test that non-existing configuration file can not be read by Cfg
     obj = Cfg(check=False, prt_fullname=False)
     obj.set_cfg(file_cfg)
     assert not obj.rd_rc()
 
     # Write configuration file. Test that an exiting cfg file is not overwritten
-    system(f'cat {file_cfg}')
+    cat(file_cfg)
     assert wr_rcfile(file_cfg, force=False)
-    system(f'cat {file_cfg}')
-    assert exists(file_cfg)
+    cat(file_cfg)
+    assert op.exists(file_cfg)
     assert not wr_rcfile(file_cfg, force=False)
 
     # Read configuration file, file_cfg
@@ -42,46 +40,46 @@ def test_cfg_desecriptive():
         f'UNEXPECTED FILENAME: EXP({file_cfg}) ACT({next(iter(cfg))})'
 
     # Test that the values for the new cfg file are the default values
-    fin_base = basename(file_cfg)
+    fin_base = op.basename(file_cfg)
     for key, actual in obj.cfgparser['pmidcite'].items():
         print(f'{fin_base}: {key} {actual}')
         expected = Cfg.dfltdct['pmidcite'][key]
         assert actual == expected, \
             f'{file_cfg}: KEY({key}) ACTUAL({actual}) != EXPECTED({expected})'
     print(stat(file_cfg))
-    system(f'rm {file_cfg}')
+    Path(file_cfg).unlink(missing_ok=True)
     print('PASSED: cfg init with comments')
 
 
 def test_cfg_example(update_example=False):
     """Test that the config file example is up-to-date"""
-    fin_cfg = join(DIR_REPO, 'test_icite.cfg')
-    fin_ex = join(DIR_REPO, 'doc/example_cfg/.pmidciterc')
+    fin_cfg = op.join(DIR_REPO, 'test_icite.cfg')
+    fin_ex = op.join(DIR_REPO, 'doc/example_cfg/.pmidciterc')
     wr_rcfile(fin_cfg, force=True)
     wr_rcfile(fin_ex, force=update_example)
     # Compare the contents of the two files
     assert filecmp.cmp(fin_cfg, fin_ex, shallow=False), \
-        f'EXP({relpath(fin_ex)}) != ACT({relpath(fin_cfg)})'
-    system('rm {fin_cfg}')
+        f'EXP({op.relpath(fin_ex)}) != ACT({op.relpath(fin_cfg)})'
+    Path(fin_cfg).unlink(missing_ok=True)
     print('PASSED: Cfg example matches cfg default')
 
 
 def test_cfg_icite():
     """Test cfg file documentation matches cfg init"""
     # NIH iCite configuration file
-    file_cfg = join(DIR_REPO, 'test_icite.cfg')
+    file_cfg = op.join(DIR_REPO, 'test_icite.cfg')
 
     # Remove test configuration file and test that it can no longer be read by Cfg
-    system(f'rm -f {file_cfg}')
+    Path(file_cfg).unlink(missing_ok=True)
     obj = Cfg(check=False, prt_fullname=False)
     obj.set_cfg(file_cfg)
     assert not obj.rd_rc()
 
     # Write configuration file
-    system(f'cat {file_cfg}')
+    cat(file_cfg)
     assert obj.wr_rc()
-    system('cat {file_cfg}')
-    assert exists(file_cfg)
+    cat(file_cfg)
+    assert op.exists(file_cfg)
     assert not obj.wr_rc()
 
     # Read configuration file
@@ -95,21 +93,21 @@ def test_cfg_icite():
          f"EXP({Cfg.dfltdct['pmidcite']['dir_icite_py']}) "
          f"ACT({obj.cfgparser['pmidcite']['dir_icite_py']})")
     assert obj.cfgparser['pmidcite']['dir_pubmed_txt'] == Cfg.dfltdct['pmidcite']['dir_pubmed_txt']
-    system(f'rm {file_cfg}')
+    Path(file_cfg).unlink(missing_ok=True)
     print('PASSED: cfg init with no comments')
 
 
 def test_cfg_eutils():
     """Test writing and reading the configuration file"""
     # NIH iCite configuration file
-    file_cfg = join(DIR_REPO, 'test_eutils.cfg')
+    file_cfg = op.join(DIR_REPO, 'test_eutils.cfg')
     obj = Cfg(check=False)
     obj.set_cfg(file_cfg)
 
     # Write configuration file
-    system(f'rm -f {file_cfg}')
+    Path(file_cfg).unlink(missing_ok=True)
     assert obj.wr_rc()
-    assert exists(file_cfg)
+    assert op.exists(file_cfg)
     assert not obj.wr_rc()
 
     # Newly created cfg file should have default values for private data
@@ -119,7 +117,7 @@ def test_cfg_eutils():
     assert obj.cfgparser['pmidcite']['apikey'] == Cfg.dfltdct['pmidcite']['apikey']
     assert obj.cfgparser['pmidcite']['tool'] == Cfg.dfltdct['pmidcite']['tool']
 
-    system(f'rm {file_cfg}')
+    Path(file_cfg).unlink(missing_ok=True)
     print('PASSED: cfg init private values are default')
 
 

tests/test_dnld_cites_refs.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 """Test downloading a paper, paper + refs, paper + cites, paper + all"""
 
-from os import system
+import os
 from sys import stdout
 from glob import glob
 from pmidcite.icite.pmid_dnlder import NIHiCiteDownloader
@@ -16,6 +16,7 @@ def test_dnld_cites_refs():
     """Test downloading a paper, paper + refs, paper + cites, paper + all"""
 
     pmid = 20640201
+    os.makedirs(DIR_ICITE, exist_ok=True)
     dnldr_top = NIHiCiteDownloader(DIR_ICITE, FORCE_DNLD)
     _run("TEST(plain)", dnldr_top, pmid)
 
@@ -45,11 +46,12 @@ def _compare(test_name, pmid, entry_lst, desc):
 
 def _run(test_name, dnldr, pmid):
     """Download PMIDs"""
-    system(f'rm -f {dnldr.dir_dnld}/p*.py')
+    globstr = f'{dnldr.dir_dnld}/p*.py'
+    for fname in glob(globstr):
+        os.remove(fname)
     pmid2icitepaper = dnldr.get_pmid2paper({pmid}, pmid2note=None)
 
     # Get filenames (p{PMID}.py) downloaded from NIH's citation database
-    globstr = f'{dnldr.dir_dnld}/p*.py'
     pmids_dnlded = glob(globstr)
     print(f'{len(pmids_dnlded)} NIH icites downloaded')
     ## print('{N} pmid2icitepaper for PMID({P}); {O} assc PMIDs'.format(