Dink0 commited on
Commit
7099dd2
·
verified ·
1 Parent(s): cfa78a7

Delete app.py

Browse files
Files changed (1) hide show
  1. app.py +0 -459
app.py DELETED
@@ -1,459 +0,0 @@
1
- from __future__ import annotations
2
-
3
- import os
4
- import re
5
- import html as html_lib
6
- from typing import List, Tuple, Union
7
-
8
- import requests
9
- from fastapi import FastAPI
10
- from fastapi.middleware.cors import CORSMiddleware
11
- from pydantic import BaseModel
12
-
13
- APP_NAME = "pr-tool-backend"
14
-
15
- VK_API_VERSION = os.getenv("VK_API_VERSION", "5.131")
16
- VK_ACCESS_TOKEN = os.getenv("VK_ACCESS_TOKEN", "")
17
- REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "15"))
18
-
19
- app = FastAPI(title=APP_NAME)
20
-
21
- # Для простоты разрешаем все источники. Можно сузить список доменов в проде.
22
- app.add_middleware(
23
- CORSMiddleware,
24
- allow_origins=["*"],
25
- allow_credentials=False,
26
- allow_methods=["POST", "GET", "OPTIONS"],
27
- allow_headers=["*"]
28
- )
29
-
30
-
31
- class ParseRequest(BaseModel):
32
- links: Union[List[str], str]
33
-
34
-
35
- class ParseResponse(BaseModel):
36
- html: str
37
- text: str
38
- telegram_total: int
39
- vk_total: int
40
- errors: List[str]
41
-
42
-
43
- # ---------- Вспомогательные функции ----------
44
-
45
- def human_format_views(num: int) -> str:
46
- if num >= 1_000_000:
47
- value = num / 1_000_000
48
- suffix = "M"
49
- else:
50
- value = num / 1000
51
- suffix = "K"
52
-
53
- rounded = round(value, 1)
54
- if rounded.is_integer():
55
- formatted = f"{int(rounded)}{suffix}"
56
- else:
57
- formatted = f"{rounded:.1f}{suffix}"
58
-
59
- return formatted.replace(".", ",")
60
-
61
-
62
- def detect_platform(link: str) -> str:
63
- link = link.strip().lower()
64
- if "t.me/" in link or "telegram.me/" in link:
65
- return "telegram"
66
- if "vk.com/wall" in link or "vk.ru/wall" in link:
67
- return "vk"
68
- return ""
69
-
70
-
71
- def canonicalize_tg_link(link: str) -> Tuple[str | None, str | None, str | None]:
72
- link = link.strip()
73
- m = re.match(r"https?://(?:t(?:elegram)?\.me)/(?:s/)?([^/]+)/(?P<id>\d+)", link, re.IGNORECASE)
74
- if not m:
75
- return None, None, None
76
- username = m.group(1)
77
- message_id = m.group("id")
78
- canonical = f"https://t.me/{username}/{message_id}"
79
- return canonical, username, message_id
80
-
81
-
82
- def canonicalize_vk_link(link: str) -> Tuple[str | None, int | None, str | None]:
83
- link = link.strip()
84
- m = re.search(r"(?:vk\.com|vk\.ru)/wall(-?\d+)_(\d+)", link)
85
- if not m:
86
- return None, None, None
87
- owner_id = int(m.group(1))
88
- post_id = m.group(2)
89
- canonical = f"https://vk.com/wall{owner_id}_{post_id}"
90
- return canonical, owner_id, post_id
91
-
92
-
93
- def process_telegram_link(channel_username: str, message_id: str, canonical_link: str) -> Tuple[str, int]:
94
- # Набор альтернативных адресов (если t.me не резолвится)
95
- urls = [
96
- f"https://t.me/{channel_username}/{message_id}?embed=1",
97
- f"https://t.me/s/{channel_username}/{message_id}",
98
- f"https://telegram.me/{channel_username}/{message_id}?embed=1",
99
- f"https://r.jina.ai/http://t.me/{channel_username}/{message_id}?embed=1",
100
- f"https://r.jina.ai/http://t.me/s/{channel_username}/{message_id}",
101
- f"https://r.jina.ai/http://telegram.me/{channel_username}/{message_id}?embed=1",
102
- ]
103
- headers = {"User-Agent": "Mozilla/5.0"}
104
-
105
- def _parse_views_number(s: str) -> int:
106
- if not s:
107
- return 0
108
- s = s.strip().replace("\u00a0", "").replace(" ", "").replace(",", ".")
109
- m2 = re.match(r"([\d\.]+)\s*([kKmM]?)", s)
110
- if not m2:
111
- return 0
112
- val = float(m2.group(1))
113
- suf = m2.group(2).lower()
114
- if suf == "k":
115
- val *= 1_000
116
- elif suf == "m":
117
- val *= 1_000_000
118
- return int(val)
119
-
120
- def _fetch_first_ok(url_list):
121
- last_err = None
122
- for u in url_list:
123
- try:
124
- resp = requests.get(u, headers=headers, timeout=REQUEST_TIMEOUT)
125
- if resp.status_code == 200 and resp.text:
126
- return resp.text
127
- except Exception as exc:
128
- last_err = exc
129
- raise last_err or Exception("Не удалось получить страницу Telegram")
130
-
131
- def _strip_emoji(s: str) -> str:
132
- # Убираем emoji/pictographs и variation selectors.
133
- s = re.sub(
134
- r"[\U0001F1E6-\U0001F1FF\U0001F300-\U0001FAFF\u2600-\u27BF\uFE0E\uFE0F]",
135
- "",
136
- s,
137
- )
138
- return re.sub(r"\s+", " ", s).strip()
139
-
140
- def _clean_title(raw: str, fallback: str) -> str:
141
- title = re.sub(r"<[^>]+>", "", (raw or "").strip())
142
- title = html_lib.unescape(title)
143
- title = re.sub(r"\s*[-|]\s*Telegram\s*$", "", title, flags=re.IGNORECASE)
144
- title = _strip_emoji(title)
145
- return title or fallback
146
-
147
- def _extract_views(html: str) -> int:
148
- patterns = [
149
- # Виджет Telegram
150
- r'class="tgme_widget_message_views[^"]*">([\d\s\.,kKmM]+)<',
151
- # Резервный вариант из meta-description, но только если рядом явн�� "views"
152
- r'content="[^"]*?([\d\s\.,kKmM]+)\s+views[^"]*"',
153
- r'content="[^"]*?(?:Просмотры|просмотры|просмотров)[:\s]*([\d\s\.,kKmM]+)[^"]*"',
154
- ]
155
- for pattern in patterns:
156
- m = re.search(pattern, html, re.IGNORECASE)
157
- if m:
158
- parsed = _parse_views_number(m.group(1))
159
- if parsed > 0:
160
- return parsed
161
- return 0
162
-
163
- try:
164
- html = _fetch_first_ok(urls)
165
-
166
- title_raw = None
167
- m_site_name = re.search(
168
- r"<meta[^>]*property=[\"']og:site_name[\"'][^>]*content=[\"']([^\"']+)[\"']",
169
- html,
170
- re.IGNORECASE,
171
- )
172
- if m_site_name:
173
- title_raw = m_site_name.group(1).strip()
174
-
175
- m_title_meta = re.search(
176
- r"<meta[^>]*property=[\"']og:title[\"'][^>]*content=[\"']([^\"']+)[\"']",
177
- html,
178
- re.IGNORECASE,
179
- )
180
- if m_title_meta and not title_raw:
181
- title_raw = m_title_meta.group(1).strip()
182
- if not title_raw:
183
- m_title = re.search(
184
- r'class="tgme_widget_message_author_name".*?<span[^>]*>(.*?)</span>',
185
- html,
186
- re.IGNORECASE | re.DOTALL,
187
- )
188
- if m_title:
189
- title_raw = m_title.group(1).strip()
190
- if not title_raw:
191
- m_title = re.search(
192
- r'class="tgme_widget_message_owner_name".*?<span[^>]*>(.*?)</span>',
193
- html,
194
- re.IGNORECASE | re.DOTALL,
195
- )
196
- if m_title:
197
- title_raw = m_title.group(1).strip()
198
- title = _clean_title(title_raw or "", channel_username)
199
-
200
- views = _extract_views(html)
201
-
202
- return title, views
203
-
204
- except Exception as exc:
205
- return f"Ошибка (TG) при обработке {canonical_link}: {exc}", 0
206
-
207
-
208
- def process_vk_link(owner_id: int, post_id: str, canonical_link: str) -> Tuple[str, int]:
209
- posts_param = f"{owner_id}_{post_id}"
210
- api_url = "https://api.vk.com/method/wall.getById"
211
-
212
- params = {
213
- "posts": posts_param,
214
- "v": VK_API_VERSION,
215
- "extended": 1,
216
- }
217
- if VK_ACCESS_TOKEN:
218
- params["access_token"] = VK_ACCESS_TOKEN
219
-
220
- try:
221
- resp = requests.get(api_url, params=params, timeout=REQUEST_TIMEOUT)
222
- except Exception as exc:
223
- return f"**Ошибка (VK)**: {exc} — {canonical_link}", 0
224
-
225
- if resp.status_code != 200:
226
- return f"**Ошибка (VK)**: HTTP {resp.status_code} — {canonical_link}", 0
227
-
228
- data = resp.json()
229
- if "error" in data:
230
- error_msg = data["error"].get("error_msg", "")
231
- return f"**Ошибка (VK)**: {error_msg or 'API error'} — {canonical_link}", 0
232
-
233
- response_data = data.get("response", {})
234
- items = response_data.get("items", [])
235
- if not items:
236
- return f"**Ошибка (VK)**: пост не найден — {canonical_link}", 0
237
-
238
- post = items[0]
239
- views = post.get("views", {}).get("count", 0)
240
- post_owner_id = post.get("owner_id", owner_id)
241
-
242
- title = None
243
- if post_owner_id < 0:
244
- group_id = -post_owner_id
245
- for group in response_data.get("groups", []):
246
- if group.get("id") == group_id:
247
- title = group.get("name")
248
- break
249
- else:
250
- user_id = post_owner_id
251
- for profile in response_data.get("profiles", []):
252
- if profile.get("id") == user_id:
253
- first_name = profile.get("first_name", "")
254
- last_name = profile.get("last_name", "")
255
- title = (first_name + " " + last_name).strip()
256
- break
257
- if not title:
258
- title = "VK пост"
259
-
260
- return title, views
261
-
262
-
263
- def normalize_links(links: Union[List[str], str]) -> List[str]:
264
- if isinstance(links, str):
265
- raw = links.splitlines()
266
- else:
267
- raw = []
268
- for item in links:
269
- raw.extend(str(item).splitlines())
270
- return [line.strip() for line in raw if line.strip()]
271
-
272
-
273
- def _escape(text: str) -> str:
274
- return html_lib.escape(text, quote=True)
275
-
276
-
277
- def build_output(lines: List[Tuple[str, str, object, object]]) -> Tuple[str, str, int, int, List[str]]:
278
- errors: List[str] = []
279
-
280
- tg_groups = {}
281
- vk_groups = {}
282
-
283
- for plat, canonical, a, b in lines:
284
- if plat == "telegram":
285
- username, mid = a, b
286
- if not username or not mid:
287
- key = canonical
288
- tg_groups.setdefault(key, {"title": "Telegram", "items": []})
289
- tg_groups[key]["items"].append((canonical, 0))
290
- else:
291
- title, views = process_telegram_link(username, mid, canonical)
292
- key = username
293
- if key not in tg_groups:
294
- tg_groups[key] = {"title": title, "items": []}
295
- if not tg_groups[key].get("title") or str(tg_groups[key]["title"]).startswith("**Ошибка"):
296
- tg_groups[key]["title"] = title
297
- tg_groups[key]["items"].append((canonical, views))
298
-
299
- elif plat == "vk":
300
- owner_id, post_id = a, b
301
- if owner_id is None or post_id is None:
302
- key = canonical
303
- vk_groups.setdefault(key, {"title": "VK пост", "items": []})
304
- vk_groups[key]["items"].append((canonical, 0))
305
- else:
306
- title, views = process_vk_link(owner_id, post_id, canonical)
307
- key = str(owner_id)
308
- if key not in vk_groups:
309
- vk_groups[key] = {"title": title, "items": []}
310
- if not vk_groups[key].get("title") or str(vk_groups[key]["title"]).startswith("**Ошибка"):
311
- vk_groups[key]["title"] = title
312
- vk_groups[key]["items"].append((canonical, views))
313
- else:
314
- errors.append(f"Неизвестная платформа: {canonical}")
315
-
316
- tg_total_views = sum(v for g in tg_groups.values() for _, v in g["items"])
317
- vk_total_views = sum(v for g in vk_groups.values() for _, v in g["items"])
318
-
319
- tg_sorted = sorted(
320
- tg_groups.items(),
321
- key=lambda kv: sum(v for _, v in kv[1]["items"]),
322
- reverse=True,
323
- )
324
- vk_sorted = sorted(
325
- vk_groups.items(),
326
- key=lambda kv: sum(v for _, v in kv[1]["items"]),
327
- reverse=True,
328
- )
329
-
330
- html_lines: List[str] = []
331
- text_lines: List[str] = []
332
-
333
- # --- Telegram ---
334
- html_lines.append("<h2>Telegram</h2>")
335
- html_lines.append(f'Суммарно посты собрали <b>{human_format_views(tg_total_views)}</b> просмотров.')
336
- text_lines.append("Telegram")
337
- text_lines.append(f"Суммарно посты собрали {human_format_views(tg_total_views)} просмотров.")
338
- if tg_sorted:
339
- html_lines.append("<ol>")
340
- for idx, (_, data) in enumerate(tg_sorted, start=1):
341
- title = data["title"] or "Telegram"
342
- items = data["items"]
343
- first_link, _first_views = items[0]
344
-
345
- title_html = _escape(title)
346
- first_link_html = _escape(first_link)
347
- line_html = f'<li><a href="{first_link_html}">{title_html}</a>'
348
- if len(items) > 1:
349
- for link2, _v in items[1:]:
350
- line_html += f' + <a href="{_escape(link2)}">ещё</a>'
351
- views_str = " + ".join(human_format_views(v) for _, v in items)
352
- line_html += f" — {views_str}</li>"
353
- html_lines.append(line_html)
354
-
355
- line_text = f"{idx}. {title} ({first_link})"
356
- if len(items) > 1:
357
- extra_links = ", ".join(link2 for link2, _v in items[1:])
358
- line_text += f" + ещё: {extra_links}"
359
- line_text += f" — {views_str}"
360
- text_lines.append(line_text)
361
- html_lines.append("</ol>")
362
- else:
363
- html_lines.append("<i>Нет ссылок на Telegram</i>")
364
- text_lines.append("Нет ссылок на Telegram")
365
- html_lines.append("<br/>")
366
- text_lines.append("")
367
-
368
- # --- VK ---
369
- html_lines.append("<h2>ВКонтакте</h2>")
370
- html_lines.append(f'Суммарно посты собрали <b>{human_format_views(vk_total_views)}</b> просмотров.')
371
- text_lines.append("ВКонтакте")
372
- text_lines.append(f"Суммарно посты собрали {human_format_views(vk_total_views)} просмотров.")
373
- if vk_sorted:
374
- html_lines.append("<ol>")
375
- for idx, (_, data) in enumerate(vk_sorted, start=1):
376
- title = data["title"] or "VK пост"
377
- items = data["items"]
378
- first_link, _first_views = items[0]
379
-
380
- title_html = _escape(title)
381
- first_link_html = _escape(first_link)
382
- line_html = f'<li><a href="{first_link_html}">{title_html}</a>'
383
- if len(items) > 1:
384
- for link2, _v in items[1:]:
385
- line_html += f' + <a href="{_escape(link2)}">ещё</a>'
386
- views_str = " + ".join(human_format_views(v) for _, v in items)
387
- line_html += f" — {views_str}</li>"
388
- html_lines.append(line_html)
389
-
390
- line_text = f"{idx}. {title} ({first_link})"
391
- if len(items) > 1:
392
- extra_links = ", ".join(link2 for link2, _v in items[1:])
393
- line_text += f" + ещё: {extra_links}"
394
- line_text += f" — {views_str}"
395
- text_lines.append(line_text)
396
- html_lines.append("</ol>")
397
- else:
398
- html_lines.append("<i>Нет ссылок на ВКонтакте</i>")
399
- text_lines.append("Нет ссылок на ВКонтакте")
400
- html_lines.append("<br/>")
401
- text_lines.append("")
402
-
403
- if errors:
404
- html_lines.append("<h2>Ошибки</h2>")
405
- html_lines.append("<ul>")
406
- for err in errors:
407
- html_lines.append(f"<li>{_escape(err)}</li>")
408
- text_lines.append(f"- {err}")
409
- html_lines.append("</ul>")
410
- html_lines.append("<br/>")
411
- text_lines.append("")
412
-
413
- return "\n".join(html_lines), "\n".join(text_lines), tg_total_views, vk_total_views, errors
414
-
415
-
416
- @app.get("/health")
417
- def health_check():
418
- return {"status": "ok"}
419
-
420
-
421
- @app.post("/parse", response_model=ParseResponse)
422
- def parse_links(payload: ParseRequest):
423
- raw_lines = normalize_links(payload.links)
424
- if not raw_lines:
425
- return ParseResponse(html="", text="", telegram_total=0, vk_total=0, errors=["Нет ссылок для обработки."])
426
-
427
- seen = set()
428
- lines: List[Tuple[str, str, object, object]] = []
429
- for link in raw_lines:
430
- plat = detect_platform(link)
431
- if plat == "telegram":
432
- canonical, username, mid = canonicalize_tg_link(link)
433
- if not canonical:
434
- canonical = link
435
- username = None
436
- mid = None
437
- if canonical in seen:
438
- continue
439
- seen.add(canonical)
440
- lines.append(("telegram", canonical, username, mid))
441
- elif plat == "vk":
442
- canonical, owner_id, post_id = canonicalize_vk_link(link)
443
- if not canonical:
444
- canonical = link
445
- if canonical in seen:
446
- continue
447
- seen.add(canonical)
448
- lines.append(("vk", canonical, owner_id, post_id))
449
- else:
450
- lines.append(("unknown", link, None, None))
451
-
452
- html, text, tg_total, vk_total, errors = build_output(lines)
453
- return ParseResponse(
454
- html=html,
455
- text=text,
456
- telegram_total=tg_total,
457
- vk_total=vk_total,
458
- errors=errors,
459
- )