diff --git a/db/migrations/046_insta_parser_settings.sql b/db/migrations/046_insta_parser_settings.sql
index 96b81b7..e6562de 100644
--- a/db/migrations/046_insta_parser_settings.sql
+++ b/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_password', 'Instagram Password', 'secret', ''),
-('insta_proxy_url', 'Instagram Proxy URL (http://user:pass@ip:port)', 'string', ''),
-('insta_fetch_count', 'Instagram: количество проверяемых последних постов', 'int', '5'),
-('insta_delay_base_minutes', 'Instagram: пауза между аккаунтами (минут)', 'int', '35'),
-('insta_delay_random_minutes', 'Instagram: разброс паузы (± минут)', 'int', '5'),
-('insta_cooldown_hours', 'Instagram: отлежка при лимитах (часов)', 'int', '12')
+ ('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_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;
diff --git a/src/vk_parser_app/admin.py b/src/vk_parser_app/admin.py
index 151717e..53acc71 100644
--- a/src/vk_parser_app/admin.py
+++ b/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-ссылку"
- if not row["error"]:
+ 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()
- 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)
-
+ 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)
@@ -2335,18 +2372,18 @@ async def sources_bulk_create(
ON CONFLICT DO NOTHING
RETURNING id
""",
- platform_val,
- str(item.get("name") or external_id_val or "").strip(),
- normalize_hash_tag(str(item.get("tag") or external_id_val or ""), external_id_val or "source"),
- url_val,
- external_id_val,
+ 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(),
+ 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 "instagram.com" in resolved_url and platform == PLATFORM_VK:
+ if platform == PLATFORM_VK and source_platform_from_url(resolved_url) == PLATFORM_INSTAGRAM:
platform = PLATFORM_INSTAGRAM
-
- if platform == PLATFORM_VK:
- 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 ""
+ if platform == PLATFORM_INSTAGRAM:
+ external_id, resolved_url = normalize_instagram_source(resolved_url)
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,
diff --git a/src/vk_parser_app/templates/source_form.html b/src/vk_parser_app/templates/source_form.html
index 3010cdd..b03d857 100644
--- a/src/vk_parser_app/templates/source_form.html
+++ b/src/vk_parser_app/templates/source_form.html
@@ -17,6 +17,7 @@
diff --git a/src/vk_parser_app/workers/ai_alerts.py b/src/vk_parser_app/workers/ai_alerts.py
index 46209e0..c4d0e59 100644
--- a/src/vk_parser_app/workers/ai_alerts.py
+++ b/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"🚨 CRITICAL PARSER ERROR\n"
- f"Parser: {parser_name}\n"
- f"Source: {source_name}\n\n"
- f"Error: {error[:1000]}"
+ "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, 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:
- logger.error("Failed to send parser alert to {}: {}", recipient_id, e)
+ except Exception as exc:
+ logger.error("Failed to send parser alert to {}: {}", recipient_id, exc)
break
finally:
await bot.session.close()
-
diff --git a/src/vk_parser_app/workers/insta_parser.py b/src/vk_parser_app/workers/insta_parser.py
index 3b7a8a5..1b2ce16 100644
--- a/src/vk_parser_app/workers/insta_parser.py
+++ b/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
-
-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 typing import Any
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:
- return hashlib.sha256((text or "").strip().lower().encode()).hexdigest()
+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
+
+
+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("insta-parser", 60)
- self.heartbeat_task: asyncio.Task | None = None
- self.client: Client | None = None
- self.session_file = Path("insta_session.json")
+ self.heartbeat = HeartbeatReporter(WORKER_INSTA_PARSER, 30)
+ self.client = None
+ self.exceptions: dict[str, type[BaseException]] = {}
- async def init(self):
+ async def init(self) -> None:
self.pool = await get_pool()
- async def get_active_source(self) -> dict | None:
- async with self.pool.acquire() as conn:
- row = await conn.fetchrow(
- """
- SELECT id, external_id, name, last_checked_at, last_parsed_at
- FROM sources
- WHERE platform = $1 AND active = TRUE
- ORDER BY last_checked_at NULLS FIRST
- LIMIT 1
- """,
- PLATFORM_INSTAGRAM
- )
+ 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 get_known_hashes(self, hashes: list[str]) -> set[str]:
- if not hashes:
- return set()
- async with self.pool.acquire() as conn:
- rows = await conn.fetch(
- """
- SELECT content_hash
- FROM posts
- WHERE content_hash = ANY($1::text[])
- """,
- hashes,
- )
- return {str(r["content_hash"]) for r in rows}
+ async def mark_source_error(self, source_id: int, message: str) -> None:
+ await self.pool.execute(
+ """
+ UPDATE sources
+ SET status=$2,
+ status_msg=$3,
+ last_checked_at=NOW(),
+ updated_at=NOW()
+ WHERE id=$1
+ """,
+ source_id,
+ SOURCE_STATUS_ERROR,
+ message[:1000],
+ )
- async def 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:
- try:
- numeric_ids = [int(pk) for pk in external_post_ids]
- rows = await conn.fetch(
- """
- SELECT vk_post_id
- FROM posts
- WHERE source_id = $1 AND vk_post_id = ANY($2::bigint[])
- """,
- source_id,
- numeric_ids,
- )
- return {str(r["vk_post_id"]) for r in rows}
- except ValueError:
- 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 deactivate_source(self, source_id: int, reason: str) -> None:
- async with self.pool.acquire() as conn:
- await conn.execute(
- """
- UPDATE sources
- SET active = FALSE,
- status = $1,
- status_msg = $2,
- last_checked_at = NOW(),
- updated_at = NOW()
- WHERE id = $3
- """,
- SOURCE_STATUS_ERROR,
- reason[:1000],
- source_id,
- )
+ 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 mark_source_ok(self, source_id: int, last_parsed_at: datetime | None = None) -> None:
- async with self.pool.acquire() as conn:
- await conn.execute(
- """
- UPDATE sources
- SET status = $1,
- status_msg = NULL,
- last_checked_at = NOW(),
- last_parsed_at = COALESCE($2, last_parsed_at),
- updated_at = NOW()
- WHERE id = $3
- """,
- SOURCE_STATUS_OK,
- last_parsed_at,
- source_id,
- )
-
- async def save_cooldown(self, hours: int) -> None:
- until = datetime.now(tz=timezone.utc) + timedelta(hours=hours)
- async with self.pool.acquire() as conn:
- await conn.execute(
- """
- INSERT INTO app_settings (key, value, value_type)
- VALUES ('insta_cooldown_until', $1, 'string')
- ON CONFLICT (key) DO UPDATE SET value = $1
- """,
- until.isoformat()
- )
+ 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 get_cooldown_until(self) -> datetime | None:
- async with self.pool.acquire() as conn:
- val = await conn.fetchval("SELECT value FROM app_settings WHERE key = 'insta_cooldown_until'")
- if val:
- try:
- dt = datetime.fromisoformat(val)
- if dt.tzinfo is None:
- dt = dt.replace(tzinfo=timezone.utc)
- return dt
- except ValueError:
- pass
- return None
+ 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
- async def save_post_and_media(self, source_id: int, post: dict, status: str, skip_reason: str | None) -> int | None:
- text = (post.get("caption_text") or "").strip()
- content_hash = make_content_hash(text)
- posted_at = post.get("taken_at")
- if not posted_at:
- posted_at = datetime.now(tz=timezone.utc)
- elif posted_at.tzinfo is None:
- posted_at = posted_at.replace(tzinfo=timezone.utc)
+ def setup_client(self, login: str, password: str, 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 path.exists():
+ client.load_settings(path)
+ client.login(login, password)
+ client.dump_settings(path)
+ self.client = client
- 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 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():
- post_id = await conn.fetchval(
+ raw_post_id = await conn.fetchval(
"""
- INSERT INTO posts (
- source_id, vk_post_id, vk_owner_id, posted_at,
- raw_text, raw_json, content_hash,
- status, skip_reason
+ 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::jsonb,$7,$8::post_status,$9)
- ON CONFLICT (source_id, vk_post_id) DO NOTHING
+ 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,
- pk,
- int(post.get("user", {}).get("pk") or 0),
- posted_at,
- text,
+ 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 post_id is None:
+ if raw_post_id is None:
return None
- for p_url in photos:
- if not p_url: continue
+ for item in media:
await conn.execute(
"""
- INSERT INTO post_media (post_id, media_type, vk_url)
- VALUES ($1, 'photo', $2)
+ 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)
""",
- post_id,
- str(p_url),
+ 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,
)
- for v_url in videos:
- if not v_url: continue
+ if status == POST_STATUS_STORAGE_PENDING and has_downloadable_media:
await conn.execute(
"""
- INSERT INTO post_media (post_id, media_type, vk_url)
- VALUES ($1, 'video', $2)
- """,
- 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')
+ INSERT INTO jobs(type, entity_type, entity_id, payload_json, status)
+ VALUES($1, 'raw_post', $2, '{}'::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:
- if self.client:
- return True
-
- if Client is None:
- logger.error("instagrapi is not installed! Cannot setup client.")
- raise Exception("Module instagrapi is not installed. Please install it.")
-
- self.client = Client()
- if proxy:
- self.client.set_proxy(proxy)
-
- if self.session_file.exists():
- try:
- self.client.load_settings(self.session_file)
- self.client.get_timeline_feed()
- logger.info("Instagram session loaded successfully")
- return True
- except Exception as e:
- logger.warning(f"Session invalid, will re-login: {e}")
- self.session_file.unlink(missing_ok=True)
-
- logger.info(f"Logging into Instagram as {login}...")
- try:
- self.client.login(login, password)
- 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_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
- 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
+ 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
- if not medias:
- await self.mark_source_ok(source_id)
- return
+ 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]
- 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)
+ 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_dt = None
+ max_seen: datetime | None = None
for post in candidates:
- try:
- post_dt = post.get("taken_at")
- if post_dt:
- if post_dt.tzinfo is None:
- post_dt = post_dt.replace(tzinfo=timezone.utc)
- if max_seen_dt is None or post_dt > max_seen_dt:
- max_seen_dt = post_dt
+ 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 = (post.get("caption_text") or "").strip()
- content_hash = make_content_hash(text)
-
- if 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"
- media_type = post.get("media_type")
- has_media = media_type in (1, 2, 8)
-
- if not has_media:
- if await self.save_post_and_media(source_id, post, POST_STATUS_SKIPPED, "no_media"):
- saved += 1
- continue
-
- if len(text) < min_text_length:
- if await self.save_post_and_media(source_id, post, POST_STATUS_SKIPPED, "text_too_short"):
- saved += 1
- continue
+ 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)
- if await self.save_post_and_media(source_id, post, POST_STATUS_STORAGE_PENDING, None):
- saved += 1
- except Exception as e:
- logger.error(f"Failed to save insta post {post.get('pk')}: {e}")
+ 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
- await self.mark_source_ok(source_id, max_seen_dt)
- logger.info(f"Source {username}: saved {saved} posts")
+ async def run_once(self) -> None:
+ enabled = await is_worker_enabled(self.pool, WORKER_INSTA_PARSER)
+ if not enabled:
+ await self.heartbeat.beat(self.pool, status="disabled", force=True)
+ return
- 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)
+ 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
+
+ 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()
+ 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))
if not login or not password:
- logger.warning("Instagram login/password not set in app_settings. Skipping.")
- 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.")
+ await self.heartbeat.beat(self.pool, status="missing_credentials")
return
try:
- await asyncio.to_thread(self.setup_client, login, password, proxy)
- except ChallengeRequired as e:
- error_msg = f"Challenge Required! Cannot login. Need manual verification.\n{e}"
- logger.error(error_msg)
- 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}")
+ await asyncio.to_thread(self.setup_client, login, password, proxy, session_path)
+ except Exception as exc:
+ await self.set_cooldown(cooldown_hours, f"login failed: {exc}")
+ await send_parser_error_alert(WORKER_INSTA_PARSER, "login", str(exc))
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)
-
- delay = delay_base * 60 + random.uniform(-delay_random * 60, delay_random * 60)
- delay = max(60, delay)
- logger.info(f"Sleeping for {delay/60:.1f} minutes before next parse.")
- 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)
+ 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)
- async def _heartbeat_loop(self) -> None:
+ async def run_loop(self) -> None:
+ await self.init()
+ logger.info("{} started", WORKER_INSTA_PARSER)
while True:
try:
- await self.heartbeat.beat(self.pool, {"state": "polling"})
- except Exception as e:
- logger.warning(f"Insta-parser heartbeat failed: {e}")
- await asyncio.sleep(self.heartbeat_interval_sec)
-
- async def run_loop(self):
- await self.init()
- 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
+ await self.run_once()
+ except Exception as exc:
+ logger.exception("Instagram parser loop error: {}", exc)
+ 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():
+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=(
- "{time:YYYY-MM-DD HH:mm:ss} | "
- "{level: <8} | "
- "{name}:{line} — {message}"
- ),
- )
asyncio.run(main())