Интеграция с Avito API: полное руководство по автоматизации объявлений

 1{
 2  "resources": [
 3    {
 4      "id": 1234567890,
 5      "title": "3-комнатная квартира, 85 м²",
 6      "status": "active",
 7      "price": 8500000,
 8      "address": "Казань, ул. Баумана, 15",
 9      "category": "Недвижимость",
10      "images": [
11        {"url": "https://acd.avito.ru/image1.jpg"}
12      ],
13      "stats": {
14        "contacts": 45,
15        "views": 1200,
16        "favorites": 32
17      }
18    }
19  ],
20  "metadata": {
21    "total": 480,
22    "page": 1,
23    "per_page": 50
24  }
25}

 1import httpx
 2import asyncio
 3from datetime import datetime, timedelta
 4
 5class AvitoOAuth:
 6    """OAuth 2.0 клиент для Avito API"""
 7    
 8    AUTH_URL = "https://www.avito.ru/web/1/oauth"
 9    TOKEN_URL = "https://www.avito.ru/web/1/oauth/token"
10    
11    def __init__(
12        self, client_id: str, client_secret: str, redirect_uri: str
13    ):
14        self.client_id = client_id
15        self.client_secret = client_secret
16        self.redirect_uri = redirect_uri
17        self.access_token = None
18        self.refresh_token = None
19        self.token_expires = None
20    
21    def get_auth_url(self) -> str:
22        """Получить URL для авторизации пользователя"""
23        return (
24            f"{self.AUTH_URL}?response_type=code"
25            f"&client_id={self.client_id}"
26            f"&redirect_uri={self.redirect_uri}"
27            f"&scope=messenger+items"
28        )
29    
30    async def exchange_code(self, code: str) -> dict:
31        """Обменять authorization code на токены"""
32        async with httpx.AsyncClient() as client:
33            response = await client.post(
34                self.TOKEN_URL,
35                data={
36                    "grant_type": "authorization_code",
37                    "code": code,
38                    "client_id": self.client_id,
39                    "client_secret": self.client_secret,
40                    "redirect_uri": self.redirect_uri
41                }
42            )
43            response.raise_for_status()
44            data = response.json()
45            
46            self.access_token = data["access_token"]
47            self.refresh_token = data["refresh_token"]
48            self.token_expires = (
49                datetime.now() 
50                + timedelta(seconds=data["expires_in"])
51            )
52            return data
53    
54    async def refresh_access_token(self) -> dict:
55        """Обновить access_token через refresh_token"""
56        if not self.refresh_token:
57            raise ValueError("Refresh token не установлен")
58        
59        async with httpx.AsyncClient() as client:
60            response = await client.post(
61                self.TOKEN_URL,
62                data={
63                    "grant_type": "refresh_token",
64                    "refresh_token": self.refresh_token,
65                    "client_id": self.client_id,
66                    "client_secret": self.client_secret
67                }
68            )
69            response.raise_for_status()
70            data = response.json()
71            
72            self.access_token = data["access_token"]
73            self.refresh_token = data["refresh_token"]
74            self.token_expires = (
75                datetime.now()
76                + timedelta(seconds=data["expires_in"])
77            )
78            return data
79    
80    async def get_valid_token(self) -> str:
81        """Получить валидный токен, обновив при необходимости"""
82        if (
83            self.token_expires 
84            and datetime.now() >= self.token_expires - timedelta(minutes=5)
85        ):
86            await self.refresh_access_token()
87        
88        if not self.access_token:
89            raise ValueError("Токен не получен. Авторизуйтесь.")
90        
91        return self.access_token
92
93
94class AvitoAPI:
95    """Клиент для работы с Avito API"""
96    
97    BASE_URL = "https://api.avito.ru"
98    
99    def __init__(self, oauth: AvitoOAuth, user_id: int):
100        self.oauth = oauth
101        self.user_id = user_id
102    
103    async def _get_headers(self) -> dict:
104        token = await self.oauth.get_valid_token()
105        return {
106            "Authorization": f"Bearer {token}",
107            "Content-Type": "application/json"
108        }
109    
110    async def get_items(self, page: int = 1, 
111                         per_page: int = 50) -> dict:
112        """Получить список объявлений"""
113        headers = await self._get_headers()
114        url = f"{self.BASE_URL}/core/v1/items"
115        params = {
116            "user_id": self.user_id,
117            "page": page,
118            "per_page": per_page
119        }
120        async with httpx.AsyncClient() as client:
121            response = await client.get(
122                url, params=params, headers=headers
123            )
124            response.raise_for_status()
125            return response.json()
126    
127    async def update_price(self, item_id: int, 
128                            price: int) -> dict:
129        """Обновить цену объявления"""
130        headers = await self._get_headers()
131        url = f"{self.BASE_URL}/core/v1/items/{item_id}"
132        async with httpx.AsyncClient() as client:
133            response = await client.patch(
134                url, json={"price": price}, headers=headers
135            )
136            response.raise_for_status()
137            return response.json()
138    
139    async def get_messages(self, item_id: int) -> dict:
140        """Получить сообщения по объявлению"""
141        headers = await self._get_headers()
142        url = f"{self.BASE_URL}/messenger/v2/messages"
143        params = {"item_id": item_id}
144        async with httpx.AsyncClient() as client:
145            response = await client.get(
146                url, params=params, headers=headers
147            )
148            response.raise_for_status()
149            return response.json()
150
151
152# Пример использования
153async def main():
154    oauth = AvitoOAuth(
155        client_id="your-client-id",
156        client_secret="your-client-secret",
157        redirect_uri="https://your-server.com/callback"
158    )
159    
160    # После авторизации пользователя:
161    # await oauth.exchange_code("code_from_callback")
162    
163    avito = AvitoAPI(oauth, user_id=123456)
164    items = await avito.get_items()
165    print(f"Объявлений: {items['metadata']['total']}")
166
167asyncio.run(main())

 1async def rotate_avito_items(avito_api: AvitoAPI):
 2    """Автоматическая ротация объявлений для повышения позиций"""
 3    
 4    # Получаем все активные объявления
 5    items = await avito_api.get_items(per_page=100)
 6    active_items = [
 7        item for item in items["resources"]
 8        if item["status"] == "active"
 9    ]
10    
11    for item in active_items:
12        # Проверяем дату последнего обновления
13        days_since_update = (
14            datetime.now() - parse_date(item["updated_at"])
15        ).days
16        
17        # Если объявление не обновлялось 3+ дня — ротируем
18        if days_since_update >= 3:
19            item_id = item["id"]
20            
21            # Сохраняем данные объявления
22            saved_data = {
23                "title": item["title"],
24                "description": item.get("description", ""),
25                "price": item["price"],
26                "images": item.get("images", []),
27            }
28            
29            # Снимаем объявление
30            await avito_api.deactivate_item(item_id)
31            
32            # Пауза для избежания rate limit
33            await asyncio.sleep(2)
34            
35            # Публикация заново с теми же данными
36            new_item = await avito_api.create_item(saved_data)
37            
38            print(
39                f"Объявление {item_id} ротировано "
40                f"→ {new_item['id']}"
41            )
42            
43            await asyncio.sleep(2)  # Лимит: 30 req/min

 1async def auto_reply_to_messages(avito_api: AvitoAPI):
 2    """Автоматические ответы на типовые вопросы"""
 3    
 4    templates = {
 5        "цена": "Здравствуйте! Цена актуальна. "
 6                "Можем обсудить при личной встрече.",
 7        "доставка": "Доставка возможна. "
 8                    "Уточните адрес, рассчитаю стоимость.",
 9        "торг": "Рассмотрим разумный торг при быстрой сделке.",
10        "availability": "Да, объект доступен. "
11                        "Когда удобно посмотреть?"
12    }
13    
14    items = await avito_api.get_items()
15    for item in items["resources"]:
16        messages = await avito_api.get_messages(item["id"])
17        
18        for msg in messages.get("messages", []):
19            if msg["direction"] == "in" and not msg.get("read"):
20                text_lower = msg["text"].lower()
21                
22                for keyword, reply in templates.items():
23                    if keyword in text_lower:
24                        await avito_api.send_message(
25                            item_id=item["id"],
26                            text=reply,
27                            chat_id=msg["chat_id"]
28                        )
29                        print(
30                            f"Автоответ на '{keyword}' "
31                            f"для объявления {item['id']}"
32                        )
33                        break

 1import re
 2from datetime import datetime
 3
 4class AvitoMessengerBot:
 5    """Интеллектуальный бот для обработки сообщений Avito"""
 6    
 7    def __init__(self, avito_api: AvitoAPI):
 8        self.avito = avito_api
 9        self.responses = {
10            "цена": {
11                "patterns": [
12                    r"цена", r"стоимость", r"сколько стоит",
13                    r"почем", r"price"
14                ],
15                "response": (
16                    "Здравствуйте! Цена актуальна, указана "
17                    "в объявлении. При быстрой сделке возможен "
18                    "небольшой торг. Хотите посмотреть?"
19                ),
20            },
21            "доступность": {
22                "patterns": [
23                    r"есть ли", r"доступн", r"свободн",
24                    r"актуальн", r"еще актуально"
25                ],
26                "response": (
27                    "Да, объект/товар доступен! "
28                    "Когда удобно посмотреть/получить?"
29                ),
30            },
31            "доставка": {
32                "patterns": [
33                    r"доставк", r"привезти", r"отправ",
34                    r"пересылк", r"сдэк", r"почта"
35                ],
36                "response": (
37                    "Доставка возможна! СДЭК, Почта России "
38                    "или самовывоз. Какой вариант удобнее?"
39                ),
40            },
41            "торг": {
42                "patterns": [
43                    r"торг", r"скидк", r"дешевле",
44                    r"уменьшить", r"скинуть"
45                ],
46                "response": (
47                    "Рассмотрим разумный торг при личной "
48                    "встрече/быстрой сделке. "
49                    "Какая сумма рассматриваете?"
50                ),
51            },
52        }
53        self.handled_messages = set()
54    
55    def _match_response(self, text: str) -> str:
56        """Определить подходящий ответ по тексту сообщения"""
57        text_lower = text.lower()
58        for category, data in self.responses.items():
59            for pattern in data["patterns"]:
60                if re.search(pattern, text_lower):
61                    return data["response"]
62        return (
63            "Спасибо за обращение! Отвечу в ближайшее время. "
64            "Если срочно — позвоните по номеру в профиле."
65        )
66    
67    async def process_new_messages(self):
68        """Обработать все новые сообщения"""
69        items = await self.avito.get_items(per_page=100)
70        
71        total_replied = 0
72        for item in items.get("resources", []):
73            messages = await self.avito.get_messages(item["id"])
74            
75            for msg in messages.get("messages", []):
76                msg_id = msg.get("id")
77                
78                # Пропускаем уже обработанные
79                if msg_id in self.handled_messages:
80                    continue
81                
82                # Обрабатываем только входящие непрочитанные
83                if msg["direction"] == "in" and not msg.get("read"):
84                    reply = self._match_response(msg["text"])
85                    
86                    await self.avito.send_message(
87                        item_id=item["id"],
88                        text=reply,
89                        chat_id=msg["chat_id"]
90                    )
91                    
92                    self.handled_messages.add(msg_id)
93                    total_replied += 1
94        
95        print(f"Обработано сообщений: {total_replied}")

 1async def collect_avito_analytics(
 2    avito_api: AvitoAPI
 3) -> list[dict]:
 4    """Собрать аналитику по всем объявлениям"""
 5    
 6    items = await avito_api.get_items(per_page=100)
 7    analytics = []
 8    
 9    for item in items.get("resources", []):
10        stats = item.get("stats", {})
11        views = stats.get("views", 0)
12        contacts = stats.get("contacts", 0)
13        favorites = stats.get("favorites", 0)
14        
15        # Рассчитываем конверсию
16        ctr = (contacts / views * 100) if views > 0 else 0
17        
18        # Определяем эффективность
19        if ctr >= 5:
20            effectiveness = "Высокая"
21        elif ctr >= 2:
22            effectiveness = "Средняя"
23        else:
24            effectiveness = "Низкая"
25        
26        analytics.append({
27            "item_id": item["id"],
28            "title": item["title"],
29            "views": views,
30            "contacts": contacts,
31            "favorites": favorites,
32            "ctr": round(ctr, 2),
33            "effectiveness": effectiveness,
34        })
35    
36    # Сортируем по CTR
37    analytics.sort(key=lambda x: x["ctr"], reverse=True)
38    
39    return analytics
40
41
42async def optimize_avito_budget(avito_api: AvitoAPI):
43    """Оптимизация бюджета на продвижение объявлений"""
44    
45    analytics = await collect_avito_analytics(avito_api)
46    
47    recommendations = []
48    for item in analytics:
49        if item["effectiveness"] == "Высокая":
50            recommendations.append({
51                "item_id": item["item_id"],
52                "action": "Увеличить бюджет",
53                "reason": f"CTR {item['ctr']}% — выше среднего",
54                "budget_change": "+50%",
55            })
56        elif item["effectiveness"] == "Низкая" and item["views"] > 500:
57            recommendations.append({
58                "item_id": item["item_id"],
59                "action": "Переписать заголовок",
60                "reason": (
61                    f"CTR {item['ctr']}% при {item['views']} "
62                    f"просмотрах — низкая конверсия"
63                ),
64                "budget_change": "-50%",
65            })
66        elif item["effectiveness"] == "Низкая":
67            recommendations.append({
68                "item_id": item["item_id"],
69                "action": "Снять и перепубликовать",
70                "reason": "Низкие просмотры и конверсия",
71                "budget_change": "Снять",
72            })
73    
74    return recommendations