from flask import session, current_app from flask_wtf import FlaskForm from wtforms import (StringField, BooleanField, SubmitField, DateField, IntegerField, SelectField, TextAreaField, URLField) from wtforms.validators import DataRequired, Length, Optional, URL, ValidationError, NumberRange from flask_wtf.file import FileField, FileRequired import json from wtforms_sqlalchemy.fields import QuerySelectField from common.extensions import cache_manager from common.models.document import Catalog from common.utils.form_assistants import validate_json from config.type_defs.catalog_types import CATALOG_TYPES from config.type_defs.processor_types import PROCESSOR_TYPES from .dynamic_form_base import DynamicFormBase def validate_catalog_name(form, field): # Controleer of een catalog met deze naam al bestaat existing_catalog = Catalog.query.filter_by(name=field.data).first() if existing_catalog: raise ValidationError(f'A Catalog with name "{field.data}" already exists. Choose another name.') class CatalogForm(FlaskForm): name = StringField('Name', validators=[DataRequired(), Length(max=50), validate_catalog_name]) description = TextAreaField('Description', validators=[Optional()]) # Select Field for Catalog Type (Uses the CATALOG_TYPES defined in config) type = SelectField('Catalog Type', validators=[DataRequired()]) # Selection fields for processing & creating embeddings min_chunk_size = IntegerField('Minimum Chunk Size (1500)', validators=[NumberRange(min=0), Optional()], default=1500) max_chunk_size = IntegerField('Maximum Chunk Size (2500)', validators=[NumberRange(min=0), Optional()], default=2500) # Metadata fields user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) system_metadata = TextAreaField('System Metadata', validators=[Optional(), validate_json]) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Dynamically populate the 'type' field using the constructor self.type.choices = [(key, value['name']) for key, value in CATALOG_TYPES.items()] class EditCatalogForm(DynamicFormBase): name = StringField('Name', validators=[DataRequired(), Length(max=50), validate_catalog_name]) description = TextAreaField('Description', validators=[Optional()]) # Select Field for Catalog Type (Uses the CATALOG_TYPES defined in config) type = StringField('Catalog Type', validators=[DataRequired()], render_kw={'readonly': True}) # Selection fields for processing & creating embeddings min_chunk_size = IntegerField('Minimum Chunk Size (2000)', validators=[NumberRange(min=0), Optional()], default=2000) max_chunk_size = IntegerField('Maximum Chunk Size (3000)', validators=[NumberRange(min=0), Optional()], default=3000) # Metadata fields user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) system_metadata = TextAreaField('System Metadata', validators=[Optional(), validate_json], ) class ProcessorForm(FlaskForm): name = StringField('Name', validators=[DataRequired(), Length(max=50)]) description = TextAreaField('Description', validators=[Optional()]) # Catalog for the Retriever catalog = QuerySelectField( 'Catalog ID', query_factory=lambda: Catalog.query.all(), allow_blank=True, get_label='name', validators=[DataRequired()], ) # Select Field for Catalog Type (Uses the CATALOG_TYPES defined in config) type = SelectField('Processor Type', validators=[DataRequired()]) sub_file_type = StringField('Sub File Type', validators=[Optional(), Length(max=50)]) min_chunk_size = IntegerField('Minimum Chunk Size (2000)', validators=[NumberRange(min=0), Optional()], default=2000) max_chunk_size = IntegerField('Maximum Chunk Size (3000)', validators=[NumberRange(min=0), Optional()], default=3000) tuning = BooleanField('Enable Embedding Tuning', default=False) # Metadata fields user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) system_metadata = TextAreaField('System Metadata', validators=[Optional(), validate_json]) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Dynamically populate the 'type' field using the constructor self.type.choices = sorted( [(key, value['name']) for key, value in PROCESSOR_TYPES.items()], key=lambda x: x[1], ) class EditProcessorForm(DynamicFormBase): name = StringField('Name', validators=[DataRequired(), Length(max=50)]) description = TextAreaField('Description', validators=[Optional()]) # Catalog for the Retriever catalog = QuerySelectField( 'Catalog ID', query_factory=lambda: Catalog.query.all(), allow_blank=True, get_label='name', validators=[Optional()], ) type = StringField('Processor Type', validators=[DataRequired()], render_kw={'readonly': True}) sub_file_type = StringField('Sub File Type', validators=[Optional(), Length(max=50)]) min_chunk_size = IntegerField('Minimum Chunk Size (2000)', validators=[NumberRange(min=0), Optional()], default=2000) max_chunk_size = IntegerField('Maximum Chunk Size (3000)', validators=[NumberRange(min=0), Optional()], default=3000) tuning = BooleanField('Enable Embedding Tuning', default=False) # Metadata fields user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) system_metadata = TextAreaField('System Metadata', validators=[Optional(), validate_json]) class RetrieverForm(FlaskForm): name = StringField('Name', validators=[DataRequired(), Length(max=50)]) description = TextAreaField('Description', validators=[Optional()]) # Catalog for the Retriever catalog = QuerySelectField( 'Catalog ID', query_factory=lambda: Catalog.query.all(), allow_blank=True, get_label='name', validators=[Optional()], ) # Select Field for Retriever Type (Uses the RETRIEVER_TYPES defined in config) type = SelectField('Retriever Type', validators=[DataRequired()]) tuning = BooleanField('Enable Tuning', default=False) # Metadata fields user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) system_metadata = TextAreaField('System Metadata', validators=[Optional(), validate_json]) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) types_dict = cache_manager.retrievers_types_cache.get_types() # Dynamically populate the 'type' field using the constructor self.type.choices = [(key, value['name']) for key, value in types_dict.items()] class EditRetrieverForm(DynamicFormBase): name = StringField('Name', validators=[DataRequired(), Length(max=50)]) description = TextAreaField('Description', validators=[Optional()]) # Catalog for the Retriever catalog = QuerySelectField( 'Catalog ID', query_factory=lambda: Catalog.query.all(), allow_blank=True, get_label='name', validators=[Optional()], ) # Select Field for Retriever Type (Uses the RETRIEVER_TYPES defined in config) type = StringField('Processor Type', validators=[DataRequired()], render_kw={'readonly': True}) tuning = BooleanField('Enable Tuning', default=False) # Metadata fields user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) system_metadata = TextAreaField('System Metadata', validators=[Optional(), validate_json]) class AddDocumentForm(DynamicFormBase): file = FileField('File', validators=[FileRequired()]) catalog = StringField('Catalog', render_kw={'readonly': True}) sub_file_type = StringField('Sub File Type', validators=[Optional(), Length(max=50)]) name = StringField('Name', validators=[Length(max=100)]) language = SelectField('Language', choices=[], validators=[Optional()]) user_context = TextAreaField('User Context', validators=[Optional()]) valid_from = DateField('Valid from', id='form-control datepicker', validators=[Optional()]) user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.language.choices = [(language, language) for language in session.get('tenant').get('allowed_languages')] if not self.language.data: self.language.data = session.get('tenant').get('default_language') self.catalog.data = session.get('catalog_name', '') class AddURLForm(DynamicFormBase): url = URLField('URL', validators=[DataRequired(), URL()]) catalog = StringField('Catalog', render_kw={'readonly': True}) sub_file_type = StringField('Sub File Type', validators=[Optional(), Length(max=50)]) name = StringField('Name', validators=[Length(max=100)]) language = SelectField('Language', choices=[], validators=[Optional()]) user_context = TextAreaField('User Context', validators=[Optional()]) valid_from = DateField('Valid from', id='form-control datepicker', validators=[Optional()]) user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.language.choices = [(language, language) for language in session.get('tenant').get('allowed_languages')] if not self.language.data: self.language.data = session.get('tenant').get('default_language') self.catalog.data = session.get('catalog_name', '') class EditDocumentForm(FlaskForm): name = StringField('Name', validators=[Length(max=100)]) valid_from = DateField('Valid from', id='form-control datepicker', validators=[Optional()]) valid_to = DateField('Valid to', id='form-control datepicker', validators=[Optional()]) submit = SubmitField('Submit') class EditDocumentVersionForm(DynamicFormBase): sub_file_type = StringField('Sub File Type', validators=[Optional(), Length(max=50)]) language = StringField('Language') user_context = TextAreaField('User Context', validators=[Optional()]) system_context = TextAreaField('System Context', validators=[Optional()]) user_metadata = TextAreaField('User Metadata', validators=[Optional(), validate_json]) system_metadata = TextAreaField('System Metadata', validators=[Optional(), validate_json]) submit = SubmitField('Submit')