import upstream GUKO 2026-06-23

This commit is contained in:
OpenClaw
2026-06-23 00:44:35 +00:00
commit 0e79f74184
30 changed files with 6419 additions and 0 deletions

21
telegram-bot/Dockerfile Normal file
View File

@@ -0,0 +1,21 @@
FROM python:3.13-slim
WORKDIR /app
ARG GUKO_VERSION=0.1.24
ENV GUKO_VERSION=${GUKO_VERSION}
LABEL org.opencontainers.image.version=${GUKO_VERSION}
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
RUN apt-get update \
&& apt-get install -y --no-install-recommends openssh-client sshpass ca-certificates fonts-dejavu-core fonts-noto-cjk curl gnupg librsvg2-bin chromium \
&& curl -fsSL https://deb.nodesource.com/setup_24.x | bash - \
&& apt-get install -y --no-install-recommends nodejs \
&& rm -rf /var/lib/apt/lists/*
COPY telegram-bot/requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir -r /app/requirements.txt
RUN npm install -g playwright@1.59.1
ENV NODE_PATH=/usr/lib/node_modules PLAYWRIGHT_BROWSERS_PATH=/ms-playwright
COPY auth.py guko.py jiaoops.py /app/
RUN mkdir -p /data/keys /data/media /data/tmp
COPY telegram-bot/bot.py telegram-bot/render_checkplace.py /app/
COPY telegram-bot/tools/ /app/tools/
ENV BGP_FETCH=/app/tools/bgp_fetch.py IPPURE_DOWNLOAD=/app/tools/download_ippure.js
CMD ["python", "/app/bot.py"]

3942
telegram-bot/bot.py Executable file

File diff suppressed because it is too large Load Diff

145
telegram-bot/render_checkplace.py Executable file
View File

@@ -0,0 +1,145 @@
#!/usr/bin/env python3
"""Render Check.Place/IPQuality SVG reports into Telegram-friendly PNG images.
This intentionally does not rely on browser/SVG font metrics. Check.Place SVGs use
terminal cells (ch/em) plus colored background rectangles; normal SVG converters
often misalign mixed CJK/Latin text. This script parses the SVG and renders it as a
native terminal-like screenshot with a fixed cell grid and CJK fallback.
"""
from __future__ import annotations
import argparse
import html
import re
import unicodedata
from pathlib import Path
from PIL import Image, ImageDraw, ImageFont
DEFAULT_LATIN = "/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf"
DEFAULT_LATIN_ITALIC = "/usr/share/fonts/truetype/dejavu/DejaVuSansMono-Oblique.ttf"
DEFAULT_CJK = "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc"
FG = {
"fa0": (0, 0, 0),
"fa1": (255, 112, 112),
"fa2": (100, 255, 116),
"fa3": (255, 232, 96),
"fa6": (96, 245, 245),
"fa7": (246, 246, 246),
}
BG = {
"ba1": (178, 22, 22),
"ba2": (14, 150, 28),
"ba3": (166, 146, 22),
"ba7": (225, 225, 225),
}
TERMINAL_BG = (8, 10, 14)
OUTPUT_SCALE = 1
def cells(ch: str) -> int:
return 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
def is_cjk(ch: str) -> bool:
o = ord(ch)
return (0x2E80 <= o <= 0x9FFF) or (0xF900 <= o <= 0xFAFF) or (0xFF00 <= o <= 0xFFEF)
def parse_svg_size(svg: str) -> tuple[int, int]:
m = re.search(r'<svg[^>]*width="([0-9.]+)ch"[^>]*height="([0-9.]+)em"', svg)
if not m:
return 74, 47
return int(float(m.group(1))), int(float(m.group(2)))
def render(svg_path: Path, out_path: Path, *, cell_w: int, cell_h: int, font_size: int, pad: int) -> None:
svg = svg_path.read_text("utf-8", errors="ignore")
width_cells, height_cells = parse_svg_size(svg)
latin = ImageFont.truetype(DEFAULT_LATIN, font_size)
latin_italic = ImageFont.truetype(DEFAULT_LATIN_ITALIC, font_size)
cjk = ImageFont.truetype(DEFAULT_CJK, font_size)
scale = OUTPUT_SCALE
cell_w *= scale
cell_h *= scale
font_size *= scale
pad *= scale
image = Image.new(
"RGB",
(pad * 2 + width_cells * cell_w, pad * 2 + height_cells * cell_h),
TERMINAL_BG,
)
draw = ImageDraw.Draw(image)
# Draw terminal background highlight blocks first, using the same cell metrics as text.
rect_re = re.compile(
r'<rect x="([0-9.]+)ch" y="([0-9.]+)em" width="([0-9.]+)ch" height="1em" class="(ba\d)"'
)
for m in rect_re.finditer(svg):
x, y, w, cls = float(m.group(1)), float(m.group(2)), float(m.group(3)), m.group(4)
color = BG.get(cls)
if not color:
continue
draw.rectangle(
[
pad + x * cell_w,
pad + y * cell_h,
pad + (x + w) * cell_w,
pad + (y + 1) * cell_h,
],
fill=color,
)
text_re = re.compile(r'<text x="0ch" y="([0-9.]+)em">(.*?)</text>', re.S)
span_re = re.compile(r'<tspan(?: class="([^"]+)")?>(.*?)</tspan>', re.S)
for tm in text_re.finditer(svg):
y = float(tm.group(1))
top = pad + y * cell_h - cell_h / 2
col = 0
for sp in span_re.finditer(tm.group(2)):
classes = (sp.group(1) or "").split()
text = html.unescape(re.sub(r"<.*?>", "", sp.group(2))).replace("\r", "")
color = FG["fa7"]
italic = "italic" in classes
underline = "underline" in classes
for cls in classes:
if cls in FG:
color = FG[cls]
for ch in text:
span = cells(ch)
x = pad + col * cell_w
font = cjk if is_cjk(ch) else (latin_italic if italic else latin)
bbox = draw.textbbox((0, 0), ch, font=font)
text_w = bbox[2] - bbox[0]
text_h = bbox[3] - bbox[1]
tx = x + (span * cell_w - text_w) / 2 - bbox[0]
ty = top + (cell_h - text_h) / 2 - bbox[1]
draw.text((tx, ty), ch, font=font, fill=color)
if underline and ch != " ":
draw.line((x, top + cell_h - 3, x + span * cell_w, top + cell_h - 3), fill=color, width=1)
col += span
out_path.parent.mkdir(parents=True, exist_ok=True)
image.save(out_path, optimize=False, compress_level=4)
def main() -> int:
parser = argparse.ArgumentParser(description="Render Check.Place SVG to terminal-like PNG")
parser.add_argument("svg", type=Path)
parser.add_argument("output", type=Path)
parser.add_argument("--cell-w", type=int, default=15, help="terminal cell width in px; wider cells make Telegram previews easier to read")
parser.add_argument("--cell-h", type=int, default=30, help="terminal cell height in px; taller rows keep larger text crisp")
parser.add_argument("--font-size", type=int, default=26, help="font size in px; tuned for readable Telegram previews")
parser.add_argument("--pad", type=int, default=10, help="padding in px")
args = parser.parse_args()
render(args.svg, args.output, cell_w=args.cell_w, cell_h=args.cell_h, font_size=args.font_size, pad=args.pad)
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -0,0 +1,3 @@
python-telegram-bot==21.9
Pillow==11.2.1
CairoSVG==2.7.1

241
telegram-bot/tools/bgp_fetch.py Executable file
View File

@@ -0,0 +1,241 @@
#!/usr/bin/env python3
import argparse, gzip, ipaddress, re, sys, time, socket, zlib
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
from html.parser import HTMLParser
OUTDIR = Path('/data/media/bgp')
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.6261.112 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
'Accept-Encoding': 'gzip, deflate',
'Referer': 'https://bgp.tools/',
'Accept-Language': 'en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7',
'Upgrade-Insecure-Requests': '1',
'Cache-Control': 'max-age=0',
'Sec-Ch-Ua': '"Chromium";v="122", "Google Chrome";v="122", "Not=A?Brand";v="99"',
'Sec-Ch-Ua-Mobile': '?0',
'Sec-Ch-Ua-Platform': '"Windows"',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none',
'Sec-Fetch-User': '?1',
'Sec-Fetch-Dest': 'document',
'Dnt': '1',
'Sec-Gpc': '1',
'Pragma': 'no-cache',
}
class TextParser(HTMLParser):
def __init__(self):
super().__init__()
self.parts=[]
def handle_data(self, data):
if data and data.strip(): self.parts.append(data.strip())
def text(self): return '\n'.join(self.parts)
def resolve_target(s):
raw = s.strip()
m = re.search(r'(\d{1,3}(?:\.\d{1,3}){3})', raw)
if m:
try:
return ipaddress.IPv4Address(m.group(1)), None
except Exception:
raise SystemExit('ERROR: invalid IPv4')
# Treat as domain/hostname: strip scheme/path/port and resolve A records.
host = re.sub(r'^[a-zA-Z][a-zA-Z0-9+.-]*://', '', raw).split('/')[0].split('?')[0].strip('[]')
if '@' in host:
host = host.rsplit('@', 1)[-1]
if ':' in host and host.count(':') == 1:
host = host.rsplit(':', 1)[0]
host = host.strip().rstrip('.')
if not host or not re.match(r'^[A-Za-z0-9.-]+$', host):
raise SystemExit('ERROR: no IPv4 or valid domain found')
try:
infos = socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM)
except socket.gaierror as e:
raise SystemExit(f'ERROR: failed to resolve domain {host}: {e}')
ips = []
for info in infos:
addr = info[4][0]
if addr not in ips:
ips.append(addr)
if not ips:
raise SystemExit(f'ERROR: no IPv4 A record found for {host}')
return ipaddress.IPv4Address(ips[0]), host
def prefixes(ip):
# Prefer real visible prefixes from bgp.tools search, sorted by highest visibility.
# Tie-breaker: more-specific first, then bgp.tools row order. Blind /24 can be wrong.
real = search_prefixes(ip)
if real:
return [net for net, _visibility, _asn in real]
p24 = ipaddress.IPv4Network(f'{ip}/24', strict=False)
p23 = ipaddress.IPv4Network(f'{ip}/23', strict=False)
res=[p24]
if p23 != p24: res.append(p23)
return res
def search_prefixes(ip):
url=f'https://bgp.tools/search?q={ip}'
try:
html,_=fetch(url)
except Exception:
return []
rows=[]
text=html.decode('utf-8','ignore')
visibility_rank={'high': 3, 'medium': 2, 'low': 1}
seen=set()
for rm in re.finditer(r'<tr\b[^>]*>(.*?)</tr>', text, re.I | re.S):
row=rm.group(1)
pm=re.search(r'/prefix/(\d{1,3}(?:\.\d{1,3}){3}/\d{1,2})', row)
if not pm:
continue
try:
net=ipaddress.IPv4Network(pm.group(1), strict=False)
except Exception:
continue
if ip not in net or net in seen:
continue
seen.add(net)
am=re.search(r'/as/(\d+)', row, re.I)
asn=f'AS{am.group(1)}' if am else ''
cells=re.findall(r'<td\b[^>]*>(.*?)</td>', row, re.I | re.S)
cell_text=[re.sub(r'<[^>]+>', ' ', c).strip() for c in cells]
visibility=''
for c in reversed(cell_text):
lc=re.sub(r'\s+', ' ', c).strip().lower()
if lc in visibility_rank:
visibility=lc
break
rows.append((net, visibility, asn, visibility_rank.get(visibility, 0), len(rows)))
# bgp.tools search may show route objects (e.g. RADB) without /prefix links or
# visibility cells. Include those containing prefixes so /21-/16 announcements
# are not missed when the visibility table is absent/noisy.
for m in re.finditer(r'\b(\d{1,3}(?:\.\d{1,3}){3}/\d{1,2})\b', text):
try:
net=ipaddress.IPv4Network(m.group(1), strict=False)
except Exception:
continue
if ip not in net or net in seen:
continue
seen.add(net)
# Unknown visibility ranks below explicit High/Medium/Low rows, but above blind fallback.
rows.append((net, 'unknown', '', 0, len(rows)))
rows.sort(key=lambda x: (x[3], x[0].prefixlen, -x[4]), reverse=True)
return [(net, visibility, asn) for net, visibility, asn, _rank, _idx in rows]
def fetch(url, timeout=20):
req=Request(url, headers=HEADERS)
with urlopen(req, timeout=timeout) as r:
data = r.read()
enc = (r.headers.get('content-encoding') or '').lower()
if enc == 'gzip':
data = gzip.decompress(data)
elif enc == 'deflate':
try:
data = zlib.decompress(data)
except zlib.error:
data = zlib.decompress(data, -zlib.MAX_WBITS)
return data, r.headers.get('content-type','')
def placeholder(svg: bytes):
txt = svg[:20000].decode('utf-8', 'ignore')
return 'Not_Visible' in txt and 'in_DFZ' in txt
def svg_to_png(svg_path: Path, png_path: Path):
# Prefer cairosvg if present, fallback to rsvg-convert, then ImageMagick.
try:
import cairosvg
cairosvg.svg2png(url=str(svg_path), write_to=str(png_path), output_width=2400)
return
except Exception as e:
last=e
import subprocess, shutil
if shutil.which('rsvg-convert'):
subprocess.check_call(['rsvg-convert','-w','2400','-f','png','-o',str(png_path),str(svg_path)])
return
if shutil.which('magick'):
subprocess.check_call(['magick','-density','300',str(svg_path),'-resize','2400x1800>',str(png_path)])
return
raise RuntimeError(f'no SVG converter available; install cairosvg/sharp/librsvg/imagemagick. last={last}')
def fetch_bgp(ip, domain=None, outdir=OUTDIR):
outdir.mkdir(parents=True, exist_ok=True)
tried=[]; ph=None
for net in prefixes(ip):
pfx=str(net)
urlip=pfx.replace('/','_')
url=f'https://bgp.tools/pathimg/rt-{urlip}?4c1db184-e649-4491-8b7f-06177bcb4f25&loggedin'
tried.append(url)
try:
data, ctype = fetch(url)
except HTTPError as e:
if e.code == 404: continue
continue
except URLError:
continue
if placeholder(data):
ph=pfx; continue
stamp=int(time.time())
base=f'bgp-{str(net).replace("/","_")}-{stamp}'
svg=outdir/(base+'.svg')
png=outdir/(base+'.png')
target_safe=re.sub(r'[^A-Za-z0-9_.-]+', '_', str(domain or ip)).strip('_') or 'target'
latest=outdir/(f'latest-{target_safe}.png')
svg.write_bytes(data)
svg_to_png(svg, png)
latest.write_bytes(png.read_bytes())
try: svg.unlink()
except Exception: pass
print(f'OK\nTARGET={domain or ip}\nIP={ip}\nPREFIX={pfx}\nPNG={png}\nLATEST={latest}\nURL=https://bgp.tools/prefix/{pfx}')
return 0
if ph:
print(f'PLACEHOLDER\nTARGET={domain or ip}\nIP={ip}\nPREFIX={ph}\nURL=https://bgp.tools/prefix/{ph}\nREASON=bgp.tools temporarily returned no path image; please retry once')
return 2
pfx=str(prefixes(ip)[0])
print(f'NONE\nTARGET={domain or ip}\nIP={ip}\nPREFIX={pfx}\nURL=https://bgp.tools/prefix/{pfx}\nREASON=no usable BGP path image found')
return 3
def tld(domain):
parts=domain.split('.')
return '.'.join(parts[-2:]) if len(parts)>=2 else domain
def fetch_dns(ip, domain=None):
for net in prefixes(ip):
pfx=str(net)
url=f'https://bgp.tools/prefix/{pfx}#dns'
try:
html,_=fetch(url)
except Exception:
continue
parser=TextParser(); parser.feed(html.decode('utf-8','ignore'))
text=parser.text()
rows=re.findall(r'(\d{1,3}(?:\.\d{1,3}){3})\s+([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})', text)
counts={}
for _,d in rows: counts[tld(d)]=counts.get(tld(d),0)+1
lines=[f'{a}\t{d}' for a,d in rows if counts.get(tld(d),0)<=2]
if lines:
print('OK_DNS')
print(f'TARGET={domain or ip}\nIP={ip}\nPREFIX={pfx}\nURL={url}')
print('DNS_LINES_BEGIN')
print('\n'.join(lines[:80]))
print('DNS_LINES_END')
return 0
print(f'NONE_DNS\nTARGET={domain or ip}\nIP={ip}\nREASON=no DNS records found')
return 3
def main():
ap=argparse.ArgumentParser()
ap.add_argument('--dns', action='store_true')
ap.add_argument('--outdir', default=str(OUTDIR), help='directory for generated BGP images')
ap.add_argument('ip')
args=ap.parse_args()
ip, domain = resolve_target(args.ip)
return fetch_dns(ip, domain) if args.dns else fetch_bgp(ip, domain, Path(args.outdir))
if __name__ == '__main__':
raise SystemExit(main())

View File

@@ -0,0 +1,118 @@
#!/usr/bin/env node
const fs = require('fs');
const path = require('path');
const { chromium } = require('playwright');
function arg(name, fallback = '') {
const i = process.argv.indexOf(`--${name}`);
if (i >= 0 && process.argv[i + 1]) return process.argv[i + 1];
return fallback;
}
(async () => {
const ip = arg('ip') || process.argv[2];
if (!ip) throw new Error('Usage: download_ippure.js --ip <IPv4> [--outdir <dir>]');
const outdir = arg('outdir', '/data/tmp/ippure-downloads');
fs.mkdirSync(outdir, { recursive: true });
const candidates = [
process.env.CHROMIUM_PATH,
'/usr/bin/chromium',
'/usr/bin/chromium-browser',
'/usr/bin/google-chrome',
'/usr/bin/google-chrome-stable',
].filter(Boolean);
const launchOptions = { headless: true, args: ['--no-sandbox'] };
for (const candidate of candidates) {
if (fs.existsSync(candidate)) {
launchOptions.executablePath = candidate;
break;
}
}
const browser = await chromium.launch(launchOptions);
const context = await browser.newContext({
acceptDownloads: true,
viewport: { width: 1440, height: 1200 },
deviceScaleFactor: 1,
locale: 'zh-CN',
userAgent: 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
});
const page = await context.newPage();
await page.route('**/*', route => {
const url = route.request().url();
if (
url.includes('/cdn-cgi/rum') ||
url.includes('/cdn-cgi/speculation') ||
url.includes('cloudflareinsights.com') ||
url.includes('/api/ads') ||
url.includes('marker-icon.png') ||
url.includes('marker-shadow.png')
) return route.abort().catch(() => {});
return route.continue().catch(() => {});
});
const url = `https://ippure.com/?ip=${encodeURIComponent(ip)}`;
await page.goto(url, { waitUntil: 'domcontentloaded', timeout: 45000 });
await page.waitForSelector('.iptable-container', { state: 'visible', timeout: 45000 });
await page.waitForSelector('button.screenshot-btn svg.lucide-camera', { state: 'visible', timeout: 45000 });
// Faster than waiting for full network idle: wait only for the official card to be populated.
await page.waitForFunction((targetIp) => {
const card = document.querySelector('.iptable-container');
const text = card?.innerText || '';
return text.includes(targetIp) && text.includes('IPPure系数') && !text.includes('Loading...');
}, ip, { timeout: 20000 }).catch(() => {});
// Linux headless Chrome doesn't have PingFang/SF Pro. After Playwright deps are installed,
// fallback font metrics can make the IPPure score badge wrap (e.g. "40%\n中性").
// Keep the official camera export path, but stabilize fonts/nowrap inside the exported card.
await page.addStyleTag({ content: `
.iptable-container, .iptable-container * {
font-family: "Noto Sans CJK SC", "Noto Sans SC", "Microsoft YaHei", "PingFang SC", Arial, sans-serif !important;
}
.iptable-container .font-mono {
font-family: "DejaVu Sans Mono", "Noto Sans Mono CJK SC", monospace !important;
}
.iptable-container .colormap-indicator-value {
white-space: nowrap !important;
min-width: max-content !important;
}
` }).catch(() => {});
// Wait for web fonts/layout to settle so the export captures the stabilized layout.
await page.evaluate(async () => {
await document.fonts?.ready?.catch?.(() => {});
await new Promise(resolve => requestAnimationFrame(() => requestAnimationFrame(resolve)));
}).catch(() => {});
// Hide/mask the queried IP before exporting the official PNG. IPPure exposes this as
// the eye button next to the camera button; click it first so downloaded images do not
// leak the full IP address.
const hideIpButton = page.locator('button.screenshot-btn').filter({ has: page.locator('svg.lucide-eye') }).first();
if (await hideIpButton.count()) {
await hideIpButton.click({ timeout: 10000 });
await page.waitForFunction((targetIp) => {
const card = document.querySelector('.iptable-container');
const text = card?.innerText || '';
return !text.includes(targetIp);
}, ip, { timeout: 5000 }).catch(() => {});
await page.evaluate(async () => {
await new Promise(resolve => requestAnimationFrame(() => requestAnimationFrame(resolve)));
}).catch(() => {});
}
const cameraButton = page.locator('button.screenshot-btn').filter({ has: page.locator('svg.lucide-camera') }).first();
const downloadPromise = page.waitForEvent('download', { timeout: 30000 });
await cameraButton.click({ timeout: 10000 });
const download = await downloadPromise;
const suggested = await download.suggestedFilename();
const filename = suggested && suggested.toLowerCase().endsWith('.png') ? suggested : `IPPure-${ip}-${Date.now()}.png`;
const out = path.join(outdir, filename);
await download.saveAs(out);
await browser.close();
console.log(out);
})().catch(err => {
console.error(err.stack || err.message || String(err));
process.exit(1);
});