From 2ba934aeca49db6e97528e061ad07ee2342e8ebd Mon Sep 17 00:00:00 2001 From: Amir Aref Date: Sat, 27 Jan 2024 23:44:25 +0330 Subject: [PATCH] add src directory and create project tree make separate files for schemas and telegram --- Dockerfile | 2 +- main.py | 89 +++++++++++++++---------------------------------- src/__init__.py | 0 src/schemas.py | 11 ++++++ src/telegram.py | 39 ++++++++++++++++++++++ 5 files changed, 78 insertions(+), 63 deletions(-) create mode 100644 src/__init__.py create mode 100644 src/schemas.py create mode 100644 src/telegram.py diff --git a/Dockerfile b/Dockerfile index 9a1edd3..7752a57 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,7 +4,7 @@ COPY requirements.txt /app/ WORKDIR /app RUN python -m pip install -U pip && pip install -r requirements.txt RUN echo '[]' > /app/tokens.json -COPY main.py /app/ +COPY . /app/ # run crond as main process of container ENTRYPOINT [ "python3" ] CMD ["main.py"] diff --git a/main.py b/main.py index eb307f0..d4c4c8e 100644 --- a/main.py +++ b/main.py @@ -2,38 +2,30 @@ import json import os import time +from typing import Any +from pathlib import Path import requests -import telegram -from pydantic import BaseModel import asyncio +import telegram +from src.schemas import AD +from src.telegram import send_telegram_message -URL = "https://api.divar.ir/v8/web-search/{SEARCH_CONDITIONS}".format(**os.environ) -BOT_TOKEN = "{BOT_TOKEN}".format(**os.environ) -BOT_CHATID = "{BOT_CHATID}".format(**os.environ) -SLEEP_SEC = "{SLEEP_SEC}".format(**os.environ) +URL = "https://api.divar.ir/v8/web-search/{}".format(os.getenv("SEARCH_CONDITIONS")) +BOT_TOKEN = os.getenv("BOT_TOKEN") +BOT_CHATID = os.getenv("BOT_CHATID") +SLEEP_SEC = os.getenv("SLEEP_SEC") proxy_url = None -if os.environ.get("PROXY_URL", ""): - proxy_url = os.environ.get("PROXY_URL") +if os.getenv("PROXY_URL", ""): + proxy_url = os.getenv("PROXY_URL") -TOKENS = list() # setup telegram bot client req_proxy = telegram.request.HTTPXRequest(proxy_url=proxy_url) bot = telegram.Bot(token=BOT_TOKEN, request=req_proxy) -# AD class model -class AD(BaseModel): - title: str - price: int - description: str = "" - district: str - images: list[str] = [] - token: str - - def get_data(page=None): api_url = URL if page: @@ -43,10 +35,6 @@ def get_data(page=None): return response.json() -def get_ads_list(data): - return data["web_widgets"]["post_list"] - - def fetch_ad_data(token: str) -> AD: # send request data = requests.get(f"https://api.divar.ir/v8/posts-v2/web/{token}").json() @@ -87,42 +75,19 @@ def fetch_ad_data(token: str) -> AD: return ad -async def send_telegram_message(ad: AD): - text = f"🗄 {ad.title}" + "\n" - text += f"📌 محل آگهی : {ad.district}" + "\n" - _price = f"{ad.price:,} تومان" if ad.price else "توافقی" - text += f"💰 قیمت : {_price}" + "\n\n" - text += f"📄 توضیحات :\n{ad.description}" + "\n" - text += f"https://divar.ir/v/a/{ad.token}" - - # send single photo - if len(ad.images) == 1: - await bot.send_photo( - caption=text, photo=ad.images[0], chat_id=BOT_CHATID, parse_mode="HTML" - ) - # send album - elif len(ad.images) > 1: - _media_list = [telegram.InputMediaPhoto(img) for img in ad.images[:10]] - try: - await bot.send_media_group( - caption=text, media=_media_list, chat_id=BOT_CHATID, parse_mode="HTML" - ) - except telegram.error.BadRequest as e: - print("Error sending photos :", e) - return - else: - # send just text - await bot.send_message(text=text, chat_id=BOT_CHATID, parse_mode="HTML") - - -def load_tokens(): - token_path = os.path.join( - os.path.dirname(os.path.realpath(__file__)), "tokens.json" - ) - with open(token_path, "r") as content: - if content == "": - return [] - return json.load(content) +def load_tokens() -> list[str]: + # load tokens list from json file + token_path = Path("tokens.json") + try: + with open(token_path) as file: + content = file.read() + except FileNotFoundError: + return [] + # check empty list + if not content: + return [] + # parse json + return json.loads(content) def save_tokns(tokens): @@ -133,9 +98,9 @@ def save_tokns(tokens): json.dump(tokens, outfile) -def get_tokens_page(page=None): +def get_tokens_page(page=None) -> list[Any]: data = get_data(page) - data = get_ads_list(data) + data = data["web_widgets"]["post_list"] data = data[::-1] # get tokens data = filter(lambda x: x["widget_type"] == "POST_ROW", data) @@ -152,7 +117,7 @@ async def process_data(tokens): print("AD - {} - {}".format(token, vars(ad))) # send message to telegram print("sending to telegram token: {}".format(ad.token)) - await send_telegram_message(ad) + await send_telegram_message(bot=bot, user_chat_id=BOT_CHATID, ad=ad) time.sleep(1) diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/schemas.py b/src/schemas.py new file mode 100644 index 0000000..4f1b06b --- /dev/null +++ b/src/schemas.py @@ -0,0 +1,11 @@ +from pydantic import BaseModel + + +# AD class model +class AD(BaseModel): + title: str + price: int + description: str = "" + district: str + images: list[str] = [] + token: str diff --git a/src/telegram.py b/src/telegram.py new file mode 100644 index 0000000..457bac2 --- /dev/null +++ b/src/telegram.py @@ -0,0 +1,39 @@ +""" +this file is for storing telegram bot functions and modules to send message from telegram +""" + + +import logging +import telegram +from src.schemas import AD + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(name=__name__) + + +async def send_telegram_message(bot: telegram.Bot, user_chat_id: int, ad: AD): + text = f"🗄 {ad.title}" + "\n" + text += f"📌 محل آگهی : {ad.district}" + "\n" + _price = f"{ad.price:,} تومان" if ad.price else "توافقی" + text += f"💰 قیمت : {_price}" + "\n\n" + text += f"📄 توضیحات :\n{ad.description}" + "\n" + text += f"https://divar.ir/v/a/{ad.token}" + + # send single photo + if len(ad.images) == 1: + await bot.send_photo( + caption=text, photo=ad.images[0], chat_id=user_chat_id, parse_mode="HTML" + ) + # send album + elif len(ad.images) > 1: + _media_list = [telegram.InputMediaPhoto(img) for img in ad.images[:10]] + try: + await bot.send_media_group( + caption=text, media=_media_list, chat_id=user_chat_id, parse_mode="HTML" + ) + except telegram.error.BadRequest as e: + logger.error(f"Error sending photos : {e}") + return + else: + # send just text + await bot.send_message(text=text, chat_id=user_chat_id, parse_mode="HTML")