fix: rebuild instagram parser integration

db/migrations/046_insta_parser_settings.sql

@@ -1,10 +1,17 @@
-INSERT INTO app_settings (key, label, value_type, value)
+INSERT INTO worker_controls(name, enabled, settings_json)
+VALUES ('insta-parser', FALSE, '{}'::jsonb)
+ON CONFLICT (name) DO NOTHING;
+
+INSERT INTO app_settings(key, value_json, value_type, title, description, category)
 VALUES
-('insta_login', 'Instagram Login', 'string', ''),
+  ('insta_login', '""'::jsonb, 'str', 'Instagram login', 'Username for the Instagram account used by instagrapi.', 'Instagram Parser'),
-('insta_password', 'Instagram Password', 'secret', ''),
+  ('insta_password', '""'::jsonb, 'secret', 'Instagram password', 'Password for the Instagram account used by instagrapi.', 'Instagram Parser'),
-('insta_proxy_url', 'Instagram Proxy URL (http://user:pass@ip:port)', 'string', ''),
+  ('insta_proxy_url', '""'::jsonb, 'str', 'Instagram proxy URL', 'Optional stable proxy, for example http://user:pass@host:port.', 'Instagram Parser'),
-('insta_fetch_count', 'Instagram: количество проверяемых последних постов', 'int', '5'),
+  ('insta_session_path', '"insta_session.json"'::jsonb, 'str', 'Instagram session path', 'Path to the persisted instagrapi session settings file.', 'Instagram Parser'),
-('insta_delay_base_minutes', 'Instagram: пауза между аккаунтами (минут)', 'int', '35'),
+  ('insta_fetch_count', '5'::jsonb, 'int', 'Posts to inspect', 'How many latest posts to inspect per account visit.', 'Instagram Parser'),
-('insta_delay_random_minutes', 'Instagram: разброс паузы (± минут)', 'int', '5'),
+  ('insta_delay_base_minutes', '35'::jsonb, 'int', 'Account visit delay, minutes', 'Base pause after checking one Instagram account.', 'Instagram Parser'),
-('insta_cooldown_hours', 'Instagram: отлежка при лимитах (часов)', 'int', '12')
+  ('insta_delay_random_minutes', '5'::jsonb, 'int', 'Delay random spread, minutes', 'Random +/- spread added to the base account visit delay.', 'Instagram Parser'),
+  ('insta_cooldown_hours', '12'::jsonb, 'int', 'Cooldown hours', 'Sleep window after challenge, login-required or rate-limit responses.', 'Instagram Parser'),
+  ('insta_request_pause_sec', '2'::jsonb, 'float', 'Request pause, sec', 'Small pause between Instagram user lookup and media fetch.', 'Instagram Parser'),
+  ('insta_dedupe_content_hash', 'true'::jsonb, 'bool', 'Dedupe by content hash', 'Skip Instagram posts whose text and media hash already exists.', 'Instagram Parser')
 ON CONFLICT (key) DO NOTHING;

src/vk_parser_app/admin.py

@@ -22,7 +22,7 @@ from fastapi.templating import Jinja2Templates
 from loguru import logger
 
 from .config import settings
-from .constants import PLATFORM_VK, PLATFORM_INSTAGRAM
+from .constants import PLATFORM_INSTAGRAM, PLATFORM_VK
 from .db import fetch_int_setting, fetch_setting, get_pool
 from .security import hash_password, new_token, token_hash, verify_password
 from .text_utils import build_publication_text, normalize_hash_tag, parse_categories
@@ -948,7 +948,7 @@ def parse_source_line(line: str) -> dict[str, str]:
         (
             i
             for i, part in enumerate(parts)
-            if any(domain in part for domain in ("vk.com/", "vk.ru/", "m.vk.com/"))
+            if any(domain in part for domain in ("vk.com/", "vk.ru/", "m.vk.com/", "instagram.com/"))
             or part.lower().startswith(("club", "public"))
         ),
         -1,
@@ -956,9 +956,9 @@ def parse_source_line(line: str) -> dict[str, str]:
     if url_index < 0:
         if len(parts) == 1:
             value = parts[0].strip()
-            url = value if value.startswith("http") else f"https://vk.com/{value}"
+            url = value if value.startswith("http") or value.startswith("@") else f"https://vk.com/{value}"
             return {"name": "", "tag": "", "url": url, "error": ""}
-        return {"name": "", "tag": "", "url": "", "error": "Не нашёл ссылку VK"}
+        return {"name": "", "tag": "", "url": "", "error": "Не нашёл ссылку"}
     url = parts[url_index].strip()
     before_url = parts[:url_index]
     if len(before_url) >= 2:
@@ -976,10 +976,47 @@ def parse_source_line(line: str) -> dict[str, str]:
     return {"name": name, "tag": normalize_hash_tag(tag, ""), "url": url, "error": ""}
 
 
+def normalize_instagram_source(value: str) -> tuple[str, str]:
+    raw = (value or "").strip()
+    if not raw:
+        return "", ""
+    if raw.startswith("@"):
+        username = raw[1:]
+        return username, f"https://www.instagram.com/{username}/"
+    if not raw.startswith(("http://", "https://")) and "instagram.com/" not in raw:
+        return raw, f"https://www.instagram.com/{raw}/"
+    if raw.startswith(("instagram.com/", "www.instagram.com/")):
+        raw = f"https://{raw}"
+    match = re.search(r"instagram\.com/([^/?#]+)/?", raw, re.IGNORECASE)
+    username = (match.group(1) if match else "").strip()
+    if username.lower() in {"p", "reel", "stories", "explore"}:
+        username = ""
+    url = f"https://www.instagram.com/{username}/" if username else raw
+    return username, url
+
+
+def source_platform_from_url(url: str) -> str:
+    value = (url or "").strip().lower()
+    return PLATFORM_INSTAGRAM if value.startswith("@") or "instagram.com/" in value else PLATFORM_VK
+
+
 async def build_sources_preview(lines_text: str, default_active: bool = True) -> list[dict[str, Any]]:
     lines = [line for line in (lines_text or "").splitlines() if line.strip()]
     parsed = [parse_source_line(line) for line in lines]
-    normalized_values = [normalize_vk_source(item["url"]) for item in parsed if item.get("url")]
+    normalized_values = []
+    normalized_urls = []
+    for item in parsed:
+        url = item.get("url") or ""
+        if not url:
+            continue
+        if source_platform_from_url(url) == PLATFORM_INSTAGRAM:
+            external_id, normalized_url = normalize_instagram_source(url)
+        else:
+            external_id = normalize_vk_source(url)
+            normalized_url = url
+        if external_id:
+            normalized_values.append(external_id)
+        normalized_urls.append(normalized_url)
     pool = await get_pool()
     existing_rows = await pool.fetch(
         """
@@ -989,7 +1026,7 @@ async def build_sources_preview(lines_text: str, default_active: bool = True) ->
           AND (lower(external_id)=ANY($1::text[]) OR lower(url)=ANY($2::text[]) OR lower(COALESCE(tag,''))=ANY($3::text[]))
         """,
         [v.lower() for v in normalized_values if v],
-        [str(item.get("url") or "").lower() for item in parsed],
+        [url.lower() for url in normalized_urls if url],
         [normalize_hash_tag(str(item.get("tag") or ""), "").lower() for item in parsed if item.get("tag")],
     )
     existing_ids = {str(row["external_id"] or "").lower() for row in existing_rows}
@@ -1003,10 +1040,14 @@ async def build_sources_preview(lines_text: str, default_active: bool = True) ->
         preview = []
         for idx, item in enumerate(parsed, start=1):
             url = item.get("url") or ""
-            external_id = normalize_vk_source(url) if url else ""
+            platform = source_platform_from_url(url)
+            if platform == PLATFORM_INSTAGRAM:
+                external_id, url = normalize_instagram_source(url)
+            else:
+                external_id = normalize_vk_source(url) if url else ""
             row = {
                 "line_no": idx,
-                "platform": PLATFORM_VK,
+                "platform": platform,
                 "name": item.get("name") or "",
                 "tag": normalize_hash_tag(item.get("tag") or external_id, external_id or "source"),
                 "url": url,
@@ -1026,8 +1067,13 @@ async def build_sources_preview(lines_text: str, default_active: bool = True) ->
             if not row["error"] and row["tag"].lower() in existing_tags:
                 row["error"] = "Такой тэг уже есть"
             if not row["error"] and not external_id:
-                row["error"] = "Не удалось разобрать VK-ссылку"
+                row["error"] = "Не удалось разобрать ссылку"
-            if not row["error"]:
+            if not row["error"] and platform == PLATFORM_INSTAGRAM:
+                row["name"] = row["name"] or external_id
+                row["ok"] = True
+                seen.add(key)
+                seen_tags.add(row["tag"].lower())
+            elif not row["error"]:
                 try:
                     screen_name, owner_id, resolved_name = await client.resolve_group(url)
                     row["external_id"] = screen_name
@@ -2318,16 +2364,7 @@ async def sources_bulk_create(
         if not isinstance(item, dict) or not item.get("ok"):
             continue
         try:
-            url_val = str(item.get("url") or "").strip()
+            platform = str(item.get("platform") or PLATFORM_VK).strip().lower() or PLATFORM_VK
-            external_id_val = str(item.get("external_id") or "").strip()
-            platform_val = PLATFORM_VK
-            if "instagram.com" in url_val:
-                platform_val = PLATFORM_INSTAGRAM
-                if not external_id_val:
-                    m = re.search(r"instagram\.com/([^/?#]+)", url_val)
-                    if m:
-                        external_id_val = m.group(1)
-
             row = await pool.fetchrow(
                 """
                 INSERT INTO sources(platform, name, tag, url, external_id, external_owner_id, active, priority, created_by)
@@ -2335,18 +2372,18 @@ async def sources_bulk_create(
                 ON CONFLICT DO NOTHING
                 RETURNING id
                 """,
-                platform_val,
+                platform,
-                str(item.get("name") or external_id_val or "").strip(),
+                str(item.get("name") or item.get("external_id") or "").strip(),
-                normalize_hash_tag(str(item.get("tag") or external_id_val or ""), external_id_val or "source"),
+                normalize_hash_tag(str(item.get("tag") or item.get("external_id") or ""), str(item.get("external_id") or "source")),
-                url_val,
+                str(item.get("url") or "").strip(),
-                external_id_val,
+                str(item.get("external_id") or "").strip(),
                 item.get("external_owner_id"),
                 bool(item.get("active", True)),
                 user["id"],
             )
             if row:
                 created += 1
-                await audit(user["id"], "source.create", "source", int(row["id"]), {"url": item.get("url"), "bulk": True})
+                await audit(user["id"], "source.create", "source", int(row["id"]), {"url": item.get("url"), "bulk": True, "platform": platform})
         except Exception:
             logger.exception("Bulk source insert failed: {}", item)
     return redirect(f"/sources?q=&status_filter=&created={created}")
@@ -2380,18 +2417,12 @@ async def source_create(
     require_csrf(user, csrf_token)
     platform = platform.strip().lower() or PLATFORM_VK
     resolved_url = url.strip()
-    
+    if platform == PLATFORM_VK and source_platform_from_url(resolved_url) == PLATFORM_INSTAGRAM:
-    if "instagram.com" in resolved_url and platform == PLATFORM_VK:
         platform = PLATFORM_INSTAGRAM
-        
+    if platform == PLATFORM_INSTAGRAM:
-    if platform == PLATFORM_VK:
+        external_id, resolved_url = normalize_instagram_source(resolved_url)
-        external_id = normalize_vk_source(resolved_url)
-    elif platform == PLATFORM_INSTAGRAM:
-        m = re.search(r"instagram\.com/([^/?#]+)", resolved_url)
-        external_id = m.group(1) if m else ""
     else:
-        external_id = ""
+        external_id = normalize_vk_source(resolved_url)
-
     external_owner_id = None
     resolved_name = name.strip()
     status_value = "new"
@@ -2472,7 +2503,13 @@ async def source_update(
         return redirect("/login")
     require_csrf(user, csrf_token)
     platform = platform.strip().lower() or PLATFORM_VK
-    external_id = normalize_vk_source(url) if platform == PLATFORM_VK else ""
+    resolved_url = url.strip()
+    if platform == PLATFORM_VK and source_platform_from_url(resolved_url) == PLATFORM_INSTAGRAM:
+        platform = PLATFORM_INSTAGRAM
+    if platform == PLATFORM_INSTAGRAM:
+        external_id, resolved_url = normalize_instagram_source(resolved_url)
+    else:
+        external_id = normalize_vk_source(resolved_url)
     pool = await get_pool()
     await pool.execute(
         """
@@ -2491,7 +2528,7 @@ async def source_update(
         platform,
         name.strip(),
         normalize_hash_tag(tag or external_id or name, external_id or "source"),
-        url.strip(),
+        resolved_url,
         external_id,
         active == "on",
         priority,

src/vk_parser_app/templates/source_form.html

@@ -17,6 +17,7 @@
           <label class="label"><span class="label-text font-bold">Площадка</span></label>
           <select name="platform" class="select select-bordered w-full">
             <option value="vk" {% if not source or source.platform == "vk" %}selected{% endif %}>VK</option>
+            <option value="instagram" {% if source and source.platform == "instagram" %}selected{% endif %}>Instagram</option>
           </select>
         </div>
         

src/vk_parser_app/workers/ai_alerts.py

@@ -83,23 +83,22 @@ async def send_parser_error_alert(parser_name: str, source_name: str, error: str
         return
 
     text = (
-        f"🚨 <b>CRITICAL PARSER ERROR</b>\n"
+        "Parser failed\n"
-        f"Parser: {parser_name}\n"
+        f"parser: {parser_name}\n"
-        f"Source: {source_name}\n\n"
+        f"source: {source_name or '-'}\n"
-        f"Error: {error[:1000]}"
+        f"error: {error[:1000]}"
     )
     bot = Bot(token=token)
     try:
         for recipient_id in recipients:
             while True:
                 try:
-                    await bot.send_message(recipient_id, text, disable_web_page_preview=True, parse_mode="HTML")
+                    await bot.send_message(recipient_id, text, disable_web_page_preview=True)
                     break
                 except TelegramRetryAfter as exc:
                     await asyncio.sleep(float(exc.retry_after) + 1)
-                except Exception as e:
+                except Exception as exc:
-                    logger.error("Failed to send parser alert to {}: {}", recipient_id, e)
+                    logger.error("Failed to send parser alert to {}: {}", recipient_id, exc)
                     break
     finally:
         await bot.session.close()
-

src/vk_parser_app/workers/insta_parser.py

@@ -4,462 +4,480 @@ import asyncio
 import hashlib
 import json
 import random
-import time
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-
+from typing import Any
-try:
-    from instagrapi import Client
-    from instagrapi.exceptions import (
-        ChallengeRequired,
-        PleaseWaitFewMinutes,
-        LoginRequired,
-        UserNotFound,
-        PrivateAccount
-    )
-except ImportError:
-    Client = None
-    ChallengeRequired = Exception
-    PleaseWaitFewMinutes = Exception
-    LoginRequired = Exception
-    UserNotFound = Exception
-    PrivateAccount = Exception
 
 from loguru import logger
 
-from ..config import settings
 from ..constants import (
+    JOB_TYPE_VK_STORAGE_COPY,
+    MEDIA_STATUS_LINK_ONLY,
+    MEDIA_STATUS_PENDING,
     PLATFORM_INSTAGRAM,
-    POST_STATUS_STORAGE_PENDING,
     POST_STATUS_SKIPPED,
+    POST_STATUS_STORAGE_PENDING,
     SOURCE_STATUS_ERROR,
     SOURCE_STATUS_OK,
     WORKER_INSTA_PARSER,
 )
-from ..db import fetch_setting, fetch_int_setting, get_pool
+from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
 from ..heartbeat import HeartbeatReporter
+from ..jobs import is_worker_enabled
 from .ai_alerts import send_parser_error_alert
 
 
-def make_content_hash(text: str) -> str:
+def make_hash(*parts: str) -> str:
-    return hashlib.sha256((text or "").strip().lower().encode()).hexdigest()
+    h = hashlib.sha256()
+    for part in parts:
+        h.update((part or "").strip().lower().encode("utf-8"))
+        h.update(b"\0")
+    return h.hexdigest()
+
+
+def as_dict(value: Any) -> dict:
+    if isinstance(value, dict):
+        return value
+    if hasattr(value, "model_dump"):
+        return value.model_dump()
+    if hasattr(value, "dict"):
+        return value.dict()
+    return {}
+
+
+def media_value(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def media_code(post: dict) -> str:
+    return str(post.get("code") or post.get("pk") or "").strip()
+
+
+def original_url(post: dict) -> str:
+    code = media_code(post)
+    return f"https://www.instagram.com/p/{code}/" if code else ""
+
+
+def media_taken_at(post: dict) -> datetime:
+    value = post.get("taken_at")
+    if isinstance(value, datetime):
+        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    return datetime.now(tz=timezone.utc)
+
+
+def media_user_pk(post: dict) -> int | None:
+    user = as_dict(post.get("user"))
+    try:
+        return int(user.get("pk") or 0) or None
+    except Exception:
+        return None
+
+
+def extract_instagram_media(post: dict) -> list[dict]:
+    items: list[dict] = []
+    resources = [as_dict(item) for item in post.get("resources") or []]
+    if not resources:
+        resources = [post]
+
+    for index, item in enumerate(resources):
+        media_type = int(item.get("media_type") or post.get("media_type") or 0)
+        if media_type == 1:
+            url = media_value(item.get("thumbnail_url") or post.get("thumbnail_url"))
+            kind = "photo"
+        elif media_type == 2:
+            url = media_value(item.get("video_url") or post.get("video_url"))
+            kind = "video"
+        else:
+            continue
+        if not url:
+            continue
+        items.append(
+            {
+                "media_type": kind,
+                "original_url": url,
+                "original_attachment_id": str(item.get("pk") or post.get("pk") or index),
+                "width": item.get("width") or post.get("width"),
+                "height": item.get("height") or post.get("height"),
+                "duration_sec": int(float(item.get("video_duration") or post.get("video_duration") or 0)) or None,
+                "sort_order": index,
+            }
+        )
+    return items
+
+
+class NeverRaised(Exception):
+    pass
+
+
+def load_instagrapi():
+    from instagrapi import Client
+    import instagrapi.exceptions as exc
+
+    return Client, {
+        "challenge": getattr(exc, "ChallengeRequired", NeverRaised),
+        "login_required": getattr(exc, "LoginRequired", getattr(exc, "ClientLoginRequired", NeverRaised)),
+        "please_wait": getattr(exc, "PleaseWaitFewMinutes", NeverRaised),
+        "user_not_found": getattr(exc, "UserNotFound", NeverRaised),
+        "private_account": getattr(exc, "PrivateAccount", NeverRaised),
+    }
 
 
 class InstaParserWorker:
-    def __init__(self):
+    def __init__(self) -> None:
         self.pool = None
-        self.heartbeat_interval_sec = 60
+        self.heartbeat = HeartbeatReporter(WORKER_INSTA_PARSER, 30)
-        self.heartbeat = HeartbeatReporter("insta-parser", 60)
+        self.client = None
-        self.heartbeat_task: asyncio.Task | None = None
+        self.exceptions: dict[str, type[BaseException]] = {}
-        self.client: Client | None = None
-        self.session_file = Path("insta_session.json")
 
-    async def init(self):
+    async def init(self) -> None:
         self.pool = await get_pool()
 
-    async def get_active_source(self) -> dict | None:
+    async def active_source(self) -> dict | None:
-        async with self.pool.acquire() as conn:
+        row = await self.pool.fetchrow(
-            row = await conn.fetchrow(
+            """
-                """
+            SELECT *
-                SELECT id, external_id, name, last_checked_at, last_parsed_at
+            FROM sources
-                FROM sources
+            WHERE platform=$1
-                WHERE platform = $1 AND active = TRUE
+              AND active=TRUE
-                ORDER BY last_checked_at NULLS FIRST
+              AND archived_at IS NULL
-                LIMIT 1
+            ORDER BY last_checked_at NULLS FIRST, priority ASC, id ASC
-                """,
+            LIMIT 1
-                PLATFORM_INSTAGRAM
+            """,
-            )
+            PLATFORM_INSTAGRAM,
+        )
         return dict(row) if row else None
 
-    async def get_known_hashes(self, hashes: list[str]) -> set[str]:
+    async def mark_source_error(self, source_id: int, message: str) -> None:
-        if not hashes:
+        await self.pool.execute(
-            return set()
+            """
-        async with self.pool.acquire() as conn:
+            UPDATE sources
-            rows = await conn.fetch(
+            SET status=$2,
-                """
+                status_msg=$3,
-                SELECT content_hash
+                last_checked_at=NOW(),
-                FROM posts
+                updated_at=NOW()
-                WHERE content_hash = ANY($1::text[])
+            WHERE id=$1
-                """,
+            """,
-                hashes,
+            source_id,
-            )
+            SOURCE_STATUS_ERROR,
-        return {str(r["content_hash"]) for r in rows}
+            message[:1000],
+        )
 
-    async def get_known_post_ids(self, source_id: int, external_post_ids: list[str]) -> set[str]:
+    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 known_post_ids(self, source_id: int, external_post_ids: list[str]) -> set[str]:
         if not external_post_ids:
             return set()
-        async with self.pool.acquire() as conn:
+        rows = await self.pool.fetch(
-            try:
+            """
-                numeric_ids = [int(pk) for pk in external_post_ids]
+            SELECT external_post_id
-                rows = await conn.fetch(
+            FROM raw_posts
-                    """
+            WHERE source_id=$1 AND external_post_id=ANY($2::text[])
-                    SELECT vk_post_id
+            """,
-                    FROM posts
+            source_id,
-                    WHERE source_id = $1 AND vk_post_id = ANY($2::bigint[])
+            external_post_ids,
-                    """,
+        )
-                    source_id,
+        return {str(row["external_post_id"]) for row in rows}
-                    numeric_ids,
-                )
-                return {str(r["vk_post_id"]) for r in rows}
-            except ValueError:
-                return set()
 
-    async def deactivate_source(self, source_id: int, reason: str) -> None:
+    async def known_content_hashes(self, hashes: list[str]) -> set[str]:
-        async with self.pool.acquire() as conn:
+        if not hashes:
-            await conn.execute(
+            return set()
-                """
+        rows = await self.pool.fetch(
-                UPDATE sources
+            """
-                SET active = FALSE,
+            SELECT content_hash
-                    status = $1,
+            FROM raw_posts
-                    status_msg = $2,
+            WHERE content_hash=ANY($1::text[])
-                    last_checked_at = NOW(),
+            """,
-                    updated_at = NOW()
+            hashes,
-                WHERE id = $3
+        )
-                """,
+        return {str(row["content_hash"]) for row in rows}
-                SOURCE_STATUS_ERROR,
-                reason[:1000],
-                source_id,
-            )
 
-    async def mark_source_ok(self, source_id: int, last_parsed_at: datetime | None = None) -> None:
+    async def set_cooldown(self, hours: int, reason: str) -> None:
-        async with self.pool.acquire() as conn:
+        until = datetime.now(tz=timezone.utc) + timedelta(hours=max(1, hours))
-            await conn.execute(
+        await self.pool.execute(
-                """
+            """
-                UPDATE sources
+            INSERT INTO app_settings(key, value_json, value_type, title, description, category)
-                SET status = $1,
+            VALUES('insta_cooldown_until', $1::jsonb, 'str', 'Cooldown until', $2, 'Instagram Parser')
-                    status_msg = NULL,
+            ON CONFLICT (key) DO UPDATE
-                    last_checked_at = NOW(),
+            SET value_json=$1::jsonb,
-                    last_parsed_at = COALESCE($2, last_parsed_at),
+                description=$2,
-                    updated_at = NOW()
+                updated_at=NOW()
-                WHERE id = $3
+            """,
-                """,
+            json.dumps(until.isoformat()),
-                SOURCE_STATUS_OK,
+            reason[:1000],
-                last_parsed_at,
+        )
-                source_id,
-            )
 
-    async def save_cooldown(self, hours: int) -> None:
+    async def cooldown_until(self) -> datetime | None:
-        until = datetime.now(tz=timezone.utc) + timedelta(hours=hours)
+        value = await fetch_setting("insta_cooldown_until", "")
-        async with self.pool.acquire() as conn:
+        if not value:
-            await conn.execute(
+            return None
-                """
+        try:
-                INSERT INTO app_settings (key, value, value_type)
+            dt = datetime.fromisoformat(str(value))
-                VALUES ('insta_cooldown_until', $1, 'string')
+            return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
-                ON CONFLICT (key) DO UPDATE SET value = $1
+        except ValueError:
-                """,
+            return None
-                until.isoformat()
-            )
 
-    async def get_cooldown_until(self) -> datetime | None:
+    def setup_client(self, login: str, password: str, proxy: str, session_path: str) -> None:
-        async with self.pool.acquire() as conn:
+        if self.client is not None:
-            val = await conn.fetchval("SELECT value FROM app_settings WHERE key = 'insta_cooldown_until'")
+            return
-        if val:
+        Client, exceptions = load_instagrapi()
-            try:
+        self.exceptions = exceptions
-                dt = datetime.fromisoformat(val)
+        client = Client()
-                if dt.tzinfo is None:
+        if proxy:
-                    dt = dt.replace(tzinfo=timezone.utc)
+            client.set_proxy(proxy)
-                return dt
+        path = Path(session_path or "insta_session.json")
-            except ValueError:
+        if path.exists():
-                pass
+            client.load_settings(path)
-        return None
+        client.login(login, password)
+        client.dump_settings(path)
+        self.client = client
 
-    async def save_post_and_media(self, source_id: int, post: dict, status: str, skip_reason: str | None) -> int | None:
+    async def save_post(self, source: dict, post: dict, status: str, skip_reason: str | None, media: list[dict]) -> int | None:
-        text = (post.get("caption_text") or "").strip()
+        source_id = int(source["id"])
-        content_hash = make_content_hash(text)
+        external_post_id = str(post["pk"])
-        posted_at = post.get("taken_at")
+        raw_text = str(post.get("caption_text") or "").strip()
-        if not posted_at:
+        media_ids = ",".join(item["original_attachment_id"] for item in media)
-            posted_at = datetime.now(tz=timezone.utc)
+        text_hash = make_hash(raw_text)
-        elif posted_at.tzinfo is None:
+        content_hash = make_hash(raw_text, media_ids)
-            posted_at = posted_at.replace(tzinfo=timezone.utc)
+        has_downloadable_media = any(item["media_type"] == "photo" for item in media)
-
-        pk = int(post["pk"])
-        
-        photos = []
-        videos = []
-        
-        media_type = post.get("media_type")
-        if media_type == 1:
-            url = post.get("thumbnail_url")
-            if url: photos.append(url)
-        elif media_type == 2:
-            url = post.get("video_url")
-            if url: videos.append(url)
-        elif media_type == 8:
-            for resource in post.get("resources", []):
-                if resource.get("media_type") == 1:
-                    photos.append(resource.get("thumbnail_url"))
-                elif resource.get("media_type") == 2:
-                    videos.append(resource.get("video_url"))
 
         async with self.pool.acquire() as conn:
             async with conn.transaction():
-                post_id = await conn.fetchval(
+                raw_post_id = await conn.fetchval(
                     """
-                    INSERT INTO posts (
+                    INSERT INTO raw_posts(
-                        source_id, vk_post_id, vk_owner_id, posted_at,
+                      source_id, platform, external_post_id, external_owner_id,
-                        raw_text, raw_json, content_hash,
+                      original_url, raw_text, raw_json, text_hash, content_hash,
-                        status, skip_reason
+                      posted_at, status, skip_reason
                     )
-                    VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8::post_status,$9)
+                    VALUES($1,$2,$3,$4,$5,$6,$7::jsonb,$8,$9,$10,$11,$12)
-                    ON CONFLICT (source_id, vk_post_id) DO NOTHING
+                    ON CONFLICT (source_id, external_post_id) DO NOTHING
                     RETURNING id
                     """,
                     source_id,
-                    pk,
+                    PLATFORM_INSTAGRAM,
-                    int(post.get("user", {}).get("pk") or 0),
+                    external_post_id,
-                    posted_at,
+                    media_user_pk(post),
-                    text,
+                    original_url(post),
+                    raw_text,
                     json.dumps(post, default=str, ensure_ascii=False),
+                    text_hash,
                     content_hash,
+                    media_taken_at(post),
                     status,
                     skip_reason,
                 )
-                if post_id is None:
+                if raw_post_id is None:
                     return None
 
-                for p_url in photos:
+                for item in media:
-                    if not p_url: continue
                     await conn.execute(
                         """
-                        INSERT INTO post_media (post_id, media_type, vk_url)
+                        INSERT INTO raw_post_media(
-                        VALUES ($1, 'photo', $2)
+                          raw_post_id, platform, media_type, original_url, original_attachment_id,
+                          width, height, duration_sec, sort_order, status, error
+                        )
+                        VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
                         """,
-                        post_id,
+                        raw_post_id,
-                        str(p_url),
+                        PLATFORM_INSTAGRAM,
+                        item["media_type"],
+                        item["original_url"],
+                        item["original_attachment_id"],
+                        item.get("width"),
+                        item.get("height"),
+                        item.get("duration_sec"),
+                        item["sort_order"],
+                        MEDIA_STATUS_LINK_ONLY if item["media_type"] == "video" else MEDIA_STATUS_PENDING,
+                        "instagram video link only" if item["media_type"] == "video" else None,
                     )
 
-                for v_url in videos:
+                if status == POST_STATUS_STORAGE_PENDING and has_downloadable_media:
-                    if not v_url: continue
                     await conn.execute(
                         """
-                        INSERT INTO post_media (post_id, media_type, vk_url)
+                        INSERT INTO jobs(type, entity_type, entity_id, payload_json, status)
-                        VALUES ($1, 'video', $2)
+                        VALUES($1, 'raw_post', $2, '{}'::jsonb, 'pending')
-                        """,
-                        post_id,
-                        str(v_url),
-                    )
-
-                if status == POST_STATUS_STORAGE_PENDING:
-                    await conn.execute(
-                        """
-                        INSERT INTO jobs (job_type, entity_type, entity_id, payload, status)
-                        VALUES ('media.process', 'post', $1, '{}'::jsonb, 'PENDING')
                         ON CONFLICT DO NOTHING
                         """,
-                        post_id,
+                        JOB_TYPE_VK_STORAGE_COPY,
+                        raw_post_id,
                     )
-        return post_id
+        return int(raw_post_id)
 
-    def setup_client(self, login: str, password: str, proxy: str | None) -> bool:
+    async def parse_source(self, source: dict, fetch_count: int, request_pause_sec: float) -> int:
-        if self.client:
+        username = str(source.get("external_id") or "").strip().lstrip("@")
-            return True
+        if not username:
+            await self.mark_source_error(int(source["id"]), "empty instagram username")
+            return 0
 
-        if Client is None:
+        user_id = await asyncio.to_thread(self.client.user_id_from_username, username)
-            logger.error("instagrapi is not installed! Cannot setup client.")
+        if request_pause_sec:
-            raise Exception("Module instagrapi is not installed. Please install it.")
+            await asyncio.sleep(request_pause_sec)
+        medias = await asyncio.to_thread(self.client.user_medias, user_id, fetch_count)
+        posts = [as_dict(item) for item in medias]
+        if not posts:
+            await self.mark_source_ok(int(source["id"]), None)
+            return 0
 
-        self.client = Client()
+        known_ids = await self.known_post_ids(int(source["id"]), [str(post.get("pk")) for post in posts if post.get("pk")])
-        if proxy:
+        candidates = [post for post in posts if str(post.get("pk")) not in known_ids]
-            self.client.set_proxy(proxy)
 
-        if self.session_file.exists():
+        dedupe_content_hash = await fetch_bool_setting("insta_dedupe_content_hash", True)
-            try:
+        hash_by_pk: dict[str, str] = {}
-                self.client.load_settings(self.session_file)
+        if dedupe_content_hash:
-                self.client.get_timeline_feed()
+            hashes = []
-                logger.info("Instagram session loaded successfully")
+            for post in candidates:
-                return True
+                media = extract_instagram_media(post)
-            except Exception as e:
+                media_ids = ",".join(item["original_attachment_id"] for item in media)
-                logger.warning(f"Session invalid, will re-login: {e}")
+                content_hash = make_hash(str(post.get("caption_text") or "").strip(), media_ids)
-                self.session_file.unlink(missing_ok=True)
+                hash_by_pk[str(post["pk"])] = content_hash
-        
+                hashes.append(content_hash)
-        logger.info(f"Logging into Instagram as {login}...")
+            known_hashes = await self.known_content_hashes(hashes)
-        try:
+        else:
-            self.client.login(login, password)
+            known_hashes = set()
-            self.client.dump_settings(self.session_file)
-            logger.info("Instagram login successful, session saved")
-            return True
-        except ChallengeRequired as e:
-            logger.error(f"Challenge Required during login: {e}")
-            raise
-        except Exception as e:
-            logger.error(f"Failed to login: {e}")
-            self.client = None
-            raise
-
-    async def parse_single_source(self, source: dict, fetch_count: int, min_text_length: int) -> None:
-        source_id = source["id"]
-        username = source["external_id"]
-        
-        logger.info(f"Parsing Instagram source: {username}")
-        
-        try:
-            user_id = await asyncio.to_thread(self.client.user_id_from_username, username)
-            medias = await asyncio.to_thread(self.client.user_medias, user_id, fetch_count)
-        except (UserNotFound, PrivateAccount) as e:
-            logger.warning(f"Source {username} is inaccessible: {e}")
-            await self.deactivate_source(source_id, str(e))
-            return
-        except Exception as e:
-            logger.error(f"Error fetching {username}: {e}")
-            raise
-
-        if not medias:
-            await self.mark_source_ok(source_id)
-            return
-
-        candidates = [m.model_dump() for m in medias]
-        pks = [str(m["pk"]) for m in candidates]
-        
-        known_pks = await self.get_known_post_ids(source_id, pks)
-        candidates = [c for c in candidates if str(c["pk"]) not in known_pks]
-        
-        candidate_hashes = [make_content_hash(c.get("caption_text") or "") for c in candidates]
-        known_hashes = await self.get_known_hashes(candidate_hashes)
 
+        min_text_length = max(0, await fetch_int_setting("parser_min_text_length", 0))
+        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)
         saved = 0
-        max_seen_dt = None
+        max_seen: datetime | None = None
 
         for post in candidates:
-            try:
+            posted_at = media_taken_at(post)
-                post_dt = post.get("taken_at")
+            if max_seen is None or posted_at > max_seen:
-                if post_dt:
+                max_seen = posted_at
-                    if post_dt.tzinfo is None:
+            content_hash = hash_by_pk.get(str(post.get("pk")))
-                        post_dt = post_dt.replace(tzinfo=timezone.utc)
+            if dedupe_content_hash and content_hash in known_hashes:
-                    if max_seen_dt is None or post_dt > max_seen_dt:
+                continue
-                        max_seen_dt = post_dt
 
-                text = (post.get("caption_text") or "").strip()
+            text = str(post.get("caption_text") or "").strip()
-                content_hash = make_content_hash(text)
+            media = extract_instagram_media(post)
+            skip_reason = 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 content_hash in known_hashes:
+            if skip_reason and not store_skipped:
-                    continue
+                continue
+            status = POST_STATUS_SKIPPED if skip_reason else POST_STATUS_STORAGE_PENDING
+            raw_id = await self.save_post(source, post, status, skip_reason, media)
+            if raw_id:
+                saved += 1
+                if content_hash:
+                    known_hashes.add(content_hash)
 
-                media_type = post.get("media_type")
+        await self.mark_source_ok(int(source["id"]), max_seen or source.get("last_parsed_at"))
-                has_media = media_type in (1, 2, 8)
+        logger.info("Parsed Instagram source {}: fetched={} known={} saved={}", username, len(posts), len(known_ids), saved)
+        return saved
 
-                if not has_media:
+    async def run_once(self) -> None:
-                    if await self.save_post_and_media(source_id, post, POST_STATUS_SKIPPED, "no_media"):
+        enabled = await is_worker_enabled(self.pool, WORKER_INSTA_PARSER)
-                        saved += 1
+        if not enabled:
-                    continue
+            await self.heartbeat.beat(self.pool, status="disabled", force=True)
+            return
 
-                if len(text) < min_text_length:
+        until = await self.cooldown_until()
-                    if await self.save_post_and_media(source_id, post, POST_STATUS_SKIPPED, "text_too_short"):
+        if until and until > datetime.now(tz=timezone.utc):
-                        saved += 1
+            await self.heartbeat.beat(self.pool, status="cooldown", meta={"until": until.isoformat()})
-                    continue
+            return
 
-                if await self.save_post_and_media(source_id, post, POST_STATUS_STORAGE_PENDING, None):
+        login = str(await fetch_setting("insta_login", "") or "").strip()
-                    saved += 1
+        password = str(await fetch_setting("insta_password", "") or "").strip()
-            except Exception as e:
+        proxy = str(await fetch_setting("insta_proxy_url", "") or "").strip()
-                logger.error(f"Failed to save insta post {post.get('pk')}: {e}")
+        session_path = str(await fetch_setting("insta_session_path", "insta_session.json") or "").strip()
-
+        fetch_count = max(1, min(20, await fetch_int_setting("insta_fetch_count", 5)))
-        await self.mark_source_ok(source_id, max_seen_dt)
+        cooldown_hours = max(1, await fetch_int_setting("insta_cooldown_hours", 12))
-        logger.info(f"Source {username}: saved {saved} posts")
+        request_pause_sec = max(0.0, await fetch_float_setting("insta_request_pause_sec", 2.0))
-
-    async def run_once(self):
-        login = str(await fetch_setting("insta_login", "")).strip()
-        password = str(await fetch_setting("insta_password", "")).strip()
-        proxy = str(await fetch_setting("insta_proxy_url", "")).strip() or None
-        fetch_count = await fetch_int_setting("insta_fetch_count", 5)
-        delay_base = await fetch_int_setting("insta_delay_base_minutes", 35)
-        delay_random = await fetch_int_setting("insta_delay_random_minutes", 5)
-        cooldown_hours = await fetch_int_setting("insta_cooldown_hours", 12)
-        min_text_length = await fetch_int_setting("min_text_length", settings.min_text_length)
 
         if not login or not password:
-            logger.warning("Instagram login/password not set in app_settings. Skipping.")
+            await self.heartbeat.beat(self.pool, status="missing_credentials")
-            return
-
-        cooldown_until = await self.get_cooldown_until()
-        if cooldown_until and cooldown_until > datetime.now(tz=timezone.utc):
-            logger.info(f"Insta parser is in cooldown until {cooldown_until}. Skipping.")
             return
 
         try:
-            await asyncio.to_thread(self.setup_client, login, password, proxy)
+            await asyncio.to_thread(self.setup_client, login, password, proxy, session_path)
-        except ChallengeRequired as e:
+        except Exception as exc:
-            error_msg = f"Challenge Required! Cannot login. Need manual verification.\n{e}"
+            await self.set_cooldown(cooldown_hours, f"login failed: {exc}")
-            logger.error(error_msg)
+            await send_parser_error_alert(WORKER_INSTA_PARSER, "login", str(exc))
-            await send_parser_error_alert("Instagram Parser", "Login Process", error_msg)
-            await self.save_cooldown(cooldown_hours)
-            return
-        except Exception as e:
-            logger.error(f"Login setup failed: {e}")
             return
 
-        source = await self.get_active_source()
+        source = await self.active_source()
         if not source:
-            logger.debug("No active instagram sources found.")
+            await self.heartbeat.beat(self.pool, status="idle")
             return
 
         try:
-            await self.parse_single_source(source, fetch_count, min_text_length)
+            await self.heartbeat.beat(self.pool, status="running", meta={"source": source.get("external_id")})
+            await self.parse_source(source, fetch_count, request_pause_sec)
+        except (self.exceptions.get("user_not_found", NeverRaised), self.exceptions.get("private_account", NeverRaised)) as exc:
+            await self.deactivate_source(int(source["id"]), str(exc))
+            await send_parser_error_alert(WORKER_INSTA_PARSER, str(source.get("external_id") or ""), str(exc))
+        except (
+            self.exceptions.get("challenge", NeverRaised),
+            self.exceptions.get("login_required", NeverRaised),
+            self.exceptions.get("please_wait", NeverRaised),
+        ) as exc:
+            self.client = None
+            await self.mark_source_error(int(source["id"]), str(exc))
+            await self.set_cooldown(cooldown_hours, str(exc))
+            await send_parser_error_alert(WORKER_INSTA_PARSER, str(source.get("external_id") or ""), str(exc))
+        except Exception as exc:
+            await self.mark_source_error(int(source["id"]), str(exc))
+            logger.exception("Unexpected Instagram source error {}: {}", source.get("external_id"), exc)
 
-            delay = delay_base * 60 + random.uniform(-delay_random * 60, delay_random * 60)
+    async def run_loop(self) -> None:
-            delay = max(60, delay)
+        await self.init()
-            logger.info(f"Sleeping for {delay/60:.1f} minutes before next parse.")
+        logger.info("{} started", WORKER_INSTA_PARSER)
-            await asyncio.sleep(delay)
-            
-        except ChallengeRequired as e:
-            error_msg = f"Challenge Required during parsing!\n{e}"
-            logger.error(error_msg)
-            await send_parser_error_alert("Instagram Parser", source.get("external_id", "Unknown"), error_msg)
-            await self.save_cooldown(cooldown_hours)
-        except PleaseWaitFewMinutes as e:
-            error_msg = f"Rate limit hit! Cooling down.\n{e}"
-            logger.warning(error_msg)
-            await send_parser_error_alert("Instagram Parser", source.get("external_id", "Unknown"), error_msg)
-            await self.save_cooldown(cooldown_hours)
-        except Exception as e:
-            logger.error(f"Unexpected error parsing {source.get('external_id')}: {e}")
-            await asyncio.sleep(60)
-
-    async def _heartbeat_loop(self) -> None:
         while True:
             try:
-                await self.heartbeat.beat(self.pool, {"state": "polling"})
+                await self.run_once()
-            except Exception as e:
+            except Exception as exc:
-                logger.warning(f"Insta-parser heartbeat failed: {e}")
+                logger.exception("Instagram parser loop error: {}", exc)
-            await asyncio.sleep(self.heartbeat_interval_sec)
+            base = max(1, await fetch_int_setting("insta_delay_base_minutes", 35))
-
+            spread = max(0, await fetch_int_setting("insta_delay_random_minutes", 5))
-    async def run_loop(self):
+            delay = max(60.0, base * 60 + random.uniform(-spread * 60, spread * 60))
-        await self.init()
+            await asyncio.sleep(delay)
-        logger.info("Instagram Parser started")
-        await self.heartbeat.beat(self.pool, {"state": "started"}, force=True)
-        self.heartbeat_task = asyncio.create_task(self._heartbeat_loop())
-
-        try:
-            while True:
-                try:
-                    await self.heartbeat.beat(self.pool, {"state": "running"})
-                    await self.run_once()
-                except Exception as e:
-                    logger.exception(f"Insta-parser loop error: {e}")
-                
-                await asyncio.sleep(10)
-        finally:
-            if self.heartbeat_task and not self.heartbeat_task.done():
-                self.heartbeat_task.cancel()
-                try:
-                    await self.heartbeat_task
-                except asyncio.CancelledError:
-                    pass
 
 
-async def main():
+async def main() -> None:
+    logger.remove()
+    logger.add(lambda msg: print(msg, end=""))
     worker = InstaParserWorker()
     await worker.run_loop()
 
 
 if __name__ == "__main__":
-    import sys
-    logger.remove()
-    logger.add(
-        sys.stdout,
-        level=settings.log_level,
-        format=(
-            "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
-            "<level>{level: <8}</level> | "
-            "<cyan>{name}</cyan>:<cyan>{line}</cyan> — <level>{message}</level>"
-        ),
-    )
     asyncio.run(main())