Commit 3518b7df authored by Pietro Albini's avatar Pietro Albini

Initial commit

parents
/*.egg-info
*.py[co]
This diff is collapsed.
include uitsmmbot/schema.sql
## Bot di Telegram per il gruppo Social Media
Questo repository contiene il sorgente del bot di Telegram del gruppo Social
Media. Il bot permette al gruppo di inviare post ai nostri vari canali
direttamente dalla chat di Telegram, semplificando il workflow generale.
Il bot è scritto in Python 3 usando [botogram][botogram], ed è rilasciato sotto
licenza GNU AGPL v3+.
### Eseguire il bot
Per eseguire il bot, scaricare il sorgente del repository ed installare il
pacchetto Python contenuto:
```
$ git clone http://code.ubuntu-it.org/ubuntu-it-socialmedia/uitsmmbot
$ python3 -m pip install uitsmmbot/
```
Per eseguirlo, è possibile chiamare l'eseguibile `uitsmmbot` passando come
argomento il token del bot Telegram e il percorso da usare per il database
SQLite:
```
$ uitsmmbot 123456:abcdefghi path/to/database.db
```
### Configurazione del bot
Al primo avvio, il bot renderà amministratore la prima persona a contattarlo.
Per poter usare il bot in altri gruppi, è necessario che l'amministratore
digiti il comando che il bot comunicherà prima di uscire dal gruppo.
Inoltre, è necessario inserire il token di Buffer: per ottenerlo bisogna creare
una nuova applicazione dalla developer console di Buffer, e copiare il campo
"Access Token" presente nei dettagli dell'applicazione. Dopo aver fatto ciò, si
può digitare in chat:
```
/buffer_access_token abcdefghijklmno
```
Se si vuole anche pubblicare in un canale Telegram, bisogna aggiungere il bot
come amministratore del canale, e successivamente digitare in chat:
```
/link_telegram @usernamedelcanale
```
Per scollegare il canale basta il comando:
```
/unlink_telegram
```
[botogram]: https://botogram.pietroalbini.org
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import setuptools
setuptools.setup(
name = "uitsmmbot",
version = "1",
license = "GNU-AGPL v3+",
author = "Pietro Albini",
author_email = "pietroalbini@ubuntu.com",
install_requires = [
"botogram",
"requests",
],
packages = [
"uitsmmbot",
],
entry_points = {
"console_scripts": [
"uitsmmbot = uitsmmbot.__main__:main",
],
},
include_package_data = True,
zip_safe = False,
classifiers = [
"Not on PyPI"
],
)
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import sys
from .bot import create_bot
def cli(args):
"""CLI of the bot"""
if len(args) != 2:
print("Usage: uitsmmbot <telegram-token> <database-path>")
exit(1)
bot = create_bot(args[0], args[1])
bot.run(workers=1)
def main():
"""Entry point"""
cli(sys.argv[1:])
if __name__ == "__main__":
main()
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import botogram
class AuthComponent(botogram.Component):
"""Check if the user is authorized to contact the bot"""
component_name = "auth"
def __init__(self, db):
self.db = db
self.add_before_processing_hook(self.check)
self.add_command("authorize", self.authorize_command)
def check(self, chat):
"""Check if a chat is authorized to talk with the bot"""
everyone = [r[0] for r in self.db.query("SELECT id FROM auth;")]
# If this is the first chat to talk with the bot, authorize it
if not everyone:
self.db.update("INSERT INTO auth (id) VALUES (?);", chat.id)
else:
# Don't do anything if the user is authorized
if chat.id in everyone:
return
chat.send("\n".join((
"*Non sei autorizzato a contattare questo bot!*",
"Chiedi all'amministratore di digitare "
"`/authorize %s`" % chat.id
)))
if chat.type in ("group", "supergroup"):
chat.leave()
return True
def authorize_command(self, message, args):
"""Autorizza un ID a contattare il bot"""
if len(args) != 1:
message.reply("*Uso:* `/authorize <id>`")
return
if self.db.query("SELECT id FROM auth WHERE id = ?;", args[0]):
message.reply("*Errore:* Chat già autorizzata!")
return
self.db.update("INSERT INTO auth (id) VALUES (?);", args[0])
message.reply("La chat è ora *autorizzata* a contattare il bot.")
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import json
import botogram
import botogram.objects.base
from .db import Database
from .auth import AuthComponent
from .buffer import Buffer
from .settings import SettingsComponent
from .post import PostComponent
class CallbackQuery(botogram.objects.base.BaseObject):
required = {
"id": str,
"from": botogram.User,
"chat_instance": str,
}
optional = {
"inline_message_id": str,
"message": botogram.Message,
"data": str,
"game_short_name": str,
}
botogram.Update.optional["callback_query"] = CallbackQuery
def create_bot(token, db_path):
"""Create a new instance of the bot"""
bot = botogram.create(token)
bot.lang = "it"
db = Database(db_path)
db.init()
buffer = Buffer(db)
auth = AuthComponent(db)
settings = SettingsComponent(db, buffer)
post = PostComponent(db, buffer)
bot.use(auth)
bot.use(settings)
bot.use(post)
bot.register_update_processor("callback_query", post.process_callbacks)
return bot
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import requests
import botogram
class InvalidBufferTokenError(Exception):
pass
class Buffer:
"""Interface with Buffer"""
def __init__(self, db, base=None):
if base is None:
base = "https://api.bufferapp.com/1"
self.base = base
self.db = db
def set_access_token(self, token):
"""Set the new access token"""
# Check if the token works
self.request("get", "profiles.json", {
"access_token": token,
})
# If the previous method didn't trigger an exception...
self.db.update(
"INSERT OR REPLACE INTO kw VALUES ('buffer_access_token', ?);",
token
)
def access_token(self):
"""Get the current access token"""
data = self.db.query(
"SELECT value FROM kw WHERE key = 'buffer_access_token';"
)
if not data:
raise InvalidBufferTokenError
return data[0]
def request(self, method, url, params=None, data=None):
"""Make a new request to Buffer"""
if params is None:
params = {}
if "access_token" not in params:
params["access_token"] = self.access_token()
res = requests.request(method,
"%s/%s" % (self.base, url), params=params, data=data
)
if res.status_code == 401:
raise InvalidBufferTokenError
return res.json()
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import threading
import sqlite3
import pkg_resources
class Database:
def __init__(self, path):
self._path = path
def init(self):
"""Initialize the database"""
# Get the path to the schema file
schema = pkg_resources.resource_filename("uitsmmbot", "schema.sql")
# Migrate the database
cursor = self.cursor()
with open(schema) as f:
cursor.executescript(f.read())
def cursor(self):
"""Get a new cursor"""
local = threading.local()
if not hasattr(local, "db"):
local.db = sqlite3.connect(self._path)
return local.db.cursor()
def query(self, query, *params, update=False):
"""Make a new query against the db"""
cursor = self.cursor()
cursor.execute(query, params)
try:
if update:
cursor.connection.commit()
else:
return cursor.fetchall()
finally:
cursor.close()
def update(self, query, *params):
"""Make a new update query against the db"""
self.query(query, *params, update=True)
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import time
import json
import botogram
from .buffer import InvalidBufferTokenError
CLEANUP_AFTER = 30 * 60
def inline_keyboard(buttons):
"""Create an inline keyboard"""
buttons_tg = []
for row in buttons:
inner = []
for text, callback in row:
inner.append({"text": text, "callback_data": callback})
buttons_tg.append(inner)
result = lambda: None
result.serialize = lambda: {"inline_keyboard": buttons_tg}
return result
class PostComponent(botogram.Component):
component_name = "post"
def __init__(self, db, buffer):
self.db = db
self.buffer = buffer
self.add_timer(2 * 60, self.cleanup)
self.add_timer(5 * 60, self.post_to_telegram)
self.add_command("post", self.post_command)
def cleanup(self):
"""Clean up old records"""
max = time.time() - CLEANUP_AFTER
self.db.update("DELETE FROM post_pending WHERE created_at < ?", max)
def post_to_telegram(self, bot):
"""Post queued messages to Telegram"""
# Return if there isn't a valid Buffer token
try:
self.buffer.access_token()
except InvalidBufferTokenError:
return
channel = self.db.query(
"SELECT value FROM kw WHERE key = 'telegram_channel';"
)
# If there isn't a channel linked discard all the pending posts
if not channel:
self.db.update("DELETE FROM to_telegram;")
return
channel = channel[0][0]
to_telegram = [d[0] for d in self.db.query(
"SELECT post_id FROM to_telegram;"
)]
for post_id in to_telegram:
data = self.buffer.request("get", "updates/%s.json" % post_id)
if data["status"] != "sent":
continue
bot.chat(channel).send(data["text_formatted"])
self.db.update(
"DELETE FROM to_telegram WHERE post_id = ?;", post_id
)
def post_command(self, chat, message):
"""Invia un nuovo post"""
try:
self.buffer.access_token()
except InvalidBufferTokenError:
message.reply("*Errore:* token di Buffer non valido!")
return
# Check if this is a reply
if message.reply_to_message is None:
message.reply("*Rispondi* al messaggio contenente il post!")
return
message_id = message.reply_to_message.message_id
# Check if there is a publishing in progress for this post
already_exists = self.db.query(
"SELECT message FROM post_pending WHERE message = ?",
message_id
)
if already_exists:
message.reply_to_message.reply(
"*Errore:* stai già per pubblicare questo post."
)
return
profiles = self.buffer.request("get", "profiles.json")
for profile in profiles:
self.db.update(
"INSERT INTO post_pending (message, social, social_id, "
"social_pretty, social_name, post_there, created_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?);",
message_id, profile["service"], profile["_id"],
profile["formatted_service"], profile["formatted_username"],
False, int(time.time()),
)
channel = self.db.query(
"SELECT value FROM kw WHERE key = 'telegram_channel';"
)
if channel:
self.db.update(
"INSERT INTO post_pending (message, social, social_id, "
"social_pretty, social_name, post_there, created_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?);",
message_id, "telegram", "__telegram__", "Telegram",
channel[0][0], False, int(time.time()),
)
message.reply_to_message.reply(
"Seleziona i social dove vuoi pubblicare il post:",
extra=self.get_keyboard(message_id),
)
def process_callbacks(self, bot, chains, update):
"""Process incoming callbacks"""
query = update.callback_query
action, data = query.data.split(":", 1)
if action == "toggle":
message, social = data.split(":")
message = int(message)
data = self.db.query(
"SELECT post_there FROM post_pending WHERE message = ? AND "
"social = ?;",
message, social,
)
if not data:
query.message.edit("*Operazione scaduta!*")
return
current = data[0][0]
self.db.update(
"UPDATE post_pending SET post_there = ? WHERE message = ? AND "
"social = ?;",
not current, message, social
)
kb = self.get_keyboard(message)
bot.api.call("editMessageReplyMarkup", {
"chat_id": query.message.chat.id,
"message_id": query.message.message_id,
"reply_markup": json.dumps(kb.serialize()),
})
elif action in ("queue", "top", "now"):
message = int(data)
to_socials = [d[0] for d in self.db.query(
"SELECT social_id FROM post_pending WHERE message = ? AND "
"post_there = 1;", message
)]
telegram = False
if "__telegram__" in to_socials:
telegram = True
to_socials.remove("__telegram__")
if not to_socials:
bot.api.call("answerCallbackQuery", {
"callback_query_id": query.id,
"text": "Devi selezionare almeno un social che non sia "
"Telegram!",
"show_alert": True
})
return
data = {
"text": query.message.reply_to_message.text,
"profile_ids": to_socials,
}
if action == "now":
data["now"] = True
elif action == "top":
data["top"] = True
res = self.buffer.request("post", "updates/create.json", data=data)
if res["success"]:
query.message.edit("\n".join((
"*Post inviato con successo!*" if action == "now" else
"*Post messo in coda!*",
res["message"],
)))
# Queue the update also for Telegram
if telegram:
channel = self.db.query(
"SELECT value FROM kw WHERE key = 'telegram_channel';"
)[0][0]
# If the post should be published now, publish it instantly
if action == "now":
bot.chat(channel).send(
res["updates"][0]["text_formatted"]
)
else:
post_id = res["updates"][0]["id"]
self.db.update(
"INSERT INTO to_telegram (post_id) VALUES (?);",
post_id
)
else:
query.message.edit("\n".join((
"*Impossibile inviare il post:*",
res["message"],
)))
self.db.update(
"DELETE FROM post_pending WHERE message = ?;", message
)
bot.api.call("answerCallbackQuery", {
"callback_query_id": query.id,
})
def get_keyboard(self, message_id, edit=False):
"""Prepare the content of the /post command"""
pending = self.db.query(
"SELECT social, social_pretty, social_name, post_there FROM "
"post_pending WHERE message = ? ORDER BY social_pretty;",
message_id,
)
buttons = []
for one in pending:
buttons.append([(
"%s%s: %s" % (
"✅ " if one[3] else "",
one[1],
one[2],
),
"toggle:%s:%s" % (message_id, one[0])
)])
buttons += [[
("Metti in coda", "queue:%s" % message_id),
("Pubblica per primo", "top:%s" % message_id),
], [
("Pubblica ora", "now:%s" % message_id),
]]
return inline_keyboard(buttons)
-- Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
--
-- This program is free software: you can redistribute it and/or modify
-- it under the terms of the GNU Affero General Public License as published
-- by the Free Software Foundation, either version 3 of the License, or
-- (at your option) any later version.
--
-- This program is distributed in the hope that it will be useful,
-- but WITHOUT ANY WARRANTY; witout even the implied warranty of
-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-- GNU Affero General Public License for more details.
--
-- You should have received a copy of the GNU Affero General Public License
-- along with this program. If not, see <http://www.gnu.org/licenses/>.
CREATE TABLE IF NOT EXISTS auth (
id INTEGER PRIMARY KEY
);
CREATE TABLE IF NOT EXISTS kw (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS post_pending (
message INTEGER,
social TEXT NOT NULL,
social_id TEXT NOT NULL,
social_pretty TEXT NOT NULL,
social_name TEXT NOT NULL,
post_there BOOLEAN NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY (message, social)
);
CREATE TABLE IF NOT EXISTS to_telegram (
post_id TEXT PRIMARY KEY
);
# Copyright (C) 2017 Pietro Albini <pietroalbini@ubuntu.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; witout even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import botogram
from .buffer import InvalidBufferTokenError
class SettingsComponent(botogram.Component):
component_name = "buffer_settings"
def __init__(self, db, buffer):
self.db = db
self.buffer = buffer
self.add_command("buffer_access_token", self.buffer_access_token)
self.add_command("link_telegram", self.link_telegram)
self.add_command("unlink_telegram", self.unlink_telegram)
def buffer_access_token(self, message, args):
"""Imposta un nuovo token di Buffer"""
if len(args) != 1:
message.reply("*Uso:* `/buffer_access_token <nuovo-token>`")
return
try:
self.buffer.set_access_token(args[0])
except InvalidBufferTokenError:
message.reply("Il token inserito *non è valido*!")
else:
message.reply("Il token di Buffer è stato *aggiornato*!")
def link_telegram(self, bot, message, args):
"""Collega un canale telegram al bot"""
if len(args) != 1:
message.reply("*Uso:* `/link_telegram <username-canale>`")
return
if self.db.query("SELECT * FROM kw WHERE key = 'telegram_channel';"):
message.reply("*Errore:* c'è già un canale collegato.")
return
try:
status = bot.api.call("getChatMember", {
"chat_id": args[0],
"user_id": bot.itself.id,
})["result"]["status"]
except botogram.APIError:
message.reply("*Errore:* il canale non esiste!")
return
if status != "administrator":
message.reply("*Errore:* il bot non è amministratore del canale!")
return
self.db.update(
"INSERT INTO kw (key, value) VALUES ('telegram_channel', ?);",
args[0],
)
message.reply("Canale *collegato* con successo!")
def unlink_telegram(self, message):
"""Rimuovi il collegamento dal canale telegram"""
self.db.update("DELETE FROM kw WHERE key = 'telegram_channel';")
message.reply("Link al canale Telegram *rimosso*.")
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment