22
1
Fork 0

Send a notification to users

Email the owners of the given repositories telling them that terms and
conditions of the service must be abided. A private redmine note is
added to an issue to keep track of these requests. This script must be
executed on the Gitea host.

Usage:

    # cat repositories.txt
    https://forge.chapril.org/user1/repo1
    https://forge.chapril.org/userX/repoZ
    # ./notify_owners.py repositories.txt
master
Pierre-Louis Bonicoli 11 months ago
parent 7feee73ac1
commit 6a422f6b70
Signed by: pilou
GPG Key ID: 06914C4A5EDAA6DD

@ -0,0 +1,194 @@
#!/usr/bin/env python3
import configparser
import datetime
from email.message import EmailMessage
import fileinput
import functools
import getpass
import json
import logging
import smtplib
from time import sleep
from string import Template
from urllib.parse import urlsplit
import requests
SUBJECT = "Licence du dépôt '$repository' hébergé sur la forge Chapril"
SUBJECT_TMPL = Template(SUBJECT)
MAIL = """\
Bonjour,
Je suis animateur bénévole du service forge.chapril.org.
Sauf erreur de ma part, le dépôt '$repository' [1]
ne spécifie pas clairement de licence et n'est donc pas conforme aux
conditions générales d'utilisation du service [2].
Nous vous demandons de bien vouloir vous conformer aux conditions
générales d'utilisation du service, à défaut ce dépôt sera supprimé dans
15 jours.
Nous sommes heureux que nos services numériques libres, éthiques et loyaux
puissent vous être utiles.
[1] https://forge.chapril.org/$user/$repository
[2] https://www.chapril.org/cgu.html#forgechaprilorg-cpu
En vous souhaitant, au nom de toute l'équipe du Chapril, un agréable
usage de nos services libres, éthique et loyaux.
--
$animsys"""
MAIL_TMPL = Template(MAIL)
AGIR_URL = "https://agir.april.org"
FORGE_ML = "forge-support@chapril.org"
def get_db_cursor():
import psycopg2
conn = psycopg2.connect(**db_config_from_gitea(), target_session_attrs="read-only")
conn.set_session(readonly=True)
assert conn.readonly
logging.debug("Connected to gitea database")
return conn.cursor()
def db_config_from_gitea():
config = configparser.ConfigParser()
with open("/etc/gitea/gitea.ini", encoding="utf-8") as giteacfg:
# skip the first lines without a section
while True:
pos = giteacfg.tell()
line = giteacfg.readline()
try:
config.read_string(line)
except configparser.MissingSectionHeaderError:
pass
if config.sections():
config.clear()
break
giteacfg.seek(pos)
config.read_file(giteacfg)
db_params = {
"options": "-c default_transaction_read_only=on",
}
if ":" in config["database"]["host"]:
db_host, db_port = config["database"]["host"].split(":")
db_params["host"] = db_host
db_params["port"] = db_port
else:
db_params["host"] = config["database"]["host"]
for param, gitea_param in (
("sslmode", "ssl_mode"),
("dbname", "name"),
("user", "user"),
("password", "passwd"),
):
if gitea_param in config["database"]:
db_params[param] = config["database"][gitea_param]
if "schema" in config["database"] and config["database"]["schema"]:
db_params["options"] += f' -c search_path={config["database"]["schema"]}'
return db_params
def fetch_mail_from_db(cur, user):
query = "select email from public.user where name = %s;"
cur.execute(query, (user,))
return cur.fetchone()[0]
def notify(smtp, repositories, agir_key, fetch_mail):
headers = {"X-Redmine-API-Key": agir_key, "Content-Type": "application/json"}
note = Template(
json.dumps(
{
"issue": {
"private_notes": True,
"notes": f'🔲 "$repository":$url: @$user@ notified by mail '
f"({datetime.date.today()})",
}
}
)
)
req = requests.get(f"{AGIR_URL}/users/current.json", headers=headers)
req.raise_for_status()
sender = req.json()["user"]
if "firstname" in sender and "lastname" in sender:
animsys = f"{sender['firstname']} {sender['lastname']} " f"({sender['login']})"
else:
animsys = sender["login"]
for url in repositories:
url = url.strip()
if not url:
continue
if not "/" in url:
raise ValueError(f"URL not recognized ({url})")
user, repository = urlsplit(url).path.split("/")[1:3]
assert user and repository, f"Can not fetch user and repository from {url}"
user_email = fetch_mail(user)
logging.debug("Mail for '%s' is '%s'", user, user_email)
subject = SUBJECT_TMPL.substitute(repository=repository)
message = MAIL_TMPL.substitute(
user=user, repository=repository, animsys=animsys
)
email = EmailMessage()
email.set_content(message)
email["Subject"] = subject
email["From"] = FORGE_ML
email["To"] = user_email
email["CC"] = "forge-support@chapril.org"
smtp.send_message(email)
logging.debug("Mail to '%s' (%s) sent for %s.", user, user_email, url)
issue_api = f"{AGIR_URL}/issues/5615.json"
req = requests.put(
issue_api,
data=note.substitute(user=user, repository=repository, url=url),
headers=headers,
)
req.raise_for_status()
logging.debug("Note added to %s for %s", issue_api, url)
sleep(1) # throttle mail and web notifications!
def main():
logging.basicConfig(level=logging.DEBUG)
# In order to use another SMTP server (in such case, fetch_mail may need to
# be re-implemented using the Gitea API)
#
# import ssl
# server = input("type your smtp server: ")
# login = input("type your smtp login: ")
# password = getpass.getpass(prompt="type your password and press enter:")
# context = ssl.create_default_context(cafile="/usr/local/share/ca-certificates/own.crt")
# with smtplib.smtp_ssl(server, 465, context=context) as smtp:
# smtp.login(login, password)
agir_key = getpass.getpass(prompt="Type agir.april.org API key:")
repositories = fileinput.input()
with smtplib.SMTP("127.0.0.1", 25) as smtp:
with get_db_cursor() as cursor:
fetch_mail = functools.partial(fetch_mail_from_db, cursor)
notify(smtp, repositories, agir_key, fetch_mail)
if __name__ == "__main__":
main()

@ -0,0 +1,2 @@
psycopg2-binary
requests
Loading…
Cancel
Save