generate_list.py

#!/usr/bin/python3

import json
import urllib.request, urllib.error
import decode_adblock

class IllegalRuleException(RuntimeError):
    pass

class ProxyRuleUpdater:

    USER_AGENT = 'ProxyRuleUpdater/0.3'
    FILE_HEADER_FORMAT = '# Generated by ProxyRuleUpdater/0.3\n\n'
    OUTPUT_PATH = '.'
    __adblock_rule_decoder = decode_adblock.AdblockRuleDecoder()

    #CRLF = True

    def get_config_value(self, array, key, default_value):
        try:
            return array[key]
        except KeyError:
            return default_value

    def get_file_header(self):
        return self.FILE_HEADER_FORMAT

    def save_file(self, filename, content, file_header = None):
        if content == file_header:
            return
        f = open(filename, 'w+', encoding='utf-8')
        f.write(content)
        f.close()

    def gen_rule(self, config_path):
        f = open(config_path)
        config_raw_content = f.read()
        f.close()
        file_header = self.get_file_header()
        config_array = json.loads(config_raw_content)
        unbound_target_dns = ''
        for i in config_array['dns_servers']:
            unbound_target_dns += '\tforward-addr: ' + i + '\n'
        rules_list = config_array['rules_list']
        for rules_url in rules_list:
            rules = []
            rules_download_url = rules_url['url']
            rules_type = rules_url['type']
            if rules_type == 'hosts':
                # 解析 hosts 文件
                action_type = self.get_config_value(rules_url, 'action_type', 'HOST-SUFFIX') # 默认匹配方式
                action = self.get_config_value(rules_url, 'action', 'REJECT') # 默认操作
                file_prefix = self.get_config_value(rules_url, 'file_prefix', rules_url['name']) # 文件前缀

                # 下载规则列表
                print('Downloading: ' + rules_url['name'])
                http_content = self.get_web_content(rules_download_url)
                print('Download finish: ' + rules_url['name'])

                # 解析规则
                rules = self.__adblock_rule_decoder.decode_adblock_rule(http_content['content'], action_type, action)

            elif rules_type == 'adblock' or rules_type == 'gfwlist':
                # 解析 AdBlock 规则和 GFWlist 规则
                default_action = self.get_config_value(rules_url, 'default_action', 'REJECT') # 默认操作
                unsupport_convert = self.get_config_value(rules_url, 'unsupport_convert', 'REGEX') # 不能完全支持规则时的匹配方式
                exclude_action = self.get_config_value(rules_url, 'exclude_action', 'DIRECT') # 排除规则的操作
                unsupport_action = self.get_config_value(rules_url, 'unsupport_action', 'REJECT') # 不能完全支持规则时的操作
                file_prefix = self.get_config_value(rules_url, 'file_prefix', rules_url['name']) # 文件前缀

                if unsupport_convert == 'REGEX' and unsupport_action != 'REJECT':
                    raise IllegalRuleException('由于软件限制，正则表达式无法使用除拒绝外的所有操作。')

                print('Downloading: ' + rules_url['name'])
                # 下载规则列表
                http_content = self.get_web_content(rules_download_url)
                print('Download finish: ' + rules_url['name'])

                # 解析规则
                if rules_type == 'adblock':
                    rules = self.__adblock_rule_decoder.decode_adblock_rule(
                        http_content['content'],
                        default_action,
                        unsupport_convert,
                        unsupport_action,
                        exclude_action
                    )
                elif rules_type == 'gfwlist':
                    rules = self.__adblock_rule_decoder.decode_gfwlist_rule(
                        http_content['content'],
                        default_action,
                        unsupport_convert,
                        unsupport_action,
                        exclude_action
                    )

            # 转换成 Quantumult(X) 的规则片段，并保存
            quantumult_rules = self.__adblock_rule_decoder.convert_rule_to_quantumult(rules)
            if quantumult_rules['hosts'] != None and quantumult_rules['hosts'] != '':
                self.save_file(
                    self.OUTPUT_PATH + '/' + file_prefix + '-quantumult_hostnames.conf',
                    quantumult_rules['hosts'],
                    file_header
                )
            if quantumult_rules['regex_rejection'] != None and quantumult_rules['regex_rejection'] != '':
                self.save_file(
                    self.OUTPUT_PATH + '/' + file_prefix + '-quantumult_regex_rejection.conf',
                    quantumult_rules['regex_rejection'],
                    file_header
                )

            # 转换成 Clash 的规则片段，并保存
            clash_rules = self.__adblock_rule_decoder.convert_rule_to_clash(rules)
            if clash_rules != 'payload:\n':
                for key, value in clash_rules.items():
                    self.save_file(
                        self.OUTPUT_PATH + '/' + file_prefix + '-clash_' + key + '_hostnames.yaml',
                        value,
                        file_header
                    )

            # 转换成 Unbound 的规则，并保存
            unbound_rules = self.__adblock_rule_decoder.convert_rule_to_unbound(rules)
            if unbound_rules['rejection'] != None and unbound_rules['rejection'] != '':
                self.save_file(
                    self.OUTPUT_PATH + '/' + file_prefix + '-rejection-unbound_dns.conf',
                    unbound_rules['rejection'],
                    file_header
                )
            if unbound_rules['forward'] != None and unbound_rules['forward'] != '':
                self.save_file(
                    self.OUTPUT_PATH + '/' + file_prefix + '-forward-unbound_dns.conf',
                    unbound_rules['forward'],
                    file_header
                )

    def generate_full_rule(self, config_path):
        f = open(config_path, 'r')
        config = json.loads(f.read())
        f.close()
        generate_missions = []
        try:
            generate_missions = config['full_rule_outputs']
        except KeyError:
            return
        for mission in generate_missions:
            for part in mission['parts']:
                part['rules_text'] = []
                for file in part['files']:
                    f = open(file, 'r')
                    part['rules_text'].append(f.read())
                    f.close()
            rule_text = self.__adblock_rule_decoder.make_full_rule(mission['parts'], mission['target_software'])
            self.save_file(self.OUTPUT_PATH + '/' + mission['output_name'], rule_text)

    def get_web_content(self, url, post_data = None, headers = None):
        if url.startswith('file://'):
            f = open(url[7:], 'r', encoding='utf-8')
            content = f.read()
            f.close()
            return {'content': content, 'code': 200, 'headers': {}}
        default_headers = {
            'User-Agent': self.USER_AGENT
        }
        if headers != None:
            default_headers.update(headers)
        request = urllib.request.Request(url, headers = default_headers)
        try:
            response = urllib.request.urlopen(request, data = post_data)
        except urllib.error.HTTPError as e:
            response = e
        content = response.read()
        headers = response.headers
        content_type = headers['Content-Type']
        if content_type != None and len(content_type) > 4 and content_type[0:4] == 'text':
            content = content.decode('utf-8')
        return {'content': content, 'code': response.code, 'headers': headers}

if __name__ == '__main__':
    foo = ProxyRuleUpdater()
    foo.OUTPUT_PATH = './generated_rules'
    config_file = './config.json'
    #foo._test_adblock_rule()
    foo.gen_rule(config_file)
    foo.generate_full_rule(config_file)