From 15db1b5f90de89bbeb4b5d3c70a8d2797344412c Mon Sep 17 00:00:00 2001 From: "Jose A. Romero" Date: Mon, 30 Sep 2024 15:56:41 +0200 Subject: [PATCH] Initial commit --- .dockerignore | 4 + .gitignore | 13 + .pre-commit-config.yaml | 68 +++ config.ini | 3 + dockerfile | 17 + license | 21 + main.py | 183 ++++++ manage.py | 126 +++++ readme.md | 64 +++ requirements.txt | 3 + src/__init__.py | 7 + src/data/media.yaml | 192 +++++++ src/data/register.yaml | 29 + src/data/stickers.yaml | 195 +++++++ src/data/telegram_config.yaml | 12 + src/sentence_generator/__init__.py | 6 + src/sentence_generator/morning.py | 210 +++++++ src/sentence_generator/sentence_generator.py | 171 ++++++ src/utils/__init__.py | 6 + src/utils/random.py | 55 ++ src/worker.py | 557 +++++++++++++++++++ 21 files changed, 1942 insertions(+) create mode 100644 .dockerignore create mode 100644 .gitignore create mode 100644 .pre-commit-config.yaml create mode 100644 config.ini create mode 100644 dockerfile create mode 100644 license create mode 100644 main.py create mode 100644 manage.py create mode 100644 readme.md create mode 100644 requirements.txt create mode 100644 src/__init__.py create mode 100644 src/data/media.yaml create mode 100644 src/data/register.yaml create mode 100644 src/data/stickers.yaml create mode 100644 src/data/telegram_config.yaml create mode 100644 src/sentence_generator/__init__.py create mode 100644 src/sentence_generator/morning.py create mode 100644 src/sentence_generator/sentence_generator.py create mode 100644 src/utils/__init__.py create mode 100644 src/utils/random.py create mode 100644 src/worker.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..89e2bf9 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,4 @@ +.venv +.vscode +**/__pycache__ +manage.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..86d1ba7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +# vscode settings +.vscode + +# python venv and pycache +.venv +__pycache__ + +# media (for saving repo space) +src/data/media/pics/* +src/data/media/videos/* + +# sensitive data +src/data/bot.session diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..0b4681b --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,68 @@ +repos: + - repo: https://github.com/pre-commit/pre-commit-hooks + rev: v4.6.0 + hooks: + - id: check-added-large-files + args: + - --maxkb=500 + - id: check-case-conflict + - id: check-docstring-first + - id: check-json + - id: check-toml + - id: check-yaml + - id: end-of-file-fixer + - id: mixed-line-ending + args: + - --fix=lf + - id: trailing-whitespace + + - repo: https://github.com/pycqa/isort + rev: 5.13.2 + hooks: + - id: isort + files: ".*" + args: + - --profile=black + - --project=telegram-auto-texter + + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.11.2 + hooks: + - id: mypy + files: ".*" + additional_dependencies: + - types-pyyaml + + - repo: https://github.com/pycqa/pydocstyle + rev: 6.3.0 + hooks: + - id: pydocstyle + args: + - --convention=google + exclude: "docs|tests" + + - repo: https://github.com/asottile/pyupgrade + rev: v3.17.0 + hooks: + - id: pyupgrade + args: + - --py310-plus + + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.6.8 + hooks: + - id: ruff + args: + - --fix + - --exit-non-zero-on-fix + - --show-fixes + - --line-length=100 + - id: ruff-format + args: + - --line-length=100 + + - repo: https://github.com/adrienverge/yamllint.git + rev: v1.35.1 + hooks: + - id: yamllint + exclude: "tests" diff --git a/config.ini b/config.ini new file mode 100644 index 0000000..5f9c6a7 --- /dev/null +++ b/config.ini @@ -0,0 +1,3 @@ +[Telegram] +api_id = API_ID +api_hash = API_HASH diff --git a/dockerfile b/dockerfile new file mode 100644 index 0000000..497e869 --- /dev/null +++ b/dockerfile @@ -0,0 +1,17 @@ +FROM ubuntu:22.04 + +ENV DEBIAN_FRONTEND=noninteractive \ + TZ="Europe/Stockholm" + +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + python3 \ + python3-pip \ + tzdata && \ + apt-get clean && \ + rm -rf /var/lib/apt/lists/* + +WORKDIR /telegram-auto-texter +ADD . . +RUN pip install -r requirements.txt +CMD ["python3", "-u", "main.py"] diff --git a/license b/license new file mode 100644 index 0000000..1c9dcbd --- /dev/null +++ b/license @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2024 Jose A. Romero + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/main.py b/main.py new file mode 100644 index 0000000..8aeeccc --- /dev/null +++ b/main.py @@ -0,0 +1,183 @@ +"""This module serves as the main entry point for the Telegram bot application. + +It handles the initialization of the bot, sets up scheduled tasks for sending messages and media, +and manages user interactions. The module integrates with the worker functions to perform various +tasks such as sending greetings, media items, and reminders, while utilizing the APScheduler for +task scheduling. +""" + +import asyncio +from configparser import ConfigParser, ExtendedInterpolation + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from telethon import TelegramClient, events +from telethon.tl.custom.message import Message + +from src import worker + +config = ConfigParser(interpolation=ExtendedInterpolation()) +config.read("config.ini") + +client = TelegramClient( + "src/data/bot", config.get("Telegram", "api_id"), config.get("Telegram", "api_hash") +).start() +print("Telegram user bot is now running...") + + +@client.on(events.NewMessage("me", pattern="/health")) +async def handle_health(event: Message | events.NewMessage): + """Handles the /health command and responds with the application's health status. + + This asynchronous function listens for messages containing the /health command and replies with + the current health status of the application. It provides a simple way for users to check if the + bot is operational. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the reply. + """ + await event.reply(worker.health()) + + +@client.on(events.NewMessage("me", pattern="/greeting_info")) +async def handle_greeting_info(event: Message | events.NewMessage): + """Handles the /greeting_info command and responds with the next greeting time. + + This asynchronous function listens for messages containing the /greeting_info command and + replies with the scheduled time for the next greeting. It provides users with information about + when they can expect the next greeting message. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the reply. + """ + await event.reply(f"Next greeting at {worker.next_greeting_time}") + + +@client.on(events.NewMessage("me", pattern="/send_greeting")) +async def handle_send_greeting(event: Message | events.NewMessage): + """Handles the /send_greeting command to send a morning greeting. + + This asynchronous function listens for messages containing the /send_greeting command and + triggers the sending of a morning greeting via the Telegram client. It then replies to the user + to confirm that the action has been completed. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the greeting or the reply. + """ + await worker.send_morning_greeting(client) + await event.reply("Done!") + + +@client.on(events.NewMessage("me", pattern="/test_greeting")) +async def handle_test_greeting(event: Message | events.NewMessage): + """Handles the /test_greeting command to send a test morning greeting. + + This asynchronous function listens for messages containing the /test_greeting command and + triggers the sending of a morning greeting via the Telegram client without marking it as used. + It then replies to the user to confirm that the action has been completed. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the greeting or the reply. + """ + await worker.send_morning_greeting(client, user_id="me", set_as_used=False) + await event.reply("Done!") + + +@client.on(events.NewMessage("me", pattern="/afternoon_media_info")) +async def handle_afternoon_media(event: Message | events.NewMessage): + """Handles the /afternoon_media_info command and responds with the next media time. + + This asynchronous function listens for messages containing the /afternoon_media_info command + and replies with the scheduled time for the next afternoon media. It provides users with + information about when they can expect the next media item. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the reply. + """ + await event.reply(f"Next greeting at {worker.next_afternoon_media_time}") + + +@client.on(events.NewMessage("me", pattern="/send_afternoon_media")) +async def handle_send_afternoon_media(event: Message | events.NewMessage): + """Handles the /send_afternoon_media command to send an afternoon media item. + + This asynchronous function listens for messages containing the /send_afternoon_media command + and triggers the sending of an afternoon media item via the Telegram client. It then replies to + the user to confirm that the action has been completed. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the media or the reply. + """ + await worker.send_afternoon_media(client) + await event.reply("Done!") + + +@client.on(events.NewMessage("me", pattern="/test_afternoon_media")) +async def handle_test_afternoon_media(event: Message | events.NewMessage): + """Handles the /test_afternoon_media command to send a test afternoon media item. + + This asynchronous function listens for messages containing the /test_afternoon_media command + and triggers the sending of an afternoon media item via the Telegram client without marking it + as used. It then replies to the user to confirm that the action has been completed. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the media or the reply. + """ + await worker.send_afternoon_media(client, user_id="me", set_as_used=False) + await event.reply("Done!") + + +@client.on(events.NewMessage("me", pattern="/stats")) +async def handle_stats(event: Message | events.NewMessage): + """Handles the /stats command to send statistics to the user. + + This asynchronous function listens for messages containing the /stats command and triggers the + sending of statistics related to the application via the Telegram client. It provides users with + insights into the current state of the application. + + Args: + event (Message | events.NewMessage): The event object representing the incoming message. + + Raises: + Exception: If there is an error while sending the statistics. + """ + await worker.send_stats(client, user_id="me") + + +async def main(): + """Main entry point for starting the Telegram bot and scheduling tasks. + + This asynchronous function initializes the scheduler and starts the various tasks for sending + morning greetings, afternoon media, and pill reminders. It then enters an infinite loop to keep + the application running and responsive. + """ + scheduler = AsyncIOScheduler() + worker.start_sending_morning_greeting(scheduler, client, try_today=True) + worker.start_sending_afternoon_media(scheduler, client, try_today=True) + worker.start_sending_pills_reminder(scheduler, client) + scheduler.start() + while True: + await asyncio.sleep(1) + + +client.loop.run_until_complete(main()) diff --git a/manage.py b/manage.py new file mode 100644 index 0000000..358d706 --- /dev/null +++ b/manage.py @@ -0,0 +1,126 @@ +"""This module serves as the main entry point for the Telegram bot application. + +It handles the initialization of the command-line interface, sets up argument parsing, and invokes +the appropriate command functions based on user input. The module integrates with various worker +functions to perform tasks such as sending messages and managing media, while providing a structured +way to interact with the application through command-line commands. +""" + +import inspect +import subprocess +import sys +from argparse import ( + ArgumentDefaultsHelpFormatter, + ArgumentParser, + Namespace, + _SubParsersAction, +) + + +def deploy(args: Namespace): + """Deploys the Docker container for the Telegram Auto Texter application. + + This function checks for any running instances of the Telegram Auto Texter container and + optionally copies the register file from the container to the local directory. It then builds + a new Docker image with the specified name and tag, and runs the container in detached mode. + + Args: + args (argparse.Namespace): The command-line arguments containing options for deployment, + including whether to copy the register and the name and tag for the Docker image. + + Raises: + subprocess.CalledProcessError: If any of the subprocess commands fail during execution. + """ + proc = subprocess.Popen( + [ + "docker", + "ps", + "-q", + "-f", + "ancestor=telegram-auto-texter", + "-f", + "status=running", + ], + stdout=subprocess.PIPE, + shell=True, + ) + out, err = proc.communicate() + cid = out.decode("utf-8").strip() + if cid != "": + if args.register: + subprocess.run( + [ + "docker", + "container", + "cp", + f"{cid}:/telegram-auto-texter/src/data/register.yaml", + "src/", + ] + ) + subprocess.run(["docker", "stop", cid]) + + subprocess.run(["docker", "build", ".", "-t" f"{args.name}:{args.tag}"]) + subprocess.run(["docker", "run", "-d", f"{args.name}:{args.tag}"]) + + +def deploy_cli(subparsers: _SubParsersAction): + """Sets up the command-line interface for the deploy command. + + This function configures the argument parser for the deploy command, allowing users to specify + deployment settings such as the image name and tag. It also includes an option to disable + registration during deployment. + + Args: + subparsers (argparse._SubParsersAction): The subparsers object used to add the deploy + command to the CLI. + """ + parser: ArgumentParser = subparsers.add_parser( + "deploy", help="Settings for deploying the project" + ) + parser.add_argument("--name", "-n", default="telegram-auto-texter", help="") + parser.add_argument("--tag", "-t", default="latest", help="") + parser.add_argument("--no-register", dest="register", action="store_false", help="") + + +def parse_args() -> Namespace: + """Parses command-line arguments for the application. + + This function sets up an argument parser with subcommands based on functions that end with + '_cli' in the current module. It returns the parsed arguments, allowing the application to + handle different commands and options provided by the user. + + Returns: + argparse.Namespace: The parsed command-line arguments. + + Raises: + SystemExit: If the command-line arguments are invalid or if a required command is not + provided. + """ + parser = ArgumentParser("qwerty", formatter_class=ArgumentDefaultsHelpFormatter) + + subparsers = parser.add_subparsers(dest="command") + for name, obj in inspect.getmembers(sys.modules[__name__]): + if inspect.isfunction(obj) and name.endswith("_cli"): + obj(subparsers) + + return parser.parse_args() + + +def main(): + """Main entry point for the application. + + This function parses the command-line arguments and invokes the corresponding command function + based on the user's input. It serves as the starting point for executing the application's + functionality. + + Raises: + AttributeError: If the command specified in the arguments does not correspond to a valid + function. + SystemExit: If there are issues with parsing the command-line arguments. + """ + args = parse_args() + getattr(sys.modules[__name__], args.command)(args) + + +if __name__ == "__main__": + main() diff --git a/readme.md b/readme.md new file mode 100644 index 0000000..5f1a257 --- /dev/null +++ b/readme.md @@ -0,0 +1,64 @@ +# Telegram Auto Texter + +Telegram Auto Texter is a Telegram user bot application designed to send automated messages, +greetings, and media at scheduled or random times. The user bot utilizes various features of the +Telegram API to interact with users and provide a seamless experience. + +## Features + +- Sends morning and afternoon messages. +- Sends media items based on user-defined schedules. +- Provides reminders for taking pills. +- Customizable greetings and media through YAML configuration files. +- Supports dynamic text generation using tokens. + +## Installation + +1. Clone the repository: + ```bash + git clone https://github.com/jromero132/telegram-auto-texter.git + cd telegram-auto-texter``` + +2. Install the required dependencies: + ```bash + pip install -r requirements.txt + ``` +3. Obtain your API id and hash and update the [config.ini](config.ini) file. + +4. Configure the user bot settings in the file: `src/data/telegram_config.yaml` + +## Usage + +To run the user bot, execute one of the following command: + +- `python -m main` +- `python main.py` + +## Commands + +- `/health`: Check if the user bot is running. +- `/send_greeting`: Send a morning greeting. +- `/send_afternoon_media`: Send an afternoon media item. +- `/greeting_info`: Get information about the time for the next greeting. +- `/afternoon_media_info`: Get information about the time for the next afternoon media. +- `/test_greeting`: Test sending a morning greeting. +- `/test_afternoon_media`: Test sending an afternoon media item. +- `/stats`: Get statistics about remaining media items. + +## Configuration + +The user bot's behavior can be customized through YAML configuration files located in the `src/data` +directory. The following files are available: + +- [telegram_config.yaml](src/data/telegram_config.yaml): Contains the user bot's configuration about +users and times. +- [stickers.yaml](src/data/stickers.yaml): Defines the sticker items available for sending. +- [media.yaml](src/data/media.yaml): Defines the media items available for sending. +- [register.yaml](src/data/register.yaml): Keeps track of sent media. + +## Contributing +Contributions are welcome! Please open an issue or submit a pull request for any enhancements or bug +fixes. + +## License +This project is licensed under the MIT License - see the [license](license) file for details. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ea8e074 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +APScheduler>=3.10.4,<3.11 +PyYAML>=6.0.2,<6.1 +Telethon>=1.36.0,<1.37 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..59d5d35 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1,7 @@ +"""This package serves as the main module for the Telegram Auto Texter application. + +It includes various submodules that handle functionality such as generating custom text, sending +scheduled messages, and managing media. The package is designed to facilitate the interaction +between different components of the application, providing a cohesive structure for the Telegram +bot's operations. +""" diff --git a/src/data/media.yaml b/src/data/media.yaml new file mode 100644 index 0000000..cae76d1 --- /dev/null +++ b/src/data/media.yaml @@ -0,0 +1,192 @@ +afternoon_media: + - path: pics/1.jpg + uid: 1 + - path: pics/2.jpg + uid: 2 + - path: pics/3.jpg + uid: 3 + - path: pics/4.jpeg + uid: 4 + - path: pics/5.jpeg + uid: 5 + - path: pics/6.jpg + uid: 6 + - path: pics/7.jpg + uid: 7 + - path: pics/8.jpg + uid: 8 + - path: pics/9.jpg + uid: 9 + - path: pics/10.jpg + uid: 10 + - path: pics/11.jpg + uid: 11 + - path: pics/12.png + uid: 12 + - path: pics/13.jpg + uid: 13 + - path: pics/14.jpg + uid: 14 + - path: pics/15.jpg + uid: 15 + - path: pics/16.jpg + uid: 16 + - path: pics/17.jpg + uid: 17 + - path: pics/18.png + uid: 18 + - path: pics/19.jpeg + uid: 19 + - path: pics/20.jpg + uid: 20 + - path: pics/21.jpg + uid: 21 + - path: pics/22.jpg + uid: 22 + - path: pics/23.jpg + uid: 23 + - path: pics/24.png + uid: 24 + - path: pics/25.png + uid: 25 + - path: pics/26.jpg + uid: 26 + - path: pics/27.jpg + uid: 27 + - path: pics/28.jpg + uid: 28 + - path: pics/29.jpg + uid: 29 + - path: pics/30.jpg + uid: 30 + - path: pics/31.png + uid: 31 + - path: pics/32.jpg + uid: 32 + - path: pics/33.jpg + uid: 33 + - path: pics/34.jpg + uid: 34 + - path: pics/35.jpg + uid: 35 + - path: pics/36.jpg + uid: 36 + - path: pics/37.png + uid: 37 + - path: pics/38.jpeg + uid: 38 + - path: pics/39.jpg + uid: 39 + - path: pics/40.jpeg + uid: 40 + - path: pics/41.jpg + uid: 41 + - path: pics/42.jpg + uid: 42 + - path: pics/43.jpg + uid: 43 + - path: pics/44.jpg + uid: 44 + - path: pics/45.jpg + uid: 45 + - path: pics/46.jpg + uid: 46 + - path: pics/47.jpg + uid: 47 + - path: pics/48.jpg + uid: 48 + - path: pics/49.jpg + uid: 49 + - path: pics/50.jpg + uid: 50 +morning_media: + - path: videos/1.mp4 + uid: 1 + - path: videos/2.mp4 + uid: 2 + - path: videos/3.mp4 + uid: 3 + - path: videos/4.mp4 + uid: 4 + - path: videos/5.mp4 + uid: 5 + - path: videos/6.gif + uid: 6 + - path: videos/7.mp4 + uid: 7 + - path: videos/8.mp4 + uid: 8 + - path: videos/9.mp4 + uid: 9 + - path: videos/10.mp4 + uid: 10 + - path: videos/11.mp4 + uid: 11 + - path: videos/12.mp4 + uid: 12 + - path: videos/13.mp4 + uid: 13 + - path: videos/14.mp4 + uid: 14 + - path: videos/15.gif + uid: 15 + - path: videos/16.mp4 + uid: 16 + - path: videos/17.mp4 + uid: 17 + - path: videos/18.mp4 + uid: 18 + - path: videos/19.mp4 + uid: 19 + - path: videos/20.mp4 + uid: 20 + - path: videos/21.mp4 + uid: 21 + - path: videos/22.mp4 + uid: 22 + - path: videos/23.mp4 + uid: 23 + - path: videos/24.mp4 + uid: 24 + - path: videos/25.mp4 + uid: 25 + - path: videos/26.mp4 + uid: 26 + - path: videos/27.mp4 + uid: 27 + - path: videos/28.mp4 + uid: 28 + - path: videos/29.mp4 + uid: 29 + - path: videos/30.mp4 + uid: 30 + - path: videos/31.mp4 + uid: 31 + - path: videos/32.mp4 + uid: 32 + - path: videos/33.mp4 + uid: 33 + - path: videos/34.mp4 + uid: 34 + - path: videos/35.mp4 + uid: 35 + - path: videos/36.mp4 + uid: 36 + - path: videos/37.mp4 + uid: 37 + - path: videos/38.mp4 + uid: 38 + - path: videos/39.mp4 + uid: 39 + - path: videos/40.mp4 + uid: 40 + - path: videos/41.mp4 + uid: 41 + - path: videos/42.mp4 + uid: 42 + - path: videos/43.mp4 + uid: 43 + - path: videos/44.mp4 + uid: 44 + - path: videos/45.mp4 + uid: 45 diff --git a/src/data/register.yaml b/src/data/register.yaml new file mode 100644 index 0000000..15b4837 --- /dev/null +++ b/src/data/register.yaml @@ -0,0 +1,29 @@ +afternoon_media: + - 7 + - 15 + - 43 + - 46 + - 49 + - 50 +morning_media: + - 1 + - 2 + - 3 + - 4 + - 5 + - 18 + - 20 + - 36 + - 37 + - 42 + - 43 +morning_stickers: + - 4 + - 9 + - 12 + - 13 + - 19 + - 20 + - 35 + - 36 + - 38 diff --git a/src/data/stickers.yaml b/src/data/stickers.yaml new file mode 100644 index 0000000..92a3d92 --- /dev/null +++ b/src/data/stickers.yaml @@ -0,0 +1,195 @@ +morning_stickers: + - access_hash: -5627161222255998402 + file_reference: "\x01\0\x1D\xCC{f\xEBw\xB6\xA6:*\x8F\xF2\r~ \xB5S\xA9yL\ + \x8F8C" + id: 2878725108866744379 + uid: 1 + - access_hash: 1032246309735062318 + file_reference: "\x01\0\x1D\xCC|f\xEBx%\x16\xD3\xE6\xA9\x17\x91L\xA4\ + \xE4ZbH\x9B\xFB\x91\xD4" + id: 2878725108866744364 + uid: 2 + - access_hash: -4402433728774906054 + file_reference: "\x01\0\x1D\xCC}f\xEBx%\xD9\x12QTmLC)G\xCC\x8F\xB0\f2`o" + id: 2878725108866744363 + uid: 3 + - access_hash: 267078162200183185 + file_reference: "\x01\0\x1D\xCC~f\xEBx%\ff\x15\xF2\xC1\x90\xF5 8|DZ\xBA\ + \x80\xE6{" + id: 2878725108866744469 + uid: 4 + - access_hash: -6461742757637301693 + file_reference: "\x01\0\x1D\xCC\x7Ff\xEBx%\NT\xCF\x819\xEF\x81\\\xD3\ + \x06i_\x92\x89\xAB`" + id: 2878725108866744401 + uid: 5 + - access_hash: -2566159793040434077 + file_reference: "\x01\0\x1D\xCC\x80f\xEBx%%\xB5\xC9\n$\x810\xB6(\xB2\x01\ + \t\xAE-\xA2E" + id: 2878725108866744412 + uid: 6 + - access_hash: 738968695546246957 + file_reference: "\x01\0\x1D\xCC\x81f\xEBx\xAF\xB0;\xB6\xCFT2\v\x84\f\xC5\ + \x10'v\xF1\x84\x82" + id: 2878725108866744419 + uid: 7 + - access_hash: -6490191748693275654 + file_reference: "\x01\0\x1D\xCC\x82f\xEBx\xAFl\x86Txf\x8D\xB6\xD7Hzf1Mt\ + \xFD\a" + id: 2878725108866744421 + uid: 8 + - access_hash: -1878428791181028524 + file_reference: "\x01\0\x1D\xCC\x83f\xEBx\xAF\xCD\xEB\xCE_\x94\xA6B\x8F[]\ + \xBC\xC3\N@AF" + id: 2878725108866744437 + uid: 9 + - access_hash: -2549081194728456176 + file_reference: "\x01\0\x1D\xCC\x84f\xEBx\xAF\xB6\x87\xB0\xB8\x91\xBBE-\ + \xCB(}\xEF\x02\xB1!\xBA" + id: 2878725108866744438 + uid: 10 + - access_hash: 6791320340357484741 + file_reference: "\x01\0\x1D\xCC\Nf\xEBx\xAF\xB3k\xC4l\xC7\xD3\xD4/Z\xAD\ + \xA1\"{\x80\x8A\xA3" + id: 2878725108866744467 + uid: 11 + - access_hash: 8522513357271411620 + file_reference: "\x01\0\x1D\xCC\x86f\xEBx\xAF\xAF\xF1\xEB\xE8b\x04\0DPf\ + \x1E\x1E\xECR\x15v" + id: 2878725108866744479 + uid: 12 + - access_hash: 767010262269838330 + file_reference: "\x01\0\x1D\xCC\x87f\xEBx\xAF5\x04U~6\xF6U\x17\x99N\xA9\ + \xC62\xC7uV" + id: 2878725108866744542 + uid: 13 + - access_hash: -2717511321736957089 + file_reference: "\x01\0\x1D\xCC\x88f\xEBx\xAFOgx\xC0@d\x0F\xE6\xDBU\t\ + \x90:_<\x8B" + id: 4902572956405728609 + uid: 14 + - access_hash: 1588505812504426031 + file_reference: "\x01\0\x1D\xCC\x89f\xEBx\xAFn\xA9\N\x88\xAAN\xCC\xBCt\ + \xC6\x04\xCB:\x84\xCA\xE3" + id: 4920563139889594958 + uid: 15 + - access_hash: -2821200479403440661 + file_reference: "\x01\0\x1D\xCC\x8Af\xEBx\xAF\xAE+b\xF8\"V\xBF52\f:m`\ + \x9BJO" + id: 4922569349178327825 + uid: 16 + - access_hash: 6320526505152251360 + file_reference: "\x01\0\x1D\xCC\x8Bf\xEBx\xAF|z\x994GcK\xDA\xD8Jy\xA2L\ + \xEE\x89\xE3" + id: 4922461270621291143 + uid: 17 + - access_hash: 50708831805944131 + file_reference: "\x01\0\x1D\xCC\x8Cf\xEBx\xAFZ\xB9x\x8ET\x883\x9D\_mm3\ + \x1C\xDE\xCC\xD9" + id: 4920193476349395294 + uid: 18 + - access_hash: -5288994271945175990 + file_reference: "\x01\0\x1D\xCC\x8Df\xEBx\xAF\x1C\x986Z\xBE\x1D\xE5\xEF\ + \x92\x8B\x81V\xA7\xC2\xA4\xA5" + id: 4922683853006439099 + uid: 19 + - access_hash: 2087271789403080725 + file_reference: "\x01\0\x1D\xCC\x8Ef\xEBx\xAFW\xC6tG%QO[\t\xB4%\x12\xE3\ + \xA4.?" + id: 4920410020010525261 + uid: 20 + - access_hash: -1838497278476067987 + file_reference: "\x01\0\x1D\xCC\x8Ff\xEBx\xAF\xA9u\b\xEC\xCA2![Hr]\xA7\ + \xB9\xD2\xB1\xEF" + id: 4920306180586209944 + uid: 21 + - access_hash: 8391769485784589995 + file_reference: "\x01\0\x1D\xCC\x90f\xEBx\xAFhtc7^5\x83@\x15\xF9\x91\xB3\ + |X\xE2s" + id: 4945187819485135528 + uid: 22 + - access_hash: 3765858981752623683 + file_reference: "\x01\0\x1D\xCC\x91f\xEBx\xAF\xEF\xED\xF7#?\xC6\xDCAa\xFA\ + \x9A\"KG\xB7\xF7" + id: 972890833634197622 + uid: 23 + - access_hash: -2743966363538393491 + file_reference: "\x01\0\x1D\xCC\x92f\xEBx\xAF,\x1C\xE2\xD5\xAA9\xBB!\ + \xF3|=\x8F\x83\x87\xE8Y" + id: 3904618420208927435 + uid: 24 + - access_hash: -8689681307713341259 + file_reference: "\x01\0\x1D\xCC\x93f\xEBx\xAF5\xD0V\xD7\x16 Cy\xCC\x8DK\ + \x12\x91\xBD\x82\xC7" + id: 1360886329340073317 + uid: 25 + - access_hash: 528553206124457461 + file_reference: "\x01\0\x1D\xCC\x94f\xEBx\xAF\xC3\f~s\x93\xC8\xBA~\xD9\ + \x8C-\xB7(\xDEc\xC3" + id: 1360886329340073311 + uid: 26 + - access_hash: 4036201942265069558 + file_reference: "\x01\0\x1D\xCC\x95f\xEBx\xAF\xF2\xCF\xF7z\xBC!\xF7l\ + \xA7J?n\x92\xA6:<" + id: 1360886329340073324 + uid: 27 + - access_hash: 4420978996600506928 + file_reference: "\x01\0\x1D\xCC\x96f\xEBx\xAF\v\xC18\x8E=\xA5\xC0@\xD6\ + \xF4\xD3\x7Fb\xCByt" + id: 1360886329340073327 + uid: 28 + - access_hash: -5815216410399984022 + file_reference: "\x01\0\x1D\xCC\x97f\xEBx\xAF\xC2r)v\\\v\xCB\x11\eb\xF5)\ + \x11P\v\xC8" + id: 1360886329340073329 + uid: 29 + - access_hash: 4660709603466034225 + file_reference: "\x01\0\x1D\xCC\x98f\xEBx\xAF\_\xA1s\x0F\x93\xE1\x98@c!\ + \xB1\x02\xBF\x8D\x9B\xC8" + id: 1360886329340073336 + uid: 30 + - access_hash: 6085271430333172616 + file_reference: "\x01\0\x1D\xCC\x99f\xEBx\xAF\b`q\xC7\xCDR\x8C\x18[\ + \xD3yy]\xD2wV" + id: 1360886329340073337 + uid: 31 + - access_hash: -9053746079009940822 + file_reference: "\x01\0\x1D\xCC\x9Af\xEBx\xAF\x94;\xC3\xC5\a\xFC\x8E+{\ + \xEE>u\xF3\xE0\x12\x04" + id: 1360886329340073338 + uid: 32 + - access_hash: -5532619797837032256 + file_reference: "\x01\0\x1D\xCC\x9Bf\xEBx\xAFvL>T\xC8m\x8F\xF7\xF0\x8D\ + \x11\xEE\xB7t\x8D\xA1" + id: 1220028319907447506 + uid: 33 + - access_hash: -2608398328893441121 + file_reference: "\x01\0\x1D\xCC\x9Cf\xEBx\xAF\xD8\x97\x0F\xEDc\a\xDEA\ + \xD4\xD8}J\"G\xE2\xB3" + id: 1220028319907447510 + uid: 34 + - access_hash: 183651946996935583 + file_reference: "\x01\0\x1D\xCC\x9Df\xEBx\xAF\xCE\x8FU[n\xCB\xEC\xB2_\xAC\ + \x8AP\xC5\x91\xBA\\" + id: 1220028319907447511 + uid: 35 + - access_hash: -306141303989158749 + file_reference: "\x01\0\x1D\xCC\x9Ef\xEBx\xAF\xC47=\xDDDmzA\xE0\xA7\xDEff\ + \xAB\xF0\x0E" + id: 1220028319907447512 + uid: 36 + - access_hash: -7364561842178901006 + file_reference: "\x01\0\x1D\xCC\x9Ff\xEBx\xAF\x027\x8Enb\xB6\x80k\xBC\ + \x80o\xD5\xFB*\xC1E" + id: 1220028319907447515 + uid: 37 + - access_hash: -6773522358705117618 + file_reference: "\x01\0\x1D\xCC\_f\xEBx\xAF\xDA\v\xAA2\x1C\x935\xE6\xEC\ + \xB1W\b\0\x1Do\xE7" + id: 1220028319907447516 + uid: 38 + - access_hash: -4893348581376023534 + file_reference: "\x01\0\x1D\xCC\xA1f\xEBx\xAFP\xE6\x89E\x95_E9\xB7\xFFM\ + \xEA\xF8\xEEOR" + id: 1220028319907448670 + uid: 39 diff --git a/src/data/telegram_config.yaml b/src/data/telegram_config.yaml new file mode 100644 index 0000000..f7bd83f --- /dev/null +++ b/src/data/telegram_config.yaml @@ -0,0 +1,12 @@ +me: + chat_id: "me" +nathy: + chat_id: CHAT_ID + morning_greeting: + start_time: "07:00:00" + end_time: "08:00:00" + afternoon_media: + start_time: "18:00:00" + end_time: "22:00:00" + pills_reminder: + time: "21:45:00" diff --git a/src/sentence_generator/__init__.py b/src/sentence_generator/__init__.py new file mode 100644 index 0000000..fbf31da --- /dev/null +++ b/src/sentence_generator/__init__.py @@ -0,0 +1,6 @@ +"""This module provides classes and functions for generating custom text and sentences. + +It includes components for creating dynamic text based on predefined tokens and relationships, +allowing for flexible and creative sentence generation. The module is designed to facilitate the +construction of sentences using various algorithms and token management strategies. +""" diff --git a/src/sentence_generator/morning.py b/src/sentence_generator/morning.py new file mode 100644 index 0000000..fd7f0ee --- /dev/null +++ b/src/sentence_generator/morning.py @@ -0,0 +1,210 @@ +"""This module contains functionality for sending morning greeting messages. + +It includes the scheduling of morning greetings using a specified time range and a Telegram client. +The module leverages the APScheduler for job scheduling and allows for dynamic configuration of +greeting times, enabling automated daily greetings to be sent at random times within the defined +interval. +""" + +from src.sentence_generator.sentence_generator import ( + CustomGeneratedText, + SentenceGenerator, + Token, +) +from src.utils.random import low_random + +amor = Token("amor") +mi_amor = Token("mi amor") + +amorcita = Token("amorcita") +mi_amorcita = Token("mi amorcita") + +amorcito = Token("amorcito") +mi_amorcito = Token("mi amorcito") + +amooor = Token(CustomGeneratedText(lambda: "am" + "o" * low_random(1, 10) + "r")) +mi_amooor = Token(CustomGeneratedText(lambda: "mi am" + "o" * low_random(1, 10) + "r")) + +amorcitaaa = Token(CustomGeneratedText(lambda: "amorcit" + "a" * low_random(1, 10))) +mi_amorcitaaa = Token(CustomGeneratedText(lambda: "mi amorcit" + "a" * low_random(1, 10))) + +amorcitooo = Token(CustomGeneratedText(lambda: "amorcit" + "o" * low_random(1, 10))) +mi_amorcitooo = Token(CustomGeneratedText(lambda: "mi amorcit" + "o" * low_random(1, 10))) + +mailob = Token("mailob") +my_love = Token("my love") + +preciosura_tropical = Token("preciosura tropical") + + +general_right_part: list[Token] = [ + amooor, + mi_amooor, + amorcitaaa, + mi_amorcitaaa, + amorcitooo, + mi_amorcitooo, + mailob, + my_love, + preciosura_tropical, +] + + +hola = Token(CustomGeneratedText(lambda: "Hol" + "a" * low_random(1, 10))).add_next_tokens( + *general_right_part +) +buenos_dias = Token("Buenos días").add_next_tokens( + *general_right_part, + Token("señorita"), + Token("princesita"), +) +wenas = ( + Token("Wenas") + .add_next_tokens( + *[ + Token( + token.value, + pick_next=lambda tokens: tokens[low_random(0, len(tokens) - 1)], + ).add_next_tokens( + Token(""), + Token(CustomGeneratedText(lambda: "wen" + "a" * low_random(1, 5) + "s")), + ) + for token in general_right_part + ] + ) + .add_next_token(preciosura_tropical) +) +guenas = Token("Güenas").add_next_tokens( + *general_right_part, +) +jelou = Token("Jelou").add_next_tokens( + *general_right_part, +) +jeloucito = Token("Jeloucito").add_next_tokens( + mi_amorcitooo, + mailob, + my_love, + preciosura_tropical, +) +gusmornin = Token("Gusmornin").add_next_tokens( + *general_right_part, +) +good_morning = Token("Good morning").add_next_tokens( + *general_right_part, +) +god_morgon = Token("God morgon").add_next_tokens(*general_right_part, Token("prinsessa")) +bonjour = Token("Bonjour").add_next_tokens( + Token("mademoiselle"), + Token("princesse"), +) +buongiorno = Token("Buongiorno").add_next_tokens( + Token("principessa"), + Token("signorina"), +) + + +sentences = SentenceGenerator( + [ + hola, + buenos_dias, + wenas, + guenas, + jelou, + gusmornin, + good_morning, + god_morgon, + bonjour, + buongiorno, + ] +) + + +def get_morning_greeting() -> str: + """Get a random good morning message along with emojis. + + This function retrieves a random good morning greeting and appends a selection of random emojis + to enhance the message. It ensures that at least one emoji is included in the final output. + + Returns: + str: A string containing a random good morning message followed by a selection of emojis. + """ + greeting: str = sentences.eval() + + emoji1: list[str] = [ + "", + "\U0001f44b", # 👋 + "\U0000270c", # Hand with 2 raised fingers + ] + emoji2: list[str] = [ + "", + "\U0001f61b", # 😛 + "\U0001f61d", # 😝 + "\U0001f92a", # 🤪 + "\U0001f60b", # 😋 + ] + emoji3: list[str] = [ + "", + "\U0001f643", # 🙃 + "\U0001f601", # 😁 + "\U0001f604", # 😄 + "\U0001f603", # 😃 + ] + emoji4: list[str] = [ + "", + "\U0001f917", # 🤗 + "\U0001f61a", # 😚 + "\U0001f60a", # 😊 + "\U0000263a", # ☺ smiling face + "\U0001f92d", # 🤭 + ] + emoji5: list[str] = [ + "", + "\U0001f970", # 🥰 + "\U0001f618", # 😘 + ] + emoji6: list[str] = [ + "", + "\U0001faf6", # 🫶 + "\U00002764", # ❤ red heart + ] + + def get_emoji(emojis: list[str], p: float) -> str: + """Gets a random emoji from a list based on a specified probability. + + This function selects an emoji from the provided list using a random selection method that + takes into account the specified probability. The probability influences the likelihood of + each emoji being chosen, allowing for customized emoji selection. + + Args: + emojis (list[str]): The list of emojis to choose from. + p (float): The probability factor that influences the selection of the emoji. Refer to + `utils.random.low_random` documentation. + + Returns: + str: The randomly selected emoji from the list. + """ + return emojis[low_random(1, len(emojis), p) - 1] + + def get_all_emojis() -> str: + """Get a random text made of emojis. + + This function generates a string composed of randomly selected emojis from predefined lists. + It combines multiple emojis to create a fun and varied emoji text output. + + Returns: + str: The random text made of emojis. + """ + return ( + get_emoji(emoji1, 3.5) + + get_emoji(emoji2, 5) + + get_emoji(emoji3, 4) + + get_emoji(emoji4, 3) + + get_emoji(emoji5, 1.3) + + get_emoji(emoji6, 2) + ) + + emojis: str = "" + while not emojis: + emojis = get_all_emojis() + + return f"{greeting} {emojis}" diff --git a/src/sentence_generator/sentence_generator.py b/src/sentence_generator/sentence_generator.py new file mode 100644 index 0000000..c18c499 --- /dev/null +++ b/src/sentence_generator/sentence_generator.py @@ -0,0 +1,171 @@ +"""This module provides classes for generating custom text and sentences using tokens. + +It includes the `CustomGeneratedText` class for creating text from a generator function, the `Token` +class for representing individual tokens in a sentence, and the `SentenceGenerator` class for +assembling sentences from a list of tokens. These components work together to allow for flexible and +dynamic sentence generation based on defined token relationships and selection algorithms. +""" + +from collections.abc import Callable +from random import choice +from typing import Union + + +class CustomGeneratedText: + """Represents a custom generated text. + + This class encapsulates a text generator function, allowing for the creation of dynamic text + based on the logic defined within the generator. It provides a simple interface to retrieve + the generated text as a string. + """ + + def __init__(self, generator: Callable[[], str]): + """Creates a custom generated text class. + + This constructor initializes the `CustomGeneratedText` instance with a specified text + generator function. + + Args: + generator (Callable[[], str]): The text generator function that produces a string when + called. + """ + self.generator = generator + + def __repr__(self) -> str: + """Return the generated text as a string. + + This method calls the generator function and returns the resulting string. + + Returns: + str: The generated text from the generator function. + """ + return str(self.generator()) + + +class Token: + """Represents a token of a sentence. + + This class encapsulates a single unit of a sentence, which can be a word or a generated text. + It manages the value of the token and the possible next tokens that can follow it, allowing + for the construction of dynamic and flexible sentence structures. + """ + + def __init__( + self, + value: str | CustomGeneratedText, + pick_next: Callable[[list["Token"]], "Token"] = choice, + ): + """Creates a token of a sentence. + + This constructor initializes the `Token` instance with a specified value and a method for + selecting the next token in the sequence. It allows for the creation of complex sentence + structures by linking multiple tokens together. + + Args: + value (str | CustomGeneratedText): The value of the token, which can be a string or a + custom generated text. + pick_next (Callable[[list[Token]], Token], optional): Algorithm to pick the next token + after this one. Defaults to random.choice. + """ + super().__init__() + self.value = value + self.pick_next = pick_next + self.next_tokens: list[Token] = [] + + def eval(self) -> str: + """Evaluate this token. + + This method processes the token's value and appends the value of the next token, if + available. It constructs a string representation of the token and its subsequent tokens, + forming part of a complete sentence. + + Returns: + str: The evaluated string representation of the token and its next tokens. + """ + left = str(self.value) + next_token = self.get_next_token() + right = "" if next_token is None else next_token.eval() + return f"{left}{' ' if left != '' and right != '' else ''}{right}" + + def get_next_token(self) -> Union["Token", None]: + """Gets the next token, if any. + + This method retrieves the next token in the sequence based on the defined selection + algorithm. If no next tokens are available, it returns None. + + Returns: + Union[Token, None]: The next token, if available; otherwise, None. + """ + return None if len(self.next_tokens) == 0 else self.pick_next(self.next_tokens) + + def add_next_token(self, next_token: "Token") -> "Token": + """Adds a possible next token to this one. + + This method allows for the addition of a single next token to the current token, enabling + the construction of a sequence of tokens. + + Args: + next_token (Token): The possible next token to be added. + + Returns: + Token: Returns itself, allowing for method chaining in a builder design pattern. + """ + self.next_tokens.append(next_token) + return self + + def add_next_tokens(self, *next_tokens: "Token") -> "Token": + """Adds multiple possible next tokens to this one. + + This method allows for the addition of multiple next tokens to the current token, + facilitating the creation of more complex token sequences. + + Args: + _ (Token): Possible next tokens to be added. + + Returns: + Token: Returns itself, allowing for method chaining in a builder design pattern. + """ + self.next_tokens.extend(next_tokens) + return self + + +class SentenceGenerator: + """Represents a sentence generator algorithm class. + + This class is responsible for generating sentences by evaluating a set of initial tokens. + It utilizes a selection algorithm to determine the root token and constructs sentences + dynamically based on the relationships between tokens, allowing for varied and creative sentence + generation. + """ + + def __init__( + self, + initial_tokens: list[Token], + pick_root: Callable[[list["Token"]], "Token"] = choice, + ): + """Creates a sentence generator algorithm class. + + This constructor initializes the `SentenceGenerator` with a list of initial tokens and a + method for selecting the root token of the sentence. It sets up the necessary components for + generating sentences based on the provided tokens. + + Args: + initial_tokens (list[Token]): List of all possible initial tokens for sentence + generation. + pick_root (Callable[[list[Token]], Token], optional): Algorithm to pick the root token + of the sentence. Defaults to random.choice. + """ + self.initial_tokens = initial_tokens + self.pick_root = pick_root + + def eval(self) -> str: + """Evaluate this sentence generator. + + This method generates a complete sentence by selecting a root token and evaluating it along + with its subsequent tokens. It constructs the final string representation of the generated + sentence. + + Returns: + str: The generated sentence based on the initial tokens and their relationships. + """ + return "" if len(self.initial_tokens) == 0 else self.pick_root(self.initial_tokens).eval() diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..c3bec65 --- /dev/null +++ b/src/utils/__init__.py @@ -0,0 +1,6 @@ +"""This module provides utility functions for the Telegram Auto Texter application. + +It includes various helper functions that assist with tasks such as random value generation, YAML +file handling, and time manipulation. These utilities are designed to simplify and streamline the +main application logic by providing reusable components. +""" diff --git a/src/utils/random.py b/src/utils/random.py new file mode 100644 index 0000000..91bdc81 --- /dev/null +++ b/src/utils/random.py @@ -0,0 +1,55 @@ +"""This module contains utility functions for generating random values. + +It includes the `low_random` function, which generates a low-biased random integer within a +specified range, and the `random_time` function, which generates a random time duration between two +specified time intervals. These functions can be used to introduce variability and randomness in +applications requiring random number generation. +""" + +import random +from datetime import timedelta + + +def low_random(a: int, b: int, p: float = 2) -> int: + """Generates a low-biased random integer between two specified values. + + This function skews the randomness towards the lower end, upper end or uniform distribution of + the specified range. + + The generated integer is calculated using a power function to adjust the distribution, + making it more likely to return values closer to the lower bound, upper bound or uniform + distribution according to the parameter p. + + Args: + a (int): The lower bound of the random integer range. + b (int): The upper bound of the random integer range. + p (float, optional): The power parameter that influences the skewness of the distribution. + Always greater than 0. + If p = 1 then this is the uniform distribution. + If p > 1 makes it more likely to return values closer to the lower bound. + If p < 1 makes it more likely to return values closer to the upper bound. + Defaults to 2. + + Returns: + int: A low-biased random integer within the specified range [a, b]. + """ + return a + int((b - a + 1) * (random.random() ** p)) + + +def random_time(start: timedelta, end: timedelta) -> timedelta: + """Generates a random time duration between two specified time intervals. + + This function returns a random timedelta that falls within the range defined by the start and + end parameters. + + The generated random time is inclusive on both the start and end time, allowing for variability + in time calculations. + + Args: + start (timedelta): The lower bound of the time interval. + end (timedelta): The upper bound of the time interval. + + Returns: + timedelta: A random timedelta representing a duration within the specified range. + """ + return timedelta(seconds=random.randint(int(start.total_seconds()), int(end.total_seconds()))) diff --git a/src/worker.py b/src/worker.py new file mode 100644 index 0000000..bb44569 --- /dev/null +++ b/src/worker.py @@ -0,0 +1,557 @@ +"""This module provides functionality for sending scheduled messages and media via Telegram. + +It includes functions for sending morning greetings and afternoon messages, media items, and +reminders, as well as managing the state of sent items. The module utilizes YAML files for +configuration and data storage, and it integrates with the APScheduler for scheduling tasks. +""" + +import bisect +import random +from datetime import datetime, timedelta +from pathlib import Path + +import yaml +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from telethon import TelegramClient +from telethon.tl.types import InputDocument + +from src.sentence_generator import morning +from src.utils.random import random_time + +DATA_PATH = Path("src/data") +MEDIA_PATH = DATA_PATH / "media" +MEDIA_YAML_PATH = DATA_PATH / "media.yaml" +REGISTER_YAML_PATH = DATA_PATH / "register.yaml" +STICKERS_YAML_PATH = DATA_PATH / "stickers.yaml" +TELEGRAM_CONFIG_PATH = DATA_PATH / "telegram_config.yaml" + +next_greeting_time: datetime = datetime.now() +next_afternoon_media_time: datetime = datetime.now() + + +class CustomYamlDumper(yaml.Dumper): + """Custom YAML dumper that modifies the indentation behavior. + + This class extends the default YAML dumper to ensure that the indentation is always increased + for block styles, making the output more readable and consistent. It overrides the + `increase_indent` method to customize the indentation settings. + """ + + def increase_indent(self, flow=False, *args, **kwargs): + """Increase the indentation level for YAML output. + + This method modifies the behavior of indentation to ensure that it is not indentless, + providing a clearer structure in the generated YAML. + + Args: + flow (bool, optional): Indicates whether the indentation is for flow style. Defaults to + False. + """ + return super().increase_indent(flow=flow, indentless=False) + + +def read_yaml(path: Path, *, encoding: str = "utf-8") -> dict: + """Reads a YAML file and returns its contents as a dictionary. + + This function loads the specified YAML file from the given path and parses its content into + a Python dictionary, allowing for easy access to configuration settings. + + Args: + path (Path): The path to the YAML file to be read. + encoding (str, optional): The character encoding to use when reading the file. Defaults to + "utf-8". + + Returns: + dict: A dictionary containing the contents of the YAML file. + + Raises: + FileNotFoundError: If the specified YAML file does not exist. + yaml.YAMLError: If there is an error parsing the YAML file. + """ + with open(path, encoding=encoding) as f: + return yaml.safe_load(f) + + +def save_yaml(data: dict, path: Path, *, encoding: str = "utf-8"): + """Saves a Python dictionary as a YAML file. + + This function serializes the provided dictionary and writes it to the specified file path + in YAML format, allowing for easy storage and retrieval of structured data. + + Args: + data (dict): The Python object to be saved as YAML. + path (Path): The path to the YAML file where the data will be saved. + encoding (str, optional): The character encoding to use when writing the file. Defaults to + "utf-8". + """ + with open(path, "w", encoding=encoding) as f: + yaml.dump(data, f, Dumper=CustomYamlDumper) + + +def filter_by_register(data: dict | list, register: list["int | str"], *, key="uid") -> list: + """Filters the data to exclude items present in the register. + + This function checks the provided data against a list of identifiers in the register and + returns a new list containing only those items that are not in the register. It allows for + flexible filtering based on a specified key. + + Args: + data (dict | list): The data to be filtered, which can be a dictionary or a list. + register (list[int | str]): A list of identifiers to filter out from the data. + key (str, optional): The key used to identify items in the data for filtering. Defaults to + "uid". + + Returns: + list: A list of items from the data that are not present in the register. + """ + _register = set(register) + return [d for d in data if (d if key is None else d[key]) not in _register] + + +def text_to_timedelta(text: str) -> timedelta: + """Converts a text representation of time into a timedelta object. + + This function takes a string formatted as "HH:MM:SS" and converts it into a timedelta object, + allowing for easy manipulation of time intervals in Python. + + Args: + text (str): The text representation of time to be parsed. + + Returns: + timedelta: A timedelta object representing the parsed time. + + Raises: + ValueError: If the input string is not in the expected format or cannot be converted to + integers. + """ + _time: list[int] = [int(x) for x in text.split(":")] + return timedelta(hours=_time[0], minutes=_time[1], seconds=_time[2]) + + +def get_next_time(tm: timedelta, timespan: timedelta, try_now=False) -> datetime: + """Calculates the next occurrence of a specified time based on the current time. + + This function determines the next datetime by adding a specified time duration to the current + time and adjusting it according to the provided time interval. It can optionally consider the + current time as a starting point for the calculation. + + Args: + tm (timedelta): The time of day to calculate the next occurrence for. + timespan (timedelta): The interval to add if the calculated time is in the past. + try_now (bool, optional): If True, the current time is used as the starting point; + otherwise, the timespan is added to the current time. Defaults to False. + + Returns: + datetime: The next occurrence of the specified time as a datetime object. + """ + now = datetime.now() + if not try_now: + now += timespan + + dt = datetime( + year=now.year, + month=now.month, + day=now.day, + hour=tm.seconds // 3600, + minute=(tm.seconds % 3600) // 60, + second=tm.seconds % 60, + ) + + if dt < datetime.now(): + dt += timespan + + return dt + + +def health() -> str: + """Returns the health status of the application. + + This function provides a simple check to indicate that the application is running and + operational. It returns a string message confirming the application's status. + + Returns: + str: A message indicating that the application is alive. + """ + return "Alive" + + +def get_morning_greeting() -> str: + """Retrieves a morning greeting message. + + This function acts as a wrapper to obtain a good morning greeting from the `morning` module. It + provides a simple interface to access the greeting functionality. + + Returns: + str: A morning greeting message. + """ + return morning.get_morning_greeting() + + +def get_morning_sticker() -> dict: + """Retrieves a random morning sticker that has not been sent yet. + + This function reads the list of morning stickers from a YAML file and filters out those that + have already been sent. It then randomly selects one of the remaining stickers and prepares it + for use by encoding its file reference. + + Returns: + dict: A dictionary representing the selected morning sticker, + including its file reference and other associated data. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + ValueError: If there are no available morning stickers to choose from. + """ + morning_stickers_sent: list = read_yaml(REGISTER_YAML_PATH).get("morning_stickers", []) + morning_stickers: list = filter_by_register( + data=read_yaml(STICKERS_YAML_PATH, encoding="latin-1")["morning_stickers"], + register=morning_stickers_sent, + ) + morning_sticker: dict = random.choice(morning_stickers) + morning_sticker["file_reference"] = morning_sticker["file_reference"].encode("latin-1") + return morning_sticker + + +def get_morning_media() -> dict: + """Retrieves a random morning media item that has not been sent yet. + + This function reads the list of morning media items from a YAML file and filters out those that + have already been sent. It then randomly selects one of the remaining media items for use. + + Returns: + dict: A dictionary representing the selected morning media item, including its associated + data. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + ValueError: If there are no available morning media items to choose from. + """ + morning_media_sent = read_yaml(REGISTER_YAML_PATH).get("morning_media", []) + morning_media = filter_by_register( + data=read_yaml(MEDIA_YAML_PATH)["morning_media"], + register=morning_media_sent, + ) + return random.choice(morning_media) + + +def get_afternoon_media() -> dict: + """Retrieves a random afternoon media item that has not been sent yet. + + This function reads the list of afternoon media items from a YAML file and filters out those + that have already been sent. It then randomly selects one of the remaining media items for use. + + Returns: + dict: A dictionary representing the selected afternoon media item, including its associated + data. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + ValueError: If there are no available afternoon media items to choose from. + """ + afternoon_media_sent = read_yaml(REGISTER_YAML_PATH).get("afternoon_media", []) + afternoon_media = filter_by_register( + data=read_yaml(MEDIA_YAML_PATH)["afternoon_media"], + register=afternoon_media_sent, + ) + return random.choice(afternoon_media) + + +def set_as_used(entry: str, uid: int, db_yaml: Path): + """Marks a specified entry as used by adding a unique ID to the register. + + This function updates the register by adding the provided unique ID to the list of used IDs for + the specified entry. If the number of used IDs matches the total entries in the database, it + clears the list for that entry. + + Args: + entry (str): The key in the register that corresponds to the entry being updated. + uid (int): The unique ID to be marked as used. + db_yaml (Path): The path to the YAML file containing the database entries. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + """ + register = read_yaml(REGISTER_YAML_PATH) + if uid not in register[entry]: + bisect.insort(register[entry], uid) + data = read_yaml(db_yaml) + if len(register[entry]) == len(data[entry]): + register[entry] = [] + + save_yaml(register, REGISTER_YAML_PATH) + + +def set_morning_sticker_as_used(uid: int): + """Marks a morning sticker as used for a specified unique ID. + + This function updates the register to indicate that the morning sticker has been used by the + provided unique ID. It calls the `set_as_used` function with the appropriate parameters to + perform the update. + + Args: + uid (int): The unique ID to be marked as having used the morning sticker. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + """ + set_as_used("morning_stickers", uid, STICKERS_YAML_PATH) + + +def set_morning_media_as_used(uid): + """Marks a morning media item as used for a specified unique ID. + + This function updates the register to indicate that the morning media item has been used by the + provided unique ID. It calls the `set_as_used` function with the appropriate parameters to + perform the update. + + Args: + uid (int): The unique ID to be marked as having used the morning media item. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + """ + set_as_used("morning_media", uid, MEDIA_YAML_PATH) + + +def set_afternoon_media_as_used(uid): + """Marks an afternoon media item as used for a specified unique ID. + + This function updates the register to indicate that the afternoon media item has been used by + the provided unique ID. It calls the `set_as_used` function with the appropriate parameters to + perform the update. + + Args: + uid (int): The unique ID to be marked as having used the afternoon media item. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + """ + set_as_used("afternoon_media", uid, MEDIA_YAML_PATH) + + +async def send_morning_greeting( + client: TelegramClient, user_id: str = "nathy", set_as_used: bool = True +): + """Sends a morning greeting message along with a sticker and media to a user. + + This asynchronous function retrieves a morning greeting message, a sticker, and media, then + sends them to the specified user via a Telegram client. It can also mark the sticker and media + as used if specified. + + Args: + client (TelegramClient): The Telegram client used to send messages. + user_id (str, optional): The identifier for the user to whom the greeting is sent. + Defaults to "nathy". + set_as_used (bool, optional): Indicates whether to mark the sticker and media as used after + sending. Defaults to True. + + Raises: + FileNotFoundError: If the configuration or media files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML configuration. + Exception: If there is an error sending messages through the Telegram client. + """ + tconfig = read_yaml(TELEGRAM_CONFIG_PATH) + user_id = tconfig.get(user_id, {}).get("chat_id") + + msg = get_morning_greeting() + sticker = get_morning_sticker() + media = get_morning_media() + + await client.send_message(user_id, msg) + await client.send_message( + user_id, + file=InputDocument( + id=sticker["id"], + access_hash=sticker["access_hash"], + file_reference=sticker["file_reference"], + ), + ) + await client.send_file(user_id, MEDIA_PATH / media["path"]) + + if set_as_used: + set_morning_sticker_as_used(sticker["uid"]) + set_morning_media_as_used(media["uid"]) + + +def start_sending_morning_greeting( + scheduler: AsyncIOScheduler, client: TelegramClient, try_today: bool = False +): + """Schedules the sending of morning greeting messages via Telegram. + + This function retrieves the configured start and end times for morning greetings, calculates a + random time within that range, and schedules the greeting to be sent using the provided + scheduler. It can also attempt to send the greeting for today if specified. + + Args: + scheduler (AsyncIOScheduler): The scheduler instance used to manage scheduled jobs. + client (TelegramClient): The Telegram client used to send messages. + try_today (bool, optional): If True, sends the greeting today if the calculated time has not + passed. Defaults to False. + """ + tconfig = read_yaml(TELEGRAM_CONFIG_PATH) + start_time = text_to_timedelta( + tconfig.get("nathy", {}).get("morning_greeting").get("start_time") + ) + end_time = text_to_timedelta(tconfig.get("nathy", {}).get("morning_greeting").get("end_time")) + tm = random_time(start=start_time, end=end_time) + dt = get_next_time(tm, timedelta(days=1), try_today) + + async def wrap(scheduler, client): + """Wraps the process of sending a morning greeting and rescheduling it. + + This asynchronous function sends a morning greeting message using the provided Telegram + client and then schedules the next morning greeting using the specified scheduler. It + ensures that the greeting process is repeated at the appropriate time. + + Args: + scheduler: The scheduler instance used to manage the scheduling of jobs. + client: The Telegram client used to send the greeting message. + """ + await send_morning_greeting(client) + start_sending_morning_greeting(scheduler, client) + + global next_greeting_time + next_greeting_time = dt + print(f"Next greeting at {dt}") + scheduler.add_job(wrap, "date", run_date=dt, args=[scheduler, client]) + + +async def send_afternoon_media(client, user_id="nathy", set_as_used=True): + """Sends an afternoon media item to a specified user. + + This asynchronous function retrieves an afternoon media item and sends it to the specified user + via a Telegram client. It can also mark the media item as used if specified. + + Args: + client: The Telegram client used to send the media. + user_id (str, optional): The identifier for the user to whom the media is sent. Defaults to + "nathy". + set_as_used (bool, optional): Indicates whether to mark the media item as used after + sending. Defaults to True. + + Raises: + FileNotFoundError: If the configuration or media files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML configuration. + Exception: If there is an error sending the media through the Telegram client. + """ + tconfig = read_yaml(TELEGRAM_CONFIG_PATH) + user_id = tconfig.get(user_id, {}).get("chat_id") + media = get_afternoon_media() + await client.send_file(user_id, MEDIA_PATH / media["path"]) + if set_as_used: + set_afternoon_media_as_used(media["uid"]) + + +def start_sending_afternoon_media(scheduler, client, try_today=False): + """Schedules the sending of afternoon media messages via Telegram. + + This function retrieves the configured start and end times for afternoon media, calculates a + random time within that range, and schedules the media to be sent using the provided scheduler. + It can also attempt to send the media for today if specified. + + Args: + scheduler: The scheduler instance used to manage scheduled jobs. + client: The Telegram client used to send messages. + try_today (bool, optional): If True, sends the media today if the calculated time has not + passed. Defaults to False. + + Raises: + FileNotFoundError: If the configuration files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML configuration. + """ + tconfig = read_yaml(TELEGRAM_CONFIG_PATH) + start_time = text_to_timedelta( + tconfig.get("nathy", {}).get("afternoon_media").get("start_time") + ) + end_time = text_to_timedelta(tconfig.get("nathy", {}).get("afternoon_media").get("end_time")) + tm = random_time(start=start_time, end=end_time) + dt = get_next_time(tm, timedelta(days=1), try_today) + + async def wrap(scheduler, client): + """Wraps the process of sending afternoon media and rescheduling it. + + This asynchronous function sends an afternoon media item using the provided Telegram client + and then schedules the next afternoon media sending using the specified scheduler. It + ensures that the media sending process is repeated at the appropriate time. + + Args: + scheduler: The scheduler instance used to manage the scheduling of jobs. + client: The Telegram client used to send the media item. + """ + await send_afternoon_media(client) + start_sending_afternoon_media(scheduler, client) + + global next_afternoon_media_time + next_afternoon_media_time = dt + print(f"Next afternoon media at {dt}") + scheduler.add_job(wrap, "date", run_date=dt, args=[scheduler, client]) + + +async def send_stats(client, user_id): + """Sends a message containing the remaining media items to a specified user. + + This asynchronous function retrieves the current state of media items and the register, + calculates the remaining items for each media type, and sends a summary message to the + specified user via the Telegram client. + + Args: + client: The Telegram client used to send the message. + user_id: The identifier for the user to whom the message is sent. + + Raises: + FileNotFoundError: If the specified YAML files cannot be found. + yaml.YAMLError: If there is an error parsing the YAML files. + Exception: If there is an error sending the message through the Telegram client. + """ + register = read_yaml(REGISTER_YAML_PATH) + media_data = read_yaml(MEDIA_YAML_PATH) + + msg = "Remaining:" + for data in (media_data,): + for field in data: + msg += f"\n - {field}: {len(data[field]) - len(register[field])}" + + await client.send_message(user_id, msg) + + +def start_sending_pills_reminder(scheduler, client): + """Schedules a daily reminder to take pills for a specified user. + + This function retrieves the user's chat ID and the reminder time from the configuration, then + sets up a scheduled job to send a reminder message via the Telegram client. The reminder message + is sent daily at the specified time. + + Args: + scheduler: The scheduler instance used to manage scheduled jobs. + client: The Telegram client used to send the reminder message. + + Raises: + FileNotFoundError: If the configuration file cannot be found. + yaml.YAMLError: If there is an error parsing the YAML configuration. + Exception: If there is an error scheduling the job or sending the message. + """ + tconfig = read_yaml(TELEGRAM_CONFIG_PATH) + user_id = tconfig.get("nathy", {}).get("chat_id") + reminder_time = tconfig.get("nathy", {}).get("pills_reminder").get("time").split(":") + + async def wrap(scheduler, client): + await client.send_message( + user_id, "💊 Amorcito, recuerda tomarte la píldora a las 10. Te amo ❤️" + ) + start_sending_pills_reminder(scheduler, client) + + scheduler.add_job( + wrap, + "cron", + hour=reminder_time[0], + minute=reminder_time[1], + second=reminder_time[2], + args=[scheduler, client], + )