user.py

from . import BaseView, button_import, get_params
from .company import company_widget
from .user_login import (regenerate_security_code, send_email_security_code, generate_api_key,)
import os
import re
import colander
from deform import (widget, )

from opensipkd.tools import create_now, SaveFile
from opensipkd.tools.report import open_rml_row, csv_response, open_rml_pdf, pdf_response, file_response
from opensipkd.models import (DBSession, User, Group, UserGroup, ResCompany,)
from pyramid.i18n import TranslationStringFactory
from pyramid.view import view_config
from sqlalchemy import (func,)
from ziggurat_foundations.models.services.user import UserService

# lib buat mendapatkan full path asset
from pyramid.path import AssetResolver


_ = TranslationStringFactory('user')

class ListSchema(colander.Schema):
    id = colander.SchemaNode(
        colander.String(),
        title="Action",
        searchable=False)
    email = colander.SchemaNode(colander.String())
    user_name = colander.SchemaNode(
        colander.String(),
        title=_("user-name", 
        default="User Name"))
    status = colander.SchemaNode(
        colander.Integer(),
        widget=widget.CheckboxWidget(),
        width=50, 
        searchable=False)
    last_login = colander.SchemaNode(
        colander.String(), 
        width=100,
        field="last_login_date",
        searchable=False)
    registered = colander.SchemaNode(
        colander.String(), 
        width=100,
        field="registered_date",
        searchable=False)


class Views(BaseView):
    def __init__(self, request):
        super(Views, self).__init__(request)
        self.list_schema = ListSchema
        self.list_route = 'user'
        self.table = User
        self.edit_schema = EditSchema
        self.add_schema = AddSchema
        self.list_buttons = self.list_buttons + self.list_report + (button_import,)
        
        path = os.path.dirname(__file__)
        path = os.path.dirname(path)
        self.report_file = os.path.join(path, 'reports', 'users.jrxml')

    def get_bindings(self, row=None):
        status_list = (
            ('1', _('Active')),
            ('0', _('Archived')))
        if row and row.api_key:
            api_key_list = (
                ('', _(row.api_key)),
                ('0', _('Hapus')))
        else:
            api_key_list = (
                ('', _('Tidak ada')),
                ('1', _('Buatkan')))

        group_list = get_group_list()
        return dict(status_list=status_list,
                    group_list=group_list,
                    api_key_list=api_key_list,
                    user=row,
                    company_list=ResCompany.get_list())

    @view_config(
        route_name='user', 
        renderer='templates/table.pt',
        permission='user-view')
    def view_list(self):
        return super(Views, self).view_list()

    # Mendapatkan path gambar
    def get_module_path(self, module_name):
        a = AssetResolver(module_name)
        resolver = a.resolve('')
        return resolver.abspath()

    def pdf_response(self, **kwargs):
        from opensipkd.base.tools.report import jasper_export
        logo = self.get_module_path('opensipkd.base')
        logo = os.path.join(logo, 'static', 'img', 'logo.png')
        parameters = {
            "judul": get_params('company', "openSIPKD"),
            "logo": logo
        }
        print(parameters)
        filename = jasper_export(self.report_file, parameters=parameters)
        return file_response(self.req, filename=filename[0])

    @view_config(
        route_name='user-act', 
        renderer='json', 
        permission='user-view')
    def view_act(self):
        # url_dict = self.req.matchdict
        # if url_dict['act'] == 'csv':
        #     query = query_register()
        #     row = query.first()
        #     header = row.keys()
        #     rows = [list(item) for item in query.all()]
        #     filename = 'user.csv'
        #     value = {
        #         'header': header,
        #         'rows': rows,
        #     }
        #     return csv_response(self.req, value, filename)
        
        # elif url_dict['act'] == 'pdf':
        #     query = query_register()
        #     import opensipkd.base
        #     base_path = os.path.dirname(opensipkd.base.__file__)
        #     path = os.path.join(base_path, 'reports')
        #     rml_row = open_rml_row(path + '/user.row.rml')
        #     rows = [rml_row.format(user_name=r.user_name, email=r.email,
        #                            registered_date=r.registered_date) for r in
        #             query.all()]
        #     pdf, filename = open_rml_pdf(path + '/user.rml', rows=rows,
        #                                  company=self.req.company,
        #                                  departement=self.req.departement,
        #                                  address=self.req.address,
        #                                  base_path=base_path)
        #     filename = os.path.basename(filename)
        #     resp = pdf_response(self.req, pdf, filename)
        #     if resp.content_length < 10:
        #         resp.content_length = len(resp.body)
        #     return resp

        return super(Views, self).view_act()

    def form_validator(self, form, value):
        if "company_id" in value and not value["company_id"]:
            value["company_id"] = None

    def save_request(self, values, row=None):
        request = self.req
        values["email"] = values['email'].lower()
        values["user_name"] = re.sub(' ', '', values['user_name'])  # .lower()
        values["security_code_date"] = create_now()
        # company_id = request.user and request.user.company_id or "company_id" in values and \
        #              values["company_id"] or None
        # values["company_id"] = company_id
        if "company_id" not in values:
            values["company_id"] = None
        if 'is_api_key' in values:
            values["api_key"] = generate_api_key()
        insert = not row
        row = self.save(values, self.req.user, row)
        if insert:
            remain = regenerate_security_code(row)
            if 'password' in values:
                data = dict(username=row.user_name)
                ts = _(
                    'user-added-with-password',
                    default='${username} berhasil ditambahkan.', mapping=data)
            else:
                send_email_security_code(
                    self.req, row, remain, 'Welcome new user', 'email-new-user',
                    'email-new-user.tpl')
                data = dict(email=row.email)
                ts = _(
                    'user-added',
                    default='${email} berhasil ditambahkan dan email untuk ubah '
                            'kata kunci sudah dikirim.',
                    mapping=data)
            self.ses.flash(ts)

        if 'password' in values:
            UserService.set_password(row, values['password'])

        DBSession.add(row)
        DBSession.flush()

        existing = user_group_set(row)
        unused = existing - values['groups']
        if unused:
            q = DBSession.query(UserGroup).filter_by(user_id=row.id).filter(
                UserGroup.group_id.in_(unused))
            q.delete(synchronize_session=False)
            for gid in unused:
                reduce_member_count(gid)
        new = values['groups'] - existing
        for gid in new:
            ug = UserGroup()
            ug.user_id = row.id
            ug.group_id = gid
            DBSession.add(ug)
            add_member_count(gid)
        return row

    @view_config(
        route_name='user-add', 
        renderer='templates/form.pt',
        permission='user-view')
    def view_add(self):
        return super(Views, self).view_add()
        # user, remain = insert(request, values)

    def get_values(self, row, istime=False):
        d = super(Views, self).get_values(row, istime)
        d["groups"] = user_group_set(row)
        return d

    @view_config(
        route_name='user-edit', 
        renderer='templates/form.pt',
        permission='user-edit')
    def view_edit(self):
        return super(Views, self).view_edit()

    @view_config(
        route_name='user-view', 
        renderer='templates/form.pt',
        permission='user-view')
    def view_view(self):
        return super(Views, self).view_view()

    @view_config(
        route_name='user-delete', 
        renderer='templates/form.pt',
        permission='user-edit')
    def view_delete(self):
        return super(Views, self).view_delete()

    def delete_msg(self, row):
        data = dict(uid=row.id, email=row.email)
        return _(
            'user-deleted',
            default='User ${email} ID ${uid} has been deleted',
            mapping=data)

    def before_delete(self, row):
        gid_list = user_group_set(row)
        for gid in gid_list:
            reduce_member_count(gid)

    def query_id(self):
        q = DBSession.query(User).filter_by(id=self.req.matchdict['id'])
        if self.req.user.company_id:
            q = q.filter_by(company_id=self.req.user.company_id)
        return q
    
    # Tambah fungsi upload/import
    @view_config(
        route_name='user-upload',
        renderer='templates/form.pt',
        permission='user'
    )
    def view_upload(self):
        self.upload_keys = ["email"]
        return super().view_upload(exts=(".csv", ".tsv"))


#######
# Add #
#######
@colander.deferred
def status_widget(node, kw):
    values = kw.get('status_list', [])
    return widget.SelectWidget(values=values)


@colander.deferred
def group_widget(node, kw):
    values = kw.get('group_list', [])
    return widget.CheckboxChoiceWidget(values=values)


@colander.deferred
def api_key_widget(node, kw):
    values = kw.get('api_key_list', [])
    return widget.SelectWidget(values=values)


class Validator:
    def __init__(self, user):
        self.user = user


class EmailValidator(colander.Email, Validator):
    def __init__(self, user):
        colander.Email.__init__(self)
        Validator.__init__(self, user)

    def __call__(self, node, value):
        def email_found():
            data = dict(email=email, uid=found.id)
            ts = _(
                'email-already-used',
                default='Email ${email} already used by user ID ${uid}',
                mapping=data)
            raise colander.Invalid(node, ts)

        if self.match_object.match(value) is None:
            raise colander.Invalid(node, _('Invalid email format'))

        email = value.lower()
        q = DBSession.query(User).filter_by(email=email)
        found = q.first()
        if found and (not self.user or self.user.email != found.email):
            email_found()

        # Cek karakter
        def email_max():
            raise colander.Invalid(node, _('Maksimal 100 karakter'))

        if (len(email) > 100):
            email_max()


REGEX_ONLY_CONTAIN = re.compile('([A-Za-z0-9-]*)')
REGEX_BEGIN_END_ALPHANUMERIC = re.compile('^[A-Za-z0-9]+(?:[-][A-Za-z0-9]+)*$')


class UsernameValidator(Validator):
    def __call__(self, node, value):
        username = value

        # Fungsi menampilkan error pada field form
        def username_max():
            raise colander.Invalid(node, _('Maksimal 128 karakter'))

        # Jika karakter melebihi 128 karakter maka, panggil fungsi
        if (len(username) > 128):
            username_max()

        if self.user and self.user.user_name == username:
            return
        match = REGEX_ONLY_CONTAIN.search(username)
        if not match or match.group(1) != username or username != value:
            ts = _(
                'username-only-contain',
                default='Only A-Z a-z, 0-9, and - characters are allowed')
            raise colander.Invalid(node, ts)
        match = REGEX_BEGIN_END_ALPHANUMERIC.search(username)
        if not match:
            ts = _(
                'username-first-end-alphanumeric',
                default='Only A-Z a-z or 0-9 at the start and end')
            raise colander.Invalid(node, ts)
        q = DBSession.query(User).filter_by(user_name=username)
        found = q.first()
        if not found:
            return

        data = dict(username=username, uid=found.id)
        ts = _(
            'username-already-used',
            default='Username ${username} already used by ID ${uid}',
            mapping=data)
        raise colander.Invalid(node, ts)


@colander.deferred
def email_validator(node, kw):
    return EmailValidator(kw['user'])


@colander.deferred
def username_validator(node, kw):
    return UsernameValidator(kw['user'])


def save_user(values, user, row=None):
    if not row:
        row = User()
        row.status = 0

    row.from_dict(values)
    DBSession.add(row)
    DBSession.flush()
    if 'password' in values and values['password']:
        UserService.set_password(row, values['password'])
    return row


class AddSchema(colander.Schema):
    email = colander.SchemaNode(
        colander.String(), 
        title=_('Email'),
        validator=email_validator)

    user_name = colander.SchemaNode(
        colander.String(), 
        title=_('Username'),
        validator=username_validator)

    groups = colander.SchemaNode(
        colander.Set(), 
        widget=group_widget, 
        title=_('Group'))

    is_api_key = colander.SchemaNode(
        colander.String(), 
        widget=api_key_widget, 
        title=_('API Key'),
        missing=colander.drop)

    password = colander.SchemaNode(
        colander.String(), 
        widget=widget.CheckedPasswordWidget(),
        missing=colander.drop)
    
    company_id = colander.SchemaNode(
        colander.Integer(), 
        widget=company_widget,
        title="Company",
        missing=colander.drop)


class EditSchema(AddSchema):
    status = colander.SchemaNode(
        colander.String(), widget=widget.CheckboxWidget(true_val="1", false_val="0"), title=_('Status'))


def get_group_list():
    r = []
    q = DBSession.query(Group).order_by(Group.group_name)
    for row in q:
        g = (str(row.id), _(row.description))
        r.append(g)
    return r


def add_member_count(gid):
    q = DBSession.query(Group).filter_by(id=gid)
    group = q.first()
    group.member_count += 1
    DBSession.add(group)


def reduce_member_count(gid):
    q = DBSession.query(Group).filter_by(id=gid)
    group = q.first()
    group.member_count -= 1
    DBSession.add(group)


def user_group_set(user):
    q = DBSession.query(UserGroup).filter_by(user_id=user.id)
    r = []
    for ug in q:
        r.append(str(ug.group_id))
    return set(r)


def query_register():
    return DBSession.query(User.user_name, User.email,
                           func.to_char(User.registered_date,
                                        "DD-MM-YYYY").label(
                               "registered_date")).order_by(
        User.user_name)


def user_list():
    qry = User.query().order_by(User.user_name)
    return [(r.id, r.user_name) for r in qry]


def user_select():
    result = user_list()
    result.insert(0, ('', 'Pilih User'))
    return result


@colander.deferred
def user_widget(node, kw):
    values = kw.get('user_list', [])
    request = kw.get("request")
    return widget.SelectWidget(values=values,
                               placeholder="Pilih User",
                               style="width:300px;")


class UserFilterSchema(colander.Schema):
    user_id = colander.SchemaNode(
        colander.Integer(),
        widget=user_widget,
        oid="user_id",
        title="User",
        missing=colander.drop,
    )