2
\$\begingroup\$

Goal: Iterate through 100+ text files, parse for specific types of rows, return the row length for the desired row ONCE. Once, because these files have about 5-6 different types of rows with varying lengths but the same type of row will have the exact length. So, once the program has found a desired row, return the row length for that first instance. Ignore other rows that meet this criteria. Move to the next row type, return the row length for the first instance of that row type.

Outcome: A report that returns the file_date, row type, and row length for each file.

Issue: We have written something that does this... However, it took 9 hours to scan 48 files. :(

This is my first stab at python so I am certain there's a better approach. I am hoping someone with much more Python experience can provide pointers for improvement.

import os
from datetime import datetime


def write_log_file(month_year, rec15_len, rec1_len, rec7_len):
    output_file = "log.txt"
    with open(output_file, "a") as f1:
        f1.write(month_year + ",1.5," + str(rec15_len))
        f1.write("\n")
        f1.write(month_year + ",1," + str(rec1_len))
        f1.write("\n")
        f1.write(month_year + ", 7," + str(rec7_len))
        f1.write("-------------------------\n")

    f1.close()


def check_rec1(row, record):

    if record == "1":
        if row[44] == "5":
            rec_len = len(row)
            # print("found 15")
            return rec_len
        else:
            rec_len = len(row)
            # print("found 1")
            return rec_len


def check_rec7(row, record):

    if record == "7":
        rec_len = len(row)
        # print("found 7")
        return rec_len


def process_files(folder):
    for folder, subfolders, filenames in os.walk(folder):
        print("the current folder is " + folder)
        for subfolder in subfolders:
            print("subfolder= " + folder + ": " + subfolder)
        for filename in filenames:
            if filename.startswith("itec") and filename.endswith(".cmp"):
                print("found file name " + filename)
                xname_cmp = filename.split(".")
                xname = xname_cmp[0].split("_")
                month_year = xname[2] + "/" + xname[1]
                print("Processing file:" + filename)
                print("Start processing time: " + str(datetime.now()))
                cmp_file = open(folder + "/" + filename, "r+")
                print("reading file ", filename)
                read_rows(cmp_file, month_year)
                cmp_file.close()
                print("End processing time: " + str(datetime.now()))
                print("-------------------------------------------------")
            else:
                print(f"Skipping {filename}")
    return


def find_rec(line):
    split_line = line.split("    ")
    record_type = split_line[1]
    # print(len(record_type))
    rec_list = record_type.lstrip()
    return rec_list


def read_rows(cmp_file, month_year):
    count = 0
    process_row = True
    rec15_len = 0
    rec1_len = 0
    rec7_len = 0
    row_list = []

    while process_row:
        if (rec1_len != 0) and (rec15_len != 0) and (rec7_len != 0):
            # print("process row false")
            process_row = False
        else:
            # print("process row true")
            line = cmp_file.readline()
            if not line:
                process_row = False
                break
            new_line = line.rstrip()
            count += 1
            row_list[:0] = find_rec(new_line)
            record_type = row_list[0]
            # print("stuff " + record_type)
            if record_type == "1":
                if row_list[1] == "5":
                    rec15_len = check_rec1(new_line, record_type)
                    # print(f"rec15 length {str(rec15_len)} {count}")
                else:
                    rec1_len = check_rec1(new_line, record_type)
                    # print(f"rec1 length {str(rec1_len)} {count}")
            elif record_type == "7":
                check_len = check_rec7(new_line, record_type)
                if check_len > 0:
                    rec7_len = check_len
                    # print(f"rec7 length {str(rec7_len)}  {count}")

    # if (type1 == True) and (type15 == True) and (type7 == True):
    print("finished processing all recs true")
    write_log_file(month_year, rec15_len, rec1_len, rec7_len)


output_file = "log.txt"
header = "file_date,record_type,length"
with open(output_file, "w") as f1:
    f1.write(header)
    f1.write("\n")
f1.close()


folder_name = "files"
process_files(folder_name)

```
\$\endgroup\$

1 Answer 1

7
\$\begingroup\$

String concatenation

Don't use + on strings, either use f-strings or str.format:

filename = 'hello'
foldername = 'world'

newfile = f'{foldername}_{filename}.txt'

# or
newfile = '{}_{}.txt'.format(foldername, filename)

Processing files

When opening files, it's best to use the with context manager, which eliminates you having to remember to close() the file handle later:

with open(myfile) as fh:
    # process fh

# close has automatically been invoked here

Next, it's usually a good pattern to iterate over a file using for rather than while:

with open(myfile) as fh:
    for line in fh:
        line = line.rstrip()
        # do something here

enumerate

You are also maintaining a counter in a loop:

while True:
    ~snip~
    count += 1

When you convert this to a for loop, you can use enumerate to deal with that for you:

for count, line in enumerate(fh):
    # do something

Process rows

In your check_rec_[n] functions you do an if record == n check. This doesn't need to happen, since you should have filtered that ahead of time to call the function. So drop them:

def check_rec1(row, record):
    if row[44] == "5":
        rec_len = len(row)
        # print("found 15")
        return rec_len
    else:
        rec_len = len(row)
        # print("found 1")
        return rec_len


def check_rec7(row, record):
    rec_len = len(row)
    # print("found 7")
    return rec_len

But examining them shows that they all do the exact same thing:

def check_rec(row, record):
    return len(row)

So record is a redundant variable. Remove it from the function:

def check_rec(row):
    return len(row)

Now this is just unnecessarily wrapping a call to len. So anywhere you have a check_rec_[n], replace it with len(row):

for count, line in enumerate(fh):
    line = line.rstrip()

    row_list[:0] = find_rec(new_line)
    record_type = row_list[0]

    if record_type == "1":
        if row_list[1] == "5":
            rec15_len = len(line)
            # print(f"rec15 length {str(rec15_len)} {count}")
        else:
            rec1_len = len(line)
            # print(f"rec1 length {str(rec1_len)} {count}")
    elif record_type == "7":
        check_len = len(line)
        if check_len > 0:
            rec7_len = check_len

Checking for empty strings

if check_len > 0 should just be if line:

    elif record_type == '7':
        if line:
           rec7_len = len(line)

Your original break condition

You are checking if all((rec1_len, rec7_len, rec15_len)). Since that will break the loop, don't use an else, use it as a guard:

for count, line in enumerate(fh):
    if all((rec1_len, rec7_len, rec15_len)):
        break

    line = line.rstrip()
    ...

However, this naming indicates there's a better way. There is, use a dictionary:

record_lens = {
    "1": 0,
    "15": 0,
    "7": 0,
}


for count, line in enumerate(fh):
     # do things

     record_lens[record] = len(line)

And your guard becomes:

for count, line in enumerate(fh):
    if all(record_lens.values()):
        break

Furthermore, if you've already populated it in the dictionary, that means that you can very simply look up the length and skip it if it's nonzero:

if record_lens[record_type]:
    # no need to re-calculate
    continue 

find_rec

You are splitting line on multiple spaces and then returning rec_list. This could be slightly refactored as the following:

def find_rec(line):
    _, record_type, *_ = line.split("    ")
    return record_type.strip()

I now think that this is small enough that I wouldn't make it a separate function.

The issue is that you are prepending to your list here:

    row_list[:0] = find_rec(new_line)

This is a bit sneaky, because the list slice will split up your record_type, so if you have two digits (the number 15, for instance`, you have to do two checks. A few things:

  • Prepending to a list is expensive
  • You don't use the collected results at all

So don't collect a list here, read the value you've returned. It's already a string!

    record_type = find_rec(new_line)

    if record_lens[record_type]:
        continue
    else:
        record_lens[record_type] = len(line)

Last, this assumes that you only have three record_types. If you have more, this will fail, so to further generalize it, let's use an empty dictionary and the dict.get function to prevent pesky KeyErrors:

mydict = {}

# the key isn't in the dictionary
# so the lookup fails
mydict['thing']
KeyError

# now the lookup succeeds
mydict.get('thing')
None

Now, the only thing that will change is the guard clause can go away, because we are skipping all rows that are currently processed:

for count, line in enumerate(fh):
    # no if check here
    line = line.rstrip()

    _, record_type, *_ = line.split("    ")

    # if a value is present, skip
    if record_lens.get(record_type):
        continue
    # otherwise, set it
    else:
        record_lens[record_type] = len(line)

So let's look at what we have so far as a refactored effort:

def read_rows(fh):
    record_lens = {}

    for count, line in enumerate(fh):
        line = line.rstrip()

        _, record_type, *_ = line.split("    ")

        if record_lens.get(record_type):
            continue
        else:
            record_lens[record_type] = len(line)
    
    # let's make sure we return this
    return record_lens

month_year is only used to log to the file, which I'd prefer to do separately. So let's drop it for now.

Combining the with for the file handle on opening:

def process_files(folder):
    for folder, subfolders, filenames in os.walk(folder):
        print("the current folder is " + folder)
        for subfolder in subfolders:
            print("subfolder= " + folder + ": " + subfolder)
        for filename in filenames:
            if filename.startswith("itec") and filename.endswith(".cmp"):
                print("found file name " + filename)
                xname_cmp = filename.split(".")
                xname = xname_cmp[0].split("_")
                month_year = xname[2] + "/" + xname[1]

                print("Processing file:" + filename)
                print("Start processing time: " + str(datetime.now()))

                with open(f'{folder}/{filename}') as fh:
                    print("reading file ", filename)
                    # we will do something with this momentarily
                    record_lens = read_rows(fh)

                
                print("End processing time: " + str(datetime.now()))
                print("-------------------------------------------------")
            else:
                print(f"Skipping {filename}")
    return
      

Using if as a guard, again

When walking your files, you can avoid unnecessary indentation by checking for the case you want to skip first and using continue:

def process_files(folder):
    for folder, subfolders, filenames in os.walk(folder):
        print("the current folder is " + folder)

        for filename in filenames:
            if not filename.startswith("itec"):
                continue
            elif not filename.endswith(".cmp"):
                continue
 
            # no else needed

Tuple unpacking

I used this earlier, but when using split, you can unpack the result into your desired variables directly:

                _, month, year, *_ = filename.split(".")[0].split('_')
                month_year = f'{month}/{year}'

Where the _ denotes a throwaway or unused quantity, and the *_ is to ensure I capture any trailing elements:

somelist = list('abcd')

_, b, c, *_ = somelist
print(b, c)
'b' 'c'

Style: Add line breaks

This function:

def process_files(folder):
    for folder, subfolders, filenames in os.walk(folder):
        print("the current folder is " + folder)
        for subfolder in subfolders:
            print("subfolder= " + folder + ": " + subfolder)
        for filename in filenames:
            if filename.startswith("itec") and filename.endswith(".cmp"):
                print("found file name " + filename)
                xname_cmp = filename.split(".")
                xname = xname_cmp[0].split("_")
                month_year = xname[2] + "/" + xname[1]
                print("Processing file:" + filename)
                print("Start processing time: " + str(datetime.now()))
                cmp_file = open(folder + "/" + filename, "r+")
                print("reading file ", filename)
                read_rows(cmp_file, month_year)
                cmp_file.close()
                print("End processing time: " + str(datetime.now()))
                print("-------------------------------------------------")
            else:
                print(f"Skipping {filename}")
    return

Is a nearly impossible to read wall-of-text. Break it up. Refactored:

def process_files(folder):
    for folder, subfolders, filenames in os.walk(folder):
        print(f"the current folder is {folder}")

        for filename in filenames:
            if not filename.startswith("itec"):
                continue
            elif not filename.endswith(".cmp"):
                continue

            print("found file name {filename}")
            xname_cmp = filename.split(".")
            xname = xname_cmp[0].split("_")
            month_year = f"{xname[2]}/{xname[1]}"

            print(f"Processing file: {filename}")
            print("Start processing time: {datetime.now()}"))

            with open(f'{folder}/{filename}') as fh:
                print("reading file ", filename)
                # we will do something with this momentarily
                record_lens = read_rows(fh)

            print(f"End processing time: {datetime.now()}"))
            print("-------------------------------------------------")
    return

I'm going to create a generator expression with the yield statement in order to log here:

def process_files(folder):
    for folder, subfolders, filenames in os.walk(folder):
        print(f"the current folder is {folder}")

        for filename in filenames:
            if not filename.startswith("itec"):
                continue
            elif not filename.endswith(".cmp"):
                continue

            print(f"found file name {filename}")
            _, month, year, *_ = filename.split(".")[0].split('_')
            date = f'{year}/{month}'

            print(f"Processing file: {filename}")
            print(f"Start processing time: {datetime.now()}")

            with open(f'{folder}/{filename}') as fh:
                print(f"reading {filename} ")
                record_lens = read_rows(fh)

            print(f"End processing time: {datetime.now()}"))
            print("-------------------------------------------------")
            yield record_lens, date

Now I can iterate over the result of this function and it evaluates lazily:

def write_log_file(date, records):
    output_file = "log.txt"
    with open(output_file, "a") as f1:
        for rec_type, val in records.items():
            f1.write(f"{date},{rec_type},{val}")
            f1.write('\n')


for filename, date, record_lens in process_files(folder_name):
    write_log_file(date, record_lens)

I've also simplified the log to file function by just taking the dictionary as a parameter.

Separating folder filtering

It might make the code a bit easier to read if we break up the process_folder function into two distinct parts:

  1. Get files
  2. Process file

Let's start with just getting the file names themselves. I'm going to lean on the generator concept again:

def get_filenames(folder):
    for folder, subfolders, filenames in os.walk(folder):
        print(f"the current folder is {folder}")

        for filename in filenames:
            if not filename.startswith("itec"):
                continue
            elif not filename.endswith(".cmp"):
                continue

            print(f"found file name {filename}")
            _, month, year, *_ = filename.split(".")[0].split('_')

            yield f'{folder}/{filename}', f'{year}/{month}'

This now very clearly just builds the paths and dates for each file in the folder.

Now, when we process each file, we can do that in the main loop without having to clutter up the process_folder function:

for file, date in get_filenames(folder):
    with open(file) as fh:
        records = read_rows(fh)

    # do something

Logging

You are repeatedly opening and closing your log file, you could use a more global file handle:

def write_log_file(fh, date, records):
    for rec_type, val in records.items():
        fh.write(f"{date},{rec_type},{val}")
        fh.write('\n')


# open this globally
with open(logfile, 'w') as fh:
    for file, date in get_filenames(folder_name):
        with open(file) as gh:
            record_lens = read_rows(gh)

        # reuse it here
        write_log_file(fh, date, record_lens)

But the best way is to use the logging module:

import logging

# create logger
logger = logging.getLogger('Process CMP Files')
logger.setLevel(logging.DEBUG)

# create file handler which logs even debug messages
fh = logging.FileHandler('log.txt')
fh.setLevel(logging.DEBUG)

# create formatter and add it to the handler
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)

# add the handler to the logger
logger.addHandler(fh)

Now, anytime you want to log:

for filepath, date get_filenames(folder_name):
    with open(filepath) as fh:
        records = read_rows(fh)

    for rec_type, val in records.items():
        logger.info(f"{date},{rec_type},{val}")

And we can just eliminate your log_to_file function entirely.

main function

Let's wrap your functionality in a main function:

def main(folder_name):
    for filepath, date in get_filenames(folder_name):
        with open(filepath) as fh:
            records = read_rows(fh)
        
        for rec_type, val in records.items():
            logger.info(f"{date},{rec_type},{val}")

And let's add an if __name__ guard to make your code more portable if you want to import any of these functions into another module:

def main(folder_name):
    for filepath, date get_filenames(folder_name):
        with open(filepath) as fh:
            records = read_rows(fh)
        
        for rec_type, val in records.items():
            logger.info(f"{date},{rec_type},{val}")


if __name__ == "__main__":
    # you can now process this as an input 
    # arg to your function
    folder_name = sys.argv[1]

    main(folder_name)

Result

Here's the end result:

from datetime import datetime
import logging
import os

# create a global logger
logger = logging.getLogger('Process CMP Files')
logger.setLevel(logging.DEBUG)

# create file handler which logs info messages
fh = logging.FileHandler('log.txt')
fh.setLevel(logging.INFO)

# Let's add a console handler to eliminate print statements
ch = logging.StreamHandler()
# take debug messages
ch.setLevel(logging.DEBUG)

# create formatter and add it to the handlers
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
ch.setFormatter(formatter)

# add the handlers to the logger
logger.addHandler(fh)
logger.addHandler(ch)


def read_rows(fh):
    record_lens = {}

    for count, line in enumerate(fh):
        line = line.rstrip()

        _, record_type, *_ = line.split("    ")

        if record_lens.get(record_type):
            continue
        else:
            record_lens[record_type] = len(line)

    return record_lens


def get_filenames(folder):
    for folder, subfolders, filenames in os.walk(folder):
        logger.debug(f"the current folder is {folder}")

        for filename in filenames:
            if not filename.startswith("itec"):
                continue
            elif not filename.endswith(".cmp"):
                continue

            logger.debug(f"found file name {filename}")
            _, month, year, *_ = filename.split(".")[0].split('_')

            yield f'{folder}/{filename}', f'{year}/{month}'


def main(folder_name):
    for filename, date in get_filenames(folder_name):
        logger.debug(f"Processing file: {filename}")
        logger.debug(f"Start processing time: {datetime.now()}")

        with open(filename) as fh:
            records = read_rows(fh)

        logger.debug(f"End processing time: {datetime.now()}")
        logger.debug("-" * 49)
        for rec_type, val in records.items():
            logger.info(f"{date},{rec_type},{val}")


if __name__ == "__main__":
    # you can now process this as an input 
    # arg to your function
    folder_name = sys.argv[1]

    main(folder_name)

Revisiting filepaths with pathlib

You can use new style paths with the pathlib module, which simplifies your os.walk loop significantly:

from pathlib import Path


def get_filenames(folder):
    path = Path(folder)

    # the '**/' allows for a recursive glob
    for file in path.glob('**/itec*.cmp'):
        logger.debug(f"Found file {file}")
        
        _, month, year, *_ = file.stem.split('_')

        # open works with Path objects
        yield file, f'{year}/{month}'
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.