git-importer/lib/user.py

from __future__ import annotations

from lib.db import DB

FAKE_ACCOUNTS = (
    "unknown",
    "buildservice-autocommit",
    "autobuild",
    "_service",
    "admin",
    "jg",
    "mrdocs",
    "AdrianSuSE",
    "$user",
    "embar-",
    "michel_mno",
    "guinux",
)


class User:
    @staticmethod
    def find(db: DB, userid: str) -> User:
        row = User.lookup(db, userid)
        self = User()
        self.userid = userid
        if row:
            (_, _, self.email, self.realname) = row
        else:
            self.email = ""
            self.realname = ""
        if not self.email:
            self.email = "null@suse.de"
        if not self.realname:
            self.realname = f"OBS User {userid}"
        return self

    def parse(self, xml, userid):
        self.userid = userid
        self.realname = xml.find("realname").text
        if self.realname is None:
            self.realname = ""
        self.email = xml.find("email").text
        if self.email is None:
            self.email = ""

        return self

    def __str__(self):
        return f"User {self.userid}: {self.realname} {self.email}"

    def __repr__(self):
        return f"[{self.__str__()}]"

    def import_into_db(self, db):
        with db.cursor() as cur:
            cur.execute(
                """INSERT INTO users (userid, realname, email) 
                    VALUES (%s,%s,%s)""",
                (
                    self.userid,
                    self.realname,
                    self.email,
                ),
            )

    @staticmethod
    def lookup(db, userid):
        with db.cursor() as cur:
            cur.execute("SELECT * FROM users where userid=%s", (userid,))
            row = cur.fetchone()
        if not row:
            return None
        return row

    @staticmethod
    def missing_users(db):
        missing_users = set()
        with db.cursor() as cur:
            cur.execute(
                """SELECT DISTINCT revisions.userid
                    FROM revisions LEFT JOIN users ON revisions.userid = users.userid
                    WHERE users.userid IS NULL AND revisions.userid NOT IN {}""".format(
                    FAKE_ACCOUNTS
                )
            )
            for row in cur.fetchall():
                missing_users.add(row[0])
            cur.execute(
                """SELECT DISTINCT requests.creator
                FROM requests LEFT JOIN users ON requests.creator=users.userid
                WHERE users.userid IS NULL AND requests.creator NOT IN {}""".format(
                    FAKE_ACCOUNTS
                )
            )
            for row in cur.fetchall():
                missing_users.add(row[0])
        return sorted(missing_users)