backend/scripts/swagger-drogon/swagger_drogon/infer.py

from __future__ import annotations

import re
from pathlib import Path
from typing import Any

from .models import InferredOperation, Route


def load_sources(source_dirs: list[Path]) -> dict[Path, str]:
    sources: dict[Path, str] = {}
    for directory in source_dirs:
        if not directory.exists():
            continue
        for path in sorted(directory.glob("*.cc")):
            sources[path] = path.read_text(encoding="utf-8")
    return sources


def infer_operation(route: Route, sources: dict[Path, str]) -> InferredOperation:
    controller_body = _find_controller_body(route, sources)
    service_body = _find_service_body(controller_body, sources)
    combined = "\n".join(part for part in (controller_body, service_body) if part)

    inferred = InferredOperation()
    inferred.query_parameters = _infer_query_parameters(controller_body)
    inferred.request_schema = _infer_request_schema(combined)
    inferred.response_schema = _infer_response_schema(combined)
    return inferred


def _find_controller_body(route: Route, sources: dict[Path, str]) -> str:
    pattern = f"void {route.controller}::{route.handler}"
    for path, text in sources.items():
        if path.name != f"{route.controller}.cc":
            continue
        index = text.find(pattern)
        if index == -1:
            return ""
        brace = text.find("{", index)
        if brace == -1:
            return ""
        return _balanced_block(text, brace)
    return ""


def _find_service_body(controller_body: str, sources: dict[Path, str]) -> str:
    match = re.search(r"\bservice\.([A-Za-z_]\w*)\s*\(", controller_body)
    if not match:
        return ""
    method = match.group(1)
    pattern = re.compile(rf"Json::Value\s+[A-Za-z_]\w*Service::{method}\s*\(")
    for path, text in sources.items():
        if not path.name.endswith("Service.cc"):
            continue
        match = pattern.search(text)
        if not match:
            continue
        brace = text.find("{", match.start())
        if brace == -1:
            return ""
        return _balanced_block(text, brace)
    return ""


def _balanced_block(text: str, open_brace: int) -> str:
    depth = 0
    for index in range(open_brace, len(text)):
        char = text[index]
        if char == "{":
            depth += 1
        elif char == "}":
            depth -= 1
            if depth == 0:
                return text[open_brace + 1:index]
    return ""


def _infer_query_parameters(controller_body: str) -> list[dict[str, Any]]:
    names = sorted(set(re.findall(r'queryValue\(\s*req\s*,\s*"([^"]+)"\s*\)', controller_body)))
    return [
        {
            "name": name,
            "in": "query",
            "required": False,
            "schema": _schema_for_field(name),
        }
        for name in names
    ]


def _infer_request_schema(text: str) -> dict[str, Any] | None:
    properties: dict[str, Any] = {}
    required: set[str] = set()

    for group in re.findall(r"missingStrings\(\s*body\s*,\s*\{([^}]+)\}\s*\)", text):
        for field in re.findall(r'"([^"]+)"', group):
            properties[field] = {"type": "string"}
            required.add(field)

    for field in re.findall(r'hasString\(\s*body\s*,\s*"([^"]+)"\s*\)', text):
        properties.setdefault(field, {"type": "string"})

    for field in re.findall(r'hasNumber\(\s*body\s*,\s*"([^"]+)"\s*\)', text):
        properties.setdefault(field, _schema_for_field(field))
        required.add(field)

    for field in re.findall(r'body\["([^"]+)"\]\.isBool\(\)', text):
        properties.setdefault(field, {"type": "boolean"})

    for field in re.findall(r'body\["([^"]+)"\]\.isArray\(\)', text):
        properties.setdefault(field, {"type": "array", "items": {}})

    for field in re.findall(r'body\["([^"]+)"\]\.isObject\(\)', text):
        properties.setdefault(field, {"type": "object", "additionalProperties": True})

    for field in re.findall(r'body\.get\(\s*"([^"]+)"', text):
        properties.setdefault(field, {"type": "string"})

    if not properties:
        return None

    schema: dict[str, Any] = {
        "type": "object",
        "additionalProperties": False,
        "properties": properties,
    }
    if required:
        schema["required"] = sorted(required)
    return schema


def _infer_response_schema(text: str) -> dict[str, Any] | None:
    properties: dict[str, Any] = {}
    if "success()" in text or 'body["ok"] = true' in text:
        properties["ok"] = {"type": "boolean"}
    for field in re.findall(r'response\["([^"]+)"\]', text):
        properties.setdefault(field, _schema_for_field(field))
    for field in re.findall(r'body\["([^"]+)"\]', text):
        properties.setdefault(field, _schema_for_field(field))
    if not properties:
        return None
    return {
        "type": "object",
        "additionalProperties": True,
        "properties": properties,
    }


def _schema_for_field(name: str) -> dict[str, Any]:
    if name == "status":
        return {"type": "string"}
    if name == "id" or name.endswith("_id") or name.endswith("_hours") or name.endswith("_minutes"):
        return {"type": "integer", "format": "int64"}
    if name in {"latitude", "longitude", "rating", "points", "total", "value"}:
        return {"type": "number"}
    if name.startswith("is_") or name in {"ok", "banned", "must_change_password"}:
        return {"type": "boolean"}
    if name.endswith("s") or name in {"roles", "permissions", "reports", "messages", "media"}:
        return {"type": "array", "items": {"type": "object", "additionalProperties": True}}
    if name.endswith("_geojson") or name in {"user", "report"}:
        return {"type": "object", "additionalProperties": True}
    return {"type": "string"}