diff --git a/.gitignore b/.gitignore index 28f65d1..5d9f36e 100644 --- a/.gitignore +++ b/.gitignore @@ -7,9 +7,9 @@ __pycache__/ logs/ uploads/ original_project/ -ai_worker.py -media_worker.py -parser.py -raw_feed_worker.py +/ai_worker.py +/media_worker.py +/parser.py +/raw_feed_worker.py vkdev_main.js sample_photo.jpg diff --git a/src/vk_parser_app/workers/parser.py b/src/vk_parser_app/workers/parser.py new file mode 100644 index 0000000..c505c13 --- /dev/null +++ b/src/vk_parser_app/workers/parser.py @@ -0,0 +1,449 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +from datetime import datetime, timedelta, timezone + +from loguru import logger + +from ..config import settings +from ..constants import ( + JOB_TYPE_VK_STORAGE_COPY, + PLATFORM_VK, + POST_STATUS_SKIPPED, + POST_STATUS_STORAGE_PENDING, + SOURCE_STATUS_ERROR, + SOURCE_STATUS_OK, + WORKER_PARSER, +) +from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, get_pool +from ..heartbeat import HeartbeatReporter +from ..jobs import is_worker_enabled +from ..vk_api import ( + VKAPIClient, + VKAPIError, + extract_media, + is_deleted_or_invalid, + is_fatal_source_error, + is_repost, + post_vk_url, +) + + +def utc_from_ts(value: int) -> datetime: + return datetime.fromtimestamp(int(value), tz=timezone.utc) + + +def make_hash(*parts: str) -> str: + h = hashlib.sha256() + for part in parts: + h.update((part or "").strip().lower().encode("utf-8")) + h.update(b"\0") + return h.hexdigest() + + +class VKParserWorker: + def __init__(self) -> None: + self.pool = None + self.heartbeat = HeartbeatReporter(WORKER_PARSER, 30) + + async def init(self) -> None: + self.pool = await get_pool() + + async def active_sources(self) -> list[dict]: + rows = await self.pool.fetch( + """ + SELECT * + FROM sources + WHERE platform=$1 + AND active=TRUE + AND archived_at IS NULL + ORDER BY last_checked_at NULLS FIRST, priority ASC, id ASC + """, + PLATFORM_VK, + ) + return [dict(r) for r in rows] + + async def mark_source_error(self, source_id: int, message: str) -> None: + await self.pool.execute( + """ + UPDATE sources + SET status=$2, + status_msg=$3, + last_checked_at=NOW(), + updated_at=NOW() + WHERE id=$1 + """, + source_id, + SOURCE_STATUS_ERROR, + message[:1000], + ) + + async def deactivate_source(self, source_id: int, message: str) -> None: + await self.pool.execute( + """ + UPDATE sources + SET active=FALSE, + status=$2, + status_msg=$3, + last_checked_at=NOW(), + updated_at=NOW() + WHERE id=$1 + """, + source_id, + SOURCE_STATUS_ERROR, + message[:1000], + ) + + async def mark_source_ok(self, source_id: int, last_parsed_at: datetime | None) -> None: + await self.pool.execute( + """ + UPDATE sources + SET status=$2, + status_msg=NULL, + last_checked_at=NOW(), + last_parsed_at=COALESCE($3, last_parsed_at), + updated_at=NOW() + WHERE id=$1 + """, + source_id, + SOURCE_STATUS_OK, + last_parsed_at, + ) + + async def resolve_source_if_needed(self, client: VKAPIClient, source: dict) -> dict: + if source.get("external_owner_id"): + return source + external_id, owner_id, resolved_name = await client.resolve_group(source["url"] or source.get("external_id") or "") + await self.pool.execute( + """ + UPDATE sources + SET external_id=$2, + external_owner_id=$3, + name=CASE WHEN name='' OR name=url THEN $4 ELSE name END, + updated_at=NOW() + WHERE id=$1 + """, + int(source["id"]), + external_id, + owner_id, + resolved_name, + ) + source["external_id"] = external_id + source["external_owner_id"] = owner_id + if not source.get("name") or source.get("name") == source.get("url"): + source["name"] = resolved_name + return source + + async def known_post_ids(self, source_id: int, external_post_ids: list[str]) -> set[str]: + if not external_post_ids: + return set() + rows = await self.pool.fetch( + """ + SELECT external_post_id + FROM raw_posts + WHERE source_id=$1 AND external_post_id=ANY($2::text[]) + """, + source_id, + external_post_ids, + ) + return {str(r["external_post_id"]) for r in rows} + + async def known_content_hashes(self, hashes: list[str]) -> set[str]: + if not hashes: + return set() + rows = await self.pool.fetch( + """ + SELECT content_hash + FROM raw_posts + WHERE content_hash=ANY($1::text[]) + """, + hashes, + ) + return {str(r["content_hash"]) for r in rows} + + async def save_post( + self, + source: dict, + post: dict, + *, + status: str, + skip_reason: str | None = None, + create_storage_job: bool = True, + ) -> int | None: + source_id = int(source["id"]) + owner_id = int(post.get("owner_id") or source["external_owner_id"]) + external_post_id = str(post["id"]) + raw_text = str(post.get("text") or "").strip() + posted_at = utc_from_ts(int(post["date"])) + media = extract_media(post) + media_ids = ",".join(m.attachment_id for m in media) + text_hash = make_hash(raw_text) + content_hash = make_hash(raw_text, media_ids) + original_url = post_vk_url(owner_id, external_post_id) + + async with self.pool.acquire() as conn: + async with conn.transaction(): + raw_post_id = await conn.fetchval( + """ + INSERT INTO raw_posts( + source_id, platform, external_post_id, external_owner_id, + original_url, raw_text, raw_json, text_hash, content_hash, + posted_at, status, skip_reason + ) + VALUES($1,$2,$3,$4,$5,$6,$7::jsonb,$8,$9,$10,$11,$12) + ON CONFLICT (source_id, external_post_id) DO NOTHING + RETURNING id + """, + source_id, + PLATFORM_VK, + external_post_id, + owner_id, + original_url, + raw_text, + json.dumps(post, ensure_ascii=False), + text_hash, + content_hash, + posted_at, + status, + skip_reason, + ) + if raw_post_id is None: + return None + + for m in media: + await conn.execute( + """ + INSERT INTO raw_post_media( + raw_post_id, platform, media_type, original_url, original_attachment_id, + width, height, duration_sec, sort_order + ) + VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9) + """, + raw_post_id, + PLATFORM_VK, + m.media_type, + m.original_url, + m.attachment_id, + m.width, + m.height, + m.duration_sec, + m.sort_order, + ) + + if create_storage_job: + await conn.execute( + """ + INSERT INTO jobs(type, entity_type, entity_id, payload_json, status) + VALUES($1, 'raw_post', $2, '{}'::jsonb, 'pending') + ON CONFLICT DO NOTHING + """, + JOB_TYPE_VK_STORAGE_COPY, + raw_post_id, + ) + return int(raw_post_id) + + async def parse_source(self, client: VKAPIClient, source: dict) -> int: + source = await self.resolve_source_if_needed(client, source) + source_id = int(source["id"]) + owner_id = int(source["external_owner_id"]) + page_size = max(1, min(100, await fetch_int_setting("vk_wall_page_size", 50))) + lookback_days = max(1, await fetch_int_setting("parser_new_source_lookback_days", 14)) + overlap_minutes = max(0, await fetch_int_setting("parser_reparse_overlap_minutes", 120)) + + last_parsed_at = source.get("last_parsed_at") + parse_from = source.get("parse_from") + if last_parsed_at: + since_dt = last_parsed_at - timedelta(minutes=overlap_minutes) + elif parse_from: + since_dt = parse_from + else: + since_dt = datetime.now(timezone.utc) - timedelta(days=lookback_days) + if since_dt.tzinfo is None: + since_dt = since_dt.replace(tzinfo=timezone.utc) + since_ts = int(since_dt.timestamp()) + + fetched: list[dict] = [] + offset = 0 + stop = False + while not stop: + response = await client.get_wall_posts(owner_id=owner_id, count=page_size, offset=offset) + items = response.get("items", []) or [] + if not items: + break + for post in items: + if is_deleted_or_invalid(post): + continue + if int(post["date"]) <= since_ts: + stop = True + break + fetched.append(post) + if len(items) < page_size: + break + offset += page_size + + known = await self.known_post_ids(source_id, [str(p["id"]) for p in fetched if p.get("id")]) + candidates = [p for p in fetched if str(p.get("id")) not in known] + saved = 0 + max_seen: datetime | None = None + min_text_length = max(0, await fetch_int_setting("parser_min_text_length", 0)) + skip_reposts = await fetch_bool_setting("parser_skip_reposts", True) + skip_empty_text = await fetch_bool_setting("parser_skip_empty_text", True) + skip_no_media = await fetch_bool_setting("parser_skip_no_media", True) + skip_short_text = await fetch_bool_setting("parser_skip_text_too_short", True) + store_skipped = await fetch_bool_setting("parser_store_skipped_posts", False) + dedupe_content_hash = await fetch_bool_setting("parser_dedupe_content_hash", True) + + hash_by_post_id: dict[str, str] = {} + if dedupe_content_hash: + candidate_hashes: list[str] = [] + for post in candidates: + if is_deleted_or_invalid(post) or (skip_reposts and is_repost(post)): + continue + media = extract_media(post) + media_ids = ",".join(m.attachment_id for m in media) + content_hash = make_hash(str(post.get("text") or "").strip(), media_ids) + hash_by_post_id[str(post["id"])] = content_hash + candidate_hashes.append(content_hash) + known_hashes = await self.known_content_hashes(candidate_hashes) + else: + known_hashes = set() + + for post in candidates: + if is_deleted_or_invalid(post): + continue + text = str(post.get("text") or "").strip() + posted_at = utc_from_ts(int(post["date"])) + if max_seen is None or posted_at > max_seen: + max_seen = posted_at + + if skip_reposts and is_repost(post): + continue + + media = extract_media(post) + content_hash = hash_by_post_id.get(str(post["id"])) + if dedupe_content_hash and content_hash in known_hashes: + continue + + skip_reason: str | None = None + if skip_empty_text and not text: + skip_reason = "empty_text" + elif skip_no_media and not media: + skip_reason = "no_media" + elif skip_short_text and len(text) < min_text_length: + skip_reason = "text_too_short" + + if skip_reason: + if not store_skipped: + continue + raw_id = await self.save_post( + source, + post, + status=POST_STATUS_SKIPPED, + skip_reason=skip_reason, + create_storage_job=False, + ) + else: + raw_id = await self.save_post( + source, + post, + status=POST_STATUS_STORAGE_PENDING, + skip_reason=None, + create_storage_job=True, + ) + if raw_id: + saved += 1 + if content_hash: + known_hashes.add(content_hash) + + await self.mark_source_ok(source_id, max_seen or last_parsed_at) + logger.info( + "Parsed source {}: fetched={} known={} candidates={} saved={}", + source.get("name"), + len(fetched), + len(known), + len(candidates), + saved, + ) + return saved + + async def run_once(self) -> None: + enabled = await is_worker_enabled(self.pool, WORKER_PARSER) + if not enabled: + await self.heartbeat.beat(self.pool, status="disabled", force=True) + return + + rps = await fetch_int_setting("vk_requests_per_second", 3) + timeout_total = await fetch_int_setting("vk_api_timeout_total_sec", 15) + timeout_connect = await fetch_int_setting("vk_api_timeout_connect_sec", 5) + rate_limit_sleep = await fetch_float_setting("vk_rate_limit_sleep_sec", 1.0) + retry_attempts = await fetch_int_setting("vk_api_retry_attempts", 3) + retry_min_delay = await fetch_float_setting("vk_api_retry_min_delay_sec", 2.0) + retry_max_delay = await fetch_float_setting("vk_api_retry_max_delay_sec", 10.0) + source_pause = max(0.0, await fetch_float_setting("parser_source_pause_sec", 0.0)) + sources = await self.active_sources() + await self.heartbeat.beat(self.pool, meta={"sources": len(sources)}) + if not sources: + logger.info("No active VK sources") + return + + logger.info( + "Parser cycle: sources={} rps={} timeout={}/{} retry={}/{}..{} rate_limit_sleep={} source_pause={}", + len(sources), + rps, + timeout_total, + timeout_connect, + retry_attempts, + retry_min_delay, + retry_max_delay, + rate_limit_sleep, + source_pause, + ) + + async with VKAPIClient( + rps=rps, + timeout_total_sec=timeout_total, + timeout_connect_sec=timeout_connect, + rate_limit_sleep_sec=rate_limit_sleep, + retry_attempts=retry_attempts, + retry_min_delay_sec=retry_min_delay, + retry_max_delay_sec=retry_max_delay, + ) as client: + for source in sources: + try: + await self.parse_source(client, source) + except VKAPIError as e: + if is_fatal_source_error(e): + await self.deactivate_source(int(source["id"]), str(e)) + logger.warning("VK source deactivated {}: {}", source.get("name"), e) + else: + await self.mark_source_error(int(source["id"]), str(e)) + logger.warning("VK source temporary error {}: {}", source.get("name"), e) + except Exception as e: + await self.mark_source_error(int(source["id"]), str(e)) + logger.exception("Unexpected source error {}: {}", source.get("name"), e) + if source_pause: + await asyncio.sleep(source_pause) + + async def run_loop(self) -> None: + await self.init() + logger.info("{} started", WORKER_PARSER) + while True: + try: + await self.run_once() + except Exception as e: + logger.exception("Parser loop error: {}", e) + interval = max(10, await fetch_int_setting("parser_interval_sec", 300)) + await asyncio.sleep(interval) + + +async def main() -> None: + logger.remove() + logger.add(lambda msg: print(msg, end=""), level=settings.log_level) + worker = VKParserWorker() + await worker.run_loop() + + +if __name__ == "__main__": + asyncio.run(main())