Skip to content

Instantly share code, notes, and snippets.

@gsw945
Last active March 14, 2019 03:01
Show Gist options
  • Select an option

  • Save gsw945/a8c47dff1dbef805faea79b38729cdb6 to your computer and use it in GitHub Desktop.

Select an option

Save gsw945/a8c47dff1dbef805faea79b38729cdb6 to your computer and use it in GitHub Desktop.
python filter text files in folder(path match glob rule and content contains specified substring)
import os
import re
import glob
import traceback
import chardet
# import threading
from multiprocessing.dummy import Pool as ThreadPool
from operator import is_not
from functools import partial
def re_escape(rule):
rule = rule.replace('[', r'\[')
rule = rule.replace(']', r'\]')
rule = rule.replace('(', r'\(')
rule = rule.replace(')', r'\)')
return rule
def file_list(folder, pattern):
for root, dirnames, filenames in os.walk(folder):
try:
rule = os.path.join(re_escape(root), pattern)
for filename in glob.iglob(rule, recursive=True):
yield re.sub(r'(^[a-z]\:)([^\\])', r'\1\\\2', filename)
except Exception as e:
# import ipdb; ipdb.set_trace()
traceback.print_exc()
pass
def all_in(searchs, content):
if not bool(searchs):
return False
is_in = True
for item in searchs:
is_in = is_in and item in content
if not is_in:
break
return is_in
def do_search(filename):
is_found = False
if os.path.basename(filename) != '__init__.py':
try:
with open(filename, 'rb') as f:
raw_content = f.read()
# 检测编码
detected = chardet.detect(raw_content)
content = ''
if isinstance(detected, dict) and 'encoding' in detected:
encoding = detected['encoding']
if encoding is not None:
content = raw_content.decode(encoding, 'ignore')
if all_in(('jquery', 'execute', 'script'), content):
# print(filename)
is_found = True
del content
except Exception as e:
# import ipdb; ipdb.set_trace()
traceback.print_exc()
pass
if is_found:
return filename
if __name__ == '__main__':
folder = r'f:'
pattern = '*.py'
files = file_list(folder, pattern)
tpool = ThreadPool(processes=20)
# 启动任务
tresult_proxy = tpool.map_async(do_search, files)
# 关闭线程池
tpool.close()
# 等到线程池所有任务完成
tpool.join()
# 获取结果
result = tresult_proxy.get()
result = list(filter(partial(is_not, None), result))
from pprint import pprint
pprint(result)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment