"""Normalize public links and render structured parser results.""" import html import re from collections import OrderedDict from urllib.parse import parse_qs, urlsplit def normalize(raw): try: url = urlsplit(raw.strip()) if url.scheme not in ("https", "http") or url.username or url.password or url.port: raise ValueError("Нужна обычная ссылка http/https") host = url.hostname.lower().removeprefix("www.").removeprefix("m.") path = url.path.rstrip("/") if host in ("t.me", "telegram.me"): if path.startswith("/c/"): raise ValueError("Закрытая ссылка Telegram: нужен публичный адрес поста") match = re.fullmatch(r"/(?:s/)?([A-Za-z][A-Za-z0-9_]{2,})/([1-9]\d*)", path) if match: slug, post = match.groups() slug = slug.lower() return {"url": f"https://t.me/{slug}/{post}", "platform": "telegram", "group": slug, "a": slug, "b": post} elif host in ("vk.com", "vk.ru"): wall = path if "w" in parse_qs(url.query): wall = "/" + parse_qs(url.query)["w"][0] match = re.fullmatch(r"/wall(-?[1-9]\d*)_([1-9]\d*)", wall) if match and "reply" not in parse_qs(url.query): owner, post = match.groups() return {"url": f"https://vk.com/wall{owner}_{post}", "platform": "vk", "group": owner, "a": int(owner), "b": post} elif host == "max.ru": match = re.fullmatch(r"/([A-Za-z0-9_]+)/([A-Za-z0-9_-]+)", path) if match: slug, post = match.groups() slug = slug.lower() return {"url": f"https://max.ru/{slug}/{post}", "platform": "max", "group": slug.lower(), "a": slug, "b": post} raise ValueError("Площадка или формат ссылки пока не поддерживаются") except (AttributeError, TypeError): raise ValueError("Некорректная ссылка") from None def human_views(value): divisor, suffix = (1_000_000, "M") if value >= 1_000_000 else (1000, "K") return f"{value / divisor:.1f}".rstrip("0").rstrip(".").replace(".", ",") + suffix def clean_title(value): # Strip legacy HTML emoji representations as well as unicode emoji clusters. value = re.sub(r"<[^>]+>", "", html.unescape(str(value))) value = re.sub(r"[\U0001F000-\U0001FAFF\u2600-\u27BF\uFE0E\uFE0F\u200D]", "", value) value = re.sub(r"[#*0-9]\u20e3", "", value) return re.sub(r"\s+", " ", value).strip() def render(records): chunks, plain, errors = [], [], [] for platform, label in (("telegram", "Telegram"), ("vk", "ВКонтакте"), ("max", "Макс")): selected = [r for r in records if r["platform"] == platform] if not selected: continue good = [r for r in selected if r.get("views") is not None and not r.get("error")] groups = OrderedDict() for record in good: groups.setdefault(record["group"], []).append(record) total = sum(r["views"] for r in good) chunks.extend([f"

{label}

", f"

Суммарно посты собрали {human_views(total)} просмотров.

"]) plain.extend([label, f"Суммарно посты собрали {human_views(total)} просмотров."]) if len(good) != len(selected): note = f"Неполный итог: получены просмотры {len(good)} из {len(selected)} постов." chunks.append(f"

{note}

") plain.append(note) chunks.append("
    ") ordered = sorted(groups.values(), key=lambda items: sum(r["views"] for r in items), reverse=True) for index, items in enumerate(ordered, 1): title = clean_title(items[0]["title"]) or items[0]["group"] links = [f'{html.escape(title if i == 0 else "ещё")}' for i, r in enumerate(items)] views = " + ".join(human_views(r["views"]) for r in items) chunks.append(f'
  1. {" + ".join(links)} — {views}
  2. ') plain.append(f'{index}. {title}{" + ещё" * (len(items) - 1)} — {views}') chunks.append("
") plain.append("") for record in records: if record.get("error"): errors.append(f'{record["url"]}: {record["error"]}') if errors: chunks.append("

Не удалось обработать

") plain.extend(["Не удалось обработать", *errors]) return {"html": "\n".join(chunks), "text": "\n".join(plain), "errors": errors, "records": records}