mirror of
https://github.com/TG-Twilight/AWAvenue-Ads-Rule.git
synced 2025-11-04 14:49:47 +08:00
添加读取原始规则的合法性 有效性校验以及修复之前的部分错误,对dns层拦截的插件添加ipv6屏蔽兼容
Some checks failed
build Rule / execute_python_script (push) Has been cancelled
Some checks failed
build Rule / execute_python_script (push) Has been cancelled
This commit is contained in:
parent
6a85fec923
commit
98c71e9b2c
1
.github/workflows/build.yaml
vendored
1
.github/workflows/build.yaml
vendored
@ -24,6 +24,7 @@ jobs:
|
||||
git clone "https://${{ github.actor }}:${{ secrets.TOKEN }}@github.com/${{ github.repository }}" -b build build
|
||||
cd build
|
||||
mkdir out
|
||||
pip install -r requirements.txt
|
||||
python main.py
|
||||
cd out
|
||||
mkdir Filters
|
||||
|
||||
36
config.py
36
config.py
@ -1,4 +1,6 @@
|
||||
import os
|
||||
import dns.resolver
|
||||
|
||||
|
||||
SCRIPT_PATH = os.path.join(os.getcwd(), "script") # 插件文件夹
|
||||
RULE_PATH = os.path.join(os.getcwd(), "rule") # 规则文件夹
|
||||
@ -8,21 +10,21 @@ regex_file=RULE_PATH + "/domain_regex.txt" # 正则规则文件
|
||||
ip_file=RULE_PATH + "/ip.txt" # IP规则文件
|
||||
ip6_file=RULE_PATH + "/ip6.txt" # IPv6规则文件
|
||||
|
||||
if not os.path.exists(OUT_PATH) or not os.path.exists(OUT_PATH):
|
||||
print("插件或者规则目录不存在!")
|
||||
exit(1)
|
||||
if not os.path.exists(OUT_PATH):
|
||||
os.makedirs(OUT_PATH)
|
||||
|
||||
class RuleList:
|
||||
# 初始化规则列表
|
||||
def __init__(self, domain_file, regex_file, ip_file, ip6_file):
|
||||
with open(domain_file, 'r') as file:
|
||||
self.domain_list = sorted(set(line.strip() for line in file))
|
||||
with open(regex_file, 'r') as file:
|
||||
self.regex_list = sorted(set(line.strip() for line in file))
|
||||
with open(ip_file, 'r') as file:
|
||||
self.ip_list = sorted(set(line.strip() for line in file))
|
||||
with open(ip6_file, 'r') as file:
|
||||
self.ip6_list = sorted(set(line.strip() for line in file))
|
||||
rule = RuleList(domain_file, regex_file, ip_file, ip6_file)
|
||||
def check_domain(domain):
|
||||
try:
|
||||
resolver = dns.resolver.Resolver()
|
||||
resolver.nameservers = ['8.8.8.8', '1.1.1.1', '223.5.5.5', '9.9.9.9', '94.140.14.140']
|
||||
resolver.timeout = 1
|
||||
resolver.lifetime = 5
|
||||
|
||||
A = resolver.resolve(domain, "A", raise_on_no_answer=False)
|
||||
AAAA = resolver.resolve(domain, "AAAA", raise_on_no_answer=False)
|
||||
# 返回包含 A 和 AAAA 记录的字典
|
||||
return {"A": bool(A.rrset), "AAAA": bool(AAAA.rrset)}
|
||||
except dns.resolver.NXDOMAIN:
|
||||
print(f"未查询到 {domain} 的解析记录")
|
||||
return {"A": False, "AAAA": False}
|
||||
except Exception as e:
|
||||
print(f"查询 {domain} 时发生错误: {e}")
|
||||
return {"A": False, "AAAA": False}
|
||||
75
main.py
75
main.py
@ -1,7 +1,9 @@
|
||||
import re
|
||||
import os
|
||||
import config
|
||||
import importlib
|
||||
import subprocess
|
||||
from config import *
|
||||
import concurrent.futures
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
@ -15,9 +17,65 @@ def get_latest_git_tag(): # 获取最新的git tag
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
class RuleList:
|
||||
def __init__(self, domain_file, regex_file, ip_file, ip6_file):
|
||||
self.domain_list, self.domainv6_list = self.domain_file(domain_file)
|
||||
self.regex_list = self.regex_file(regex_file)
|
||||
self.ip_list = self.ip_file(ip_file)
|
||||
self.ip6_list = self.ip6_file(ip6_file)
|
||||
|
||||
|
||||
## 以下内容由deepseek提供技术支持(bushi
|
||||
|
||||
def domain_file(self, filename):
|
||||
with open(filename, 'r') as file:
|
||||
domains = {line.strip() for line in file}
|
||||
|
||||
valid_domains = set()
|
||||
valid_domains_v6 = set()
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
|
||||
future_to_domain = {executor.submit(config.check_domain, domain): domain for domain in domains}
|
||||
for future in concurrent.futures.as_completed(future_to_domain):
|
||||
domain = future_to_domain[future]
|
||||
result = future.result()
|
||||
if result["A"] or result["AAAA"]:
|
||||
valid_domains.add(domain)
|
||||
if result["AAAA"]:
|
||||
valid_domains_v6.add(domain)
|
||||
return sorted(valid_domains), sorted(valid_domains_v6)
|
||||
|
||||
def regex_file(self, filename):
|
||||
with open(filename, 'r') as file:
|
||||
return sorted({line.strip() for line in file})
|
||||
|
||||
def ip_file(self, filename):
|
||||
ipv4_pattern = re.compile(r'^(\d{1,3}\.){3}\d{1,3}$')
|
||||
ips = set()
|
||||
with open(filename, 'r') as file:
|
||||
for line in file:
|
||||
ip = line.strip()
|
||||
if ipv4_pattern.match(ip):
|
||||
parts = ip.split('.')
|
||||
if all(0 <= int(part) <= 255 for part in parts):
|
||||
ips.add(ip)
|
||||
return sorted(ips)
|
||||
|
||||
def ip6_file(self, filename):
|
||||
ipv6_pattern = re.compile(r'^([0-9a-fA-F]{0,4}:){2,7}[0-9a-fA-F]{0,4}$')
|
||||
ips = set()
|
||||
with open(filename, 'r') as file:
|
||||
for line in file:
|
||||
ip = line.strip()
|
||||
if ipv6_pattern.match(ip):
|
||||
ips.add(ip)
|
||||
return sorted(ips)
|
||||
|
||||
|
||||
|
||||
def WriteFile(name, text, suffix, comment, module_total): # 写入文件
|
||||
try:
|
||||
with open(OUT_PATH + "/AWAvenue-Ads-Rule-" + name + suffix, 'w', encoding="utf-8") as file:
|
||||
with open(config.OUT_PATH + "/AWAvenue-Ads-Rule-" + name + suffix, 'w', encoding="utf-8") as file:
|
||||
|
||||
if comment != "":
|
||||
now = datetime.now()
|
||||
@ -43,13 +101,16 @@ def WriteFile(name, text, suffix, comment, module_total): # 写入文件
|
||||
|
||||
|
||||
def RunScript():
|
||||
for filename in os.listdir(SCRIPT_PATH): # 遍历script目录下的所有文件
|
||||
rule = RuleList(config.domain_file, config.regex_file, config.ip_file, config.ip6_file)
|
||||
config.rule = rule
|
||||
|
||||
for filename in os.listdir(config.SCRIPT_PATH): # 遍历script目录下的所有文件
|
||||
if filename.endswith(".py"):
|
||||
plugins_name = filename[:-3]
|
||||
full_plugins_name = f"script.{plugins_name}" # 拼接完整的插件名
|
||||
|
||||
try:
|
||||
plugins = importlib.import_module(full_plugins_name).build(rule) # 传入规则列表(config.RuleList)类的实例
|
||||
plugins = importlib.import_module(full_plugins_name).build(rule=rule) # 传入规则列表(config.RuleList)类的实例
|
||||
|
||||
if plugins['list'] == True:
|
||||
print(f"{plugins_name}转换成功")
|
||||
@ -67,5 +128,9 @@ def RunScript():
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
|
||||
if not os.path.exists(config.RULE_PATH) or not os.path.exists(config.SCRIPT_PATH):
|
||||
print("插件或者规则目录不存在!")
|
||||
exit(1)
|
||||
if not os.path.exists(config.OUT_PATH):
|
||||
os.makedirs(config.OUT_PATH)
|
||||
RunScript()
|
||||
|
||||
1
requirements.txt
Normal file
1
requirements.txt
Normal file
@ -0,0 +1 @@
|
||||
dnspython
|
||||
@ -1,11 +1,15 @@
|
||||
|
||||
def format_domain(List):
|
||||
def format_domain(List, Listv6):
|
||||
domain = []
|
||||
for line in List:
|
||||
domain_lines = f"address=/{line.strip()}/0.0.0.0"
|
||||
domain.append(domain_lines)
|
||||
|
||||
for line in Listv6:
|
||||
domain_lines = f"address=/{line.strip()}/::"
|
||||
domain.append(domain_lines)
|
||||
return domain
|
||||
|
||||
|
||||
def build(rule):
|
||||
return {'list': format_domain(rule.domain_list), 'suffix': '.conf', 'comment': '#', 'total': len(rule.domain_list)}
|
||||
return {'list': format_domain(rule.domain_list, rule.domainv6_list), 'suffix': '.conf', 'comment': '#', 'total': len(rule.domain_list)}
|
||||
@ -14,40 +14,10 @@
|
||||
|
||||
API:
|
||||
- rule.domain_list(获取域名列表)
|
||||
- rule.domain_v6_list(获取支持ipv6域名列表)
|
||||
- rule.regex_list(获取正则表达式的域名列表)
|
||||
- rule.ip_list(获取ip列表)
|
||||
- rule.ip6_list(获取ipv6列表)
|
||||
|
||||
> 所有变量均为列表 也可以导入config.py获取
|
||||
|
||||
模板:
|
||||
```python
|
||||
|
||||
def format_domain(List): # 转换域名规则
|
||||
domain = []
|
||||
for line in List:
|
||||
domain_lines = f" - DOMAIN,{line.strip()}"
|
||||
domain.append(domain_lines)
|
||||
return domain
|
||||
|
||||
def format_regex(List): # 转换正则表达式规则
|
||||
regex = []
|
||||
for line in List:
|
||||
regex_lines = f" - DOMAIN-REGEX,'{line.strip()}'"
|
||||
regex.append(regex_lines)
|
||||
return regex
|
||||
|
||||
def format_ip(List): # 转换ip列表
|
||||
ip = []
|
||||
for line in List:
|
||||
ip_lines = f" - IP-CIDR,{line.strip()}"
|
||||
ip.append(ip_lines)
|
||||
return ip
|
||||
|
||||
def build(rule): # 入口函数
|
||||
clash_list = ["payload:"] + format_ip(rule.ip_list) + format_domain(rule.domain_list) + format_regex(rule.regex_list)
|
||||
return clash_list, ".yaml", "#", len(clash_list)
|
||||
#
|
||||
|
||||
```
|
||||
> 输出文件名=插件名
|
||||
输出文件名=插件名
|
||||
@ -1,12 +1,16 @@
|
||||
import json
|
||||
|
||||
def format_domain(List):
|
||||
def format_domain(List, Listv6):
|
||||
domain = ["0.0.0.0 localhost", "::1 localhost", "", ""]
|
||||
for line in List:
|
||||
domain_lines = f"0.0.0.0 {line.strip()}"
|
||||
domain.append(domain_lines)
|
||||
|
||||
for line in Listv6:
|
||||
domain_lines = f":: {line.strip()}"
|
||||
domain.append(domain_lines)
|
||||
return domain
|
||||
|
||||
|
||||
def build(rule):
|
||||
return {'list': format_domain(rule.domain_list), 'suffix': '.txt', 'comment': '#', 'total': len(rule.domain_list)}
|
||||
return {'list': format_domain(rule.domain_list, rule.domainv6_list), 'suffix': '.txt', 'comment': '#', 'total': len(rule.domain_list)}
|
||||
@ -1,11 +1,15 @@
|
||||
import json
|
||||
|
||||
def format_domain(List):
|
||||
def format_domain(List, Listv6):
|
||||
domain = []
|
||||
for line in List:
|
||||
domain_line = f"ip dns static add address=240.0.0.1 name={line.strip()}"
|
||||
domain.append(domain_line)
|
||||
|
||||
for line in Listv6:
|
||||
domain_line = f"ip dns static add address=:: name={line.strip()}"
|
||||
domain.append(domain_line)
|
||||
return domain
|
||||
|
||||
def build(rule):
|
||||
return {'list': format_domain(rule.domain_list), 'suffix': '.txt', 'comment': '!', 'total': len(rule.regex_list)}
|
||||
return {'list': format_domain(rule.domain_list, rule.domainv6_list), 'suffix': '.txt', 'comment': '!', 'total': len(rule.regex_list)}
|
||||
|
||||
@ -30,4 +30,4 @@ def build(rule):
|
||||
}
|
||||
|
||||
json_data = [json.dumps(List, indent=2)]
|
||||
return {'list': json_data, 'suffix': '.json', 'comment': '//', 'total': len(json_data)}
|
||||
return {'list': json_data, 'suffix': '.json', 'comment': '', 'total': len(json_data)}
|
||||
@ -21,4 +21,4 @@ def build(rule):
|
||||
}
|
||||
|
||||
json_data = [json.dumps(rule, indent=2)]
|
||||
return {'list': json_data, 'suffix': '.json', 'comment': '//', 'total': len(json_data)}
|
||||
return {'list': json_data, 'suffix': '.json', 'comment': '', 'total': len(json_data)}
|
||||
@ -1,11 +1,15 @@
|
||||
|
||||
def format_domain(List):
|
||||
def format_domain(List, Listv6):
|
||||
domain = ["127.0.0.1 localhost", "::1 localhost", "", ""]
|
||||
for line in List:
|
||||
domain_lines = f"0.0.0.0 {line.strip()}"
|
||||
domain.append(domain_lines)
|
||||
|
||||
for line in Listv6:
|
||||
domain_lines = f":: {line.strip()}"
|
||||
domain.append(domain_lines)
|
||||
return domain
|
||||
|
||||
|
||||
def build(rule):
|
||||
return {'list': format_domain(rule.domain_list), 'suffix': '.txt', 'comment': '!', 'total': len(rule.domain_list)}
|
||||
return {'list': format_domain(rule.domain_list, rule.domainv6_list), 'suffix': '.txt', 'comment': '!', 'total': len(rule.domain_list)}
|
||||
Loading…
Reference in New Issue
Block a user