parse huge table from html

parse_table.py

import os
import re
import codecs
import concurrent.futures
import multiprocessing

import pandas as pd


def worker(task_q):
    while not task_q.empty():
        fpath = task_q.get()
        fdir, fname = os.path.split(fpath)
        fdir = os.path.split(fdir)[1]
        hfname = os.path.splitext(os.path.split(fname)[1])[0]
        regx = re.compile('\<tr.+\<\/tr\>')
        df = pd.DataFrame([])
        res = ''
        chunk_i = 0

        # load file content
        with codecs.open(fpath, 'r', 'gbk') as fp:
            while True:
                chunk_i += 1
                fcontent = fp.read(1000000)
                if not fcontent:
                    break

                table = regx.findall(res + fcontent)
                if not table:
                    res += fcontent
                    continue
                else:
                    table = table[0]

                res = fcontent[fcontent.rfind('</tr>'):]
                table = ''.join(['<table>', table, '</table>'])

                try:
                    df_chunk = pd.read_html(table)[0]
                    print('load html chunk')
                except:
                    import traceback
                    traceback.print_exc()
                    print('{} wrong'.format(fpath))
                    continue
                else:
                    print('chunk_i {} ok'.format(chunk_i))
                    df = pd.concat([df, df_chunk])

        if df.values.size == 0:
            continue

        df.to_csv(
            os.path.join('../out', fdir + '_' + hfname + '.csv'),
            encoding='utf-8'
        )
        print('{} ok'.format(fpath))


all_path = []
m = multiprocessing.Manager()
task_q = m.Queue()
for path, dirname, fnames in os.walk('../raw'):
    for fname in fnames:
        if fname not in all_path:
            all_path.append(os.path.join(path, fname))
            task_q.put(os.path.join(path, fname))

print('all_path', all_path)

with concurrent.futures.ProcessPoolExecutor(max_workers=7) as executor:
    futures = [executor.submit(worker, task_q) for _ in range(15)]
    for future in concurrent.futures.as_completed(futures):
        try:
            future.result()
            print('future done')
        except:
            import traceback
            traceback.print_exc()
            continue

print('all ok')