Track parser worker

.gitignore

@@ -7,9 +7,9 @@ __pycache__/
 logs/
 uploads/
 original_project/
-ai_worker.py
+/ai_worker.py
-media_worker.py
+/media_worker.py
-parser.py
+/parser.py
-raw_feed_worker.py
+/raw_feed_worker.py
 vkdev_main.js
 sample_photo.jpg

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())