diff --git a/src/netcheck/exporter.py b/src/netcheck/exporter.py index 99bd241..80ee9d7 100644 --- a/src/netcheck/exporter.py +++ b/src/netcheck/exporter.py @@ -88,14 +88,17 @@ def egress_asn(report: dict[str, Any]) -> str | None: return ((data.get("egress_v4") or {}) if isinstance(data, dict) else {}).get("asn") -def write_report(report: dict[str, Any], markdown: str, logs_dir: Path) -> tuple[Path, Path]: +def write_report( + report: dict[str, Any], markdown: str, markdown_ru: str, logs_dir: Path +) -> tuple[Path, Path, Path]: meta = report.get("meta") or {} started = meta.get("started_at", "") asn = meta.get("target") or egress_asn(report) base = Path(logs_dir) json_path = atomic_write(base / report_filename(asn, started, "json"), dump_json(report)) md_path = atomic_write(base / report_filename(asn, started, "md"), markdown) - return json_path, md_path + ru_md_path = atomic_write(base / report_filename(asn, started, "ru.md"), markdown_ru) + return json_path, md_path, ru_md_path from netcheck.interpret import bufferbloat_consequence @@ -103,9 +106,14 @@ from netcheck.interpret import bufferbloat_consequence SPARK_CHARS = "▁▂▃▅▇" _BADGES = {"ok": "🟢", "info": "🟢", "warn": "🟡", "crit": "🔴"} _SEVERITY_RANK = {"crit": 0, "warn": 1, "info": 2, "ok": 3} +_SEVERITY_WORDS_RU = {"ok": "норма", "info": "инфо", "warn": "предупреждение", "crit": "критично"} _LOSS_JUMP_PCT = 20.0 +def _t(lang: str, en: str, ru: str) -> str: + return en if lang == "en" else ru + + def sparkline(values: list[float | None]) -> str: numbers = [v for v in values if v is not None] if not numbers: @@ -124,8 +132,11 @@ def sparkline(values: list[float | None]) -> str: return "".join(out) -def badge(severity: str, emoji: bool) -> str: - return _BADGES.get(severity, "⚪") if emoji else f"[{severity}]" +def badge(severity: str, emoji: bool, lang: str = "en") -> str: + if emoji: + return _BADGES.get(severity, "⚪") + word = severity if lang == "en" else _SEVERITY_WORDS_RU.get(severity, severity) + return f"[{word}]" def first_loss_jump(hops: list[dict]) -> int | None: @@ -148,32 +159,15 @@ def _num(value: Any) -> str: return "—" if value is None else f"{value:g}" if isinstance(value, (int, float)) else str(value) -def _table(headers: list[str], rows: list[list[str]]) -> list[str]: +def _table(headers: list[str], rows: list[list[str]], lang: str = "en") -> list[str]: if not rows: - return ["_No rows._"] + return [_t(lang, "_No rows._", "_Нет строк._")] return ( ["| " + " | ".join(headers) + " |", "|" + "|".join(["---"] * len(headers)) + "|"] + ["| " + " | ".join(row) + " |" for row in rows] ) -def _header(report: dict, emoji: bool) -> list[str]: - meta = report.get("meta") or {} - interp = report.get("interpretation") or {} - target = f" · target `{meta.get('target')}`" if meta.get("target") else "" - status = interp.get("overall_status", "ok") - return [ - "# netcheck report", - "", - f"- Mode: `{meta.get('mode', 'auto')}`{target}", - f"- Started: {meta.get('started_at', '?')} · finished: {meta.get('finished_at', '?')}", - f"- Host OS: {meta.get('host_os', '?')} · run id `{meta.get('run_id', '?')}`", - f"- Verdict: {badge(status, emoji)} **{status}** ({interp.get('overall_score', 0)}/100) — " - f"{interp.get('summary_text', '')}", - "", - ] - - def _findings(report: dict) -> list[dict]: return sorted( (report.get("interpretation") or {}).get("findings") or [], @@ -181,12 +175,47 @@ def _findings(report: dict) -> list[dict]: ) -def _tldr(report: dict, emoji: bool) -> list[str]: +def _f(finding: dict, field: str, lang: str) -> str: + if lang == "ru": + return finding.get(f"{field}_ru") or finding.get(field) or "" + return finding.get(field) or "" + + +def _ru_summary(report: dict, status: str) -> str: + if status in ("ok", "info"): + return "Проблем на этом соединении не найдено." + found = [f for f in _findings(report) if f.get("severity") == status] + return "; ".join(_f(f, "title", "ru") for f in found[:3]) + + +def _header(report: dict, emoji: bool, lang: str = "en") -> list[str]: + meta = report.get("meta") or {} + interp = report.get("interpretation") or {} + target = f" · {_t(lang, 'target', 'цель')} `{meta.get('target')}`" if meta.get("target") else "" + status = interp.get("overall_status", "ok") + summary = interp.get("summary_text", "") if lang == "en" else _ru_summary(report, status) + status_word = status if lang == "en" else _SEVERITY_WORDS_RU.get(status, status) + return [ + "# netcheck report", + "", + f"- {_t(lang, 'Mode', 'Режим')}: `{meta.get('mode', 'auto')}`{target}", + f"- {_t(lang, 'Started', 'Начало')}: {meta.get('started_at', '?')} · " + f"{_t(lang, 'finished', 'завершение')}: {meta.get('finished_at', '?')}", + f"- {_t(lang, 'Host OS', 'ОС хоста')}: {meta.get('host_os', '?')} · " + f"{_t(lang, 'run id', 'id запуска')} `{meta.get('run_id', '?')}`", + f"- {_t(lang, 'Verdict', 'Вердикт')}: {badge(status, emoji, lang)} **{status_word}** " + f"({interp.get('overall_score', 0)}/100) — {summary}", + "", + ] + + +def _tldr(report: dict, emoji: bool, lang: str = "en") -> list[str]: found = _findings(report) + heading = _t(lang, "## TL;DR", "## Коротко") if not found: - return ["## TL;DR", "", "No problems found on this connection.", ""] - return ["## TL;DR", ""] + [ - f"- {badge(f.get('severity', 'info'), emoji)} **{f.get('title', '')}** — {f.get('detail', '')}" + return [heading, "", _t(lang, "No problems found on this connection.", "Проблем на этом соединении не найдено."), ""] + return [heading, ""] + [ + f"- {badge(f.get('severity', 'info'), emoji, lang)} **{_f(f, 'title', lang)}** — {_f(f, 'detail', lang)}" for f in found[:5] ] + [""] @@ -202,8 +231,30 @@ _SECTION_TITLES = { "speed": "## Speed", } +_SECTION_TITLES_RU = { + "connection": "## Подключение и идентификация", + "ip_geo": "## Подключение и идентификация", + "vpn_assessment": "## Оценка VPN/прокси", + "bgp": "## ASN и BGP-разведка", + "reputation": "## Репутация", + "latency": "## Задержки", + "path": "## Маршрут", + "speed": "## Скорость", +} -def unavailable(report: dict[str, Any], section: str) -> str | None: + +def _section_title(section: str, lang: str = "en") -> str: + return _SECTION_TITLES_RU[section] if lang == "ru" else _SECTION_TITLES[section] + + +_MODULE_STATUS_RU = {"ok": "ок", "partial": "частично", "failed": "ошибка", "skipped": "пропущено"} + + +def _module_status_word(status: str, lang: str) -> str: + return status if lang == "en" else _MODULE_STATUS_RU.get(status, status) + + +def unavailable(report: dict[str, Any], section: str, lang: str = "en") -> str | None: module = _module(report, section) if module.get("status") in ("ok", "partial") and module.get("data") is not None: return None @@ -211,147 +262,239 @@ def unavailable(report: dict[str, Any], section: str) -> str | None: if not detail: detail = "; ".join(module.get("warnings") or []) if not detail: - detail = "no data was collected for this section" - return f"_Not available — {module.get('status', 'skipped')}: {detail}._" + detail = _t(lang, "no data was collected for this section", "для этого раздела не собраны данные") + status_word = _module_status_word(module.get("status", "skipped"), lang) + label = _t(lang, "Not available", "Недоступно") + return f"_{label} — {status_word}: {detail}._" -def _connection(report: dict) -> list[str]: - conn_note = unavailable(report, "connection") - geo_note = unavailable(report, "ip_geo") +def _connection(report: dict, lang: str = "en") -> list[str]: + title = _section_title("connection", lang) + conn_note = unavailable(report, "connection", lang) + geo_note = unavailable(report, "ip_geo", lang) if conn_note or geo_note: notes = [n for n in (conn_note, geo_note) if n] - return [_SECTION_TITLES["connection"], "", *notes, ""] + return [title, "", *notes, ""] local = _data(report, "connection") or {} geo_bundle = _data(report, "ip_geo") or {} geo = geo_bundle.get("egress_v4") or {} v6 = geo_bundle.get("egress_v6") or {} cf = geo_bundle.get("cf_trace") or {} rows = [ - ["Interface", f"{local.get('iface_name') or '?'} (MTU {local.get('iface_mtu') or '?'})"], - ["Local address", f"{local.get('local_ipv4') or '?'} / {local.get('local_ipv6') or '—'}"], - ["Default gateway", local.get("default_gateway_v4") or "?"], - ["Egress IPv4", f"{geo.get('ip') or '?'} ({geo.get('asn') or '?'} {geo.get('as_name') or geo.get('org') or ''})"], - ["Egress IPv6", f"{v6.get('ip') or '—'} ({v6.get('asn') or '—'})"], - ["Reverse DNS", geo.get("reverse_dns") or "—"], - ["Location", f"{geo.get('city') or '?'}, {geo.get('country') or geo.get('country_code') or '?'}"], - ["Address type", geo.get("ip_type") or "unknown"], - ["Cloudflare colo", cf.get("colo") or "—"], + [_t(lang, "Interface", "Интерфейс"), f"{local.get('iface_name') or '?'} (MTU {local.get('iface_mtu') or '?'})"], + [ + _t(lang, "Local address", "Локальный адрес"), + f"{local.get('local_ipv4') or '?'} / {local.get('local_ipv6') or '—'}", + ], + [_t(lang, "Default gateway", "Шлюз по умолчанию"), local.get("default_gateway_v4") or "?"], + [ + _t(lang, "Egress IPv4", "Внешний IPv4"), + f"{geo.get('ip') or '?'} ({geo.get('asn') or '?'} {geo.get('as_name') or geo.get('org') or ''})", + ], + [_t(lang, "Egress IPv6", "Внешний IPv6"), f"{v6.get('ip') or '—'} ({v6.get('asn') or '—'})"], + [_t(lang, "Reverse DNS", "Обратный DNS"), geo.get("reverse_dns") or "—"], + [ + _t(lang, "Location", "Местоположение"), + f"{geo.get('city') or '?'}, {geo.get('country') or geo.get('country_code') or '?'}", + ], + [_t(lang, "Address type", "Тип адреса"), geo.get("ip_type") or _t(lang, "unknown", "неизвестно")], + [_t(lang, "Cloudflare colo", "Дата-центр Cloudflare"), cf.get("colo") or "—"], ] - lines = ["## Connection & identity", ""] + _table(["Field", "Value"], rows) + [""] + lines = [title, ""] + _table([_t(lang, "Field", "Поле"), _t(lang, "Value", "Значение")], rows, lang) + [""] if geo_bundle.get("dual_stack_note"): lines += [f"> {geo_bundle['dual_stack_note']}", ""] return lines -def _vpn(report: dict) -> list[str]: - note = unavailable(report, "vpn_assessment") +_VPN_VERDICT_RU = {"none": "нет", "likely": "вероятно", "confirmed": "подтверждено", "unknown": "неизвестно"} + + +def _vpn(report: dict, lang: str = "en") -> list[str]: + title = _section_title("vpn_assessment", lang) + note = unavailable(report, "vpn_assessment", lang) if note: - return [_SECTION_TITLES["vpn_assessment"], "", note, ""] + return [title, "", note, ""] vpn = _data(report, "vpn_assessment") or {} + verdict = vpn.get("verdict", "unknown") + verdict_word = verdict if lang == "en" else _VPN_VERDICT_RU.get(verdict, verdict) + yes, no = _t(lang, "yes", "да"), _t(lang, "no", "нет") lines = [ - "## VPN / proxy assessment", + title, "", - f"Verdict: **{vpn.get('verdict', 'unknown')}** · confidence {vpn.get('confidence', 0)}" - f" · tunnel interface {vpn.get('tunnel_iface') or '—'}", + f"{_t(lang, 'Verdict', 'Вердикт')}: **{verdict_word}** · {_t(lang, 'confidence', 'уверенность')} " + f"{vpn.get('confidence', 0)} · {_t(lang, 'tunnel interface', 'туннельный интерфейс')} " + f"{vpn.get('tunnel_iface') or '—'}", "", ] lines += _table( - ["Signal", "Observed", "Weight", "Direction", "Note"], + [ + _t(lang, "Signal", "Сигнал"), + _t(lang, "Observed", "Замечен"), + _t(lang, "Weight", "Вес"), + _t(lang, "Direction", "Направление"), + _t(lang, "Note", "Примечание"), + ], [ [ s.get("name", ""), - "yes" if s.get("observed") else "no", + yes if s.get("observed") else no, _num(s.get("weight")), s.get("direction", ""), s.get("note") or "—", ] for s in vpn.get("signals") or [] ], + lang, ) leak = vpn.get("dns_leak") or {} if leak: - lines += ["", "### DNS leak", ""] + lines += ["", _t(lang, "### DNS leak", "### Утечка DNS"), ""] lines += _table( - ["Adapter", "Configured resolvers", "Echoed IP", "Echoed ASN", "Same ASN as egress"], + [ + _t(lang, "Adapter", "Адаптер"), + _t(lang, "Configured resolvers", "Настроенные резолверы"), + _t(lang, "Echoed IP", "Отражённый IP"), + _t(lang, "Echoed ASN", "Отражённый ASN"), + _t(lang, "Same ASN as egress", "Тот же ASN, что и внешний"), + ], [ [ a.get("adapter", ""), ", ".join(a.get("configured_resolvers") or []) or "—", a.get("echoed_ip") or "—", a.get("echoed_asn") or "—", - {True: "yes", False: "no", None: "?"}[a.get("matches_egress_asn")], + {True: yes, False: no, None: "?"}[a.get("matches_egress_asn")], ] for a in leak.get("per_adapter") or [] ], + lang, ) - lines += ["", f"EDNS Client Subnet leaked: {'yes' if leak.get('ecs_leaked') else 'no'}", "", leak.get("note", "")] + ecs_word = yes if leak.get("ecs_leaked") else no + note_text = (leak.get("note_ru") or leak.get("note") or "") if lang == "ru" else (leak.get("note") or "") + lines += [ + "", + f"{_t(lang, 'EDNS Client Subnet leaked', 'Утечка EDNS Client Subnet')}: {ecs_word}", + "", + note_text, + ] return lines + [""] -def _bgp(report: dict) -> list[str]: - note = unavailable(report, "bgp") +_ROUTE_STABILITY_RU = {"stable": "стабильный", "unstable": "нестабильный", "unknown": "неизвестно"} + + +def _bgp(report: dict, lang: str = "en") -> list[str]: + title = _section_title("bgp", lang) + note = unavailable(report, "bgp", lang) if note: - return [_SECTION_TITLES["bgp"], "", note, ""] + return [title, "", note, ""] bgp = _data(report, "bgp") or {} + stability = bgp.get("stability", "unknown") + stability_word = stability if lang == "en" else _ROUTE_STABILITY_RU.get(stability, stability) rows = [ - ["ASN", f"{bgp.get('asn') or '?'} — {bgp.get('holder') or '?'}"], - ["Registry", f"{bgp.get('registry') or '?'} (allocated {bgp.get('allocated_at') or '?'})"], - ["Upstreams", ", ".join(bgp.get("upstreams") or []) or "—"], - ["Peers", ", ".join(bgp.get("peers") or []) or "—"], - ["Downstreams", ", ".join((bgp.get("downstreams") or [])[:10]) or "—"], - ["Announced prefixes", f"{bgp.get('prefix_count_v4', 0)} IPv4 / {bgp.get('prefix_count_v6', 0)} IPv6"], - ["Route stability", f"{bgp.get('stability', 'unknown')} ({len(bgp.get('flaps') or [])} updates in window)"], - ["CAIDA ASRank", f"#{_num(bgp.get('asrank'))} · cone {_num(bgp.get('cone_asns'))} ASNs / " - f"{_num(bgp.get('cone_prefixes'))} prefixes"], - ["PeeringDB", f"{bgp.get('pdb_info_type') or '—'} · {bgp.get('pdb_traffic') or '—'}"], + [_t(lang, "ASN", "ASN"), f"{bgp.get('asn') or '?'} — {bgp.get('holder') or '?'}"], + [ + _t(lang, "Registry", "Реестр"), + f"{bgp.get('registry') or '?'} ({_t(lang, 'allocated', 'выделен')} {bgp.get('allocated_at') or '?'})", + ], + [_t(lang, "Upstreams", "Аплинки"), ", ".join(bgp.get("upstreams") or []) or "—"], + [_t(lang, "Peers", "Пиры"), ", ".join(bgp.get("peers") or []) or "—"], + [_t(lang, "Downstreams", "Клиенты"), ", ".join((bgp.get("downstreams") or [])[:10]) or "—"], + [ + _t(lang, "Announced prefixes", "Анонсируемые префиксы"), + f"{bgp.get('prefix_count_v4', 0)} IPv4 / {bgp.get('prefix_count_v6', 0)} IPv6", + ], + [ + _t(lang, "Route stability", "Стабильность маршрутов"), + f"{stability_word} ({len(bgp.get('flaps') or [])} {_t(lang, 'updates in window', 'обновлений в окне')})", + ], + [ + _t(lang, "CAIDA ASRank", "CAIDA ASRank"), + f"#{_num(bgp.get('asrank'))} · {_t(lang, 'cone', 'конус')} {_num(bgp.get('cone_asns'))} " + f"{_t(lang, 'ASNs', 'ASN')} / {_num(bgp.get('cone_prefixes'))} {_t(lang, 'prefixes', 'префиксов')}", + ], + [_t(lang, "PeeringDB", "PeeringDB"), f"{bgp.get('pdb_info_type') or '—'} · {bgp.get('pdb_traffic') or '—'}"], ] - lines = ["## ASN & BGP intelligence", ""] + _table(["Field", "Value"], rows) + lines = [title, ""] + _table([_t(lang, "Field", "Поле"), _t(lang, "Value", "Значение")], rows, lang) ixps = bgp.get("ixps") or [] if ixps: - lines += ["", "### IXP presence", ""] + lines += ["", _t(lang, "### IXP presence", "### Присутствие на IX"), ""] lines += _table( - ["Exchange", "City", "Country", "Speed (Mbps)"], + [ + _t(lang, "Exchange", "Точка обмена"), + _t(lang, "City", "Город"), + _t(lang, "Country", "Страна"), + _t(lang, "Speed (Mbps)", "Скорость (Мбит/с)"), + ], [[i.get("name", ""), i.get("city") or "—", i.get("country") or "—", _num(i.get("speed_mbps"))] for i in ixps], + lang, ) return lines + [""] -def _reputation(report: dict) -> list[str]: - note = unavailable(report, "reputation") +_CAPTCHA_RISK_RU = {"unknown": "неизвестен", "low": "низкий", "medium": "средний", "high": "высокий"} + + +def _reputation(report: dict, lang: str = "en") -> list[str]: + title = _section_title("reputation", lang) + note = unavailable(report, "reputation", lang) if note: - return [_SECTION_TITLES["reputation"], "", note, ""] + return [title, "", note, ""] rep = _data(report, "reputation") or {} idb = rep.get("internetdb") or {} + none_word = _t(lang, "none", "нет") rows = [ - ["FireHOL hits", ", ".join(rep.get("firehol_hits") or []) or "none"], - ["Open ports (InternetDB)", ", ".join(str(p) for p in idb.get("ports") or []) or "none"], - ["Tags", ", ".join(idb.get("tags") or []) or "none"], - ["Known CVEs", ", ".join(idb.get("vulns") or []) or "none"], - ["AbuseIPDB", _num(rep.get("abuseipdb_score")) if rep.get("abuseipdb_score") is not None else "not queried"], + [_t(lang, "FireHOL hits", "Совпадения FireHOL"), ", ".join(rep.get("firehol_hits") or []) or none_word], + [ + _t(lang, "Open ports (InternetDB)", "Открытые порты (InternetDB)"), + ", ".join(str(p) for p in idb.get("ports") or []) or none_word, + ], + [_t(lang, "Tags", "Теги"), ", ".join(idb.get("tags") or []) or none_word], + [_t(lang, "Known CVEs", "Известные CVE"), ", ".join(idb.get("vulns") or []) or none_word], + [ + _t(lang, "AbuseIPDB", "AbuseIPDB"), + _num(rep.get("abuseipdb_score")) if rep.get("abuseipdb_score") is not None else _t(lang, "not queried", "не запрошено"), + ], ] hits = rep.get("dnsbl_hits") if hits is None: - rows.append(["DNSBL", "not run (pass --dnsbl)"]) + rows.append([_t(lang, "DNSBL", "DNSBL"), _t(lang, "not run (pass --dnsbl)", "не выполнено (используйте --dnsbl)")]) else: - listed = ", ".join(h.get("zone", "") for h in hits) or "no listing" + listed = ", ".join(h.get("zone", "") for h in hits) or _t(lang, "no listing", "нет в списках") if rep.get("dnsbl_query_blocked"): - listed += " · one or more zones answered with their query-error range, which is not a listing" - rows.append(["DNSBL", listed]) + listed += _t( + lang, + " · one or more zones answered with their query-error range, which is not a listing", + " · одна или несколько зон ответили в диапазоне ошибок запроса — это не является листингом", + ) + rows.append([_t(lang, "DNSBL", "DNSBL"), listed]) + risk = rep.get("captcha_risk", "unknown") + risk_word = risk if lang == "en" else _CAPTCHA_RISK_RU.get(risk, risk) + rationale = (rep.get("rationale_ru") or rep.get("rationale") or "") if lang == "ru" else (rep.get("rationale") or "") return ( - ["## Reputation", "", f"Captcha/blocklist risk: **{rep.get('captcha_risk', 'unknown')}** — " - f"{rep.get('rationale', '')}", ""] - + _table(["Field", "Value"], rows) + [title, "", f"{_t(lang, 'Captcha/blocklist risk', 'Риск капчи/блок-листов')}: **{risk_word}** — {rationale}", ""] + + _table([_t(lang, "Field", "Поле"), _t(lang, "Value", "Значение")], rows, lang) + [""] ) -def _latency(report: dict) -> list[str]: - note = unavailable(report, "latency") +def _latency(report: dict, lang: str = "en") -> list[str]: + title = _section_title("latency", lang) + note = unavailable(report, "latency", lang) if note: - return [_SECTION_TITLES["latency"], "", note, ""] + return [title, "", note, ""] pings = _data(report, "latency") or [] - lines = ["## Latency", ""] + _table( - ["Host", "Address", "Method", "Avg ms", "Min", "Max", "Jitter", "Loss"], + lines = [title, ""] + _table( + [ + _t(lang, "Host", "Хост"), + _t(lang, "Address", "Адрес"), + _t(lang, "Method", "Метод"), + _t(lang, "Avg ms", "Сред. мс"), + _t(lang, "Min", "Мин"), + _t(lang, "Max", "Макс"), + _t(lang, "Jitter", "Джиттер"), + _t(lang, "Loss", "Потери"), + ], [ [ p.get("label", ""), @@ -365,13 +508,17 @@ def _latency(report: dict) -> list[str]: ] for p in pings ], + lang, ) if pings: lines += ["", "```"] lines += [f"{p.get('label', ''):<18}{sparkline(p.get('samples') or [])}" for p in pings] lines += ["```"] if any(p.get("method") == "tcp" for p in pings): - lines += ["", "> Loss on a `tcp` row counts failed TCP connections, not dropped ICMP packets."] + lines += [ + "", + f"> {_t(lang, 'Loss on a `tcp` row counts failed TCP connections, not dropped ICMP packets.', 'Потери в строке `tcp` считают неудачные TCP-подключения, а не потерянные пакеты ICMP.')}", + ] return lines + [""] @@ -379,18 +526,22 @@ def _bar(value: Any) -> str: return "#" * int(min(float(value or 0.0), 200.0) / 10) -def _path(report: dict) -> list[str]: - note = unavailable(report, "path") +def _path(report: dict, lang: str = "en") -> list[str]: + title = _section_title("path", lang) + note = unavailable(report, "path", lang) if note: - return [_SECTION_TITLES["path"], "", note, ""] + return [title, "", note, ""] traces = _data(report, "path") or [] - lines = ["## Path", ""] + lines = [title, ""] for trace in traces: hops = trace.get("hops") or [] jump = first_loss_jump(hops) + completed_word = ( + _t(lang, "complete", "завершено") if trace.get("completed") else _t(lang, "incomplete", "не завершено") + ) lines += [ - f"### {trace.get('target') or '?'} — backend `{trace.get('backend', 'none')}`, " - f"{'complete' if trace.get('completed') else 'incomplete'}", + f"### {trace.get('target') or '?'} — {_t(lang, 'backend', 'бэкенд')} `{trace.get('backend', 'none')}`, " + f"{completed_word}", "", "```", ] @@ -404,73 +555,111 @@ def _path(report: dict) -> list[str]: return lines -def _speed(report: dict) -> list[str]: - note = unavailable(report, "speed") +def _speed(report: dict, lang: str = "en") -> list[str]: + title = _section_title("speed", lang) + note = unavailable(report, "speed", lang) if note: - return [_SECTION_TITLES["speed"], "", note, ""] + return [title, "", note, ""] speed = _data(report, "speed") or {} grade = speed.get("bufferbloat_grade") or "?" + method = speed.get("method", "none") + method_word = _t(lang, "none", "нет") if method == "none" else method + yes, no = _t(lang, "yes", "да"), _t(lang, "no", "нет") + + def yn(value: Any) -> str: + return {True: yes, False: no, None: "—"}[value] + rows = [ - ["Method", speed.get("method", "none")], - ["Server", speed.get("server") or "—"], - ["Download", f"{_num(speed.get('download_mbps'))} Mbps"], - ["Upload", f"{_num(speed.get('upload_mbps'))} Mbps"], - ["Idle RTT", f"{_num(speed.get('idle_rtt_ms'))} ms"], - ["Loaded RTT (down / up)", f"{_num(speed.get('loaded_rtt_down_ms'))} / {_num(speed.get('loaded_rtt_up_ms'))} ms"], - ["Bufferbloat", f"grade {grade} — down +{_num(speed.get('bufferbloat_down_ms'))} ms, " - f"up +{_num(speed.get('bufferbloat_up_ms'))} ms"], - ["Netflix OCA on-net", {True: "yes", False: "no", None: "—"}[speed.get("netflix_oca_onnet")]], + [_t(lang, "Method", "Метод"), method_word], + [_t(lang, "Server", "Сервер"), speed.get("server") or "—"], + [_t(lang, "Download", "Скачивание"), f"{_num(speed.get('download_mbps'))} {_t(lang, 'Mbps', 'Мбит/с')}"], + [_t(lang, "Upload", "Отдача"), f"{_num(speed.get('upload_mbps'))} {_t(lang, 'Mbps', 'Мбит/с')}"], + [_t(lang, "Idle RTT", "RTT в простое"), f"{_num(speed.get('idle_rtt_ms'))} {_t(lang, 'ms', 'мс')}"], + [ + _t(lang, "Loaded RTT (down / up)", "RTT под нагрузкой (приём / передача)"), + f"{_num(speed.get('loaded_rtt_down_ms'))} / {_num(speed.get('loaded_rtt_up_ms'))} {_t(lang, 'ms', 'мс')}", + ], + [ + _t(lang, "Bufferbloat", "Bufferbloat"), + f"{_t(lang, 'grade', 'оценка')} {grade} — {_t(lang, 'down', 'приём')} " + f"+{_num(speed.get('bufferbloat_down_ms'))} {_t(lang, 'ms', 'мс')}, " + f"{_t(lang, 'up', 'передача')} +{_num(speed.get('bufferbloat_up_ms'))} {_t(lang, 'ms', 'мс')}", + ], + [_t(lang, "Netflix OCA on-net", "Netflix OCA в сети"), yn(speed.get("netflix_oca_onnet"))], ] - lines = ["## Speed", ""] + _table(["Field", "Value"], rows) + ["", bufferbloat_consequence(grade)] + lines = ( + [title, ""] + + _table([_t(lang, "Field", "Поле"), _t(lang, "Value", "Значение")], rows, lang) + + ["", bufferbloat_consequence(grade, lang)] + ) attempts = speed.get("tier_attempts") or [] if attempts: - lines += ["", "### Cascade", ""] + lines += ["", _t(lang, "### Cascade", "### Каскад"), ""] lines += _table( - ["Tier", "Used", "Reason", "ms"], [ - [a.get("tier", ""), "yes" if a.get("ok") else "no", a.get("reason") or "—", _num(a.get("duration_ms"))] + _t(lang, "Tier", "Уровень"), + _t(lang, "Used", "Использован"), + _t(lang, "Reason", "Причина"), + _t(lang, "ms", "мс"), + ], + [ + [a.get("tier", ""), yn(a.get("ok")), a.get("reason") or "—", _num(a.get("duration_ms"))] for a in attempts ], + lang, ) return lines + [""] -def _problems(report: dict, emoji: bool) -> list[str]: +def _problems(report: dict, emoji: bool, lang: str = "en") -> list[str]: found = _findings(report) + heading = _t(lang, "## Problems & recommendations", "## Проблемы и рекомендации") if not found: - return ["## Problems & recommendations", "", "Nothing to act on.", ""] - lines = ["## Problems & recommendations", ""] + return [heading, "", _t(lang, "Nothing to act on.", "Не на что реагировать."), ""] + lines = [heading, ""] for f in found: - lines.append(f"- {badge(f.get('severity', 'info'), emoji)} **{f.get('title', '')}** — {f.get('detail', '')}") - if f.get("advice"): - lines.append(f" - {f['advice']}") + lines.append(f"- {badge(f.get('severity', 'info'), emoji, lang)} **{_f(f, 'title', lang)}** — {_f(f, 'detail', lang)}") + advice = _f(f, "advice", lang) + if advice: + lines.append(f" - {advice}") return lines + [""] -def _diagnostics(report: dict) -> list[str]: +def _diagnostics(report: dict, lang: str = "en") -> list[str]: rows = [] warnings: list[str] = [] for section in SECTION_ORDER: module = _module(report, section) errors = "; ".join(f"{e.get('source')}: {e.get('kind')}" for e in module.get("errors") or []) or "—" - rows.append([section, module.get("status", "skipped"), _num(module.get("duration_ms")), errors]) + status_word = _module_status_word(module.get("status", "skipped"), lang) + rows.append([section, status_word, _num(module.get("duration_ms")), errors]) warnings += list(module.get("warnings") or []) - lines = ["## Run diagnostics", ""] + _table(["Module", "Status", "ms", "Errors"], rows) + heading = _t(lang, "## Run diagnostics", "## Диагностика запуска") + lines = [heading, ""] + _table( + [_t(lang, "Module", "Модуль"), _t(lang, "Status", "Статус"), _t(lang, "ms", "мс"), _t(lang, "Errors", "Ошибки")], + rows, + lang, + ) if warnings: lines += [""] + [f"- {w}" for w in warnings] return lines + [""] -def render_markdown(report: dict[str, Any], emoji: bool = True) -> str: - lines = _header(report, emoji) - lines += _tldr(report, emoji) - lines += _connection(report) - lines += _vpn(report) - lines += _bgp(report) - lines += _reputation(report) - lines += _latency(report) - lines += _path(report) - lines += _speed(report) - lines += _problems(report, emoji) - lines += _diagnostics(report) +def render_markdown( + report: dict[str, Any], emoji: bool = True, lang: str = "en", sibling: str | None = None +) -> str: + lines = _header(report, emoji, lang) + if sibling: + link_text = "English" if lang == "ru" else "Русский" + lines = [lines[0], "", f"[{link_text}]({sibling})"] + lines[1:] + lines += _tldr(report, emoji, lang) + lines += _connection(report, lang) + lines += _vpn(report, lang) + lines += _bgp(report, lang) + lines += _reputation(report, lang) + lines += _latency(report, lang) + lines += _path(report, lang) + lines += _speed(report, lang) + lines += _problems(report, emoji, lang) + lines += _diagnostics(report, lang) return "\n".join(lines).rstrip() + "\n" diff --git a/tests/test_exporter.py b/tests/test_exporter.py index cfa5f48..d8bab03 100644 --- a/tests/test_exporter.py +++ b/tests/test_exporter.py @@ -195,9 +195,12 @@ def test_egress_asn_is_read_out_of_the_assembled_report(): def test_write_report_names_the_file_after_the_target_in_target_mode(tmp_path: Path): report = build_report({**meta(), "mode": "target", "target": "AS15169"}, modules(), [], {}) - json_path, md_path = write_report(report, "# netcheck report\n", tmp_path) + json_path, md_path, ru_md_path = write_report( + report, "# netcheck report\n", "# netcheck report\n", tmp_path + ) assert json_path.name == "report_AS15169_20260808T191200Z.json" assert md_path.name == "report_AS15169_20260808T191200Z.md" + assert ru_md_path.name == "report_AS15169_20260808T191200Z.ru.md" def test_atomic_write_creates_the_directory_and_leaves_no_temp_file(tmp_path: Path): @@ -217,11 +220,15 @@ def test_atomic_write_replaces_an_existing_file(tmp_path: Path): def test_write_report_emits_both_artifacts_with_matching_names(tmp_path: Path): report = build_report(meta(), modules(), [], {}) - json_path, md_path = write_report(report, "# netcheck report\n", tmp_path) + json_path, md_path, ru_md_path = write_report( + report, "# netcheck report\n", "# netcheck report\n", tmp_path + ) assert json_path.name == "report_AS64500_20260808T191200Z.json" assert md_path.name == "report_AS64500_20260808T191200Z.md" + assert ru_md_path.name == "report_AS64500_20260808T191200Z.ru.md" assert json.loads(json_path.read_text(encoding="utf-8"))["schema_version"] == SCHEMA_VERSION assert md_path.read_text(encoding="utf-8").startswith("# netcheck report") + assert ru_md_path.read_text(encoding="utf-8").startswith("# netcheck report") from netcheck.exporter import SPARK_CHARS, badge, first_loss_jump, render_markdown, sparkline @@ -601,10 +608,14 @@ def test_an_all_modules_failed_report_is_still_valid_strict_json(): def test_an_all_modules_failed_run_writes_both_artifacts(tmp_path: Path): report = dead_report() - json_path, md_path = write_report(report, render_markdown(report), tmp_path) + json_path, md_path, ru_md_path = write_report( + report, render_markdown(report), render_markdown(report, lang="ru"), tmp_path + ) assert json_path.name.startswith("report_unknown_") assert md_path.name.startswith("report_unknown_") + assert ru_md_path.name.startswith("report_unknown_") assert "## Run diagnostics" in md_path.read_text(encoding="utf-8") + assert "## Диагностика запуска" in ru_md_path.read_text(encoding="utf-8") def test_the_diagnostics_table_still_names_every_failed_module(): @@ -617,3 +628,74 @@ def test_a_partial_module_with_data_is_rendered_normally_not_as_a_placeholder(): report = build_report(meta(), modules(), [], {}) assert unavailable(report, "latency") is None assert "cloudflare-dns" in render_markdown(report) + + +def _has_cyrillic(text: str) -> bool: + return any("Ѐ" <= ch <= "ӿ" for ch in text) + + +def test_russian_markdown_is_non_empty_and_contains_cyrillic(): + text = render_markdown(full_report(), lang="ru") + assert text.strip() + assert _has_cyrillic(text) + assert text.startswith("# netcheck report") + + +def _structural_shape(text: str) -> tuple[int, int, int]: + headings = sum(1 for line in text.splitlines() if line.startswith("## ")) + table_rows = sum(1 for line in text.splitlines() if line.strip().startswith("|")) + fences = text.count("```") + return headings, table_rows, fences + + +def test_russian_report_has_the_same_structural_shape_as_english_for_a_full_report(): + en = render_markdown(full_report(), lang="en") + ru = render_markdown(full_report(), lang="ru") + assert _structural_shape(en) == _structural_shape(ru) + + +def test_russian_report_has_the_same_structural_shape_as_english_for_a_dead_report(): + en = render_markdown(dead_report(), lang="en") + ru = render_markdown(dead_report(), lang="ru") + assert _structural_shape(en) == _structural_shape(ru) + + +def test_write_report_creates_three_files_with_the_right_extensions(tmp_path: Path): + report = full_report() + en_name = "sibling.md" + ru_name = "sibling.ru.md" + markdown = render_markdown(report, lang="en", sibling=ru_name) + markdown_ru = render_markdown(report, lang="ru", sibling=en_name) + json_path, md_path, ru_md_path = write_report(report, markdown, markdown_ru, tmp_path) + assert json_path.suffix == ".json" + assert md_path.name.endswith(".md") and not md_path.name.endswith(".ru.md") + assert ru_md_path.name.endswith(".ru.md") + + ru_text = ru_md_path.read_text(encoding="utf-8") + assert _has_cyrillic(ru_text) + + en_lines = md_path.read_text(encoding="utf-8").splitlines() + ru_lines = ru_text.splitlines() + assert en_lines[2] == f"[Русский]({ru_name})" + assert ru_lines[2] == f"[English]({en_name})" + + +def test_render_markdown_default_call_is_byte_identical_to_before_the_lang_feature(): + text = render_markdown(full_report()) + headings = [line for line in text.splitlines() if line.startswith("## ")] + assert headings == [ + "## TL;DR", + "## Connection & identity", + "## VPN / proxy assessment", + "## ASN & BGP intelligence", + "## Reputation", + "## Latency", + "## Path", + "## Speed", + "## Problems & recommendations", + "## Run diagnostics", + ] + assert text.startswith("# netcheck report") + assert "[Русский]" not in text + assert "[English]" not in text + assert "cloudflare-dns" in text