"""Account guard: detects FB account threats and suspends scraping.
Monitors Apify results for signs of blocking (checkpoint, captcha, private
group errors, consecutive failures). On threat detection:
- Sets account_suspended flag in DB
- Sends email alarm via SendGrid
- scheduled_run checks flag on start and skips the cycle
Unsuspend is MANUAL (delete the flag row or run unsuspend()).
"""
from __future__ import annotations
import json
import logging
import sqlite3
from datetime import datetime
from pathlib import Path
from agent_samochodowy.config import Settings
logger = logging.getLogger(__name__)
# Patterns in Apify error items that signal FB account trouble
_THREAT_PATTERNS = [
"checkpoint",
"captcha",
"verify your identity",
"verify your account",
"blocked",
"suspended",
"temporarily locked",
"unusual activity",
"confirm your identity",
"PRIVATE GROUP",
"login",
"log in to continue",
]
# How many consecutive all-groups-zero-posts cycles before alarm
_ZERO_POSTS_ALARM_CYCLES = 2
# How many consecutive run failures before alarm
_CONSECUTIVE_FAILURES_ALARM = 3
class AccountGuard:
"""Monitors FB scraping health and suspends on threats."""
def __init__(self, settings: Settings) -> None:
self._settings = settings
self._db_path = settings.db_path
self._consecutive_failures = 0
self._groups_with_zero_posts = 0
self._total_groups_this_cycle = 0
self._conn = sqlite3.connect(self._db_path)
self._ensure_schema()
# ------------------------------------------------------------------
# Schema
# ------------------------------------------------------------------
def _ensure_schema(self) -> None:
self._conn.execute("""
CREATE TABLE IF NOT EXISTS account_guard (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""")
self._conn.execute("""
CREATE TABLE IF NOT EXISTS daily_scan_count (
date TEXT PRIMARY KEY,
count INTEGER NOT NULL DEFAULT 0
)
""")
self._conn.commit()
# ------------------------------------------------------------------
# Suspension check (call at cycle start)
# ------------------------------------------------------------------
def is_suspended(self) -> bool:
"""Check if account is suspended. Pipeline should skip if True."""
row = self._conn.execute(
"SELECT value FROM account_guard WHERE key = 'suspended'"
).fetchone()
return row is not None and row[0] == "true"
def suspend(self, reason: str) -> None:
"""Suspend the account and send alarm email."""
now = datetime.now().isoformat()
self._conn.execute(
"INSERT OR REPLACE INTO account_guard (key, value, updated_at) "
"VALUES ('suspended', 'true', ?)",
(now,),
)
self._conn.execute(
"INSERT OR REPLACE INTO account_guard (key, value, updated_at) "
"VALUES ('suspend_reason', ?, ?)",
(reason, now),
)
self._conn.commit()
logger.error("ACCOUNT SUSPENDED: %s", reason)
self._send_alarm(reason)
def unsuspend(self) -> None:
"""Manually unsuspend the account."""
self._conn.execute("DELETE FROM account_guard WHERE key = 'suspended'")
self._conn.execute("DELETE FROM account_guard WHERE key = 'suspend_reason'")
self._conn.execute("DELETE FROM account_guard WHERE key = 'consecutive_failures'")
self._conn.execute("DELETE FROM account_guard WHERE key = 'zero_post_cycles'")
self._conn.commit()
logger.info("Account unsuspended manually")
# ------------------------------------------------------------------
# Daily scan cap
# ------------------------------------------------------------------
def check_daily_cap(self) -> bool:
"""Returns True if we're still under the daily scan limit."""
today = datetime.now().strftime("%Y-%m-%d")
row = self._conn.execute(
"SELECT count FROM daily_scan_count WHERE date = ?", (today,)
).fetchone()
current = row[0] if row else 0
return current < self._settings.max_daily_scans
def increment_daily_count(self) -> int:
"""Increment and return the daily scan count."""
today = datetime.now().strftime("%Y-%m-%d")
self._conn.execute(
"INSERT INTO daily_scan_count (date, count) VALUES (?, 1) "
"ON CONFLICT(date) DO UPDATE SET count = count + 1",
(today,),
)
self._conn.commit()
row = self._conn.execute(
"SELECT count FROM daily_scan_count WHERE date = ?", (today,)
).fetchone()
return row[0]
def daily_count(self) -> int:
today = datetime.now().strftime("%Y-%m-%d")
row = self._conn.execute(
"SELECT count FROM daily_scan_count WHERE date = ?", (today,)
).fetchone()
return row[0] if row else 0
# ------------------------------------------------------------------
# Threat detection (call per group result)
# ------------------------------------------------------------------
def check_items(self, group_name: str, raw_items: list[dict]) -> list[dict]:
"""Check Apify items for threat signals. Returns clean items (no errors).
If threat detected, suspends immediately and returns empty list.
"""
clean = []
for item in raw_items:
error = item.get("error") or item.get("errorDescription") or ""
if error:
if self._is_threat(str(error)):
self.suspend(
f"FB threat detected in group '{group_name}': {str(error)[:200]}"
)
return []
# Non-threat error item — skip it but don't alarm
logger.warning("Apify error item for %s: %s", group_name, str(error)[:200])
continue
clean.append(item)
return clean
def record_group_result(self, posts_count: int) -> None:
"""Track per-group results for zero-posts detection."""
self._total_groups_this_cycle += 1
if posts_count == 0:
self._groups_with_zero_posts += 1
def check_run_failure(self, run_status: str) -> None:
"""Track consecutive Apify run failures."""
if run_status != "SUCCEEDED":
self._consecutive_failures += 1
self._persist_counter("consecutive_failures", self._consecutive_failures)
if self._consecutive_failures >= _CONSECUTIVE_FAILURES_ALARM:
self.suspend(
f"Apify: {self._consecutive_failures} consecutive run failures "
f"(last status: {run_status})"
)
else:
self._consecutive_failures = 0
self._persist_counter("consecutive_failures", 0)
def end_of_cycle_check(self) -> None:
"""Call at end of scraping phase. Checks zero-posts across all groups."""
if self._total_groups_this_cycle == 0:
return
if self._groups_with_zero_posts >= self._total_groups_this_cycle:
# All groups returned 0 posts this cycle
prev = self._load_counter("zero_post_cycles")
new_count = prev + 1
self._persist_counter("zero_post_cycles", new_count)
if new_count >= _ZERO_POSTS_ALARM_CYCLES:
self.suspend(
f"All groups returned 0 posts for {new_count} consecutive cycles. "
f"Possible silent block by Facebook."
)
else:
# At least some groups had posts — reset counter
self._persist_counter("zero_post_cycles", 0)
# Reset per-cycle counters
self._groups_with_zero_posts = 0
self._total_groups_this_cycle = 0
# ------------------------------------------------------------------
# Warmup: dynamic group count
# ------------------------------------------------------------------
def warmup_group_count(self) -> int | None:
"""Return group count based on warmup schedule, or None if not in warmup."""
warmup_until = self._settings.warmup_until
if not warmup_until:
return None
try:
end_date = datetime.strptime(warmup_until, "%Y-%m-%d")
except ValueError:
return None
now = datetime.now()
if now >= end_date:
return None # warmup over
days_remaining = (end_date - now).days
total_warmup_days = 14 # assumed 2-week warmup
day_number = total_warmup_days - days_remaining
if day_number <= 3:
return 1
elif day_number <= 7:
return 2
elif day_number <= 10:
return 3
else:
return 4
# ------------------------------------------------------------------
# Internal
# ------------------------------------------------------------------
def _is_threat(self, text: str) -> bool:
lower = text.lower()
return any(p.lower() in lower for p in _THREAT_PATTERNS)
def _persist_counter(self, key: str, value: int) -> None:
self._conn.execute(
"INSERT OR REPLACE INTO account_guard (key, value, updated_at) "
"VALUES (?, ?, ?)",
(key, str(value), datetime.now().isoformat()),
)
self._conn.commit()
def _load_counter(self, key: str) -> int:
row = self._conn.execute(
"SELECT value FROM account_guard WHERE key = ?", (key,)
).fetchone()
return int(row[0]) if row else 0
def _send_alarm(self, reason: str) -> None:
"""Send alarm email via SendGrid. Failures are logged, not raised."""
if not self._settings.sendgrid_api_key or not self._settings.notify_email_to:
logger.warning("Cannot send alarm email: SendGrid not configured")
return
try:
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
recipients = [
e.strip()
for e in self._settings.notify_email_to.split(",")
if e.strip()
]
for recipient in recipients:
message = Mail(
from_email=self._settings.notify_email_from,
to_emails=recipient,
subject="ALARM: Agent Samochodowy — konto FB zagrożone",
html_content=(
f"Konto FB zostało zawieszone
"
f"Powód: {reason}
"
f"Skanowanie zostało automatycznie wstrzymane.
"
f"Po weryfikacji konta FB, odblokuj ręcznie.
"
f"
"
f"Agent Samochodowy — alarm automatyczny"
f"
"
),
)
sg = SendGridAPIClient(self._settings.sendgrid_api_key)
sg.send(message)
logger.info("Alarm email sent to %d recipient(s)", len(recipients))
except Exception:
logger.exception("Failed to send alarm email")
def close(self) -> None:
self._conn.commit()
self._conn.close()