exostring/new_vk_parser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: ingest site videos through raw pipeline

Browse Source
This commit is contained in:
Your Name
2026-08-10 22:35:33 +05:00
parent 7889d4d376
commit faded0132b
9 changed files with 231 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+12
View File
@@ -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 и таймаут. Внешний модуль запускается командой:
db/migrations/049_generic_video_uploader.sql
+9
View File
@@ -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';
site_parser_worker/app.py
+104 -11
View File
@@ -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,
site_parser_worker/test_parser.py
+3 -1
View File
@@ -4,12 +4,14 @@ from app import parse_rss
def test_parse_rss() -> None:
xml = """<rss><channel><item><guid>1</guid><title>Title</title>
<link>https://example.test/1</link>
<description>&lt;p&gt;Body&lt;/p&gt;&lt;img src="https://example.test/1.jpg"&gt;</description>
<description>&lt;p&gt;Body&lt;/p&gt;&lt;img src="/1.jpg"&gt;
&lt;iframe src="https://www.youtube.com/embed/dQw4w9WgXcQ?feature=oembed"&gt;&lt;/iframe&gt;</description>
<pubDate>Mon, 10 Aug 2026 10:00:00 +0000</pubDate></item></channel></rss>"""
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__":
src/vk_parser_app/source_adapters.py
+12 -1
View File
@@ -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:
src/vk_parser_app/templates/source_form.html
+1
View File
@@ -52,6 +52,7 @@
"max_items": 20
}</pre>
<p class="mt-2"><code>access</code>: <code>auto</code> сначала пробует обычный запрос и при Cloudflare использует RuCaptcha; <code>http</code> запрещает браузер; <code>cloudflare</code> сразу запускает браузер.</p>
<p class="mt-2">Если RSS не содержит текст или медиа, добавьте <code>"follow_links": true</code> и CSS-селектор содержимого страницы, например <code>"content_selector": "article.full"</code>.</p>
</details>
</div>
</div>
src/vk_parser_app/workers/vk_storage_uploader.py
+50 -27
View File
@@ -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:
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)
tests/test_source_adapters.py
+25
View File
@@ -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()
tests/test_storage_uploader.py
+13
View File
@@ -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()