agent/app/services.py - master - pv_hum/schemaforge

agent/app/services.py
216 строк12 KB
pv_hum
Initial commit
21 май 2026, 22:25
21 май 2026, 22:250de9fa8
import re
import sqlparse
from .models import *

def normalize_name(value: str) -> str:
    value = re.sub(r'[^a-zA-Z0-9_]+', '_', value.strip()).lower()
    value = re.sub(r'_+', '_', value).strip('_')
    return value or 'unnamed'

def normalize_schema(schema: SchemaModel, source_mode: str | None = None, dialect: str | None = None) -> SchemaModel:
    entities = []
    seen_entities = set()
    for idx, entity in enumerate(schema.entities):
        name = normalize_name(entity.name)
        if name in seen_entities:
            name = f'{name}_{idx+1}'
        seen_entities.add(name)
        attrs = []
        seen_attrs = set()
        has_pk = False
        for attr in entity.attributes:
            attr_name = normalize_name(attr.name)
            if attr_name in seen_attrs:
                continue
            seen_attrs.add(attr_name)
            pk = bool(attr.primaryKey)
            has_pk = has_pk or pk
            ref = attr.references
            attrs.append(Attribute(
                name=attr_name,
                type=(attr.type or 'varchar(255)').lower(),
                nullable=bool(attr.nullable) if not pk else False,
                primaryKey=pk,
                unique=bool(attr.unique),
                foreignKey=bool(attr.foreignKey or ref is not None),
                references=AttributeReference(entity=normalize_name(ref.entity), attribute=normalize_name(ref.attribute)) if ref else None,
            ))
        if not has_pk:
            attrs.insert(0, Attribute(name='id', type='bigint', nullable=False, primaryKey=True))
        entities.append(Entity(name=name, displayName=entity.displayName or entity.name.title(), description=entity.description, attributes=attrs, x=entity.x, y=entity.y))

    relationships = []
    dedupe = set()
    for rel in schema.relationships:
        key = (normalize_name(rel.fromEntity), normalize_name(rel.toEntity), rel.fromField or '', rel.toField or '')
        if key in dedupe:
            continue
        dedupe.add(key)
        relationships.append(Relationship(
            name=normalize_name(rel.name) if rel.name else None,
            fromEntity=normalize_name(rel.fromEntity),
            toEntity=normalize_name(rel.toEntity),
            type=rel.type,
            fromField=normalize_name(rel.fromField) if rel.fromField else None,
            toField=normalize_name(rel.toField) if rel.toField else None,
            label=rel.label,
            junctionEntity=normalize_name(rel.junctionEntity) if rel.junctionEntity else None,
        ))

    lookup = {e.name: e for e in entities}
    for entity in entities:
        for attr in entity.attributes:
            if attr.references and attr.references.entity in lookup:
                relationships.append(Relationship(
                    fromEntity=entity.name,
                    toEntity=attr.references.entity,
                    type='many-to-one',
                    fromField=attr.name,
                    toField=attr.references.attribute,
                    label=f'{entity.name}.{attr.name} -> {attr.references.entity}.{attr.references.attribute}'
                ))
    uniq = []
    seen = set()
    for rel in relationships:
        key = (rel.fromEntity, rel.toEntity, rel.fromField, rel.toField, rel.type)
        if key not in seen:
            seen.add(key)
            uniq.append(rel)
    return SchemaModel(entities=entities, relationships=uniq, metadata=schema.metadata or {}, sourceMode=source_mode or schema.sourceMode, sqlDialect=dialect or schema.sqlDialect)

def validate_schema(schema: SchemaModel) -> ValidationResult:
    issues: list[ValidationIssue] = []
    if not schema.entities:
        issues.append(ValidationIssue(level='error', code='NO_ENTITIES', message='Schema must contain at least one entity.'))
    names = [e.name for e in schema.entities]
    if len(names) != len(set(names)):
        issues.append(ValidationIssue(level='error', code='DUPLICATE_ENTITY', message='Entity names must be unique.'))
    entity_map = {e.name: e for e in schema.entities}
    for entity in schema.entities:
        if not entity.attributes:
            issues.append(ValidationIssue(level='error', code='NO_ATTRIBUTES', message=f'Entity {entity.name} has no attributes.', entity=entity.name))
        pk_count = sum(1 for a in entity.attributes if a.primaryKey)
        if pk_count == 0:
            issues.append(ValidationIssue(level='error', code='NO_PRIMARY_KEY', message=f'Entity {entity.name} has no primary key.', entity=entity.name))
        attr_names = [a.name for a in entity.attributes]
        if len(attr_names) != len(set(attr_names)):
            issues.append(ValidationIssue(level='warning', code='DUPLICATE_ATTRIBUTE', message=f'Entity {entity.name} has duplicate attributes.', entity=entity.name))
        for attr in entity.attributes:
            if attr.references:
                target = entity_map.get(attr.references.entity)
                if not target:
                    issues.append(ValidationIssue(level='error', code='BAD_REFERENCE_ENTITY', message=f'{entity.name}.{attr.name} references missing entity {attr.references.entity}.', entity=entity.name, attribute=attr.name))
                elif attr.references.attribute not in [x.name for x in target.attributes]:
                    issues.append(ValidationIssue(level='error', code='BAD_REFERENCE_ATTRIBUTE', message=f'{entity.name}.{attr.name} references missing attribute {attr.references.attribute}.', entity=entity.name, attribute=attr.name))
    for rel in schema.relationships:
        if rel.fromEntity not in entity_map or rel.toEntity not in entity_map:
            issues.append(ValidationIssue(level='error', code='BAD_RELATIONSHIP', message=f'Relationship endpoints must exist: {rel.fromEntity} -> {rel.toEntity}.'))
    if not issues:
        issues.append(ValidationIssue(level='info', code='VALID', message='Schema passed core validation checks.'))
    return ValidationResult(valid=not any(i.level == 'error' for i in issues), issues=issues)

def schema_to_sql(schema: SchemaModel, dialect: str = 'postgresql') -> str:
    lines: list[str] = []
    serial_type = 'BIGSERIAL' if dialect == 'postgresql' else 'BIGINT AUTO_INCREMENT'
    for entity in schema.entities:
        col_lines = []
        pk_cols = []
        table_constraints = []
        for attr in entity.attributes:
            atype = attr.type
            if attr.primaryKey and atype == 'bigint' and dialect in ('postgresql', 'mysql'):
                atype = serial_type
            col = f'  {attr.name} {atype}'
            if not attr.nullable:
                col += ' NOT NULL'
            if attr.unique and not attr.primaryKey:
                col += ' UNIQUE'
            if attr.primaryKey and 'AUTO_INCREMENT' not in atype and 'SERIAL' not in atype:
                pk_cols.append(attr.name)
            if attr.primaryKey and ('SERIAL' in atype or 'AUTO_INCREMENT' in atype):
                col += ' PRIMARY KEY'
            if attr.references:
                col += f' REFERENCES {attr.references.entity}({attr.references.attribute})'
            col_lines.append(col)
        if pk_cols:
            table_constraints.append(f'  PRIMARY KEY ({", ".join(pk_cols)})')
        create_stmt = ',\n'.join(col_lines + table_constraints)
        lines.append(f'CREATE TABLE {entity.name} (\n{create_stmt}\n);')
    return '\n\n'.join(lines)

def explain_changes(old: SchemaModel, new: SchemaModel) -> ChangeSummary:
    items: list[ChangeItem] = []
    old_entities = {e.name: e for e in old.entities}
    new_entities = {e.name: e for e in new.entities}
    for name in new_entities.keys() - old_entities.keys():
        items.append(ChangeItem(type='added', target=name, message=f'Added entity: {name}'))
    for name in old_entities.keys() - new_entities.keys():
        items.append(ChangeItem(type='removed', target=name, message=f'Removed entity: {name}'))
    for name in old_entities.keys() & new_entities.keys():
        old_attrs = {a.name: a for a in old_entities[name].attributes}
        new_attrs = {a.name: a for a in new_entities[name].attributes}
        for attr in new_attrs.keys() - old_attrs.keys():
            items.append(ChangeItem(type='added', target=f'{name}.{attr}', message=f'Added attribute: {name}.{attr}'))
        for attr in old_attrs.keys() - new_attrs.keys():
            items.append(ChangeItem(type='removed', target=f'{name}.{attr}', message=f'Removed attribute: {name}.{attr}'))
        for attr in old_attrs.keys() & new_attrs.keys():
            if old_attrs[attr].model_dump() != new_attrs[attr].model_dump():
                items.append(ChangeItem(type='updated', target=f'{name}.{attr}', message=f'Updated attribute: {name}.{attr}'))
    if not items:
        items.append(ChangeItem(type='updated', target='schema', message='No structural differences detected.'))
    return ChangeSummary(items=items)

def parse_sql_to_schema(sql: str, dialect: str = 'postgresql') -> SchemaModel:
    entities: list[Entity] = []
    relationships: list[Relationship] = []
    statements = [s.strip() for s in sqlparse.split(sql) if s.strip()]
    for stmt in statements:
        normalized = stmt.strip().rstrip(';')
        m = re.search(r'create\s+table\s+([a-zA-Z_][\w]*)\s*\((.*)\)$', normalized, re.IGNORECASE | re.DOTALL)
        if not m:
            continue
        table_name = normalize_name(m.group(1))
        body = m.group(2)
        parts = []
        depth = 0
        buf = []
        for ch in body:
            if ch == '(':
                depth += 1
            elif ch == ')':
                depth -= 1
            if ch == ',' and depth == 0:
                parts.append(''.join(buf).strip())
                buf = []
            else:
                buf.append(ch)
        if buf:
            parts.append(''.join(buf).strip())
        attrs: list[Attribute] = []
        table_pk: list[str] = []
        for part in parts:
            if re.match(r'^primary\s+key', part, re.IGNORECASE):
                cols = re.findall(r'\((.*?)\)', part)
                if cols:
                    table_pk.extend([normalize_name(x.strip()) for x in cols[0].split(',')])
                continue
            col_match = re.match(r'^([a-zA-Z_][\w]*)\s+(.+)$', part, re.IGNORECASE | re.DOTALL)
            if not col_match:
                continue
            col_name = normalize_name(col_match.group(1))
            rest = col_match.group(2)
            type_match = re.match(r'^([a-zA-Z]+(?:\s*\([^)]*\))?(?:\s+[a-zA-Z]+)?)', rest, re.IGNORECASE)
            col_type = type_match.group(1) if type_match else 'varchar(255)'
            primary = bool(re.search(r'primary\s+key', rest, re.IGNORECASE))
            nullable = not bool(re.search(r'not\s+null', rest, re.IGNORECASE))
            unique = bool(re.search(r'\bunique\b', rest, re.IGNORECASE))
            ref_match = re.search(r'references\s+([a-zA-Z_][\w]*)\s*\(([^)]+)\)', rest, re.IGNORECASE)
            ref = None
            if ref_match:
                ref = AttributeReference(entity=normalize_name(ref_match.group(1)), attribute=normalize_name(ref_match.group(2)))
                relationships.append(Relationship(fromEntity=table_name, toEntity=normalize_name(ref_match.group(1)), type='many-to-one', fromField=col_name, toField=normalize_name(ref_match.group(2)), label=f'{table_name}.{col_name} -> {normalize_name(ref_match.group(1))}.{normalize_name(ref_match.group(2))}'))
            attrs.append(Attribute(name=col_name, type=col_type.lower(), nullable=nullable, primaryKey=primary, unique=unique, foreignKey=ref is not None, references=ref))
        if table_pk:
            attrs = [a.model_copy(update={'primaryKey': a.name in table_pk, 'nullable': False if a.name in table_pk else a.nullable}) for a in attrs]
        entities.append(Entity(name=table_name, displayName=table_name.replace('_', ' ').title(), attributes=attrs))
    return normalize_schema(SchemaModel(entities=entities, relationships=relationships, sourceMode='sql', sqlDialect=dialect), source_mode='sql', dialect=dialect)