from __future__ import annotations import json import random import time from typing import Any from app.db import execute, get_bool, get_int, query_all, query_one, transaction, utcnow from .follow_service import FollowService from .instagram_service import InstagramService from .errors import classify_exception from .log_service import activity, debug, error from .media_service import MediaService from .normalizer import TextNormalizer, matches class AutomationEngine: def __init__(self): self.instagram = InstagramService() self.follow = FollowService() self.media = MediaService() def _automation_matches_media(self, automation_id: int, scope: str, media_id: str | None) -> bool: if scope == 'all' or not media_id: return True row = query_one('SELECT 1 FROM automation_media WHERE automation_id=? AND media_id=?', (automation_id, media_id)) return bool(row) def find_match(self, trigger_type: str, text: str, media_id: str | None = None) -> tuple[Any, Any] | tuple[None, None]: rows = query_all( "SELECT * FROM automations WHERE enabled=1 AND (trigger_type=? OR trigger_type='both') ORDER BY id ASC", (trigger_type,), ) for automation in rows: if trigger_type == 'comment' and not self._automation_matches_media(automation['id'], automation['media_scope'], media_id): continue keywords = query_all('SELECT * FROM automation_keywords WHERE automation_id=? ORDER BY id', (automation['id'],)) for keyword in keywords: if matches(text, keyword['keyword'], keyword['match_type']): return automation, keyword return None, None def dry_run_match(self, trigger_type: str, text: str, media_id: str | None = None) -> dict: automation, keyword = self.find_match(trigger_type, text, media_id) if not automation: return {'matched': False, 'original': text, 'normalized': TextNormalizer.normalize(text), 'steps': []} steps = query_all('SELECT * FROM automation_steps WHERE automation_id=? ORDER BY position, id', (automation['id'],)) return { 'matched': True, 'original': text, 'normalized': TextNormalizer.normalize(text), 'automation': dict(automation), 'keyword': dict(keyword), 'steps': [dict(s) for s in steps], } def _choose_comment_reply(self, automation) -> str | None: replies = query_all('SELECT * FROM comment_replies WHERE automation_id=? ORDER BY position, id', (automation['id'],)) if not replies: return None if automation['reply_mode'] == 'sequential': with transaction() as conn: current = conn.execute('SELECT sequential_index FROM automations WHERE id=?', (automation['id'],)).fetchone()['sequential_index'] chosen = replies[current % len(replies)]['text'] conn.execute('UPDATE automations SET sequential_index=?, updated_at=? WHERE id=?', ((current + 1) % len(replies), utcnow(), automation['id'])) return chosen return random.choice(replies)['text'] def _message_pause(self, dry_run: bool): low = max(0.0, float(get_int('message_delay_min', 1))) high = max(low, float(get_int('message_delay_max', 3))) seconds = random.uniform(low, high) if not dry_run and seconds > 0: time.sleep(min(seconds, 5.0)) return round(seconds, 2) def _save_state(self, user_id: str, automation_id: int, state: str, payload: dict): execute( '''INSERT INTO conversation_states(instagram_user_id, automation_id, state, payload_json, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(instagram_user_id, automation_id) DO UPDATE SET state=excluded.state, payload_json=excluded.payload_json, updated_at=excluded.updated_at''', (str(user_id), automation_id, state, json.dumps(payload, ensure_ascii=False), utcnow()), ) def _clear_state(self, user_id: str, automation_id: int): execute('DELETE FROM conversation_states WHERE instagram_user_id=? AND automation_id=?', (str(user_id), automation_id)) def try_resume(self, client, user_id: str, username: str, thread_id: str, text: str, run_id: int | None = None) -> dict | None: states = query_all('SELECT * FROM conversation_states WHERE instagram_user_id=? ORDER BY updated_at DESC', (str(user_id),)) for state in states: payload = json.loads(state['payload_json'] or '{}') keys = payload.get('keywords', []) if keys and not any(matches(text, key, 'exact') for key in keys): continue automation = query_one('SELECT * FROM automations WHERE id=? AND enabled=1', (state['automation_id'],)) if not automation: self._clear_state(user_id, state['automation_id']) continue context = { 'user_id': str(user_id), 'username': username, 'thread_id': thread_id, 'trigger_type': 'dm', 'resume_keywords': keys, } result = self.execute(client, automation, context, start_position=int(payload.get('next_position', 0)), run_id=run_id) result['automation_id'] = automation['id'] return result return None def execute(self, client, automation, context: dict, start_position: int = 0, run_id: int | None = None) -> dict: dry_run = get_bool('dry_run', False) steps = query_all( 'SELECT * FROM automation_steps WHERE automation_id=? AND position>=? ORDER BY position, id', (automation['id'], start_position), ) result = {'sent': 0, 'actions': [], 'stopped': False} for step in steps: step_type = step['step_type'] cfg = json.loads(step['config_json'] or '{}') debug(run_id, step_type.upper(), 'OK' if dry_run else 'STARTED', 'dry-run' if dry_run else '') try: if step_type == 'send_comment_reply': if not context.get('comment_id'): result['actions'].append({'step': step_type, 'result': 'skipped'}) continue text = cfg.get('text') or self._choose_comment_reply(automation) if not text: result['actions'].append({'step': step_type, 'result': 'skipped'}) continue if not dry_run: self.instagram.reply_comment(client, context['media_id'], context['comment_id'], text) result['sent'] += 1 result['actions'].append({'step': step_type, 'text': text, 'result': 'dry_run' if dry_run else 'sent'}) self._message_pause(dry_run) elif step_type in {'send_text_dm', 'send_link'}: text = cfg.get('text', '').strip() if not text: continue if not dry_run: sent_message = self.instagram.send_text(client, context['user_id'], text, context.get('thread_id')) result['sent'] += 1 if not context.get('thread_id'): new_thread_id = getattr(sent_message, 'thread_id', None) if new_thread_id: context['thread_id'] = str(new_thread_id) result['actions'].append({'step': step_type, 'text': text, 'result': 'dry_run' if dry_run else 'sent'}) self._message_pause(dry_run) elif step_type in {'send_image_dm', 'send_video_dm'}: media_id = int(cfg.get('media_file_id') or 0) if not media_id: continue path = self.media.path_for(media_id) if not dry_run: if step_type == 'send_image_dm': self.instagram.send_photo(client, context['user_id'], path, context.get('thread_id')) else: self.instagram.send_video(client, context['user_id'], path, context.get('thread_id')) result['sent'] += 1 result['actions'].append({'step': step_type, 'media_file_id': media_id, 'result': 'dry_run' if dry_run else 'sent'}) self._message_pause(dry_run) elif step_type == 'delay': low = max(0.0, float(cfg.get('min_seconds', 1))) high = max(low, float(cfg.get('max_seconds', low))) seconds = random.uniform(low, high) if not dry_run: time.sleep(min(seconds, 10.0)) result['actions'].append({'step': step_type, 'seconds': round(seconds, 2), 'result': 'dry_run' if dry_run else 'waited'}) elif step_type == 'wait_for_user_message': keywords = cfg.get('keywords') or [] if isinstance(keywords, str): keywords = [k.strip() for k in keywords.replace(',', '\n').splitlines() if k.strip()] self._save_state( context['user_id'], automation['id'], 'WAITING_FOR_USER_MESSAGE', {'next_position': step['position'] + 1, 'keywords': keywords}, ) result['actions'].append({'step': step_type, 'result': 'waiting', 'keywords': keywords}) result['stopped'] = True break elif step_type == 'check_follow': if dry_run: follow_result = cfg.get('dry_run_result', 'followed') else: follow_result = self.follow.check(client, context['user_id'], context.get('username', '')) result['actions'].append({'step': step_type, 'follow_result': follow_result}) if follow_result != 'followed': if follow_result == 'unknown': text = cfg.get('unknown_text') or '\u0641\u0639\u0644\u0627\u064b \u0627\u0645\u06a9\u0627\u0646 \u0628\u0631\u0631\u0633\u06cc \u0641\u0627\u0644\u0648 \u0648\u062c\u0648\u062f \u0646\u062f\u0627\u0631\u062f. \u0686\u0646\u062f \u0644\u062d\u0638\u0647 \u0628\u0639\u062f \u062f\u0648\u0628\u0627\u0631\u0647 \u0627\u0645\u062a\u062d\u0627\u0646 \u06a9\u0646.' else: text = cfg.get('not_followed_text') or '\u0647\u0646\u0648\u0632 \u0641\u0627\u0644\u0648 \u0634\u062f\u0646\u062a \u0631\u0627 \u0646\u0645\u06cc\u200c\u0628\u06cc\u0646\u0645. \u0641\u0627\u0644\u0648 \u06a9\u0646 \u0648 \u062f\u0648\u0628\u0627\u0631\u0647 \u067e\u06cc\u0627\u0645 \u0628\u062f\u0647.' if text and not dry_run: self.instagram.send_text(client, context['user_id'], text, context.get('thread_id')) result['sent'] += 1 keys = context.get('resume_keywords') or cfg.get('retry_keywords') or ['done', '\u0627\u0646\u062c\u0627\u0645 \u0634\u062f', '\u0641\u0627\u0644\u0648 \u06a9\u0631\u062f\u0645'] self._save_state( context['user_id'], automation['id'], 'WAITING_FOR_FOLLOW_CONFIRM', {'next_position': step['position'], 'keywords': keys}, ) result['stopped'] = True break self._clear_state(context['user_id'], automation['id']) elif step_type == 'stop_flow': self._clear_state(context['user_id'], automation['id']) result['actions'].append({'step': step_type, 'result': 'stopped'}) result['stopped'] = True break else: result['actions'].append({'step': step_type, 'result': 'skipped_unknown'}) debug(run_id, step_type.upper(), 'OK', json.dumps(result['actions'][-1] if result['actions'] else {}, ensure_ascii=False)[:2000]) except Exception as exc: et = classify_exception(exc) if et == 'UNKNOWN_ERROR': et = { 'send_comment_reply': 'COMMENT_SEND_ERROR', 'send_text_dm': 'DM_SEND_ERROR', 'send_link': 'DM_SEND_ERROR', 'send_image_dm': 'MEDIA_UPLOAD_ERROR', 'send_video_dm': 'MEDIA_UPLOAD_ERROR', 'check_follow': 'FOLLOW_CHECK_ERROR', }.get(step_type, et) et = error( exc, 'AutomationEngine', step_type, error_type=et, user_id=context.get('user_id'), automation_id=automation['id'], comment_id=context.get('comment_id'), message_id=context.get('message_id'), ) debug(run_id, step_type.upper(), 'FAILED', f'{et}: {exc}') result['actions'].append({'step': step_type, 'result': 'failed', 'error_type': et, 'message': str(exc)}) if et in {'LOGIN_REQUIRED', 'RATE_LIMIT', 'DATABASE_ERROR'}: result['critical_error'] = et result['stopped'] = True break if not result['stopped']: self._clear_state(context['user_id'], automation['id']) activity( 'automation', 'execute', f"Automation {automation['id']} executed", {'automation': automation['name'], 'user_id': context.get('user_id'), 'actions': result['actions']}, ) return result