from __future__ import annotations import json import time from datetime import datetime, timedelta, timezone from app.db import execute, get_bool, get_int, get_setting, query_all, query_one, set_setting, transaction, utcnow from .automation_engine import AutomationEngine from .errors import classify_exception from .instagram_service import InstagramService from .lock_service import WorkerLock from .log_service import activity, cleanup_old_logs, debug, error from .normalizer import TextNormalizer class WorkerService: def __init__(self): self.instagram = InstagramService() self.engine = AutomationEngine() def run(self) -> dict: lock = WorkerLock() if not lock.acquire(): return {'ok': True, 'skipped': True, 'reason': 'worker_locked'} started = time.monotonic() started_at = utcnow() run_id = execute('INSERT INTO worker_runs(started_at, result) VALUES (?, ?)', (started_at, 'running')) stats = {'comments_checked': 0, 'new_comments': 0, 'dm_checked': 0, 'new_dm': 0, 'automations_triggered': 0, 'messages_sent': 0, 'errors': 0} details = [] try: execute("UPDATE cron_status SET last_run_at=?, updated_at=? WHERE id=1", (started_at, started_at)) debug(run_id, 'LOAD_SESSION', 'STARTED') backoff_until = get_setting('instagram_backoff_until', '') or '' if backoff_until: try: if datetime.fromisoformat(backoff_until) > datetime.now(timezone.utc): debug(run_id, 'LOAD_SESSION', 'SKIPPED', 'Instagram backoff active') return self._finish(run_id, started, stats, details, 'backoff') except ValueError: set_setting('instagram_backoff_until', '') if not get_bool('bot_enabled', True): debug(run_id, 'LOAD_SESSION', 'SKIPPED', 'Bot disabled') return self._finish(run_id, started, stats, details, 'disabled') client = self.instagram.load_client() debug(run_id, 'LOAD_SESSION', 'OK') own_user_id = str(getattr(client, 'user_id', '') or '') if not own_user_id: account = self.instagram.account_row() own_user_id = str(account['instagram_user_id'] if account else '') post_limit = max(1, min(get_int('posts_check_limit', 5), 20)) comment_limit = max(1, min(get_int('comment_check_limit', 20), 100)) dm_limit = max(1, min(get_int('dm_check_limit', 20), 100)) debug(run_id, 'FETCH_COMMENTS', 'STARTED') try: medias = self._temporary_call(self.instagram.recent_media, client, post_limit) media_map = {m['id']: m for m in medias} selected = query_all("SELECT DISTINCT am.media_id FROM automation_media am JOIN automations a ON a.id=am.automation_id WHERE a.enabled=1 AND a.media_scope='selected'") for row in selected: media_map.setdefault(row['media_id'], {'id': row['media_id']}) for media in media_map.values(): comments = self._temporary_call(self.instagram.fetch_comments, client, media['id'], comment_limit) stats['comments_checked'] += len(comments) for comment in reversed(comments): if comment['user_id'] == own_user_id or not comment['id']: continue if query_one('SELECT 1 FROM processed_comments WHERE comment_id=?', (comment['id'],)): continue stats['new_comments'] += 1 self._process_comment(client, comment, stats, run_id) debug(run_id, 'FETCH_COMMENTS', 'OK', f"checked={stats['comments_checked']}") except Exception as exc: stats['errors'] += 1 et = error(exc, 'WorkerService', 'fetch_comments') debug(run_id, 'FETCH_COMMENTS', 'FAILED', f'{et}: {exc}') if et == 'RATE_LIMIT': self._activate_rate_limit_backoff() if et in {'LOGIN_REQUIRED', 'RATE_LIMIT'}: return self._finish(run_id, started, stats, details, et.lower(), last_error=str(exc)) debug(run_id, 'FETCH_DM', 'STARTED') try: messages = self._temporary_call(self.instagram.fetch_direct_messages, client, dm_limit, message_limit=10) stats['dm_checked'] = len(messages) for message in messages: if message['user_id'] == own_user_id or not message['id']: continue if query_one('SELECT 1 FROM processed_messages WHERE message_id=?', (message['id'],)): continue stats['new_dm'] += 1 self._process_dm(client, message, stats, run_id) debug(run_id, 'FETCH_DM', 'OK', f"checked={stats['dm_checked']}") except Exception as exc: stats['errors'] += 1 et = error(exc, 'WorkerService', 'fetch_dm') debug(run_id, 'FETCH_DM', 'FAILED', f'{et}: {exc}') if et == 'RATE_LIMIT': self._activate_rate_limit_backoff() if et in {'LOGIN_REQUIRED', 'RATE_LIMIT'}: return self._finish(run_id, started, stats, details, et.lower(), last_error=str(exc)) cleanup_old_logs() return self._finish(run_id, started, stats, details, 'success') except Exception as exc: stats['errors'] += 1 et = error(exc, 'WorkerService', 'run') debug(run_id, 'FINISH', 'FAILED', f'{et}: {exc}') return self._finish(run_id, started, stats, details, 'failed', last_error=str(exc)) finally: lock.release() def _temporary_call(self, func, *args, **kwargs): last_exc = None for attempt in range(3): try: return func(*args, **kwargs) except Exception as exc: last_exc = exc et = classify_exception(exc) if et not in {'NETWORK_ERROR', 'TIMEOUT'} or attempt >= 2: raise time.sleep(1 + attempt * 2) raise last_exc def _activate_rate_limit_backoff(self): until = datetime.now(timezone.utc) + timedelta(minutes=15) set_setting('instagram_backoff_until', until.isoformat(timespec='seconds')) execute("UPDATE instagram_accounts SET rate_limit_status='limited', updated_at=? WHERE id=1", (utcnow(),)) def _process_comment(self, client, comment: dict, stats: dict, run_id: int): normalized = TextNormalizer.normalize(comment['text']) debug(run_id, 'NORMALIZE_TEXT', 'OK', f"comment {comment['id']}: {normalized}") automation, keyword = self.engine.find_match('comment', comment['text'], comment['media_id']) debug(run_id, 'MATCH_AUTOMATION', 'OK' if automation else 'SKIPPED', f"automation={automation['id'] if automation else None}, keyword={keyword['keyword'] if keyword else None}") reply_status = 'no_match' err = None automation_id = None try: if automation: automation_id = automation['id'] stats['automations_triggered'] += 1 result = self.engine.execute(client, automation, { 'trigger_type': 'comment', 'user_id': comment['user_id'], 'username': comment['username'], 'media_id': comment['media_id'], 'comment_id': comment['id'], }, run_id=run_id) stats['messages_sent'] += result['sent'] reply_status = 'dry_run' if get_bool('dry_run', False) else ('sent' if result['sent'] else 'processed') if result.get('critical_error'): err = result['critical_error'] state = query_one('SELECT state FROM conversation_states WHERE instagram_user_id=? AND automation_id=?', (comment['user_id'], automation_id)) debug(run_id, 'SAVE_STATE', 'OK' if state else 'SKIPPED', state['state'] if state else '') execute( '''INSERT OR IGNORE INTO processed_comments(comment_id, automation_id, user_id, username, media_id, original_text, normalized_text, reply_status, error, processed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (comment['id'], automation_id, comment['user_id'], comment['username'], comment['media_id'], comment['text'], normalized, reply_status, err, utcnow()), ) except Exception as exc: stats['errors'] += 1 et = error(exc, 'WorkerService', 'process_comment', user_id=comment['user_id'], automation_id=automation_id, comment_id=comment['id']) execute( '''INSERT OR IGNORE INTO processed_comments(comment_id, automation_id, user_id, username, media_id, original_text, normalized_text, reply_status, error, processed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (comment['id'], automation_id, comment['user_id'], comment['username'], comment['media_id'], comment['text'], normalized, 'failed', et, utcnow()), ) def _process_dm(self, client, message: dict, stats: dict, run_id: int): normalized = TextNormalizer.normalize(message['text']) debug(run_id, 'NORMALIZE_TEXT', 'OK', f"message {message['id']}: {normalized}") automation_id = None reply_status = 'no_match' err = None try: result = self.engine.try_resume(client, message['user_id'], message['username'], message['thread_id'], message['text'], run_id=run_id) if result is not None: automation_id = result.get('automation_id') stats['automations_triggered'] += 1 stats['messages_sent'] += result['sent'] reply_status = 'resumed' else: automation, keyword = self.engine.find_match('dm', message['text']) debug(run_id, 'MATCH_AUTOMATION', 'OK' if automation else 'SKIPPED', f"automation={automation['id'] if automation else None}, keyword={keyword['keyword'] if keyword else None}") if automation: automation_id = automation['id'] stats['automations_triggered'] += 1 result = self.engine.execute(client, automation, { 'trigger_type': 'dm', 'user_id': message['user_id'], 'username': message['username'], 'thread_id': message['thread_id'], 'message_id': message['id'], }, run_id=run_id) stats['messages_sent'] += result['sent'] reply_status = 'dry_run' if get_bool('dry_run', False) else ('sent' if result['sent'] else 'processed') if result.get('critical_error'): err = result['critical_error'] if automation_id: state = query_one('SELECT state FROM conversation_states WHERE instagram_user_id=? AND automation_id=?', (message['user_id'], automation_id)) debug(run_id, 'SAVE_STATE', 'OK' if state else 'SKIPPED', state['state'] if state else '') execute( '''INSERT OR IGNORE INTO processed_messages(message_id, automation_id, user_id, username, thread_id, original_text, normalized_text, reply_status, error, processed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (message['id'], automation_id, message['user_id'], message['username'], message['thread_id'], message['text'], normalized, reply_status, err, utcnow()), ) except Exception as exc: stats['errors'] += 1 et = error(exc, 'WorkerService', 'process_dm', user_id=message['user_id'], automation_id=automation_id, message_id=message['id']) execute( '''INSERT OR IGNORE INTO processed_messages(message_id, automation_id, user_id, username, thread_id, original_text, normalized_text, reply_status, error, processed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (message['id'], automation_id, message['user_id'], message['username'], message['thread_id'], message['text'], normalized, 'failed', et, utcnow()), ) def _finish(self, run_id: int, started_monotonic: float, stats: dict, details: list, result: str, last_error: str | None = None) -> dict: ended = utcnow() duration_ms = int((time.monotonic() - started_monotonic) * 1000) execute( '''UPDATE worker_runs SET ended_at=?, duration_ms=?, result=?, comments_checked=?, new_comments=?, dm_checked=?, new_dm=?, automations_triggered=?, messages_sent=?, errors=?, details_json=? WHERE id=?''', (ended, duration_ms, result, stats['comments_checked'], stats['new_comments'], stats['dm_checked'], stats['new_dm'], stats['automations_triggered'], stats['messages_sent'], stats['errors'], json.dumps(details, ensure_ascii=False), run_id), ) if result in {'success', 'disabled', 'backoff'}: execute("UPDATE cron_status SET last_success_at=?, last_error=NULL, updated_at=? WHERE id=1", (ended, ended)) else: execute("UPDATE cron_status SET last_error=?, updated_at=? WHERE id=1", ((last_error or result)[:1000], ended)) debug(run_id, 'FINISH', 'OK' if result in {'success', 'disabled', 'backoff'} else 'FAILED', result) activity('worker', 'run', f'Worker #{run_id}: {result}', {'duration_ms': duration_ms, **stats}) if result == 'success': set_setting('instagram_backoff_until', '') execute("UPDATE instagram_accounts SET rate_limit_status='ok', updated_at=? WHERE id=1", (ended,)) return {'ok': result in {'success', 'disabled', 'backoff'}, 'run_id': run_id, 'result': result, 'duration_ms': duration_ms, **stats}