Python ruamel.yaml 模块，YAML 实例源码

def load_bgp_database(filename):
    """ load bgp database
    we load our existing bgp information from the yaml file generated from
    our first script here in order to graph that information """

    # initialize our bgp db
    bgp_db = {}

    # load our existing bgp databses
    with open(filename, 'r') as fn:
        yaml=YAML()
        bgp_db = yaml.load(fn)

    return bgp_db

def load_spec_dict(spec_path):
    with open(spec_path, 'r') as spec_file:
        return YAML().load(spec_file.read())

def _get_lang_data(self, lang):
        with open(f'./resources/lang/{lang}.yml') as f:
            return YAML(typ='safe').load(f)

def _setUp(self, bf, yml):
        self.bf = bf
        self.yml = yml
        self.name = self.__class__.__name__[7:].lower()
        YAML = yaml.YAML()
        self.cm = YAML.load(self.yml)
        if self.cm is None:
            self.cm = CommentedMap()
        self.non_empty = []
        for a in BuildFlags.attrs:
            if getattr(bf, a):
                self.non_empty.append(a)

def load_settings():
        conandir = os.path.expanduser("~/.conan/")
        if not os.path.exists(conandir):
            return
        settings_file = os.path.join(conandir, 'settings.yml')
        with open(settings_file) as f:
            txt = f.read()
            YAML = yaml.YAML()
            data = YAML.load(txt)
            settings = odict(data)
            return settings

def load_txt(yml_txt):
    """load a yml txt into a compilers, flags pair"""
    YAML = yaml.YAML()
    dump = YAML.load(yml_txt)
    fa = dump.get('flag_aliases', dump)
    return load_yml(fa)

def _load_yml(self, yml):
        YAML = yaml.YAML()
        dump = YAML.load(yml)
        dump = odict(dump)
        for i in ('project', 'config', 'flag_aliases'):
            if dump.get(i) is None:
                dump[i] = CommentedMap()
            setattr(self, i, dump[i])
        self._dump = dump
        from . import flags as c4flags
        self.flag_aliases = c4flags.FlagAliases(yml=dump.get('flag_aliases', CommentedMap()))

def setup_cache(self):
        if not self.cache_path or not os.path.isfile(self.cache_path):
            cache = {}
        else:
            yaml = YAML()
            cache = yaml.load(open(self.cache_path, 'r'))

        if 'ho_cookies' not in cache:
            cache['ho_cookies'] = {}
        if 'admin_channels' not in cache:
            cache['admin_channels'] = {}
        if 'joined_conversations' not in cache:
            cache['joined_conversations'] = {}

        return cache

def save_cache(self):
        if self.cache_path:
            self.cache['joined_conversations'] = dict(self.joined_conversations)
            yaml = YAML()
            yaml.dump(self.cache, open(self.cache_path, 'w'))

def bgp_parse_logic(filename):
    """ bgp parse logic
    parse function to run in this example exercise """

    # initialize final reliability datastructure
    bgp_db = {}

    # initialize basic variables
    route_address = ''
    next_hop_address = ''

    with open(filename, 'r') as fn:
        # iterate through the file
        for line in fn:
            # store valid addresses
            valid_addresses = flexible_parse(line)
            print(valid_addresses)
            #valid_addresses = strict_parse(line)

            # check number of valid addresses is 2
            if len(valid_addresses) == 2:
                # initialize the valid addresses
                route_address = valid_addresses[0]
                next_hop_address = valid_addresses[1]

                # initialize route address in dictionary if not already done
                if route_address not in bgp_db:
                    bgp_db[route_address] = 0

                # increment redundancy count
                bgp_db[route_address] += 1

            # check number of valid addresses is 1
            # this should only be invoked with a different format of
            # 'show ip bgp' - aka CISCO DEVICES
            # this will also break strict parsing FYI
            elif len(valid_addresses) == 1:
                # initialize valid addresses
                next_hop_address = valid_addresses[0]

                # initialize route address in dictionary if not already done
                if route_address not in bgp_db:
                    bgp_db[route_address] = 0

                # increment redundancy count
                bgp_db[route_address] += 1


    # save our data as a yaml file to be reused for graping purposes!
    with open('fullbgpredundancy.yml', 'w') as fn:
        yaml=YAML()
        yaml.default_flow_style = False
        yaml.dump(bgp_db, fn)

def result(self, url, txt1, txt2):
        fn = open(txt1,"r").read()
        catalog = fn
        fn2 = open(txt2,"r").read()
        extract = fn2

        yaml = YAML()
        import sys
        catalog =  yaml.load(catalog)
        yaml.dump(catalog, sys.stdout)   
        """                
        catalog = {
            "default":{
                "index":   "index|node|/[a-z0-9]+$",
                "content": "content|[/_]20[01][0-9][/_-]?\d\d[/_-]?\d\d",
            },
            "auto.china.com.cn":{
                "content": "/20[01][0-9]\d{4}/\d+\.shtml",
            },
        }
        """
        fetcher = Fetcher()
        extract_rule = yaml.load(extract) 
        print (">>>>>",type(extract_rule))
        links = fetcher.build(url,'.china.com.cn', catalog, iframe_a=iframe_as.link)
        lll = sorted(links.items() , key=lambda t: t[0]+t[1].catalog) 
        result_list = []
        content_List = []
        urlNum = 1
        contentNum = 1
        for k, v in lll:
            if True:
                result_list.append( (urlNum,v.catalog, v.netloc, v.title, k))
                urlNum += 1
                if v.netloc in catalog['accepted']:
                    res = fetcher.detail_page(k, extract_rule)
                    if len(res) == 7:
                        content_List.append((contentNum,res["author"],res["editor"],res["title"], k,res["crumbs"],res["date"],res['source'], res['content']))
                        contentNum += 1 
                    else:
                        print(res)     
        self.list_result = result_list
        self.content_result = content_List 
        return result_list,content_List