Last active
March 14, 2019 03:01
-
-
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)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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