Add optional TGStat archive fallback and fractional K counters for ReportBridge

reports.py

@@ -39,8 +39,6 @@ def normalize(raw):
 
 
 def human_views(value):
-    if value < 1000:
-        return str(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
 
@@ -74,7 +72,7 @@ def render(records):
         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["url"], quote=True)}">{html.escape(title if i == 0 else "ещё")}</a>' for i, r in enumerate(items)]
+            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}')

service.py

@@ -14,6 +14,7 @@ from pydantic import BaseModel, Field
 
 import legacy_parser as engine
 from reports import normalize, render
+from tgstat_api import archive_link, archived_post
 
 log = logging.getLogger(__name__)
 API_KEY = os.getenv("BRIDGE_API_KEY", "")
@@ -51,6 +52,15 @@ def fetch_one(record):
         result.update(title=title, views=views)
     except Exception:
         result.update(title=record["group"], views=None, error="Не удалось получить просмотры публичного поста")
+        if record["platform"] == "telegram":
+            result["display_url"] = archive_link(record["a"], record["b"])
+            try:
+                fallback = archived_post(record["a"], record["b"], record["group"])
+                if fallback:
+                    result.update(fallback, error=None)
+            except Exception:
+                # Never log API request URLs: TGStat credentials are query parameters.
+                pass
     return result
 
 
@@ -102,7 +112,7 @@ def root():
 @app.get("/health")
 def health():
     return {"status": "ok", "service": "ReportBridge", "configured": {
-        "bridge": bool(API_KEY), "vk": bool(engine.VK_ACCESS_TOKEN), "max": bool(os.getenv("MAX_ACCESS_TOKEN"))}}
+        "bridge": bool(API_KEY), "vk": bool(engine.VK_ACCESS_TOKEN), "max": bool(os.getenv("MAX_ACCESS_TOKEN")), "tgstat": bool(os.getenv("TGSTAT_API_TOKEN"))}}
 
 
 @app.post("/jobs", dependencies=[Depends(authorize)])

tests/test_reports.py

@@ -11,7 +11,7 @@ class ReportTests(unittest.TestCase):
                 normalize(link)
 
     def test_format_without_zero(self):
-        for value, expected in ((2000, '2K'), (10600, '10,6K'), (1000000, '1M'), (35, '35'), (0, '0')):
+        for value, expected in ((2000, '2K'), (10600, '10,6K'), (1000000, '1M'), (845, '0,8K'), (100, '0,1K'), (35, '0K'), (0, '0K')):
             self.assertEqual(human_views(value), expected)
 
     def test_errors_are_not_zero_views(self):

tests/test_tgstat_api.py

@@ -0,0 +1,54 @@
+import unittest
+from unittest.mock import Mock, patch
+
+from reports import normalize, render
+from tgstat_api import archive_link, archived_post
+import service
+
+
+class TGStatTests(unittest.TestCase):
+    def test_archive_link_and_validation(self):
+        self.assertEqual(archive_link('memtc', '1553'), 'https://tgstat.ru/channel/@memtc/1553')
+        with self.assertRaises(ValueError):
+            archive_link('../secret', '1')
+
+    def test_missing_token_does_not_request_anything(self):
+        get = Mock()
+        with patch.dict('os.environ', {'TGSTAT_API_TOKEN': ''}):
+            self.assertIsNone(archived_post('memtc', '1553', 'Канал', get))
+        get.assert_not_called()
+
+    def test_exact_counter_and_post_identity(self):
+        response = Mock()
+        response.json.return_value = {'status': 'ok', 'response': {'views': 845, 'link': 't.me/memtc/1553', 'is_deleted': 1}}
+        get = Mock(return_value=response)
+        with patch.dict('os.environ', {'TGSTAT_API_TOKEN': 'test-only'}):
+            post = archived_post('memtc', '1553', 'Канал', get)
+            self.assertEqual(post['views'], 845)
+            self.assertIn('(TGStat)', post['title'])
+            self.assertFalse(get.call_args.kwargs['allow_redirects'])
+            for views in (None, '845', True, -1):
+                response.json.return_value['response']['views'] = views
+                self.assertIsNone(archived_post('memtc', '1553', 'Канал', get))
+            response.json.return_value['response'].update(views=845, link='t.me/other/1')
+            self.assertIsNone(archived_post('memtc', '1553', 'Канал', get))
+
+    def test_fallback_only_after_telegram_failure_and_preserves_original_job_key(self):
+        record = normalize('https://t.me/memtc/1553')
+        fallback = dict(title='memtc (TGStat)', views=845, display_url=archive_link('memtc', '1553'), source='tgstat')
+        with patch.object(service.engine, 'process_telegram_link', return_value=('Ошибка', 0)), patch.object(service, 'archived_post', return_value=fallback):
+            result = service.fetch_one(record)
+        self.assertEqual(result['url'], record['url'])
+        self.assertIsNone(result['error'])
+        output = render([result])
+        self.assertIn('href="https://tgstat.ru/channel/@memtc/1553"', output['html'])
+        self.assertIn('0,8K', output['html'])
+        with patch.object(service.engine, 'process_telegram_link', return_value=('Канал', 845)), patch.object(service, 'archived_post') as archive:
+            service.fetch_one(record)
+        archive.assert_not_called()
+
+    def test_archive_api_failure_keeps_unknown_views_with_archive_link(self):
+        with patch.object(service.engine, 'process_telegram_link', side_effect=TimeoutError), patch.object(service, 'archived_post', side_effect=TimeoutError):
+            result = service.fetch_one(normalize('https://t.me/memtc/1553'))
+        self.assertIsNone(result['views'])
+        self.assertEqual(result['display_url'], archive_link('memtc', '1553'))

tgstat_api.py

@@ -0,0 +1,35 @@
+"""Optional archived Telegram counters from the official TGStat API."""
+import os
+import re
+
+import requests
+
+
+def archive_link(username, message_id):
+    if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{2,}", username) or not re.fullmatch(r"[1-9]\d*", str(message_id)):
+        raise ValueError("Invalid public Telegram post")
+    return f"https://tgstat.ru/channel/@{username}/{message_id}"
+
+
+def archived_post(username, message_id, fallback_title, get=requests.get):
+    link = archive_link(username, message_id)
+    token = os.getenv("TGSTAT_API_TOKEN", "").strip()
+    if not token:
+        return None
+    # Do not scrape challenge pages or guess a number from unrelated page content.
+    response = get("https://api.tgstat.ru/posts/get", params={
+        "token": token, "postId": f"https://t.me/{username}/{message_id}"},
+        timeout=(4, 8), allow_redirects=False)
+    response.raise_for_status()
+    data = response.json()
+    post = data.get("response")
+    if data.get("status") != "ok" or not isinstance(post, dict):
+        return None
+    views = post.get("views")
+    if type(views) is not int or views < 0:
+        return None
+    source = str(post.get("link", "")).removeprefix("https://").removeprefix("http://")
+    match = re.fullmatch(r"(?:t.me|telegram.me)/([A-Za-z][A-Za-z0-9_]{2,})/([1-9]\d*)/?", source)
+    if not match or match[1].lower() != username.lower() or match[2] != str(message_id):
+        return None
+    return {"views": views, "title": f"{fallback_title} (TGStat)", "display_url": link, "source": "tgstat"}