fix: keep instagrapi compatible with pydantic stack

STATE.md

@@ -185,6 +185,9 @@ As of 2026-07-29:
 - Prefer small targeted file reads with `rg` and narrow ranges.
 - Deployment status polling should be sparse. Coolify rebuilds can take 7-11 minutes
   because the Docker build currently runs without cache and reinstalls system deps.
+- LXC 105 Docker cleanup: old app images can be removed safely because rollback
+  is by git SHA + rebuild. Use `scripts/coolify_docker_cleanup.sh`; it removes
+  only unused Coolify app images for FN-8/RAA and never prunes Docker volumes.
 - Avoid dumping long post texts from DB unless explicitly needed.
 - For runtime checks, use the current Coolify/Gitea/LXC 105 path from `D:\DEVELOPMENT\.infra.md`.
 - Do not `git reset --hard` or revert user/server hotfixes.

db/migrations/046_insta_parser_settings.sql

@@ -0,0 +1,18 @@
+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', '""'::jsonb, 'str', 'Instagram login', 'Username for the Instagram account used by instagrapi.', 'Instagram Parser'),
+  ('insta_password', '""'::jsonb, 'secret', 'Instagram password', 'Password for the Instagram account used by instagrapi.', 'Instagram Parser'),
+  ('insta_proxy_url', '""'::jsonb, 'str', 'Instagram proxy URL', 'Optional stable proxy, for example http://user:pass@host:port.', 'Instagram Parser'),
+  ('insta_session_path', '"insta_session.json"'::jsonb, 'str', 'Instagram session path', 'Path to the persisted instagrapi session settings file.', 'Instagram Parser'),
+  ('insta_auth_status', '""'::jsonb, 'str', 'Instagram auth status', 'Last manual Instagram login result.', 'Instagram Parser'),
+  ('insta_fetch_count', '5'::jsonb, 'int', 'Posts to inspect', 'How many latest posts to inspect per account visit.', 'Instagram Parser'),
+  ('insta_delay_base_minutes', '35'::jsonb, 'int', 'Account visit delay, minutes', 'Base pause after checking one Instagram account.', 'Instagram Parser'),
+  ('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;

db/migrations/047_insta_manual_auth.sql

@@ -0,0 +1,12 @@
+INSERT INTO app_settings(key, value_json, value_type, title, description, category)
+VALUES
+  ('insta_auth_status', '""'::jsonb, 'str', 'Instagram auth status', 'Last manual Instagram login result.', 'Instagram Parser')
+ON CONFLICT (key) DO NOTHING;
+
+UPDATE app_settings
+SET description='Used only by the manual Instagram login button. The parser reads the saved session file and does not auto-login.'
+WHERE key='insta_login';
+
+UPDATE app_settings
+SET description='Used only by the manual Instagram login button. The parser reads the saved session file and does not auto-login.'
+WHERE key='insta_password';

requirements.txt

@@ -10,3 +10,4 @@ Pillow==11.3.0
 python-multipart==0.0.20
 uvicorn[standard]==0.35.0
 yt-dlp==2026.6.9
+instagrapi==2.1.2

scripts/coolify_docker_cleanup.sh

@@ -0,0 +1,44 @@
+#!/bin/sh
+set -eu
+
+# Safe cleanup for the Coolify Docker destination.
+# Removes old app images that are not used by running containers.
+# Volumes are intentionally never pruned: they hold Postgres/app data.
+
+KEEP_PER_REPO="${KEEP_PER_REPO:-1}"
+CONTAINER_UNTIL="${CONTAINER_UNTIL:-24h}"
+BUILDER_UNTIL="${BUILDER_UNTIL:-24h}"
+
+if [ "$#" -gt 0 ]; then
+  REPOS="$*"
+else
+  REPOS="n6cnr60anmruiwkiey0pukye korokrhpoyqxf0nw2y8vk7lp"
+fi
+
+used_images="$(docker ps --format '{{.Image}}' | sort -u)"
+
+for repo in $REPOS; do
+  count=0
+  docker image ls "$repo" --format '{{.Repository}}:{{.Tag}} {{.ID}}' |
+  while read -r image image_id; do
+    [ -n "$image" ] || continue
+    if printf '%s\n' "$used_images" | grep -qx "$image"; then
+      count=$((count + 1))
+      echo "keep running image: $image"
+      continue
+    fi
+
+    count=$((count + 1))
+    if [ "$count" -le "$KEEP_PER_REPO" ]; then
+      echo "keep recent image: $image"
+      continue
+    fi
+
+    echo "remove old image: $image"
+    docker rmi "$image_id" || true
+  done
+done
+
+docker container prune -f --filter "until=$CONTAINER_UNTIL"
+docker image prune -f
+docker builder prune -f --filter "until=$BUILDER_UNTIL"

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
+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
@@ -37,6 +37,7 @@ from .workers.tg_poster import TelegramPoster
 from .workers.tg_reactor import TelegramReactor
 from .workers.vk_poster import VKPoster
 from .workers.vk_storage_uploader import TelegramStorageUploader
+from .workers.insta_parser import InstaParserWorker, InstagramCodeRequired, instagram_login
 
 COOKIE_NAME = "vk_parser_admin"
 VK_OAUTH_VERIFIER_COOKIE = "vk_oauth_verifier"
@@ -195,6 +196,7 @@ CATEGORY_TITLES = {
     "MAX Poster": "MAX-постер",
     "Publishing": "Публикации",
     "Daily Report": "Ежедневный отчет",
+    "Instagram Parser": "Instagram-парсер",
     "Parser": "Парсер",
     "Uploader": "Аплоадер",
     "VK": "VK API",
@@ -211,6 +213,7 @@ CATEGORY_ORDER = {
     "Site Poster": 52,
     "Publishing": 53,
     "Daily Report": 55,
+    "Instagram Parser": 56,
     "Parser": 60,
     "VK": 70,
     "Uploader": 80,
@@ -947,7 +950,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,
@@ -955,9 +958,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:
@@ -975,10 +978,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(
         """
@@ -988,7 +1028,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}
@@ -1002,10 +1042,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 ""
+            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,
@@ -1025,8 +1069,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
@@ -1811,6 +1860,7 @@ async def startup() -> None:
         ("tg-reactor", TelegramReactor()),
         ("vk-poster", VKPoster()),
         ("vk-storage-uploader", TelegramStorageUploader()),
+        ("insta-parser", InstaParserWorker()),
     ]
     for name, worker in workers:
         asyncio.create_task(start_worker_task(worker, name))
@@ -2316,6 +2366,7 @@ async def sources_bulk_create(
         if not isinstance(item, dict) or not item.get("ok"):
             continue
         try:
+            platform = str(item.get("platform") or PLATFORM_VK).strip().lower() or PLATFORM_VK
             row = await pool.fetchrow(
                 """
                 INSERT INTO sources(platform, name, tag, url, external_id, external_owner_id, active, priority, created_by)
@@ -2323,7 +2374,7 @@ async def sources_bulk_create(
                 ON CONFLICT DO NOTHING
                 RETURNING id
                 """,
-                PLATFORM_VK,
+                platform,
                 str(item.get("name") or item.get("external_id") or "").strip(),
                 normalize_hash_tag(str(item.get("tag") or item.get("external_id") or ""), str(item.get("external_id") or "source")),
                 str(item.get("url") or "").strip(),
@@ -2334,7 +2385,7 @@ async def sources_bulk_create(
             )
             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}")
@@ -2367,10 +2418,15 @@ async def source_create(
         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)
     external_owner_id = None
     resolved_name = name.strip()
-    resolved_url = url.strip()
     status_value = "new"
     status_msg = None
     if platform == PLATFORM_VK and external_id:
@@ -2449,7 +2505,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(
         """
@@ -2468,7 +2530,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,
@@ -3561,6 +3623,8 @@ async def workers(request: Request):
             vk_schedule=await vk_poster_schedule_rows(),
             category_titles=CATEGORY_TITLES,
             prompt_hints=PROMPT_HINTS,
+            instagram_auth_status=setting_values.get("insta_auth_status") or "",
+            instagram_cooldown_until=setting_values.get("insta_cooldown_until") or "",
         ),
     )
 
@@ -3889,6 +3953,80 @@ async def worker_toggle(request: Request, worker_name: str, csrf_token: str = Fo
     return redirect("/workers")
 
 
+@app.post("/instagram-auth/login")
+async def instagram_auth_login(request: Request, csrf_token: str = Form(...), verification_code: str = Form("")):
+    user = await get_current_user(request)
+    if not user:
+        return redirect("/login")
+    require_csrf(user, csrf_token)
+    login = str(await fetch_setting("insta_login", "") or "").strip()
+    password = str(await fetch_setting("insta_password", "") or "").strip()
+    proxy = str(await fetch_setting("insta_proxy_url", "") or "").strip()
+    session_path = str(await fetch_setting("insta_session_path", "insta_session.json") or "").strip()
+    pool = await get_pool()
+    if not login or not password:
+        status_text = "missing login or password"
+    else:
+        try:
+            await asyncio.to_thread(instagram_login, login, password, proxy, session_path, verification_code.strip())
+            status_text = f"ok: session saved to {session_path or 'insta_session.json'} at {datetime.now(timezone.utc).isoformat()}"
+            await pool.execute(
+                """
+                INSERT INTO app_settings(key, value_json, value_type, title, description, category)
+                VALUES('insta_cooldown_until', '""'::jsonb, 'str', 'Cooldown until', '', 'Instagram Parser')
+                ON CONFLICT (key) DO UPDATE
+                SET value_json='""'::jsonb,
+                    description='',
+                    updated_at=NOW()
+                """
+            )
+        except InstagramCodeRequired as exc:
+            status_text = f"code_required: {exc.choice}"
+        except Exception as exc:
+            exc_name = exc.__class__.__name__
+            if exc_name in {"TwoFactorRequired", "ChallengeRequired"}:
+                status_text = f"code_required: {exc_name}"
+            else:
+                status_text = f"failed: {exc_name}: {str(exc)[:1000]}"
+    await pool.execute(
+        """
+        INSERT INTO app_settings(key, value_json, value_type, title, description, category, updated_by)
+        VALUES('insta_auth_status', $1::jsonb, 'str', 'Instagram auth status', 'Last manual Instagram login result.', 'Instagram Parser', $2)
+        ON CONFLICT (key) DO UPDATE
+        SET value_json=$1::jsonb,
+            updated_by=$2,
+            updated_at=NOW()
+        """,
+        json.dumps(status_text),
+        user["id"],
+    )
+    await audit(user["id"], "instagram.login", "setting", None, {"status": status_text})
+    return redirect("/workers")
+
+
+@app.post("/instagram-auth/clear-cooldown")
+async def instagram_auth_clear_cooldown(request: Request, csrf_token: str = Form(...)):
+    user = await get_current_user(request)
+    if not user:
+        return redirect("/login")
+    require_csrf(user, csrf_token)
+    pool = await get_pool()
+    await pool.execute(
+        """
+        INSERT INTO app_settings(key, value_json, value_type, title, description, category, updated_by)
+        VALUES('insta_cooldown_until', '""'::jsonb, 'str', 'Cooldown until', '', 'Instagram Parser', $1)
+        ON CONFLICT (key) DO UPDATE
+        SET value_json='""'::jsonb,
+            description='manual reset',
+            updated_by=$1,
+            updated_at=NOW()
+        """,
+        user["id"],
+    )
+    await audit(user["id"], "instagram.cooldown_reset", "setting", None)
+    return redirect("/workers")
+
+
 @app.post("/settings/save")
 async def settings_save(request: Request, csrf_token: str = Form(...), key: str = Form(...), value: str = Form(...)):
     user = await get_current_user(request)

src/vk_parser_app/constants.py

@@ -1,4 +1,5 @@
 PLATFORM_VK = "vk"
+PLATFORM_INSTAGRAM = "instagram"
 
 SOURCE_STATUS_NEW = "new"
 SOURCE_STATUS_OK = "ok"
@@ -38,3 +39,4 @@ WORKER_VK_POSTER = "vk-poster"
 WORKER_MAX_POSTER = "max-poster"
 WORKER_SITE_POSTER = "site-poster"
 WORKER_DAILY_REPORT = "daily-report"
+WORKER_INSTA_PARSER = "insta-parser"

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/templates/workers.html

@@ -294,12 +294,50 @@
 
 <div class="flex flex-col gap-6">
   {% for category, rows in settings|groupby("category") %}
-  <details class="card group/details" data-workers-details="settings:{{ category }}">
+  <details class="card group/details" data-workers-details="settings:{{ category }}" {% if category == "Instagram Parser" %}open{% endif %}>
     <summary class="p-4 flex items-center justify-between cursor-pointer select-none hover:bg-app-surfaceHover transition-colors border-b border-app-border list-none">
       <div class="text-lg font-bold text-white">{{ category_titles.get(category, category) }}</div>
       <i data-lucide="chevron-down" class="w-5 h-5 text-app-textMuted transition-transform group-open/details:rotate-180"></i>
     </summary>
     <div class="p-6 bg-app-bg/30">
+      {% if category == "Instagram Parser" %}
+      <div class="mb-8 p-5 bg-app-surface border border-app-border rounded-xl flex flex-col gap-4">
+        <div class="flex flex-col lg:flex-row lg:items-start lg:justify-between gap-4">
+          <div class="min-w-0">
+            <div class="font-bold text-white flex items-center gap-2">
+              <i data-lucide="instagram" class="w-5 h-5 text-pink-400"></i>
+              Авторизация Instagram
+            </div>
+            <div class="text-xs text-app-textMuted mt-2 break-words">
+              Статус: <span class="font-mono text-app-textMain">{{ instagram_auth_status or "—" }}</span>
+            </div>
+            {% if instagram_cooldown_until %}
+            <div class="text-xs text-app-warning mt-2 break-words">
+              Cooldown: <span class="font-mono">{{ instagram_cooldown_until }}</span>
+            </div>
+            {% endif %}
+          </div>
+          <form method="post" action="/instagram-auth/clear-cooldown" class="m-0">
+            <input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
+            <button class="btn btn-surface btn-sm" type="submit">
+              <i data-lucide="timer-reset" class="w-4 h-4"></i>
+              Сбросить cooldown
+            </button>
+          </form>
+        </div>
+        <form method="post" action="/instagram-auth/login" class="grid grid-cols-1 md:grid-cols-[1fr_auto] gap-3 items-end">
+          <input type="hidden" name="csrf_token" value="{{ user.csrf_token }}">
+          <div>
+            <label class="block text-[10px] uppercase font-bold text-app-textMuted mb-1">Код из SMS/email/2FA, если Instagram его просит</label>
+            <input name="verification_code" class="input w-full font-mono" autocomplete="one-time-code" placeholder="Оставь пустым для первой попытки">
+          </div>
+          <button class="btn btn-primary" type="submit">
+            <i data-lucide="key-round" class="w-4 h-4"></i>
+            Войти
+          </button>
+        </form>
+      </div>
+      {% endif %}
       
       <div class="flex flex-col gap-8">
         {% for s in rows %}
@@ -428,7 +466,7 @@
     const form = event.target;
     if (!(form instanceof HTMLFormElement)) return;
     const action = form.getAttribute("action") || "";
-    if (action.startsWith("/workers") || action.startsWith("/settings") || action.startsWith("/branding") || action.includes("-schedule/")) {
+    if (action.startsWith("/workers") || action.startsWith("/instagram-auth") || action.startsWith("/settings") || action.startsWith("/branding") || action.includes("-schedule/")) {
       saveDetails();
       sessionStorage.setItem(scrollKey, String(window.scrollY));
     }

src/vk_parser_app/workers/ai_alerts.py

@@ -69,3 +69,36 @@ async def send_ai_worker_error_alert(worker_name: str, model: str, post_ids: lis
                     await asyncio.sleep(float(exc.retry_after) + 1)
     finally:
         await bot.session.close()
+
+
+async def send_parser_error_alert(parser_name: str, source_name: str, error: str) -> None:
+    token = (
+        str(await fetch_setting("daily_report_bot_token", "") or "").strip()
+        or str(await fetch_setting("tg_poster_bot_token", "") or "").strip()
+        or settings.tg_bot_token
+    )
+    recipients = parse_recipients(await fetch_setting("daily_report_recipient_ids", [442509142]))
+    if not token or not recipients:
+        logger.warning("Parser alert skipped: token or recipients are empty")
+        return
+
+    text = (
+        "Parser failed\n"
+        f"parser: {parser_name}\n"
+        f"source: {source_name or '-'}\n"
+        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)
+                    break
+                except TelegramRetryAfter as exc:
+                    await asyncio.sleep(float(exc.retry_after) + 1)
+                except Exception as exc:
+                    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

@@ -0,0 +1,521 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import random
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+from loguru import logger
+
+from ..constants import (
+    JOB_TYPE_VK_STORAGE_COPY,
+    MEDIA_STATUS_LINK_ONLY,
+    MEDIA_STATUS_PENDING,
+    PLATFORM_INSTAGRAM,
+    POST_STATUS_SKIPPED,
+    POST_STATUS_STORAGE_PENDING,
+    SOURCE_STATUS_ERROR,
+    SOURCE_STATUS_OK,
+    WORKER_INSTA_PARSER,
+)
+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_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()
+
+
+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
+
+
+class InstagramAuthRequired(Exception):
+    pass
+
+
+class InstagramCodeRequired(Exception):
+    def __init__(self, choice: Any) -> None:
+        self.choice = choice
+        super().__init__(f"Instagram requested verification code: {choice}")
+
+
+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),
+    }
+
+
+def instagram_login(login: str, password: str, proxy: str, session_path: str, verification_code: str = "") -> None:
+    Client, _ = load_instagrapi()
+    client = Client()
+    if proxy:
+        client.set_proxy(proxy)
+    if verification_code:
+        client.challenge_code_handler = lambda username, choice: verification_code
+    else:
+        def challenge_code_handler(username: str, choice: Any) -> str:
+            raise InstagramCodeRequired(choice)
+
+        client.challenge_code_handler = challenge_code_handler
+    path = Path(session_path or "insta_session.json")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    login_kwargs = {"verification_code": verification_code} if verification_code else {}
+    client.login(login, password, **login_kwargs)
+    client.dump_settings(path)
+
+
+class InstaParserWorker:
+    def __init__(self) -> None:
+        self.pool = None
+        self.heartbeat = HeartbeatReporter(WORKER_INSTA_PARSER, 30)
+        self.client = None
+        self.exceptions: dict[str, type[BaseException]] = {}
+
+    async def init(self) -> None:
+        self.pool = await get_pool()
+
+    async def active_source(self) -> dict | None:
+        row = await self.pool.fetchrow(
+            """
+            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
+            LIMIT 1
+            """,
+            PLATFORM_INSTAGRAM,
+        )
+        return dict(row) if row else None
+
+    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 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(row["external_post_id"]) for row 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(row["content_hash"]) for row in rows}
+
+    async def set_cooldown(self, hours: int, reason: str) -> None:
+        until = datetime.now(tz=timezone.utc) + timedelta(hours=max(1, hours))
+        await self.pool.execute(
+            """
+            INSERT INTO app_settings(key, value_json, value_type, title, description, category)
+            VALUES('insta_cooldown_until', $1::jsonb, 'str', 'Cooldown until', $2, 'Instagram Parser')
+            ON CONFLICT (key) DO UPDATE
+            SET value_json=$1::jsonb,
+                description=$2,
+                updated_at=NOW()
+            """,
+            json.dumps(until.isoformat()),
+            reason[:1000],
+        )
+
+    async def cooldown_until(self) -> datetime | None:
+        value = await fetch_setting("insta_cooldown_until", "")
+        if not value:
+            return None
+        try:
+            dt = datetime.fromisoformat(str(value))
+            return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
+        except ValueError:
+            return None
+
+    def setup_client(self, proxy: str, session_path: str) -> None:
+        if self.client is not None:
+            return
+        Client, exceptions = load_instagrapi()
+        self.exceptions = exceptions
+        client = Client()
+        if proxy:
+            client.set_proxy(proxy)
+        path = Path(session_path or "insta_session.json")
+        if not path.exists():
+            raise InstagramAuthRequired(f"Instagram session file not found: {path}")
+        client.load_settings(path)
+        client.get_timeline_feed()
+        self.client = client
+
+    async def save_post(self, source: dict, post: dict, status: str, skip_reason: str | None, media: list[dict]) -> int | None:
+        source_id = int(source["id"])
+        external_post_id = str(post["pk"])
+        raw_text = str(post.get("caption_text") or "").strip()
+        media_ids = ",".join(item["original_attachment_id"] for item in media)
+        text_hash = make_hash(raw_text)
+        content_hash = make_hash(raw_text, media_ids)
+        has_downloadable_media = any(item["media_type"] == "photo" for item in media)
+
+        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_INSTAGRAM,
+                    external_post_id,
+                    media_user_pk(post),
+                    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 raw_post_id is None:
+                    return None
+
+                for item 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, status, error
+                        )
+                        VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
+                        """,
+                        raw_post_id,
+                        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,
+                    )
+
+                if status == POST_STATUS_STORAGE_PENDING and has_downloadable_media:
+                    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, source: dict, fetch_count: int, request_pause_sec: float) -> int:
+        username = str(source.get("external_id") or "").strip().lstrip("@")
+        if not username:
+            await self.mark_source_error(int(source["id"]), "empty instagram username")
+            return 0
+
+        user_id = await asyncio.to_thread(self.client.user_id_from_username, username)
+        if request_pause_sec:
+            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
+
+        known_ids = await self.known_post_ids(int(source["id"]), [str(post.get("pk")) for post in posts if post.get("pk")])
+        candidates = [post for post in posts if str(post.get("pk")) not in known_ids]
+
+        dedupe_content_hash = await fetch_bool_setting("insta_dedupe_content_hash", True)
+        hash_by_pk: dict[str, str] = {}
+        if dedupe_content_hash:
+            hashes = []
+            for post in candidates:
+                media = extract_instagram_media(post)
+                media_ids = ",".join(item["original_attachment_id"] for item in media)
+                content_hash = make_hash(str(post.get("caption_text") or "").strip(), media_ids)
+                hash_by_pk[str(post["pk"])] = content_hash
+                hashes.append(content_hash)
+            known_hashes = await self.known_content_hashes(hashes)
+        else:
+            known_hashes = set()
+
+        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: datetime | None = None
+
+        for post in candidates:
+            posted_at = media_taken_at(post)
+            if max_seen is None or posted_at > max_seen:
+                max_seen = posted_at
+            content_hash = hash_by_pk.get(str(post.get("pk")))
+            if dedupe_content_hash and content_hash in known_hashes:
+                continue
+
+            text = str(post.get("caption_text") or "").strip()
+            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 skip_reason and not store_skipped:
+                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)
+
+        await self.mark_source_ok(int(source["id"]), max_seen or source.get("last_parsed_at"))
+        logger.info("Parsed Instagram source {}: fetched={} known={} saved={}", username, len(posts), len(known_ids), saved)
+        return saved
+
+    async def run_once(self) -> bool:
+        enabled = await is_worker_enabled(self.pool, WORKER_INSTA_PARSER)
+        if not enabled:
+            await self.heartbeat.beat(self.pool, status="disabled", force=True)
+            return False
+
+        until = await self.cooldown_until()
+        if until and until > datetime.now(tz=timezone.utc):
+            await self.heartbeat.beat(self.pool, status="cooldown", meta={"until": until.isoformat()})
+            return False
+
+        proxy = str(await fetch_setting("insta_proxy_url", "") or "").strip()
+        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)))
+        cooldown_hours = max(1, await fetch_int_setting("insta_cooldown_hours", 12))
+        request_pause_sec = max(0.0, await fetch_float_setting("insta_request_pause_sec", 2.0))
+
+        try:
+            await asyncio.to_thread(self.setup_client, proxy, session_path)
+        except InstagramAuthRequired as exc:
+            await self.heartbeat.beat(self.pool, status="auth_required", meta={"error": str(exc)})
+            await self.set_cooldown(cooldown_hours, str(exc))
+            await send_parser_error_alert(WORKER_INSTA_PARSER, "auth", str(exc))
+            return False
+        except Exception as exc:
+            self.client = None
+            await self.heartbeat.beat(self.pool, status="auth_required", meta={"error": str(exc)})
+            await self.set_cooldown(cooldown_hours, f"session failed: {exc}")
+            await send_parser_error_alert(WORKER_INSTA_PARSER, "session", str(exc))
+            return False
+
+        source = await self.active_source()
+        if not source:
+            await self.heartbeat.beat(self.pool, status="idle")
+            return False
+
+        try:
+            await self.heartbeat.beat(self.pool, status="running", meta={"source": source.get("external_id")})
+            await self.parse_source(source, fetch_count, request_pause_sec)
+            return True
+        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))
+            return True
+        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))
+            return True
+        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)
+            return True
+
+    async def run_loop(self) -> None:
+        await self.init()
+        logger.info("{} started", WORKER_INSTA_PARSER)
+        while True:
+            try:
+                visited_source = await self.run_once()
+            except Exception as exc:
+                logger.exception("Instagram parser loop error: {}", exc)
+                visited_source = False
+            if not visited_source:
+                await asyncio.sleep(10)
+                continue
+            base = max(1, await fetch_int_setting("insta_delay_base_minutes", 35))
+            spread = max(0, await fetch_int_setting("insta_delay_random_minutes", 5))
+            delay = max(60.0, base * 60 + random.uniform(-spread * 60, spread * 60))
+            await asyncio.sleep(delay)
+
+
+async def main() -> None:
+    logger.remove()
+    logger.add(lambda msg: print(msg, end=""))
+    worker = InstaParserWorker()
+    await worker.run_loop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())