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.models.document import Catalog from config.type_defs.catalog_types import CATALOG_TYPES from config.type_defs.processor_types import PROCESSOR_TYPES from config.type_defs.retriever_types import RETRIEVER_TYPES from .dynamic_form_base import DynamicFormBase def validate_json(form, field): if field.data: try: json.loads(field.data) except json.JSONDecodeError: raise ValidationError('Invalid JSON format') class CatalogForm(FlaskForm): name = StringField('Name', validators=[DataRequired(), Length(max=50)]) description = TextAreaField('Description', validators=[Optional()]) # Select Field for Catalog Type (Uses the CATALOG_TYPES defined in config) type = SelectField('Catalog Type', validators=[DataRequired()]) 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]) 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)]) 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}) 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=[Optional()], ) # 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) # Dynamically populate the 'type' field using the constructor self.type.choices = [(key, value['name']) for key, value in RETRIEVER_TYPES.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')