from __future__ import annotations import os from pathlib import Path from typing import Any from app.config import SESSION_FILE from app.db import execute, query_one, utcnow from .errors import classify_exception from .log_service import activity, error class InstagramService: def _new_client(self): from instagrapi import Client client = Client() client.delay_range = [1, 3] return client def account_row(self): return query_one('SELECT * FROM instagram_accounts WHERE id = 1') def connect(self, username: str, password: str, verification_code: str | None = None) -> dict: username = (username or '').strip() if not username or not password: return {'ok': False, 'type': 'LOGIN_REQUIRED', 'message': 'Username and password are required.'} client = self._new_client() if SESSION_FILE.exists(): try: client.load_settings(str(SESSION_FILE)) except Exception: pass if verification_code: client.challenge_code_handler = lambda _username, _choice: verification_code try: kwargs = {'verification_code': verification_code} if verification_code else {} ok = client.login(username, password, **kwargs) if not ok: raise RuntimeError('Instagram login returned false') SESSION_FILE.parent.mkdir(parents=True, exist_ok=True) client.dump_settings(str(SESSION_FILE)) try: os.chmod(SESSION_FILE, 0o600) except OSError: pass info = client.account_info() now = utcnow() execute( '''UPDATE instagram_accounts SET username=?, instagram_user_id=?, profile_pic_url=?, status='connected', session_status='valid', challenge_status='none', rate_limit_status='ok', last_login_at=?, last_successful_request_at=?, last_error=NULL, login_failures=0, updated_at=? WHERE id=1''', (username, str(getattr(info, 'pk', client.user_id)), str(getattr(info, 'profile_pic_url', '') or ''), now, now, now), ) execute( '''INSERT INTO instagram_sessions(id, file_path, created_at, updated_at, valid) VALUES(1, ?, ?, ?, 1) ON CONFLICT(id) DO UPDATE SET file_path=excluded.file_path, updated_at=excluded.updated_at, valid=1''', (str(SESSION_FILE), now, now), ) activity('instagram', 'login', 'Instagram session saved', {'username': username}) return {'ok': True, 'username': username, 'user_id': str(client.user_id)} except Exception as exc: error_type = classify_exception(exc) challenge = 'required' if error_type in {'CHALLENGE_REQUIRED', 'TWO_FACTOR_REQUIRED'} else 'none' execute( '''UPDATE instagram_accounts SET username=?, status='disconnected', session_status='invalid', challenge_status=?, last_error=?, login_failures=login_failures+1, updated_at=? WHERE id=1''', (username, challenge, str(exc)[:1000], utcnow()), ) error(exc, 'InstagramService', 'connect', error_type=error_type) return {'ok': False, 'type': error_type, 'message': str(exc)} def load_client(self): if not SESSION_FILE.exists(): raise RuntimeError('Instagram session file does not exist') client = self._new_client() try: client.load_settings(str(SESSION_FILE)) account = self.account_row() if account and account['username']: client.username = account['username'] if account and account['instagram_user_id']: try: client.user_id = int(account['instagram_user_id']) except (ValueError, TypeError): pass return client except Exception as exc: execute("UPDATE instagram_accounts SET session_status='invalid', status='disconnected', last_error=?, updated_at=? WHERE id=1", (str(exc), utcnow())) raise def test_connection(self) -> dict: try: client = self.load_client() info = client.account_info() now = utcnow() execute( "UPDATE instagram_accounts SET status='connected', session_status='valid', last_successful_request_at=?, last_error=NULL, updated_at=? WHERE id=1", (now, now), ) return {'ok': True, 'username': getattr(info, 'username', None), 'user_id': str(getattr(info, 'pk', client.user_id))} except Exception as exc: et = error(exc, 'InstagramService', 'test_connection') execute( "UPDATE instagram_accounts SET status='disconnected', last_error=?, updated_at=? WHERE id=1", (str(exc)[:1000], utcnow()), ) return {'ok': False, 'type': et, 'message': str(exc)} def recent_media(self, client=None, amount: int = 6) -> list[dict[str, Any]]: client = client or self.load_client() account = self.account_row() user_id = (account['instagram_user_id'] if account else None) or client.user_id medias = client.user_medias(user_id, amount=amount) result = [] for media in medias: result.append({ 'id': str(media.id), 'pk': str(media.pk), 'caption': (getattr(media, 'caption_text', '') or '')[:140], 'thumbnail_url': str(getattr(media, 'thumbnail_url', '') or ''), 'taken_at': str(getattr(media, 'taken_at', '') or ''), }) return result def fetch_comments(self, client, media_id: str, amount: int) -> list[dict[str, Any]]: comments = client.media_comments(media_id, amount=amount) result = [] for c in comments: user = getattr(c, 'user', None) result.append({ 'id': str(getattr(c, 'pk', '')), 'text': getattr(c, 'text', '') or '', 'user_id': str(getattr(user, 'pk', '') or ''), 'username': getattr(user, 'username', '') or '', 'media_id': media_id, }) return result def fetch_direct_messages(self, client, thread_limit: int, message_limit: int = 10) -> list[dict[str, Any]]: threads = client.direct_threads(amount=thread_limit) result = [] for thread in threads: thread_id = str(getattr(thread, 'id', getattr(thread, 'pk', ''))) messages = client.direct_messages(thread_id, amount=message_limit) users_by_id = {str(getattr(u, 'pk', '')): getattr(u, 'username', '') for u in getattr(thread, 'users', [])} for m in messages: mid = str(getattr(m, 'id', getattr(m, 'pk', ''))) user_id = str(getattr(m, 'user_id', '') or '') text = getattr(m, 'text', None) if not mid or not user_id or text is None: continue result.append({ 'id': mid, 'thread_id': thread_id, 'user_id': user_id, 'username': users_by_id.get(user_id, ''), 'text': text, 'timestamp': getattr(m, 'timestamp', None), }) result.sort(key=lambda item: str(item.get('timestamp') or '')) return result def reply_comment(self, client, media_id: str, comment_id: str, text: str): return client.media_comment(media_id, text, replied_to_comment_id=int(comment_id)) def send_text(self, client, user_id: str, text: str, thread_id: str | None = None): if thread_id: return client.direct_answer(int(thread_id), text) return client.direct_send(text, user_ids=[int(user_id)]) def send_photo(self, client, user_id: str, path: Path, thread_id: str | None = None): if thread_id: return client.direct_send_photo(path, thread_ids=[int(thread_id)]) return client.direct_send_photo(path, user_ids=[int(user_id)]) def send_video(self, client, user_id: str, path: Path, thread_id: str | None = None): if thread_id: return client.direct_send_video(path, thread_ids=[int(thread_id)]) return client.direct_send_video(path, user_ids=[int(user_id)])