spotted.data package

Submodules

spotted.data.config module

Read the bot configuration from the settings.yaml and the autoreplies.yaml files

class spotted.data.config.Config[source]

Bases: object

Configurations

AUTOREPLIES_PATH = 'autoreplies.yaml'
DEFAULT_AUTOREPLIES_PATH = '/opt/hostedtoolcache/Python/3.12.6/x64/lib/python3.12/site-packages/spotted/config/yaml/autoreplies.yaml'
DEFAULT_SETTINGS_PATH = '/opt/hostedtoolcache/Python/3.12.6/x64/lib/python3.12/site-packages/spotted/config/yaml/settings.yaml'
SETTINGS_PATH = 'settings.yaml'
classmethod autoreplies_get(*keys, default=None)[source]

Get the value of the specified key in the autoreplies configuration dictionary. If the key is a tuple, it will return the value of the nested key. If the key is not present, it will return the default value.

Parameters:
  • key – key to get

  • default (Any, default: None) – default value to return if the key is not present

Returns:

dict – value of the key or default value

classmethod debug_get(key, default=None)[source]

Get the value of the specified key in the configuration under the ‘debug’ section. If the key is not present, it will return the default value.

Parameters:
  • key (Literal['local_log', 'reset_on_load', 'log_file', 'log_error_file', 'db_file', 'crypto_key']) – key to get

  • default (Any, default: None) – default value to return if the key is not present

Returns:

Any – value of the key or default value

classmethod override_settings(config)[source]

Overrides the settings with the configuration provided in the config dict.

Parameters:

config (dict) – configuration dict used to override the current settings

classmethod post_get(key, default=None)[source]

Get the value of the specified key in the configuration under the ‘post’ section. If the key is not present, it will return the default value.

Parameters:
  • key (Literal['community_group_id', 'channel_id', 'channel_tag', 'comments', 'admin_group_id', 'n_votes', 'remove_after_h', 'report', 'report_wait_mins', 'replace_anonymous_comments', 'delete_anonymous_comments', 'blacklist_messages']) – key to get

  • default (Any, default: None) – default value to return if the key is not present

Returns:

Any – value of the key or default value

classmethod reload(force_reload=False)[source]

Reset the configuration. The next time a setting parameter is required, the whole configuration will be reloaded. If force_reload is True, the configuration will be reloaded immediately.

Parameters:

force_reload (bool, default: False) – if True, the configuration will be reloaded immediately

classmethod settings_get(*keys, default=None)[source]

Get the value of the specified key in the configuration. If the key is a tuple, it will return the value of the nested key. If the key is not present, it will return the default value.

Parameters:
  • key – key to get

  • default (Any, default: None) – default value to return if the key is not present

Returns:

Any – value of the key or default value

spotted.data.data_reader module

Read data from files

spotted.data.data_reader.get_abs_path(*root_file_path)[source]

Get the abs path from the root directory of the project to the requested path

Parameters:

root_file_path (str) – path from the root project directory

Returns:

str – corresponding abs path

spotted.data.data_reader.read_file(*root_file_path)[source]

Read the contents of the file

Parameters:

root_file_path (str) – path of the file to read from the root project directory

Returns:

str – contents of the file

spotted.data.data_reader.read_md(file_name)[source]

Read the contents of a markdown file. The path is data/markdown. It also will replace the following parts of the text:

  • {channel_tag} -> Config.settings[‘post’][‘channel_tag’]

  • {bot_tag} -> Config.settings[‘bot_tag’]

Parameters:

file_name (str) – name of the file

Returns:

str – contents of the file

spotted.data.db_manager module

Handles the management of databases

class spotted.data.db_manager.DbManager[source]

Bases: object

Class that handles the management of databases

classmethod count_from(table_name, select='*', where='', where_args=None)[source]

Returns the number of rows found with the query. Executes “SELECT COUNT(select) FROM table_name [WHERE where (with where_args)]”

Parameters:
  • table_name (str) – name of the table used in the FROM

  • select (str, default: '*') – columns considered for the query

  • where (str, default: '') – where clause, with %s placeholders for the where_args

  • where_args (tuple | None, default: None) – args used in the where clause

Returns:

int – number of rows

classmethod delete_from(table_name, where='', where_args=None)[source]

Deletes the rows from the specified table, where the condition, when set, is satisfied. Executes “DELETE FROM table_name [WHERE where (with where_args)]”

Parameters:
  • table_name (str) – name of the table used in the DELETE FROM

  • where (str, default: '') – where clause, with %s placeholders for the where args

  • where_args (tuple | None, default: None) – args used in the where clause

classmethod get_db()[source]

Creates the connection to the database. It can be sqlite or postgres

Returns:

tuple[Connection, Cursor] – sqlite database connection and cursor

classmethod insert_into(table_name, values, columns='', multiple_rows=False)[source]

Inserts the specified values in the database. Executes “INSERT INTO table_name ([columns]) VALUES (placeholders)”

Parameters:
  • table_name (str) – name of the table used in the INSERT INTO

  • values (tuple) – values to be inserted. If multiple_rows is true, tuple of tuples of values to be inserted

  • columns (tuple | str, default: '') – columns that will be inserted, as a tuple of strings

  • multiple_rows (bool, default: False) – whether or not multiple rows will be inserted at the same time

classmethod query_from_file(*file_path)[source]

Commits all the queries in the specified file. The queries must be separated by a —– string Should not be used to select something

Parameters:

file_path (str) – path of the text file containing the queries

classmethod query_from_string(*queries)[source]

Commits all the queries in the string Should not be used to select something

Parameters:

queries (str) – tuple of queries

static register_adapters_and_converters()[source]

Registers the adapter and converters for the datetime type. Needed from python 3.12 onwards, as the default option has been deprecated

static row_factory(cursor, row)[source]

Converts the rows from the database into a dictionary

Parameters:
  • cursor (Cursor) – database cursor

  • row (dict) – row from the database

Returns:

dict – dictionary containing the row. The keys are the column names

classmethod select_from(table_name, select='*', where='', where_args=None, group_by='', order_by='')[source]

Returns the results of a query. Executes “SELECT select FROM table_name [WHERE where (with where_args)] [GROUP_BY group_by] [ORDER BY order_by]”

Parameters:
  • table_name (str) – name of the table used in the FROM

  • select (str, default: '*') – columns considered for the query

  • where (str, default: '') – where clause, with %s placeholders for the where_args

  • where_args (tuple | None, default: None) – args used in the where clause

  • group_by (str, default: '') – group by clause

  • order_by (str, default: '') – order by clause

Returns:

list – rows from the select

classmethod update_from(table_name, set_clause, where='', args=None)[source]

Updates the rows from the specified table, where the condition, when set, is satisfied. Executes “UPDATE table_name SET set_clause (with args) [WHERE where (with args)]”

Parameters:
  • table_name (str) – name of the table used in the DELETE FROM

  • set_clause (str) – set clause, with %s placeholders

  • where (str, default: '') – where clause, with %s placeholders for the where args

  • args (tuple | None, default: None) – args used both in the set clause and in the where clause, in this order

spotted.data.pending_post module

Pending post management

class spotted.data.pending_post.PendingPost(user_id, u_message_id, g_message_id, admin_group_id, date)[source]

Bases: object

Class that represents a pending post

Parameters:
  • user_id (int) – id of the user that sent the post

  • u_message_id (int) – id of the original message of the post

  • g_message_id (int) – id of the post in the group

  • admin_group_id (int) – id of the admin group

  • date (datetime) – when the post was sent

admin_group_id: int
classmethod create(user_message, g_message_id, admin_group_id)[source]

Creates a new post and inserts it in the table of pending posts

Parameters:
  • user_message (Message) – message sent by the user that contains the post

  • g_message_id (int) – id of the post in the group

  • admin_group_id (int) – id of the admin group

Returns:

PendingPost – instance of the class

date: datetime
delete_post()[source]

Removes all entries on a post that is no longer pending

classmethod from_group(g_message_id, admin_group_id)[source]

Retrieves a pending post from the info related to the admin group

Parameters:
  • g_message_id (int) – id of the post in the group

  • admin_group_id (int) – id of the admin group

Returns:

PendingPost | None – instance of the class

classmethod from_user(user_id)[source]

Retrieves a pending post from the user_id

Parameters:

user_id (int) – id of the author of the post

Returns:

PendingPost | None – instance of the class

g_message_id: int
static get_all(admin_group_id, before=None)[source]

Gets the list of pending posts in the specified admin group. If before is specified, returns only the one sent before that timestamp

Parameters:
  • admin_group_id (int) – id of the admin group

  • before (datetime | None, default: None) – timestamp before which messages will be considered

Returns:

list[PendingPost] – list of ids of pending posts

get_list_admin_votes(vote=None)[source]

Gets the list of admins that approved or rejected the post

Parameters:

vote (bool | None, default: None) – whether you look for the approve or reject votes, or None if you want all the votes

Returns:

list[int] | list[tuple[int, bool]] – list of admins that approved or rejected a pending post

get_votes(vote)[source]

Gets all the votes of a specific kind (approve or reject)

Parameters:

vote (bool) – whether you look for the approve or reject votes

Returns:

int – number of votes

save_post()[source]

Saves the pending_post in the database

Return type:

PendingPost

set_admin_vote(admin_id, approval)[source]

Adds the vote of the admin on a specific post, or update the existing vote, if needed

Parameters:
  • admin_id (int) – id of the admin that voted

  • approval (bool) – whether the vote is approval or reject

Returns:

int – number of similar votes (all the approve or the reject), or -1 if the vote wasn’t updated

u_message_id: int
user_id: int

spotted.data.post_data module

Data management for the bot

class spotted.data.post_data.PostData[source]

Bases: object

Class that handles the management of persistent data fetch or manipulation in the post bot

static get_n_posts()[source]

Gets the total number of posts

Returns:

int – total number of posts

spotted.data.published_post module

Published post management

class spotted.data.published_post.PublishedPost(channel_id, c_message_id, date)[source]

Bases: object

Class that represents a published post

Parameters:
  • channel_id (int) – id of the channel

  • c_message_id (int) – id of the post in the channel

c_message_id: int
channel_id: int
classmethod create(channel_id, c_message_id)[source]

Inserts a new post in the table of published posts

Parameters:
  • channel_id (int) – id of the channel

  • c_message_id (int) – id of the post in the channel

Returns:

PublishedPost – instance of the class

date: datetime
classmethod from_channel(channel_id, c_message_id)[source]

Retrieves a published post from the info related to the channel

Parameters:
  • channel_id (int) – id of the channel

  • c_message_id (int) – id of the post in the channel

Returns:

PublishedPost | None – instance of the class

save_post()[source]

Saves the published_post in the database

Return type:

PublishedPost

spotted.data.report module

Reports management

class spotted.data.report.Report(user_id, admin_group_id, g_message_id, channel_id=None, c_message_id=None, target_username=None, date=None)[source]

Bases: object

Class that represents a report

Parameters:
  • user_id (int) – id of the user that reported

  • admin_group_id (int) – id of the admin group

  • g_message_id (int) – id of the post in the group

  • channel_id (int, default: None) – id of the channel

  • c_message_id (int, default: None) – id of the post in question in the channel

  • target_username (str, default: None) – username of the reported user

  • date (datetime, default: None) – when the report happened

admin_group_id: int
c_message_id: int = None
channel_id: int = None
classmethod create_post_report(user_id, channel_id, c_message_id, admin_message)[source]

Adds the report of the user on a specific post

Parameters:
  • user_id (int) – id of the user that reported

  • channel_id (int) – id of the channel

  • c_message_id (int) – id of the post in question in the channel

  • admin_message (Message) – message received in the admin group that references the report

Returns:

Report | None – instance of the class or None if the report was not created

classmethod create_user_report(user_id, target_username, admin_message)[source]

Adds the report of the user targeting another user

Parameters:
  • user_id (int) – id of the user that reported

  • target_username (str) – username of reported user

  • admin_message (Message) – message received in the admin group that references the report

Returns:

Report – instance of the class

date: datetime = None
classmethod from_group(admin_group_id, g_message_id)[source]

Gets a report of any type related to the specified message in the admin group

Parameters:
  • admin_group_id (int) – id of the admin group

  • g_message_id (int) – id of the report in the group

Returns:

Report | None – instance of the class or None if the report was not present

g_message_id: int
classmethod get_last_user_report(user_id)[source]

Gets the last user report of a specific user

Parameters:

user_id (int) – id of the user that reported

Returns:

Report | None – instance of the class or None if the report was not present

classmethod get_post_report(user_id, channel_id, c_message_id)[source]

Gets the report of a specific user on a published post

Parameters:
  • user_id (int) – id of the user that reported

  • channel_id (int) – id of the channel

  • c_message_id (int) – id of the post in question in the channel

Returns:

Report | None – instance of the class or None if the report was not present

property minutes_passed: float

Amount of minutes elapsed from when the report was submitted, if applicable

Type:

float

save_report()[source]

Saves the report in the database

Return type:

Report

target_username: str = None
user_id: int

spotted.data.user module

Users management

class spotted.data.user.User(user_id, private_message_id=None, ban_date=None, follow_date=None)[source]

Bases: object

Class that represents a user

Parameters:
  • user_id (int) – id of the user

  • private_message_id (int | None, default: None) – id of the private message sent by the user to the bot. Only used for following

  • ban_date (datetime | None, default: None) – datetime of when the user was banned. Only used for banned users

  • follow_date (datetime | None, default: None) – datetime of when the user started following a post. Only used for following users

ban()[source]

Adds the user to the banned list

ban_date: datetime | None = None
classmethod banned_users()[source]

Returns a list of all the banned users

Return type:

list[User]

become_anonym()[source]

Removes the user from the credited list, if he was present

Returns:

bool – whether the user was already anonym

become_credited()[source]

Adds the user to the credited list, if he wasn’t already credited

Returns:

bool – whether the user was already credited

classmethod credited_users()[source]

Returns a list of all the credited users

Return type:

list[User]

follow_date: datetime | None = None
classmethod following_users(message_id)[source]

Returns a list of all the users following the post with the associated private message id used by the bot to send updates about the post by replying to it

Parameters:

message_id (int) – id of the post the users are following

Returns:

list[User] – list of users with private_message_id set to the id of the private message in the user’s conversation with the bot

get_follow_private_message_id(message_id)[source]

Verifies if the user is following a post

Parameters:

message_id (int) – id of the post

Returns:

int | None – whether the user is following the post or not

async get_user_sign(bot)[source]

Generates a sign for the user. It will be a random name for an anonym user

Parameters:

bot (Bot) – telegram bot

Returns:

str – the sign of the user

property is_banned: bool

If the user is banned or not

property is_credited: bool

If the user is in the credited list

is_following(message_id)[source]

Verifies if the user is following a post

Parameters:

message_id (int) – id of the post

Returns:

bool – whether the user is following the post or not

property is_pending: bool

If the user has a post already pending or not

private_message_id: int | None = None
sban()[source]

Removes the user from the banned list

Returns:

bool – whether the user was present in the banned list before the sban or not

set_follow(message_id, private_message_id)[source]

Sets the follow status of the user. If the private_message_id is None, the user is not following the post anymore, and the record is deleted from the database. Otherwise, the user is following the post and a new record is created.

Parameters:
  • message_id (int) – id of the post

  • private_message_id (int | None) – id of the private message. If None, the record is deleted

user_id: int

Module contents

Modules that work with the data section

spotted.data.init_db()[source]

Initialize the database. If the debug.reset_on_load setting is True, it will delete the database and create a new one.