fix(backend): watched bot enriches a group at most once in six hours
This commit is contained in:
1 parent
7d7cf3cbbb
commit
efc431c969
2 files changed
+24
-4
No files matched your search
@@ -1,4 +1,5 @@
|
|||||||
import json
|
import json
|
||||||
|
from datetime import timedelta
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import asyncpg
|
import asyncpg
|
||||||
@@ -7,15 +8,22 @@ JOBS_CHANGED_CHANNEL = "jobs_changed"
|
|||||||
|
|
||||||
|
|
||||||
async def enqueue(
|
async def enqueue(
|
||||||
pool: asyncpg.Pool, account_id: int, kind: str, params: dict[str, Any]
|
pool: asyncpg.Pool,
|
||||||
|
account_id: int,
|
||||||
|
kind: str,
|
||||||
|
params: dict[str, Any],
|
||||||
|
*,
|
||||||
|
cooldown: timedelta | None = None,
|
||||||
) -> int:
|
) -> int:
|
||||||
params_json = json.dumps(params)
|
params_json = json.dumps(params)
|
||||||
existing = await pool.fetchval(
|
existing = await pool.fetchval(
|
||||||
"SELECT id FROM jobs WHERE account_id = $1 AND kind = $2 "
|
"SELECT id FROM jobs WHERE account_id = $1 AND kind = $2 "
|
||||||
"AND params = $3::jsonb AND status IN ('pending', 'running') LIMIT 1",
|
"AND params = $3::jsonb AND (status IN ('pending', 'running') "
|
||||||
|
"OR created_at > now() - $4::interval) LIMIT 1",
|
||||||
account_id,
|
account_id,
|
||||||
kind,
|
kind,
|
||||||
params_json,
|
params_json,
|
||||||
|
cooldown,
|
||||||
)
|
)
|
||||||
if existing is not None:
|
if existing is not None:
|
||||||
return existing
|
return existing
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
import contextlib
|
import contextlib
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
from datetime import timedelta
|
||||||
|
|
||||||
import asyncpg
|
import asyncpg
|
||||||
from pyrogram import Client, enums
|
from pyrogram import Client, enums
|
||||||
@@ -30,6 +31,7 @@ TOKEN_MISSING = "Токен не найден на диске" # noqa: S105
|
|||||||
INVALID_TOKEN_ERRORS = ("ACCESS_TOKEN_INVALID", "ACCESS_TOKEN_EXPIRED")
|
INVALID_TOKEN_ERRORS = ("ACCESS_TOKEN_INVALID", "ACCESS_TOKEN_EXPIRED")
|
||||||
BACKLOG_ATTEMPTS = 3
|
BACKLOG_ATTEMPTS = 3
|
||||||
BACKLOG_RETRY_PAUSE = 60
|
BACKLOG_RETRY_PAUSE = 60
|
||||||
|
ENRICH_COOLDOWN = timedelta(hours=6)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@@ -70,14 +72,24 @@ def _handlers(pool: asyncpg.Pool) -> tuple:
|
|||||||
)
|
)
|
||||||
if message.chat.type is not enums.ChatType.PRIVATE:
|
if message.chat.type is not enums.ChatType.PRIVATE:
|
||||||
await enqueue(
|
await enqueue(
|
||||||
pool, ctx.account_id, "enrich_chat", {"chat_id": meta.chat_id}
|
pool,
|
||||||
|
ctx.account_id,
|
||||||
|
"enrich_chat",
|
||||||
|
{"chat_id": meta.chat_id},
|
||||||
|
cooldown=ENRICH_COOLDOWN,
|
||||||
)
|
)
|
||||||
|
|
||||||
async def on_member(client: Client, update) -> None: # noqa: ANN001
|
async def on_member(client: Client, update) -> None: # noqa: ANN001
|
||||||
ctx = getattr(client, "capture", None)
|
ctx = getattr(client, "capture", None)
|
||||||
if ctx is None or update.chat is None:
|
if ctx is None or update.chat is None:
|
||||||
return
|
return
|
||||||
await enqueue(pool, ctx.account_id, "enrich_chat", {"chat_id": update.chat.id})
|
await enqueue(
|
||||||
|
pool,
|
||||||
|
ctx.account_id,
|
||||||
|
"enrich_chat",
|
||||||
|
{"chat_id": update.chat.id},
|
||||||
|
cooldown=ENRICH_COOLDOWN,
|
||||||
|
)
|
||||||
|
|
||||||
return (
|
return (
|
||||||
MessageHandler(on_message),
|
MessageHandler(on_message),
|
||||||
|
|||||||
Reference in new issue
Block a user