From faded0132b9b9854b7d2b17b6c198f1449c98068 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 10 Aug 2026 22:35:33 +0500 Subject: [PATCH] feat: ingest site videos through raw pipeline --- README.md | 12 ++ db/migrations/049_generic_video_uploader.sql | 9 ++ site_parser_worker/app.py | 115 ++++++++++++++++-- site_parser_worker/test_parser.py | 4 +- src/vk_parser_app/source_adapters.py | 13 +- src/vk_parser_app/templates/source_form.html | 1 + .../workers/vk_storage_uploader.py | 81 +++++++----- tests/test_source_adapters.py | 25 ++++ tests/test_storage_uploader.py | 13 ++ 9 files changed, 231 insertions(+), 42 deletions(-) create mode 100644 db/migrations/049_generic_video_uploader.sql create mode 100644 tests/test_storage_uploader.py diff --git a/README.md b/README.md index b4235c6..ed925a4 100644 --- a/README.md +++ b/README.md @@ -124,6 +124,18 @@ https://vk.com/wall-239548476_123 } ``` +Если RSS отдаёт только заголовки и ссылки, можно явно дочитывать страницы записей: + +```json +{ + "format": "rss", + "access": "cloudflare", + "max_items": 20, + "follow_links": true, + "content_selector": "article.full" +} +``` + В глобальном разделе `Site Parser` задаются URL воркера, его токен, ключ RuCaptcha и таймаут. Внешний модуль запускается командой: diff --git a/db/migrations/049_generic_video_uploader.sql b/db/migrations/049_generic_video_uploader.sql new file mode 100644 index 0000000..4c39de5 --- /dev/null +++ b/db/migrations/049_generic_video_uploader.sql @@ -0,0 +1,9 @@ +UPDATE app_settings +SET description='Максимальная высота видео VK и YouTube перед загрузкой в Telegram.', + updated_at=NOW() +WHERE key='video_max_height'; + +UPDATE app_settings +SET description='Таймаут скачивания видео VK и YouTube.', + updated_at=NOW() +WHERE key='uploader_yt_dlp_timeout_sec'; diff --git a/site_parser_worker/app.py b/site_parser_worker/app.py index 3986f2b..c0c5310 100644 --- a/site_parser_worker/app.py +++ b/site_parser_worker/app.py @@ -2,10 +2,12 @@ from __future__ import annotations import asyncio import os +import re import secrets from datetime import datetime, timezone from time import struct_time from typing import Annotated, Any +from urllib.parse import parse_qs, urljoin, urlparse import feedparser import httpx @@ -59,28 +61,59 @@ def iso_date(value: struct_time | None) -> str | None: return datetime(*value[:6], tzinfo=timezone.utc).isoformat() -def clean_html(value: Any) -> tuple[str, str, list[dict[str, str]]]: +def youtube_url(value: str, base_url: str = "") -> str | None: + url = urljoin(base_url, str(value or "").strip()) + parsed = urlparse(url) + host = (parsed.hostname or "").lower().removeprefix("www.").removeprefix("m.") + video_id = "" + if host == "youtu.be": + video_id = parsed.path.strip("/").split("/", 1)[0] + elif host in {"youtube.com", "youtube-nocookie.com"}: + if parsed.path == "/watch": + video_id = (parse_qs(parsed.query).get("v") or [""])[0] + elif parsed.path.startswith(("/embed/", "/shorts/", "/live/")): + video_id = parsed.path.strip("/").split("/", 1)[1] + if not re.fullmatch(r"[A-Za-z0-9_-]{6,20}", video_id): + return None + return f"https://www.youtube.com/watch?v={video_id}" + + +def clean_html(value: Any, base_url: str = "") -> tuple[str, str, list[dict[str, str]]]: html = str(value or "").strip() soup = BeautifulSoup(html, "html.parser") + for element in soup.find_all(["script", "style", "noscript"]): + element.decompose() text = soup.get_text("\n", strip=True) - media = [{"type": "photo", "url": str(image["src"])} for image in soup.find_all("img", src=True)] + media = [ + {"type": "photo", "url": urljoin(base_url, str(image["src"]))} + for image in soup.find_all("img", src=True) + ] + for element in soup.find_all(["video", "source"], src=True): + media.append({"type": "video", "url": urljoin(base_url, str(element["src"]))}) + for element in soup.find_all(["iframe", "a"]): + url = youtube_url(str(element.get("src") or element.get("href") or ""), base_url) + if url: + media.append({"type": "video", "url": url}) return html, text, media def parse_rss(xml: str, max_items: int) -> list[ParsedItem]: - feed = feedparser.parse(xml) + feed = feedparser.parse(xml, sanitize_html=False) if feed.bozo and not feed.entries: raise ValueError(f"Invalid RSS: {feed.bozo_exception}") items = [] for entry in feed.entries[:max_items]: title = BeautifulSoup(str(entry.get("title") or ""), "html.parser").get_text(" ", strip=True) - html, text, media = clean_html(entry.get("description") or entry.get("summary") or "") - for enclosure in entry.get("enclosures") or []: - url = str(enclosure.get("href") or enclosure.get("url") or "").strip() - media_type = str(enclosure.get("type") or "") - if url and (not media_type or media_type.startswith("image/")): - media.append({"type": "photo", "url": url}) url = str(entry.get("link") or "").strip() + content = (entry.get("content") or [{}])[0].get("value") or entry.get("description") or entry.get("summary") or "" + html, text, media = clean_html(content, url) + for enclosure in entry.get("enclosures") or []: + enclosure_url = urljoin(url, str(enclosure.get("href") or enclosure.get("url") or "").strip()) + media_type = str(enclosure.get("type") or "") + if enclosure_url and (not media_type or media_type.startswith("image/")): + media.append({"type": "photo", "url": enclosure_url}) + elif enclosure_url and media_type.startswith("video/"): + media.append({"type": "video", "url": enclosure_url}) items.append( ParsedItem( external_id=str(entry.get("id") or entry.get("guid") or url).strip(), @@ -113,8 +146,8 @@ async def fetch_in_browser( async_two_captcha_client=AsyncTwoCaptcha(rucaptcha_token), max_attempts=1, ) as solver: - response = await page.goto(url, wait_until="domcontentloaded", timeout=60_000) - if response and response.status in {403, 429, 503} and "Just a moment" in await page.title(): + await page.goto(url, wait_until="domcontentloaded", timeout=60_000) + if "just a moment" in (await page.title()).lower(): await solver.solve_captcha( captcha_container=page, captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL, @@ -129,6 +162,47 @@ async def fetch_in_browser( return xml, state +async def enrich_items_in_browser( + items: list[ParsedItem], + rucaptcha_token: str, + browser_state: dict[str, Any] | None, + content_selector: str, +) -> tuple[list[ParsedItem], dict[str, Any]]: + async with browser_lock: + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(channel="chrome", headless=False) + context = await browser.new_context(storage_state=browser_state) if browser_state else await browser.new_context() + page = await context.new_page() + async with TwoCaptchaSolver( + framework=FrameworkType.PLAYWRIGHT, + page=page, + async_two_captcha_client=AsyncTwoCaptcha(rucaptcha_token), + max_attempts=1, + ) as solver: + for item in items: + if item.text and item.media: + continue + await page.goto(item.url, wait_until="domcontentloaded", timeout=60_000) + if "just a moment" in (await page.title()).lower(): + await solver.solve_captcha( + captcha_container=page, + captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL, + ) + try: + content = page.locator(content_selector).first + await content.wait_for(state="visible", timeout=60_000) + except Exception as exc: + raise RuntimeError(f"Item content not found: {item.url} ({content_selector})") from exc + html, text, media = clean_html(await content.inner_html(), item.url) + if text: + item.html = html + item.text = text + item.media = list({entry["url"]: entry for entry in [*item.media, *media]}.values()) + state = await context.storage_state() + await browser.close() + return items, state + + @app.get("/health") async def health() -> dict[str, bool]: return {"ok": True} @@ -154,6 +228,12 @@ async def parse_source( raise HTTPException(status_code=422, detail="config.max_items must be an integer") from exc if not 1 <= max_items <= 100: raise HTTPException(status_code=422, detail="config.max_items must be between 1 and 100") + follow_links = config.get("follow_links", False) + if not isinstance(follow_links, bool): + raise HTTPException(status_code=422, detail="config.follow_links must be true or false") + content_selector = str(config.get("content_selector") or "").strip() + if follow_links and not content_selector: + raise HTTPException(status_code=422, detail="config.content_selector is required when follow_links=true") url = str(request.url) xml = "" state = request.browser_state @@ -184,6 +264,19 @@ async def parse_source( items = parse_rss(xml, max_items) except ValueError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc + if follow_links: + if not request.rucaptcha_token: + raise HTTPException(status_code=422, detail="rucaptcha_token is required when follow_links=true") + try: + items, state = await enrich_items_in_browser( + items, + request.rucaptcha_token.get_secret_value(), + state, + content_selector, + ) + except Exception as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + fetched_via += "+item-pages" return { "source_url": url, "fetched_via": fetched_via, diff --git a/site_parser_worker/test_parser.py b/site_parser_worker/test_parser.py index 46d6143..281336c 100644 --- a/site_parser_worker/test_parser.py +++ b/site_parser_worker/test_parser.py @@ -4,12 +4,14 @@ from app import parse_rss def test_parse_rss() -> None: xml = """1Title https://example.test/1 - <p>Body</p><img src="https://example.test/1.jpg"> + <p>Body</p><img src="/1.jpg"> + <iframe src="https://www.youtube.com/embed/dQw4w9WgXcQ?feature=oembed"></iframe> Mon, 10 Aug 2026 10:00:00 +0000""" item = parse_rss(xml, 20)[0] assert item.external_id == "1" assert item.text == "Body" assert item.media[0]["url"] == "https://example.test/1.jpg" + assert item.media[1] == {"type": "video", "url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"} if __name__ == "__main__": diff --git a/src/vk_parser_app/source_adapters.py b/src/vk_parser_app/source_adapters.py index 6855a34..722005e 100644 --- a/src/vk_parser_app/source_adapters.py +++ b/src/vk_parser_app/source_adapters.py @@ -55,6 +55,13 @@ def validate_source_config(platform: str, config: dict[str, Any]) -> None: raise ValueError("max_items должен быть целым числом") from exc if not 1 <= max_items <= 100: raise ValueError("max_items должен быть от 1 до 100") + follow_links = config.get("follow_links", False) + if not isinstance(follow_links, bool): + raise ValueError("follow_links должен быть true или false") + if follow_links and not str(config.get("content_selector") or "").strip(): + raise ValueError("При follow_links=true нужен content_selector") + + def _posted_at(value: Any) -> datetime: try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) @@ -111,7 +118,11 @@ class SiteParserClient: for raw in data.get("items") or []: title = str(raw.get("title") or "").strip() body = str(raw.get("text") or "").strip() - text = "\n\n".join(part for part in (title, body) if part) + body_starts_with_title = title and ( + body.casefold() == title.casefold() + or body.casefold().startswith(f"{title}\n".casefold()) + ) + text = body if body_starts_with_title else "\n\n".join(part for part in (title, body) if part) url = str(raw.get("url") or source["url"]).strip() external_id = str(raw.get("external_id") or url).strip() if not external_id: diff --git a/src/vk_parser_app/templates/source_form.html b/src/vk_parser_app/templates/source_form.html index 76988d7..a79c643 100644 --- a/src/vk_parser_app/templates/source_form.html +++ b/src/vk_parser_app/templates/source_form.html @@ -52,6 +52,7 @@ "max_items": 20 }

access: auto сначала пробует обычный запрос и при Cloudflare использует RuCaptcha; http запрещает браузер; cloudflare сразу запускает браузер.

+

Если RSS не содержит текст или медиа, добавьте "follow_links": true и CSS-селектор содержимого страницы, например "content_selector": "article.full".

diff --git a/src/vk_parser_app/workers/vk_storage_uploader.py b/src/vk_parser_app/workers/vk_storage_uploader.py index 5237e6a..1010605 100644 --- a/src/vk_parser_app/workers/vk_storage_uploader.py +++ b/src/vk_parser_app/workers/vk_storage_uploader.py @@ -7,6 +7,7 @@ import re import sys import time from pathlib import Path +from urllib.parse import urlparse import aiohttp from aiogram import Bot @@ -52,6 +53,20 @@ def parse_vk_video_url(vk_url: str) -> tuple[int, int] | None: return int(match.group(1)), int(match.group(2)) +def video_provider(url: str) -> str | None: + host = (urlparse(str(url or "")).hostname or "").lower().removeprefix("www.").removeprefix("m.") + if host == "vk.com" and parse_vk_video_url(url): + return "vk" + if host in {"youtube.com", "youtube-nocookie.com", "youtu.be"}: + return "youtube" + return None + + +def remove_download_files(output_path: str) -> None: + for path in TMP_DIR.glob(f"{Path(output_path).name}*"): + path.unlink(missing_ok=True) + + def split_message_chunks(text: str, limit: int) -> list[str]: text = str(text or "").strip() if not text: @@ -320,37 +335,43 @@ class TelegramStorageUploader: error[:1000], ) - async def download_video(self, vk_url: str, output_path: str) -> dict | None: - parsed = parse_vk_video_url(vk_url) - if not parsed: + async def download_video(self, video_url: str, output_path: str) -> dict | None: + provider = video_provider(video_url) + if not provider: return None - owner_id, video_id = parsed max_size = self.video_max_size_mb * 1024 * 1024 - netrc_path = f"{output_path}.netrc" - fd = os.open(netrc_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - with os.fdopen(fd, "w", encoding="utf-8") as fh: - fh.write(f"machine vk.com login vk_token password {settings.vk_access_token}\n") + target_url = video_url cmd = [ sys.executable, "-m", "yt_dlp", - "--netrc-location", - netrc_path, - f"https://vk.com/video{owner_id}_{video_id}", - "-o", - output_path, + ] + netrc_path = None + if provider == "vk": + owner_id, video_id = parse_vk_video_url(video_url) or (0, 0) + target_url = f"https://vk.com/video{owner_id}_{video_id}" + netrc_path = f"{output_path}.netrc" + fd = os.open(netrc_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(f"machine vk.com login vk_token password {settings.vk_access_token}\n") + cmd.extend(["--netrc-location", netrc_path]) + cmd.extend([ + target_url, + "-o", output_path, "--no-playlist", - "-f", - ( + "--match-filter", f"duration <= {self.video_max_duration_sec}", + "--merge-output-format", "mp4", + "-f", ( f"best[height<={self.video_max_height}][filesize<{max_size}]" f"/best[height<={self.video_max_height}]" f"/bestvideo[height<={self.video_max_height}][filesize<{max_size}]+bestaudio/best" f"/bestvideo[height<={self.video_max_height}]+bestaudio/best" f"/best[filesize<{max_size}]" ), - "--quiet", - "--no-warnings", - ] + "--quiet", "--no-warnings", + ]) + proc = None + stderr = b"" try: proc = await asyncio.create_subprocess_exec( *cmd, @@ -359,20 +380,24 @@ class TelegramStorageUploader: ) _, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.yt_dlp_timeout_sec) except asyncio.TimeoutError: - proc.kill() - await proc.communicate() + if proc: + proc.kill() + await proc.communicate() + remove_download_files(output_path) return {"error": "video download timeout", "permanent": False} finally: - try: - os.unlink(netrc_path) - except FileNotFoundError: - pass + if netrc_path: + Path(netrc_path).unlink(missing_ok=True) if proc.returncode != 0 or not os.path.exists(output_path): err = (stderr or b"").decode("utf-8", errors="ignore").lower() - permanent = any(marker in err for marker in ("removed", "unavailable", "private", "access denied")) + permanent = any(marker in err for marker in ( + "removed", "unavailable", "private", "access denied", "does not pass filter", "sign in", + )) + remove_download_files(output_path) return {"error": "video unavailable or download failed", "permanent": permanent} size = os.path.getsize(output_path) if size > max_size: + remove_download_files(output_path) return {"error": "video too large", "permanent": True} return {"path": output_path, "size_bytes": size} @@ -581,8 +606,7 @@ class TelegramStorageUploader: tmp_path = item.get("tmp_path") if tmp_path: try: - if os.path.exists(tmp_path): - os.remove(tmp_path) + remove_download_files(tmp_path) except Exception: pass if not blocking_error.startswith("media still pending:"): @@ -600,8 +624,7 @@ class TelegramStorageUploader: tmp_path = item.get("tmp_path") if tmp_path: try: - if os.path.exists(tmp_path): - os.remove(tmp_path) + remove_download_files(tmp_path) except Exception: pass await self.mark_post_ready(raw_post_id, message_ids, meta_message_id) diff --git a/tests/test_source_adapters.py b/tests/test_source_adapters.py index ff519a9..00d235e 100644 --- a/tests/test_source_adapters.py +++ b/tests/test_source_adapters.py @@ -28,11 +28,23 @@ class FakeResponse: } +class FollowedPageResponse(FakeResponse): + async def json(self, **_kwargs): + data = await super().json(**_kwargs) + data["items"][0]["text"] = "Title\nAuthor" + return data + + class FakeSession: def post(self, *_args, **_kwargs): return FakeResponse() +class FollowedPageSession(FakeSession): + def post(self, *_args, **_kwargs): + return FollowedPageResponse() + + class SourceAdapterTests(unittest.IsolatedAsyncioTestCase): async def test_worker_response_is_normalized(self) -> None: client = SiteParserClient(FakeSession(), "http://worker", "token", "captcha", 30) @@ -46,10 +58,23 @@ class SourceAdapterTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(items[0].media[0].url, "https://example.test/1.jpg") self.assertEqual(state["browser_state"]["cookies"][0]["name"], "cf_clearance") + async def test_followed_page_does_not_repeat_title(self) -> None: + client = SiteParserClient(FollowedPageSession(), "http://worker", "token", "captcha", 30) + items, _ = await client.fetch({ + "url": "https://example.test/rss.xml", + "settings_json": '{"format":"rss","follow_links":true,"content_selector":"article.full"}', + "runtime_state_json": "{}", + }) + self.assertEqual(items[0].text, "Title\nAuthor") + def test_site_config_is_required(self) -> None: with self.assertRaisesRegex(ValueError, "нужен конфиг"): validate_source_config("site", {}) + def test_follow_links_requires_selector(self) -> None: + with self.assertRaisesRegex(ValueError, "content_selector"): + validate_source_config("site", {"format": "rss", "follow_links": True}) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_storage_uploader.py b/tests/test_storage_uploader.py new file mode 100644 index 0000000..c6d12ac --- /dev/null +++ b/tests/test_storage_uploader.py @@ -0,0 +1,13 @@ +from vk_parser_app.workers.vk_storage_uploader import video_provider + + +def test_video_provider() -> None: + assert video_provider("https://vk.com/video-1_2") == "vk" + assert video_provider("https://youtu.be/dQw4w9WgXcQ") == "youtube" + assert video_provider("https://www.youtube.com/watch?v=dQw4w9WgXcQ") == "youtube" + assert video_provider("file:///etc/passwd") is None + assert video_provider("http://127.0.0.1/video.mp4") is None + + +if __name__ == "__main__": + test_video_provider()