92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
"""
|
|
fetch_metro_smart.py -- decides which file to grab, instead of always
|
|
sending the same button.
|
|
|
|
Route 2: since we're authenticated as a real user (not a bot token), we can
|
|
read the FULL history of this bot chat, not just react to new messages.
|
|
Every file the bot has ever sent you (from any button you've pressed before)
|
|
is sitting in that history with a caption. So instead of guessing which
|
|
button maps to which request, we search what's already there.
|
|
|
|
Usage (n8n passes the LLM-parsed keyword as the one argument):
|
|
python3 fetch_metro_smart.py "خطوط"
|
|
|
|
Prints one line of JSON to stdout: {"files": [...], "matched": true/false}
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import pathlib
|
|
from typing import Any
|
|
|
|
from aiobale import Client, Dispatcher
|
|
from aiobale.enums import ChatType
|
|
|
|
# ---- upstream library workaround ---------------------------------------
|
|
# aiobale hard-types TextMessage.value as `str`, but Bale's servers attach a
|
|
# view/seen-tracking payload (a dict) to some messages from large verified
|
|
# bots -- history scans on those chats crash with a pydantic ValidationError
|
|
# before we ever see the message. This loosens that one field so parsing
|
|
# succeeds; matches() below coerces to str defensively since the value can
|
|
# now legitimately be a dict instead of text.
|
|
import aiobale.types.message_content as _mc
|
|
|
|
_mc.TextMessage.model_fields["value"].annotation = Any
|
|
_mc.TextMessage.model_rebuild(force=True)
|
|
# -------------------------------------------------------------------------
|
|
|
|
# ---- hardcoded demo config --------------------------------------------
|
|
SESSION_FILE = "/mnt/storage/bale-downloads/metro_session.bale"
|
|
PHONE_NUMBER = 989381393442
|
|
TARGET_CHAT_ID = 1381242617 # متروی تهران و حومه
|
|
TARGET_CHAT_TYPE = ChatType.BOT
|
|
DOWNLOAD_DIR = pathlib.Path("/mnt/storage/bale-downloads")
|
|
MAX_MESSAGES = 200 # how far back into this chat's history to scan
|
|
# -------------------------------------------------------------------------
|
|
|
|
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
dp = Dispatcher()
|
|
client = Client(dp, session_file=SESSION_FILE, phone_number=PHONE_NUMBER)
|
|
|
|
|
|
def matches(doc, msg, keyword: str) -> bool:
|
|
keyword = (keyword or "").lower().strip()
|
|
if not keyword:
|
|
return False # no keyword = match nothing (avoid grabbing everything by accident)
|
|
caption = doc.caption.content if doc.caption else ""
|
|
msg_text = msg.text
|
|
if not isinstance(msg_text, str):
|
|
msg_text = "" # some messages carry a non-text payload in this field -- see workaround above
|
|
haystack = " ".join(filter(None, [doc.name or "", caption or "", msg_text])).lower()
|
|
return keyword in haystack
|
|
|
|
|
|
async def main(keyword: str):
|
|
await client.start(run_in_background=True)
|
|
await asyncio.sleep(2)
|
|
|
|
saved = []
|
|
try:
|
|
async for msg in client.iter_messages(
|
|
chat_id=TARGET_CHAT_ID, chat_type=TARGET_CHAT_TYPE, limit=MAX_MESSAGES
|
|
):
|
|
doc = msg.document
|
|
if not doc:
|
|
continue
|
|
if matches(doc, msg, keyword):
|
|
dest = DOWNLOAD_DIR / (doc.name or f"{msg.message_id}.bin")
|
|
if str(dest) in saved:
|
|
continue # already matched this exact file earlier in history
|
|
await msg.download(destination=dest)
|
|
saved.append(str(dest))
|
|
finally:
|
|
await client.stop()
|
|
|
|
print(json.dumps({"files": saved, "matched": len(saved) > 0}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
kw = sys.argv[1] if len(sys.argv) > 1 else ""
|
|
asyncio.run(main(kw)) |