سؤال

I have a primary folder filled with subfolders, and each contains files with a particular naming scheme. I have unit-tested a function for creating and editing a text document within a single directory based on information in these files, but am now running into issues trying to get this function to iterate over every subdirectory.

Problem: I am getting a "KeyError" for line 38 if (row["r_id"]) in filters:. This is because the file br_ids.csv is not being created. In the unit test this was functioning fine, so I can only assume it is some issue with how I am using os.walk.

Code:

import csv
import os

with open('hasf.txt','w') as hf:
    for root, subFolders, files in os.walk('/path/to/topdir/'):
#if folder contains 'f_r.csv', list the path in 'hasf.txt'
        if 'f_r.csv' in files:
            hf.write("%s\n" % root)

        if 'r.csv' in files:
            with open(os.path.join(root, "r.csv")) as inf, open(os.path.join(root, "br_ids.csv"), "w") as output:
                reader = csv.DictReader(inf, quotechar='"')
                headers = ["r_id"]
                writer_br = csv.DictWriter(output, headers, extrasaction='ignore')
                writer_br.writeheader()
                for row in reader:
                    if int(row["r_type"]) == 3:
                        writer_br.writerow(row)
        # End creating br_ids

        # parse the data you're about to filter with
            with open(os.path.join(root, 'br_ids.csv'), 'r') as f:
                filters = {(row["r_id"]) for row in csv.DictReader(f, delimiter=',', quotechar='"')}

            with open(os.path.join(root, 'bt_ids.csv'), 'w') as out_f:
                headers = ["t_id"]
                out = csv.DictWriter(out_f, headers, extrasaction='ignore')
                out.writeheader()

        # go thru your rows and see if the matching(row[r_id]) is
        # found in the previously parsed set of filters; if yes, skip the row 
                with open(os.path.join(root, 't.csv'), 'r') as f:
                    for row in csv.DictReader(f, delimiter=','):
                       if (row["r_id"]) in filters:
                            out.writerow(row)

I have gone through a few similar questions here, but none of them have directly hit on creating, editing, and using a file inside of each location of an os.walk. This is my first time using Python, and I am at somewhat of a loss. Also, if there is any way to make my other code more pythonic, I am all ears.

Thanks!

هل كانت مفيدة؟

المحلول

It turns out the issue was directly the KeyError - in some of the folders, br_id.csv had zero entries, and was throwing a KeyError because of this. The way I solved it was with try, like so:

    # parse the data you're about to filter with
        with open(os.path.join(root, 'br_ids.csv'), 'r') as f:
            filters = {(row["r_id"]) for row in csv.DictReader(f, delimiter=',', quotechar='"')}

        with open(os.path.join(root, 'bt_ids.csv'), 'w') as out_f:
            headers = ["t_id"]
            out = csv.DictWriter(out_f, headers, extrasaction='ignore')
            out.writeheader()

    # go thru your rows and see if the matching(row[r_id]) is
    # found in the previously parsed set of filters; if yes, skip the row 
            with open(os.path.join(root, 't.csv'), 'r') as f:
                for row in csv.DictReader(f, delimiter=','):
                    try:
                        if (row["r_id"]) in filters:
                            out.writerow(row)
                    except KeyError:
                        continue

In another case I had a if (row["r_id"]) not in filters: and bypassed this using the same method, except that if it returned a KeyError, then it went ahead and did out.writerow(row).

مرخصة بموجب: CC-BY-SA مع الإسناد
لا تنتمي إلى StackOverflow
scroll top