Proxy configuration manager
| 5 | from ..core.models import ProxyConfig |
| 6 | |
| 7 | class ProxyManager: |
| 8 | """Proxy configuration manager""" |
| 9 | |
| 10 | def __init__(self, db: Database): |
| 11 | self.db = db |
| 12 | |
| 13 | def _parse_proxy_line(self, line: str) -> Optional[str]: |
| 14 | """将用户输入代理转换为标准 URL 格式。 |
| 15 | |
| 16 | 支持格式： |
| 17 | - http://user:pass@host:port |
| 18 | - https://user:pass@host:port |
| 19 | - socks5://user:pass@host:port |
| 20 | - socks5h://user:pass@host:port |
| 21 | - socks5://host:port:user:pass |
| 22 | - st5 host:port:user:pass |
| 23 | - host:port |
| 24 | - host:port:user:pass |
| 25 | """ |
| 26 | if not line: |
| 27 | return None |
| 28 | |
| 29 | line = line.strip() |
| 30 | if not line: |
| 31 | return None |
| 32 | |
| 33 | # st5 host:port:user:pass |
| 34 | st5_match = re.match(r"^st5\s+(.+)$", line, re.IGNORECASE) |
| 35 | if st5_match: |
| 36 | rest = st5_match.group(1).strip() |
| 37 | if "@" in rest: |
| 38 | return f"socks5://{rest}" |
| 39 | parts = rest.split(":") |
| 40 | if len(parts) >= 4 and parts[1].isdigit(): |
| 41 | host = parts[0] |
| 42 | port = parts[1] |
| 43 | username = parts[2] |
| 44 | password = ":".join(parts[3:]) |
| 45 | return f"socks5://{username}:{password}@{host}:{port}" |
| 46 | return None |
| 47 | |
| 48 | # 协议前缀格式 |
| 49 | if line.startswith(("http://", "https://", "socks5://", "socks5h://")): |
| 50 | |
| 51 | # 已是标准 user:pass@host:port（或 host:port） |
| 52 | if "@" in line: |
| 53 | return line |
| 54 | |
| 55 | # 兼容 protocol://host:port:user:pass |
| 56 | try: |
| 57 | protocol_end = line.index("://") + 3 |
| 58 | protocol = line[:protocol_end] |
| 59 | rest = line[protocol_end:] |
| 60 | parts = rest.split(":") |
| 61 | if len(parts) >= 4 and parts[1].isdigit(): |
| 62 | host = parts[0] |
| 63 | port = parts[1] |
| 64 | username = parts[2] |