#!/usr/bin/env python3 # Copyright © 2023 Lee Yingtong Li (RunasSudo) # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import itertools import sqlite3 # Open database con = sqlite3.connect('database.db') con.row_factory = sqlite3.Row cur = con.cursor() # Populate pbs_restriction.criteria_rendered # sql.js is too slow to do the database JOINs at runtime class Operator: def __init__(self, operator, nodes): self.operator = operator self.nodes = nodes def flatten(self): # If contains just 1 element, return it as itself, not as an Operator if len(self.nodes) == 1: return self.nodes[0] # Otherwise flatten recursively flattened = [n.flatten() if isinstance(n, Operator) else n for n in self.nodes] return Operator(self.operator, flattened) @staticmethod def render(op): if isinstance(op, str): return '

' + op + '

' elif isinstance(op, Operator): if len(op.nodes) == 1: # Single simple string return '

' + op.nodes[0] + '

' if not any(isinstance(n, Operator) for n in op.nodes): # All simple strings return '' else: # Mix of Operator +/- simple strings return '

{}

'.format({'all': 'AND', 'any': 'OR', 'one-of': 'OR'}[op.operator]).join(Operator.render(n) for n in op.nodes) else: # Must be a list of a single item return Operator.render(op[0]) cur.execute('SELECT * FROM pbs_restriction INNER JOIN pbs_restriction_criteria ON pbs_restriction.code = pbs_restriction_criteria.restriction_code INNER JOIN pbs_criteria ON pbs_restriction_criteria.criteria_code = pbs_criteria.code INNER JOIN pbs_criteria_parameter ON pbs_criteria.code = pbs_criteria_parameter.criteria_code') flat_parameters = cur.fetchall() parameters_by_restriction = itertools.groupby(flat_parameters, lambda p: p['restriction_code']) for restriction_code, restriction_parameters in parameters_by_restriction: restriction_parameters = list(restriction_parameters) # Build tree of nodes representing parameters restriction_operator = Operator(restriction_parameters[0]['criteria_operator'], []) parameters_by_criteria = itertools.groupby(restriction_parameters, lambda p: p['criteria_code']) for criteria_code, criteria_parameters in parameters_by_criteria: criteria_parameters = list(criteria_parameters) criteria_operator = Operator(criteria_parameters[0]['parameters_operator'], [p['text'] for p in criteria_parameters]) restriction_operator.nodes.append(criteria_operator) # Flatten and render the tree restriction_operator = restriction_operator.flatten() rendered = Operator.render(restriction_operator) cur.execute('UPDATE pbs_restriction SET criteria_rendered = ? WHERE code = ?', (rendered, restriction_code)) con.commit()