perf(ip)

example/config.ini

@@ -137,13 +137,16 @@ output = ${DEST}/ss.acl
 
 ; ip输出
 [job:ip]
-output = ${DEST}/cnip.txt
+output = ${DEST}/ip-cn.txt
 ip-cc = cn
 ip-family = all
-ip-data-local = ${DEST}/ipdata.txt
-ip-data-update-local = true
 _order = 100
 
+[job:ip]
+output = ${DEST}/ipv6-us.txt
+ip-cc = us
+ip-family = 6
+_order = 100
 
 [job:quantumultx]
 output = ${DEST}/quantumultx.conf

pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
  "requests[socks]==2.31.0",
  "publicsuffixlist",
  "PyYAML==6.0.1",
- "IPy==1.1"
+ "netaddr==1.3.0"
 ]
 optional-dependencies.server = [
  "Werkzeug==2.2.2",

src/genpac/format/ip.py

@@ -1,16 +1,44 @@
 import re
-import math
 
-from IPy import IP, IPSet
+from netaddr import IPNetwork, IPRange, AddrFormatError
 
 from .base import formater, FmtBase
-from ..util import conv_lower, conv_path
-from ..util import logger, write_file, read_file
+from ..util import FatalError
+from ..util import conv_lower
+from ..util import logger
 
 IP_CC_DEF = 'CN'
-IP_DATA_DEF = 'https://ftp.apnic.net/apnic/stats/apnic/delegated-apnic-latest'
 IP_FAMILIES = ['4', '6', 'all']
 
+# NOTE: 中国地区的数据来自IP_DATA_ASN 其它来自 IP_DATA_GEOLITE2
+# REF: https://github.com/gaoyifan/china-operator-ip/
+# https://github.com/sapics/ip-location-db/tree/main/geolite2-country
+IP_DATA_ASN = {
+ 4: 'https://github.com/gaoyifan/china-operator-ip/raw/ip-lists/china.txt',
+ 6: 'https://github.com/gaoyifan/china-operator-ip/raw/ip-lists/china6.txt'
+}
+IP_DATA_GEOLITE2 = {
+ 4: 'https://github.com/sapics/ip-location-db/raw/main/geolite2-country/geolite2-country-ipv4.csv',
+ 6: 'https://github.com/sapics/ip-location-db/raw/main/geolite2-country/geolite2-country-ipv6.csv'
+}
+
+
+class IPList(list):
+ def add(self, item):
+ if isinstance(item, IPNetwork):
+ self.append(item)
+ elif isinstance(item, IPRange):
+ self.extend(item.cidrs())
+ else:
+ raise ValueError('ONLY IPNetwork or IPRange')
+
+ @property
+ def size(self):
+ return sum(item.size for item in self)
+
+ def iter_cidrs(self):
+ return self
+
 
 @formater('ip', desc="IP地址列表")
 class FmtIP(FmtBase):
@@ -24,72 +52,103 @@ def arguments(cls, parser):
  families = ', '.join(IP_FAMILIES)
  group = super(FmtIP, cls).arguments(parser)
  group.add_argument('--ip-cc', metavar='CC',
- help=f'国家代码(ISO 3166-2) 默认: {IP_CC_DEF}')
+ help=f'国家代码(ISO 3166-1) 默认: {IP_CC_DEF}')
  group.add_argument('--ip-family', metavar='FAMILY',
  type=lambda s: s.lower(),
  choices=IP_FAMILIES,
- default='4',
  help=f'IP类型 可选: {families} 默认: 4')
- group.add_argument('--ip-data-url', metavar='URL',
- help=f'IP数据地址 \n默认: {IP_DATA_DEF}')
- group.add_argument('--ip-data-local', metavar='FILE',
- help='IP数据本地, 当在线地址获取失败时使用')
- group.add_argument('--ip-data-update-local', action='store_true',
- help='当在线IP数据成功获取且--ip-data-local参数存在时, '
- '更新IP数据本地文件内容')
  return group
 
  @classmethod
  def config(cls, options):
  options['ip-cc'] = {'conv': conv_lower, 'default': IP_CC_DEF}
  options['ip-family'] = {'conv': conv_lower, 'default': '4'}
- options['ip-data-url'] = {'default': IP_DATA_DEF}
- options['ip-data-local'] = {'conv': conv_path}
- options['ip-data-update-local'] = {'default': False}
 
  def generate(self, replacements):
- content = self._fetch_data()
- cc = self.options.ip_cc or r'[a-z]{2}'
-
- ipset = IPSet()
-
- # IPv4
- if self.options.ip_family in ['4', 'all']:
- ipv4_record = 0
- for item in re.finditer(r'\|' + cc + r'\|ipv4\|([0-9\.]+)\|([0-9]+)\|',
- content, re.IGNORECASE):
- ipv4_record = ipv4_record + 1
- ipset.add(IP('{}/{:d}'.format(item.group(1),
- int(32 - math.log(float(item.group(2)), 2)))))
- logger.debug(f'IPv4[{cc}]: Nums: {ipset.len():.2e} '
- f'Record: {ipv4_record} => {len(ipset.prefixes)}')
-
- # IPv6
- if self.options.ip_family in ['6', 'all']:
- cur_set_nums = ipset.len()
- cur_set_record = len(ipset.prefixes)
- ipv6_record = 0
- for item in re.finditer(r'\|' + cc + r'\|ipv6\|([0-9a-ff:]+)\|([0-9]+)\|',
- content, re.IGNORECASE):
- ipv6_record = ipv6_record + 1
- ipset.add(IP(f'{item.group(1)}/{item.group(2)}'))
-
- logger.debug(f'IPv6[{cc}]: Nums: {ipset.len() - cur_set_nums:.2e} '
- f'Record: {ipv6_record} => {len(ipset.prefixes) - cur_set_record}')
-
- return '\n'.join([str(i) for i in ipset])
-
- def _fetch_data(self):
+ cc = self.options.ip_cc
+
+ print(self.options)
+
+ ip4s, ip6s = self._generate_by_cc(cc)
+
+ output = ip4s + ip6s
+
+ return '\n'.join([str(i) for i in output])
+
+ @property
+ def _ipv4(self):
+ return self.options.ip_family in [4, '4', 'all']
+
+ @property
+ def _ipv6(self):
+ return self.options.ip_family in [6, '6', 'all']
+
+ def _ip_network(self, data):
  try:
- content = self.generator.fetch(self.options.ip_data_url)
- if not content:
- raise ValueError()
- if self.options.ip_data_local \
- and self.options.ip_data_update_local:
- write_file(self.options.ip_data_local, content,
- fail_msg='更新本地IPData文件{path}失败')
- except Exception:
- if self.options.ip_data_local:
- content = read_file(self.options.ip_data_local,
- fail_msg='读取本地IPData文件{path}失败')
- return content
+ if isinstance(data, str):
+ return IPNetwork(data)
+ elif isinstance(data, tuple):
+ first, last = data
+ return IPRange(first, last)
+ raise ValueError('IP数据类型错误')
+ except Exception as e:
+ logger.warning(f'解析IP地址错误: {data} {e} {type(e)}')
+ return None
+
+ def _generate_by_cc(self, cc):
+ ip4s = IPList()
+ ip6s = IPList()
+
+ record = 0
+
+ if self._ipv4:
+ for d in self._fetch_data(4, cc):
+ ip_net = self._ip_network(d)
+ if ip_net:
+ ip4s.add(ip_net)
+ record = record + 1
+ logger.debug(f'IPv4[{cc}]: Nums: {ip4s.size:.2e} '
+ f'Record: {record} => {len(ip4s.iter_cidrs())}')
+
+ record = 0
+ if self._ipv6:
+ record = 0
+ for d in self._fetch_data(6, cc):
+ ip_net = self._ip_network(d)
+ if ip_net:
+ ip6s.add(ip_net)
+ record = record + 1
+
+ logger.debug(f'IPv6[{cc}]: Nums: {ip6s.size:.2e} '
+ f'Record: {record} => {len(ip6s.iter_cidrs())}')
+
+ return ip4s, ip6s
+
+ def _fetch_data_cn(self, family):
+ url = IP_DATA_ASN[int(family)]
+ content = self.generator.fetch(url)
+ if not content:
+ raise FatalError('获取IP数据失败')
+ for ip in content.splitlines():
+ ip = ip.strip()
+ if not ip:
+ continue
+ yield ip
+
+ def _fetch_data(self, family, cc):
+ if cc.lower() == 'cn':
+ yield from self._fetch_data_cn(family)
+ return
+
+ expr = re.compile(f'^[0-9a-f:,]+,{cc}' if family == 6 else f'^[0-9\\.,]+,{cc}',
+ flags=re.IGNORECASE)
+ url = IP_DATA_GEOLITE2[int(family)]
+ content = self.generator.fetch(url)
+ if not content:
+ raise FatalError('获取IP数据失败')
+ for d in content.splitlines():
+ d = d.strip()
+ if not d or not expr.fullmatch(d):
+ continue
+ first, last, _ = d.split(',')
+ yield (first, last)