<commit_before>import os
import shutil
import json
import sql_pool
import MySQLdb
import sys
import re
import hashlib
from user import User
from model_factory import ModelFactory

class Job:
    __dbfields = ["user_id","name","error_message","status_id","creation_date","modified_date", "is_active"]
    __extfields = ["status", "role"]

    def __init__(self, job_id, meta=None):
        self.job_id = job_id
        map(lambda x: setattr(self, x, None), self.__dbfields + self.__extfields)
        self.root_path = "" 
        self.users = []
        self.meta = meta

    def get_adjusted_phenotypes(self):
        phe_file = self.relative_path("output.phe")
        phenos = {}
        if os.path.exists(phe_file):
            with open(phe_file) as f:
                for line in f:
                    (sample, val) = line.split()
                    phenos[sample] = float(val)
        return phenos

    def relative_path(self, *args):
        return os.path.expanduser(os.path.join(self.root_path, *args))

    def get_genotype_id(self):
        return self.meta.get("genotype", None) 

    def get_model(self):
        return ModelFactory.get(self.meta.get("type", None), self.root_path, None)

    def get_output_files(self):
        files = []
        def add_if_exists(rel_path, display_name, primary=False):
            file_path = self.relative_path(rel_path)
            if os.path.exists(file_path):
                files.append({"path": rel_path, "size": os.path.getsize(file_path), 
                    "name": display_name, "primary": primary})
        add_if_exists("output.epacts.gz", "Epacts Results", True)
        add_if_exists("results.txt.gz", "SAIGE Results", True)
        add_if_exists("output.filtered.001.gz", "Filtered Results (p-val<0.001)")
        return files

    def get_output_file_path(self):
        files = [x for x in self.get_output_files() if x.get("primary", False)]
        if len(files)==1:
            return self.relative_path(files[0]["path"])
        else:
            return None

    def get_owner(self):
        return User.from_id(self.user_id) 

    def get_param_hash(self):
        return Job.calc_param_hash(self.meta)

    def as_object(self):
        obj = {key: getattr(self, key) for key in self.__dbfields  + self.__extfields if hasattr(self, key)} 
        obj["job_id"] = self.job_id
        obj["users"] = self.users
        obj["output_files"] = self.get_output_files()
        if self.meta:
            details = self.meta.copy()
            model = self.get_model()
            if model:
                details["model_desc"] = model.model_desc
                if "variant_filter" in details:
                    details["variant_filter_desc"] = model.get_filter_desc(details["variant_filter"])
            obj["details"] = details
            obj["hash"] = self.get_param_hash()
        return obj

    @staticmethod
    def get(job_id, config):
        if not re.match('[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}', job_id):
            return None
        job_folder = os.path.join(config.get("JOB_DATA_FOLDER", "./"), job_id)
        meta_path = os.path.expanduser(os.path.join(job_folder, "job.json"))
        if os.path.exists(meta_path):
            with open(meta_path) as meta_file:
                meta = json.load(meta_file)
        else:
           meta = dict()
        j = Job(job_id, meta)
        j.root_path = job_folder
        db = sql_pool.get_conn()
        cur = db.cursor(MySQLdb.cursors.DictCursor)
        sql = """
            SELECT
              bin_to_uuid(jobs.id) AS id,
              jobs.name AS name, jobs.user_id as user_id,
              jobs.status_id as status_id, statuses.name AS status,
              jobs.error_message AS error_message,
              DATE_FORMAT(jobs.creation_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS creation_date,
              DATE_FORMAT(jobs.modified_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS modified_date,
              jobs.is_active
            FROM jobs
            LEFT JOIN statuses ON jobs.status_id = statuses.id
            WHERE jobs.id = uuid_to_bin(%s)
            """
        cur.execute(sql, (job_id,))
        result = cur.fetchone()
        if result is not None:
            map(lambda x: setattr(j, x, result[x]), \
                (val for val in Job.__dbfields + Job.__extfields if val in result))
            sql = """
                SELECT 
                    ju.user_id as user_id, ju.role_id as role_id,
                    role.role_name as role, users.email, 
                    DATE_FORMAT(ju.modified_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS modified_date
                FROM job_users as ju
                    LEFT JOIN job_user_roles role on role.id = ju.role_id
                    LEFT JOIN users on users.id = ju.user_id
                WHERE 
                    ju.job_id = uuid_to_bin(%s)
                """
            cur.execute(sql, (job_id,))
            result = cur.fetchall()
            j.users = result
            return j
        else:
            return None

    @staticmethod
    def list_all_for_user(user_id, config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "jobs.id IN (SELECT job_id from job_users where user_id=%s) AND jobs.is_active=1", (user_id, ))
        return results 

    @staticmethod
    def list_all_for_phenotype(pheno_id, config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "jobs.pheno_id = uuid_to_bin(%s) AND jobs.is_active=1", (pheno_id, ))
        return results 

    @staticmethod
    def list_all_for_user_by_genotype(user_id, geno_id, config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "jobs.geno_id = uuid_to_bin(%s) " + 
            "AND jobs.id IN (SELECT job_id from job_users where user_id=%s) " + 
            "AND jobs.is_active=1", (geno_id, user_id))
        return results 

    @staticmethod
    def list_all_for_user_by_hash(user_id, param_hash, config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "jobs.param_hash = %s " + 
            "AND jobs.id IN (SELECT job_id from job_users where user_id=%s) " + 
            "AND jobs.is_active=1", (param_hash, user_id))
        return results 

    @staticmethod
    def list_pending(config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "(statuses.name='queued' OR statuses.name='started')")
        return results

    @staticmethod
    def list_all(config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db)
        return results

    @staticmethod
    def __list_by_sql_where(db, where="", vals=(), order=""):
        cur = db.cursor(MySQLdb.cursors.DictCursor)
        sql = """
            SELECT bin_to_uuid(jobs.id) AS id, jobs.name AS name, 
              statuses.name AS status, 
              DATE_FORMAT(jobs.creation_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS creation_date, 
              DATE_FORMAT(jobs.modified_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS modified_date,
              users.email as user_email,
              jobs.is_active
            FROM jobs
            LEFT JOIN statuses ON jobs.status_id = statuses.id
            LEFT JOIN users ON jobs.user_id = users.id"""
        if where:
            sql += " WHERE " + where
        if order:<commit_msg>Ignore canceled job when looking for dups<commit_after>import os
import shutil
import json
import sql_pool
import MySQLdb
import sys
import re
import hashlib
from user import User
from model_factory import ModelFactory

class Job:
    __dbfields = ["user_id","name","error_message","status_id","creation_date","modified_date", "is_active"]
    __extfields = ["status", "role"]

    def __init__(self, job_id, meta=None):
        self.job_id = job_id
        map(lambda x: setattr(self, x, None), self.__dbfields + self.__extfields)
        self.root_path = "" 
        self.users = []
        self.meta = meta

    def get_adjusted_phenotypes(self):
        phe_file = self.relative_path("output.phe")
        phenos = {}
        if os.path.exists(phe_file):
            with open(phe_file) as f:
                for line in f:
                    (sample, val) = line.split()
                    phenos[sample] = float(val)
        return phenos

    def relative_path(self, *args):
        return os.path.expanduser(os.path.join(self.root_path, *args))

    def get_genotype_id(self):
        return self.meta.get("genotype", None) 

    def get_model(self):
        return ModelFactory.get(self.meta.get("type", None), self.root_path, None)

    def get_output_files(self):
        files = []
        def add_if_exists(rel_path, display_name, primary=False):
            file_path = self.relative_path(rel_path)
            if os.path.exists(file_path):
                files.append({"path": rel_path, "size": os.path.getsize(file_path), 
                    "name": display_name, "primary": primary})
        add_if_exists("output.epacts.gz", "Epacts Results", True)
        add_if_exists("results.txt.gz", "SAIGE Results", True)
        add_if_exists("output.filtered.001.gz", "Filtered Results (p-val<0.001)")
        return files

    def get_output_file_path(self):
        files = [x for x in self.get_output_files() if x.get("primary", False)]
        if len(files)==1:
            return self.relative_path(files[0]["path"])
        else:
            return None

    def get_owner(self):
        return User.from_id(self.user_id) 

    def get_param_hash(self):
        return Job.calc_param_hash(self.meta)

    def as_object(self):
        obj = {key: getattr(self, key) for key in self.__dbfields  + self.__extfields if hasattr(self, key)} 
        obj["job_id"] = self.job_id
        obj["users"] = self.users
        obj["output_files"] = self.get_output_files()
        if self.meta:
            details = self.meta.copy()
            model = self.get_model()
            if model:
                details["model_desc"] = model.model_desc
                if "variant_filter" in details:
                    details["variant_filter_desc"] = model.get_filter_desc(details["variant_filter"])
            obj["details"] = details
            obj["hash"] = self.get_param_hash()
        return obj

    @staticmethod
    def get(job_id, config):
        if not re.match('[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}', job_id):
            return None
        job_folder = os.path.join(config.get("JOB_DATA_FOLDER", "./"), job_id)
        meta_path = os.path.expanduser(os.path.join(job_folder, "job.json"))
        if os.path.exists(meta_path):
            with open(meta_path) as meta_file:
                meta = json.load(meta_file)
        else:
           meta = dict()
        j = Job(job_id, meta)
        j.root_path = job_folder
        db = sql_pool.get_conn()
        cur = db.cursor(MySQLdb.cursors.DictCursor)
        sql = """
            SELECT
              bin_to_uuid(jobs.id) AS id,
              jobs.name AS name, jobs.user_id as user_id,
              jobs.status_id as status_id, statuses.name AS status,
              jobs.error_message AS error_message,
              DATE_FORMAT(jobs.creation_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS creation_date,
              DATE_FORMAT(jobs.modified_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS modified_date,
              jobs.is_active
            FROM jobs
            LEFT JOIN statuses ON jobs.status_id = statuses.id
            WHERE jobs.id = uuid_to_bin(%s)
            """
        cur.execute(sql, (job_id,))
        result = cur.fetchone()
        if result is not None:
            map(lambda x: setattr(j, x, result[x]), \
                (val for val in Job.__dbfields + Job.__extfields if val in result))
            sql = """
                SELECT 
                    ju.user_id as user_id, ju.role_id as role_id,
                    role.role_name as role, users.email, 
                    DATE_FORMAT(ju.modified_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS modified_date
                FROM job_users as ju
                    LEFT JOIN job_user_roles role on role.id = ju.role_id
                    LEFT JOIN users on users.id = ju.user_id
                WHERE 
                    ju.job_id = uuid_to_bin(%s)
                """
            cur.execute(sql, (job_id,))
            result = cur.fetchall()
            j.users = result
            return j
        else:
            return None

    @staticmethod
    def list_all_for_user(user_id, config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "jobs.id IN (SELECT job_id from job_users where user_id=%s) AND jobs.is_active=1", (user_id, ))
        return results 

    @staticmethod
    def list_all_for_phenotype(pheno_id, config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "jobs.pheno_id = uuid_to_bin(%s) AND jobs.is_active=1", (pheno_id, ))
        return results 

    @staticmethod
    def list_all_for_user_by_genotype(user_id, geno_id, config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "jobs.geno_id = uuid_to_bin(%s) " + 
            "AND jobs.id IN (SELECT job_id from job_users where user_id=%s) " + 
            "AND jobs.is_active=1", (geno_id, user_id))
        return results 

    @staticmethod
    def list_all_for_user_by_hash(user_id, param_hash, find_canceled=False, config=None):
        db = sql_pool.get_conn()
        where = ("jobs.param_hash = %s " +
            "AND jobs.id IN (SELECT job_id from job_users where user_id=%s) " +
            "AND jobs.is_active=1")
        if not find_canceled:
            where += " AND jobs.status_id not in (select id from statuses where name='cancelled')"
        results = Job.__list_by_sql_where(db, where, (param_hash, user_id))
        return results 

    @staticmethod
    def list_pending(config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db, "(statuses.name='queued' OR statuses.name='started')")
        return results

    @staticmethod
    def list_all(config=None):
        db = sql_pool.get_conn()
        results = Job.__list_by_sql_where(db)
        return results

    @staticmethod
    def __list_by_sql_where(db, where="", vals=(), order=""):
        cur = db.cursor(MySQLdb.cursors.DictCursor)
        sql = """
            SELECT bin_to_uuid(jobs.id) AS id, jobs.name AS name, 
              statuses.name AS status, 
              DATE_FORMAT(jobs.creation_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS creation_date, 
              DATE_FORMAT(jobs.modified_date, '%%Y-%%m-%%d %%H:%%i:%%s') AS modified_date,
              users.email as user_email,
              jobs.is_active
            FROM jobs
            LEFT JOIN statuses ON jobs.status_id = statuses.id
            LEFT JOIN users ON jobs.user_id = users.id"""
        if where:
            sql += " WHERE " + where
        if order:
{'fraction_non_alphanumeric': 0.11897751605995717, 'fraction_numerical': 0.0032119914346895075, 'mean_word_length': 2.134437919463087, 'pattern_counts': {'":': 8, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2916095', 'n_tokens_mistral': 4936, 'n_tokens_neox': 4716, 'n_words': 1028}