Skip to content

route_rules.source

Modules:

Classes:

Functions:

Attributes:

PRESETS module-attribute 

PRESETS: list[PresetConfig] = [
    PresetConfig(
        "ads",
        "🛑 ADs",
        [
            "blackmatrix7:Advertising",
            "DustinWin/geosite-all:ads",
            "MetaCubeX/geosite:*-ads,*-ads-all,*@ads",
        ],
        [],
    ),
    PresetConfig(
        "private",
        "🔒 Private",
        [
            "blackmatrix7:Lan,NTPService",
            "DustinWin/geoip-all:private",
            "DustinWin/geosite-all:private",
            "MetaCubeX/geoip:private",
            "MetaCubeX/geosite:category-ntp*,private",
        ],
        ["preset:ads"],
    ),
    PresetConfig(
        "cn",
        "🇨🇳 CN",
        [
            "blackmatrix7:ChinaMax,Direct",
            "DustinWin/geoip-all:cn",
            "DustinWin/geosite-all:cn",
            "liblaf:cn",
            "MetaCubeX/geoip:cn",
            "MetaCubeX/geosite:cn,*-cn,*@cn",
        ],
        ["liblaf:!cn", "preset:ads", "preset:private"],
    ),
    PresetConfig(
        "proxy",
        "✈️ Proxy",
        [
            "blackmatrix7:Global",
            "DustinWin/geosite-all:proxy",
            "MetaCubeX/geosite:*!cn*",
        ],
        ["preset:ads", "preset:cn", "preset:private"],
    ),
    PresetConfig(
        "ai",
        "🤖 AI",
        [
            "blackmatrix7:Claude,Copilot,Gemini,OpenAI",
            "DustinWin/geosite-all:ai",
            "MetaCubeX/geosite:openai",
        ],
        ["preset:ads", "preset:cn", "preset:private"],
    ),
    PresetConfig(
        "download",
        "☁️ Download",
        [
            "blackmatrix7:Download,OneDrive",
            "MetaCubeX/geosite:onedrive",
        ],
        ["preset:ads", "preset:cn", "preset:private"],
    ),
    PresetConfig(
        "emby",
        "🍟 Emby",
        ["liblaf:emby", "NotSFC:Emby"],
        ["preset:ads", "preset:cn", "preset:private"],
    ),
    PresetConfig(
        "media",
        "📺 Media",
        [
            "blackmatrix7:GlobalMedia",
            "DustinWin/geosite-all:youtube",
            "MetaCubeX/geosite-lite:proxymedia,youtube",
            "MetaCubeX/geosite:youtube",
        ],
        ["preset:ads", "preset:cn", "preset:private"],
    ),
]

ClashClassicalText

Bases: Source

Methods:

Attributes:

Source code in src/route_rules/source/_clash.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class ClashClassicalText(Source):
    name: str
    dpath: Path
    url: Template

    def __init__(self, name: str, url: str | Template, dpath: StrPath) -> None:
        super().__init__()
        self.name = name
        if isinstance(url, str):
            self.url = Template(url)
        else:
            self.url = url
        self.dpath = Path(dpath)

    async def _get_nocache(self, key: str) -> Rule:
        filepath: Path = await rr.utils.download(
            self.url.substitute({"key": key}), self.dpath / f"{key}.list"
        )
        return ClashClassicalText.from_file(filepath)

    async def _keys_nocache(self) -> list[str]:
        raise NotImplementedError

    @staticmethod
    def from_file(fpath: StrPath) -> Rule:
        text: str = Path(fpath).read_text()
        rule: Rule = Rule()
        for line in rr.utils.strip_comments(text):
            words: list[str] = rr.utils.split_strip(line)
            match words[0]:
                case "DOMAIN":
                    rule.domain.add(words[1])
                case "DOMAIN-SUFFIX":
                    rule.domain_suffix.add(words[1])
                case "DOMAIN-KEYWORD":
                    rule.domain_keyword.add(words[1])
                case "DOMAIN-REGEX":
                    rule.domain_regex.add(words[1])
                case "IP-CIDR" | "IP-CIDR6":
                    rule.ip_cidr.add(words[1])
                case "IP-ASN":
                    pass  # TODO
                case "PROCESS-NAME":
                    pass  # TODO
                case _:
                    msg: str = f"Unknown rule: {line}"
                    raise ValueError(msg)
        return rule

dpath instance-attribute 

dpath: Path = Path(dpath)

name instance-attribute 

name: str = name

url instance-attribute 

url: Template

__init__ 

__init__(
    name: str, url: str | Template, dpath: StrPath
) -> None
Source code in src/route_rules/source/_clash.py 
14
15
16
17
18
19
20
21
def __init__(self, name: str, url: str | Template, dpath: StrPath) -> None:
    super().__init__()
    self.name = name
    if isinstance(url, str):
        self.url = Template(url)
    else:
        self.url = url
    self.dpath = Path(dpath)

from_file staticmethod 

from_file(fpath: StrPath) -> Rule
Source code in src/route_rules/source/_clash.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@staticmethod
def from_file(fpath: StrPath) -> Rule:
    text: str = Path(fpath).read_text()
    rule: Rule = Rule()
    for line in rr.utils.strip_comments(text):
        words: list[str] = rr.utils.split_strip(line)
        match words[0]:
            case "DOMAIN":
                rule.domain.add(words[1])
            case "DOMAIN-SUFFIX":
                rule.domain_suffix.add(words[1])
            case "DOMAIN-KEYWORD":
                rule.domain_keyword.add(words[1])
            case "DOMAIN-REGEX":
                rule.domain_regex.add(words[1])
            case "IP-CIDR" | "IP-CIDR6":
                rule.ip_cidr.add(words[1])
            case "IP-ASN":
                pass  # TODO
            case "PROCESS-NAME":
                pass  # TODO
            case _:
                msg: str = f"Unknown rule: {line}"
                raise ValueError(msg)
    return rule

get async 

get(*key: str) -> Rule
Source code in src/route_rules/source/_abc.py 
17
18
async def get(self, *key: str) -> Rule:
    return Rule().union(*(await asyncio.gather(*(self._get(k) for k in key))))

keys async 

keys() -> list[str]
Source code in src/route_rules/source/_abc.py 
20
21
22
23
24
async def keys(self) -> list[str]:
    if self._key_cache is not None:
        return self._key_cache
    self._key_cache = await self._keys_nocache()
    return self._key_cache

GeoIP

Bases: Source

Methods:

Attributes:

Source code in src/route_rules/source/_geoip.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class GeoIP(Source):
    name: str
    dpath: Path
    url: str

    def __init__(self, name: str, url: str, dpath: StrPath) -> None:
        super().__init__()
        self.name = name
        self.dpath = Path(dpath)
        self.url = url

    @property
    def fpath(self) -> Path:
        return self.dpath / "geoip.db"

    async def _get_nocache(self, key: str) -> Rule:
        await rr.utils.download(self.url, self.fpath)
        output: Path = self.dpath / f"{key}.json"
        args: list[StrPath] = [
            "sing-box",
            "geoip",
            "export",
            key,
            "--output",
            output,
            "--file",
            self.fpath,
        ]
        proc: asp.Process = await asyncio.create_subprocess_exec(
            *args, stdin=asp.DEVNULL
        )
        ret: int = await proc.wait()
        if ret != 0:
            raise sp.CalledProcessError(ret, args)
        return Rule.from_file(output)

    async def _keys_nocache(self) -> list[str]:
        await rr.utils.download(self.url, self.fpath)
        args: list[StrPath] = ["sing-box", "geoip", "list", "--file", self.fpath]
        proc: asp.Process = await asyncio.create_subprocess_exec(
            *args, stdin=asp.DEVNULL, stdout=asp.PIPE
        )
        stdout: bytes
        stdout, _ = await proc.communicate()
        ret: int = await proc.wait()
        if ret != 0:
            raise sp.CalledProcessError(ret, args)
        categories: dict[str, int] = {}
        for line in stdout.decode().splitlines():
            if m := re.match(r"(?P<name>.*) \((?P<count>\d+)\)", line):
                categories[m["name"]] = int(m["count"])
        return list(categories.keys())

dpath instance-attribute 

dpath: Path = Path(dpath)

fpath property 

fpath: Path

name instance-attribute 

name: str = name

url instance-attribute 

url: str = url

__init__ 

__init__(name: str, url: str, dpath: StrPath) -> None
Source code in src/route_rules/source/_geoip.py 
17
18
19
20
21
def __init__(self, name: str, url: str, dpath: StrPath) -> None:
    super().__init__()
    self.name = name
    self.dpath = Path(dpath)
    self.url = url

get async 

get(*key: str) -> Rule
Source code in src/route_rules/source/_abc.py 
17
18
async def get(self, *key: str) -> Rule:
    return Rule().union(*(await asyncio.gather(*(self._get(k) for k in key))))

keys async 

keys() -> list[str]
Source code in src/route_rules/source/_abc.py 
20
21
22
23
24
async def keys(self) -> list[str]:
    if self._key_cache is not None:
        return self._key_cache
    self._key_cache = await self._keys_nocache()
    return self._key_cache

GeoSite

Bases: Source

Methods:

Attributes:

Source code in src/route_rules/source/_geosite.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class GeoSite(Source):
    name: str
    dpath: Path
    url: str

    def __init__(self, name: str, url: str, dpath: StrPath) -> None:
        super().__init__()
        self.name = name
        self.dpath = Path(dpath)
        self.url = url

    @property
    def fpath(self) -> Path:
        return self.dpath / "geosite.db"

    async def _get_nocache(self, key: str) -> Rule:
        await rr.utils.download(self.url, self.fpath)
        output: Path = self.dpath / f"{key}.json"
        args: list[StrPath] = [
            "sing-box",
            "geosite",
            "export",
            key,
            "--output",
            output,
            "--file",
            self.fpath,
        ]
        proc: asp.Process = await asyncio.create_subprocess_exec(
            *args, stdin=asp.DEVNULL
        )
        ret: int = await proc.wait()
        if ret != 0:
            raise sp.CalledProcessError(ret, args)
        return Rule.from_file(output)

    async def _keys_nocache(self) -> list[str]:
        await rr.utils.download(self.url, self.fpath)
        args: list[StrPath] = ["sing-box", "geosite", "list", "--file", self.fpath]
        proc: asp.Process = await asyncio.create_subprocess_exec(
            *args, stdin=asp.DEVNULL, stdout=asp.PIPE
        )
        stdout: bytes
        stdout, _ = await proc.communicate()
        ret: int = await proc.wait()
        if ret != 0:
            raise sp.CalledProcessError(ret, args)
        categories: dict[str, int] = {}
        for line in stdout.decode().splitlines():
            if m := re.match(r"(?P<name>.*) \((?P<count>\d+)\)", line):
                categories[m["name"]] = int(m["count"])
        return list(categories.keys())

dpath instance-attribute 

dpath: Path = Path(dpath)

fpath property 

fpath: Path

name instance-attribute 

name: str = name

url instance-attribute 

url: str = url

__init__ 

__init__(name: str, url: str, dpath: StrPath) -> None
Source code in src/route_rules/source/_geosite.py 
17
18
19
20
21
def __init__(self, name: str, url: str, dpath: StrPath) -> None:
    super().__init__()
    self.name = name
    self.dpath = Path(dpath)
    self.url = url

get async 

get(*key: str) -> Rule
Source code in src/route_rules/source/_abc.py 
17
18
async def get(self, *key: str) -> Rule:
    return Rule().union(*(await asyncio.gather(*(self._get(k) for k in key))))

keys async 

keys() -> list[str]
Source code in src/route_rules/source/_abc.py 
20
21
22
23
24
async def keys(self) -> list[str]:
    if self._key_cache is not None:
        return self._key_cache
    self._key_cache = await self._keys_nocache()
    return self._key_cache

Preset

Bases: Source

Methods:

Attributes:

Source code in src/route_rules/source/preset/_preset.py 
14
15
16
17
18
19
20
21
class Preset(Source):
    name: str = "preset"

    async def _get_nocache(self, key: str) -> Rule:
        return await get_preset(key)

    async def _keys_nocache(self) -> list[str]:
        return [preset.id for preset in PRESETS]

name class-attribute instance-attribute 

name: str = 'preset'

__init__ 

__init__() -> None
Source code in src/route_rules/source/_abc.py 
14
15
def __init__(self) -> None:
    self._rule_cache = LRU()

get async 

get(*key: str) -> Rule
Source code in src/route_rules/source/_abc.py 
17
18
async def get(self, *key: str) -> Rule:
    return Rule().union(*(await asyncio.gather(*(self._get(k) for k in key))))

keys async 

keys() -> list[str]
Source code in src/route_rules/source/_abc.py 
20
21
22
23
24
async def keys(self) -> list[str]:
    if self._key_cache is not None:
        return self._key_cache
    self._key_cache = await self._keys_nocache()
    return self._key_cache

PresetConfig

Bases: NamedTuple

Attributes:

Source code in src/route_rules/source/preset/_const.py 
 7
 8
 9
10
11
class PresetConfig(NamedTuple):
    id: str
    name: str
    include: list[str]
    exclude: list[str]

exclude instance-attribute 

exclude: list[str]

id instance-attribute 

id: str

include instance-attribute 

include: list[str]

name instance-attribute 

name: str

SingBoxRuleSet

Bases: Source

Methods:

Attributes:

Source code in src/route_rules/source/_singbox.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class SingBoxRuleSet(Source):
    name: str
    dpath: Path
    url: Template

    def __init__(self, name: str, url: str | Template, dpath: StrPath) -> None:
        super().__init__()
        self.name = name
        if isinstance(url, str):
            self.url = Template(url)
        else:
            self.url = url
        self.dpath = Path(dpath)

    async def _get_nocache(self, key: str) -> Rule:
        filepath: Path = await rr.utils.download(
            self.url.substitute({"key": key}), self.dpath / f"{key}.json"
        )
        return Rule.from_file(filepath)

    async def _keys_nocache(self) -> list[str]:
        raise NotImplementedError

dpath instance-attribute 

dpath: Path = Path(dpath)

name instance-attribute 

name: str = name

url instance-attribute 

url: Template

__init__ 

__init__(
    name: str, url: str | Template, dpath: StrPath
) -> None
Source code in src/route_rules/source/_singbox.py 
14
15
16
17
18
19
20
21
def __init__(self, name: str, url: str | Template, dpath: StrPath) -> None:
    super().__init__()
    self.name = name
    if isinstance(url, str):
        self.url = Template(url)
    else:
        self.url = url
    self.dpath = Path(dpath)

get async 

get(*key: str) -> Rule
Source code in src/route_rules/source/_abc.py 
17
18
async def get(self, *key: str) -> Rule:
    return Rule().union(*(await asyncio.gather(*(self._get(k) for k in key))))

keys async 

keys() -> list[str]
Source code in src/route_rules/source/_abc.py 
20
21
22
23
24
async def keys(self) -> list[str]:
    if self._key_cache is not None:
        return self._key_cache
    self._key_cache = await self._keys_nocache()
    return self._key_cache

Source

Bases: ABC

Methods:

Attributes:

Source code in src/route_rules/source/_abc.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class Source(abc.ABC):
    name: str
    _key_cache: list[str] | None = None
    _rule_cache: LRU[str, Rule]

    def __init__(self) -> None:
        self._rule_cache = LRU()

    async def get(self, *key: str) -> Rule:
        return Rule().union(*(await asyncio.gather(*(self._get(k) for k in key))))

    async def keys(self) -> list[str]:
        if self._key_cache is not None:
            return self._key_cache
        self._key_cache = await self._keys_nocache()
        return self._key_cache

    @abc.abstractmethod
    async def _get_nocache(self, key: str) -> Rule: ...

    async def _get(self, key: str) -> Rule:
        if (r := self._rule_cache.get(key)) is not None:
            return r
        rule: Rule = await self._get_nocache(key)
        self._rule_cache[key] = rule
        return rule

    @abc.abstractmethod
    async def _keys_nocache(self) -> list[str]: ...

name instance-attribute 

name: str

__init__ 

__init__() -> None
Source code in src/route_rules/source/_abc.py 
14
15
def __init__(self) -> None:
    self._rule_cache = LRU()

get async 

get(*key: str) -> Rule
Source code in src/route_rules/source/_abc.py 
17
18
async def get(self, *key: str) -> Rule:
    return Rule().union(*(await asyncio.gather(*(self._get(k) for k in key))))

keys async 

keys() -> list[str]
Source code in src/route_rules/source/_abc.py 
20
21
22
23
24
async def keys(self) -> list[str]:
    if self._key_cache is not None:
        return self._key_cache
    self._key_cache = await self._keys_nocache()
    return self._key_cache

get_rule async 

get_rule(*spec: str) -> Rule
Source code in src/route_rules/source/preset/_rule.py 
10
11
async def get_rule(*spec: str) -> Rule:
    return Rule().union(*(await asyncio.gather(*(_get_rule(s) for s in spec))))

get_source 

get_source(name: str) -> Source
Source code in src/route_rules/source/_const.py 
49
50
51
52
53
def get_source(name: str) -> Source:
    for source in SOURCES:
        if source.name == name:
            return source
    raise KeyError(name)