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)