Spaces:
Sleeping
Sleeping
| """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"<h2>{label}</h2>", f"<p>Суммарно посты собрали <b>{human_views(total)}</b> просмотров.</p>"]) | |
| plain.extend([label, f"Суммарно посты собрали {human_views(total)} просмотров."]) | |
| if len(good) != len(selected): | |
| note = f"Неполный итог: получены просмотры {len(good)} из {len(selected)} постов." | |
| chunks.append(f"<p><i>{note}</i></p>") | |
| plain.append(note) | |
| chunks.append("<ol>") | |
| 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'<a href="{html.escape(r.get("display_url") or r["url"], quote=True)}">{html.escape(title if i == 0 else "ещё")}</a>' for i, r in enumerate(items)] | |
| views = " + ".join(human_views(r["views"]) for r in items) | |
| chunks.append(f'<li>{" + ".join(links)} — {views}</li>') | |
| plain.append(f'{index}. {title}{" + ещё" * (len(items) - 1)} — {views}') | |
| chunks.append("</ol>") | |
| plain.append("") | |
| for record in records: | |
| if record.get("error"): | |
| errors.append(f'{record["url"]}: {record["error"]}') | |
| if errors: | |
| chunks.append("<h2>Не удалось обработать</h2><ul>") | |
| chunks.extend(f"<li>{html.escape(error)}</li>" for error in errors) | |
| chunks.append("</ul>") | |
| plain.extend(["Не удалось обработать", *errors]) | |
| return {"html": "\n".join(chunks), "text": "\n".join(plain), "errors": errors, "records": records} | |