Upload 300 files

9438a57 verified 6 days ago

7.2 kB

"""API ve iç veri modelleri."""
from __future__ import annotations

import re

from typing import Any, Literal, Optional

from pydantic import BaseModel, Field, field_validator, model_validator

Language = Literal["python", "javascript", "typescript", "java"]
TestKind = Literal["unit", "interface", "ui"]


# judge: hata bildirimleri hakemsiz kabul edilir · prune: ayıklama yok · restart: sıfırdan yazma yok
# repair: taşıma onarımı yok · planfree: düzeltme turlarında analist planı da verilir
# best: en iyi sürüm yerine son doğrulanan aday teslim edilir
# mutfix: kaçan mutantlarla hedefli ek tur yapılmaz
Ablation = Literal["judge", "prune", "restart", "repair", "planfree", "best", "mutfix"]


_TS_RE = re.compile(
    r"^\s*(?:export\s+)?(?:declare\s+)?(?:interface|enum)\s+\w+"          # interface / enum bildirimi
    r"|^\s*(?:export\s+)?type\s+\w+(?:<[^>]*>)?\s*="                     # type Takma = ...
    r"|\b(?:implements|readonly|private|protected|public)\s+\w+"           # sınıf erişim belirteçleri
    r"|\)\s*:\s*(?:string|number|boolean|void|any|unknown|never|Promise<)"  # dönüş tipi
    r"|\b\w+\s*:\s*(?:string|number|boolean|any|unknown)(?:\[\])?\s*[,)=;]",  # parametre/alan tipi
    re.M)


def looks_typescript(code: str) -> bool:
    return bool(_TS_RE.search(code or ""))


class JobRequest(BaseModel):
    language: Language
    code: str = Field(..., min_length=1, max_length=200_000)
    test_kind: TestKind = "unit"
    # UI testleri için isteğe bağlı HTML (kod JS ise); kod HTML ise doğrudan kullanılır
    html: Optional[str] = None
    model: Optional[str] = None
    max_iterations: Optional[int] = Field(default=None, ge=1, le=10)
    coverage_target: Optional[float] = Field(default=None, ge=0, le=100)
    run_mutation: bool = True
    # Tur sınırında hâlâ başarısız olan testleri çıkarıp kalan paketi yeniden doğrula
    prune_failing: bool = True
    max_mutants: Optional[int] = Field(default=None, ge=0, le=200)
    # Mutasyon testinden sonra kaçan mutantları ajana gösterip onları yakalayan test ekletme turu sayısı
    mutation_rounds: int = Field(default=1, ge=0, le=3)
    temperature: float = Field(default=0.2, ge=0, le=1.5)
    title: Optional[str] = None
    # Ablasyon: kapatılacak mekanizmalar (deneylerde her birinin katkısını ölçmek için)
    ablate: list[Ablation] = []
    # Çok dosyalı girdi: test edilen dosyanın yanına konan yardımcı dosyalar (ad → içerik) ve özgün dosya adı.
    # Testler yalnızca `code` için yazılır; kapsama ve mutasyon yalnızca onda ölçülür.
    extra_files: dict[str, str] = Field(default_factory=dict)
    source_name: Optional[str] = Field(default=None, max_length=200)
    # Arayüz dili: olay başlıkları, durma nedeni ve analistin açıklamaları bu dilde üretilir
    lang: Literal["tr", "en"] = "tr"
    # Hesaplama cihazı (yalnızca Ollama modelleri): boş = Bağlantılar penceresindeki varsayılan
    device: Optional[Literal["auto", "gpu", "cpu"]] = None

    @model_validator(mode="after")
    def _typescript_in_js(self) -> "JobRequest":
        """.js uzantılı ama TypeScript sözdizimi içeren kod (interface, tip açıklamaları) TypeScript olarak koşulur;
        aksi hâlde Babel derleyemez ve hiç test çalışmaz."""
        if self.language == "javascript" and self.test_kind != "ui" and looks_typescript(self.code):
            self.language = "typescript"
        return self

    @field_validator("extra_files")
    @classmethod
    def _limit_extra(cls, v: dict[str, str]) -> dict[str, str]:
        if len(v) > 30 or sum(len(c) for c in v.values()) > 600_000:
            raise ValueError("en fazla 30 yardımcı dosya ve toplam 600 KB")
        return v


class Coverage(BaseModel):
    percent: Optional[float] = None
    covered_lines: list[int] = []
    missed_lines: list[int] = []
    total_lines: int = 0


class TestCaseResult(BaseModel):
    name: str
    status: Literal["passed", "failed", "error", "skipped"]
    message: str = ""


class RunResult(BaseModel):
    compiled: bool = True
    compile_output: str = ""
    passed: int = 0
    failed: int = 0
    errors: int = 0
    skipped: int = 0
    total: int = 0
    duration_s: float = 0.0
    output: str = ""
    cases: list[TestCaseResult] = []
    coverage: Coverage = Coverage()
    timed_out: bool = False

    @property
    def all_green(self) -> bool:
        return self.compiled and not self.timed_out and self.total > 0 and self.failed == 0 and self.errors == 0

    def failing_names(self) -> set[str]:
        return {c.name for c in self.cases if c.status in ("failed", "error")}


class Mutant(BaseModel):
    id: int
    line: int
    operator: str
    original: str
    mutated: str
    status: Literal["killed", "survived", "invalid", "timeout", "pending"] = "pending"
    note: str = ""
    full_line: str = Field(default="", exclude=True)


class MutationReport(BaseModel):
    total: int = 0
    killed: int = 0
    survived: int = 0
    invalid: int = 0
    timeout: int = 0
    score: Optional[float] = None  # killed / (killed + survived + timeout)
    mutants: list[Mutant] = []


class IterationRecord(BaseModel):
    iteration: int
    compiled: bool
    passed: int
    failed: int
    total: int
    coverage: Optional[float]
    action: str = ""


class Analysis(BaseModel):
    summary: str = ""
    explanation: str = ""
    units: list[dict[str, Any]] = []
    edge_cases: list[str] = []
    test_plan: list[str] = []
    risks: list[str] = []


class JobResult(BaseModel):
    analysis: Analysis = Analysis()
    test_code: str = ""
    test_filename: str = ""
    final_run: Optional[RunResult] = None
    mutation: Optional[MutationReport] = None
    # Hedefli mutasyon turundan önceki ölçüm (tur yapılmadıysa None) ve tur özeti
    mutation_before: Optional[MutationReport] = None
    mutation_round: dict[str, Any] = {}
    iterations: list[IterationRecord] = []
    suspected_bugs: list[dict[str, str]] = []
    pruned_tests: int = 0  # tur sonunda düzeltilemeyip çıkarılan test sayısı
    rejected_bug_reports: list[dict[str, str]] = []  # hakemin reddettiği "kodda hata var" bildirimleri
    llm_calls: int = 0
    tool_calls: int = 0
    duration_s: float = 0.0
    stop_reason: str = ""
    stop_reason_i18n: dict[str, str] = {}  # {"tr": ..., "en": ...}
    # Kod yorumunun sonradan üretilen çevirileri: {"en": {"analysis": {...}, "bug_reasons": [...], ...}}
    translations: dict[str, dict[str, Any]] = {}
    # Modelin çalıştığı donanım: {"model", "gpu_percent", "vram_gb", "size_gb", "tokens_per_s"}
    compute: dict[str, Any] = {}


class Event(BaseModel):
    seq: int
    ts: float
    type: str  # phase | thought | tool_call | tool_result | llm | status | result | error | done
    phase: Optional[str] = None
    title: str = ""
    data: dict[str, Any] = {}


class JobState(BaseModel):
    id: str
    status: Literal["queued", "running", "done", "failed", "cancelled"] = "queued"
    created_at: float
    request: JobRequest
    events: list[Event] = []
    result: Optional[JobResult] = None
    error: Optional[str] = None
    source_filename: str = ""

	"""API ve iç veri modelleri."""
	from __future__ import annotations

	import re

	from typing import Any, Literal, Optional

	from pydantic import BaseModel, Field, field_validator, model_validator

	Language = Literal["python", "javascript", "typescript", "java"]
	TestKind = Literal["unit", "interface", "ui"]


	# judge: hata bildirimleri hakemsiz kabul edilir · prune: ayıklama yok · restart: sıfırdan yazma yok
	# repair: taşıma onarımı yok · planfree: düzeltme turlarında analist planı da verilir
	# best: en iyi sürüm yerine son doğrulanan aday teslim edilir
	# mutfix: kaçan mutantlarla hedefli ek tur yapılmaz
	Ablation = Literal["judge", "prune", "restart", "repair", "planfree", "best", "mutfix"]


	_TS_RE = re.compile(
	r"^\s*(?:export\s+)?(?:declare\s+)?(?:interface\|enum)\s+\w+" # interface / enum bildirimi
	r"\|^\s(?:export\s+)?type\s+\w+(?:<[^>]>)?\s*=" # type Takma = ...
	r"\|\b(?:implements\|readonly\|private\|protected\|public)\s+\w+" # sınıf erişim belirteçleri
	r"\|\)\s:\s(?:string\|number\|boolean\|void\|any\|unknown\|never\|Promise<)" # dönüş tipi
	r"\|\b\w+\s:\s(?:string\|number\|boolean\|any\|unknown)(?:\[\])?\s*[,)=;]", # parametre/alan tipi
	re.M)


	def looks_typescript(code: str) -> bool:
	return bool(_TS_RE.search(code or ""))


	class JobRequest(BaseModel):
	language: Language
	code: str = Field(..., min_length=1, max_length=200_000)
	test_kind: TestKind = "unit"
	# UI testleri için isteğe bağlı HTML (kod JS ise); kod HTML ise doğrudan kullanılır
	html: Optional[str] = None
	model: Optional[str] = None
	max_iterations: Optional[int] = Field(default=None, ge=1, le=10)
	coverage_target: Optional[float] = Field(default=None, ge=0, le=100)
	run_mutation: bool = True
	# Tur sınırında hâlâ başarısız olan testleri çıkarıp kalan paketi yeniden doğrula
	prune_failing: bool = True
	max_mutants: Optional[int] = Field(default=None, ge=0, le=200)
	# Mutasyon testinden sonra kaçan mutantları ajana gösterip onları yakalayan test ekletme turu sayısı
	mutation_rounds: int = Field(default=1, ge=0, le=3)
	temperature: float = Field(default=0.2, ge=0, le=1.5)
	title: Optional[str] = None
	# Ablasyon: kapatılacak mekanizmalar (deneylerde her birinin katkısını ölçmek için)
	ablate: list[Ablation] = []
	# Çok dosyalı girdi: test edilen dosyanın yanına konan yardımcı dosyalar (ad → içerik) ve özgün dosya adı.
	# Testler yalnızca `code` için yazılır; kapsama ve mutasyon yalnızca onda ölçülür.
	extra_files: dict[str, str] = Field(default_factory=dict)
	source_name: Optional[str] = Field(default=None, max_length=200)
	# Arayüz dili: olay başlıkları, durma nedeni ve analistin açıklamaları bu dilde üretilir
	lang: Literal["tr", "en"] = "tr"
	# Hesaplama cihazı (yalnızca Ollama modelleri): boş = Bağlantılar penceresindeki varsayılan
	device: Optional[Literal["auto", "gpu", "cpu"]] = None

	@model_validator(mode="after")
	def _typescript_in_js(self) -> "JobRequest":
	""".js uzantılı ama TypeScript sözdizimi içeren kod (interface, tip açıklamaları) TypeScript olarak koşulur;
	aksi hâlde Babel derleyemez ve hiç test çalışmaz."""
	if self.language == "javascript" and self.test_kind != "ui" and looks_typescript(self.code):
	self.language = "typescript"
	return self

	@field_validator("extra_files")
	@classmethod
	def _limit_extra(cls, v: dict[str, str]) -> dict[str, str]:
	if len(v) > 30 or sum(len(c) for c in v.values()) > 600_000:
	raise ValueError("en fazla 30 yardımcı dosya ve toplam 600 KB")
	return v


	class Coverage(BaseModel):
	percent: Optional[float] = None
	covered_lines: list[int] = []
	missed_lines: list[int] = []
	total_lines: int = 0


	class TestCaseResult(BaseModel):
	name: str
	status: Literal["passed", "failed", "error", "skipped"]
	message: str = ""


	class RunResult(BaseModel):
	compiled: bool = True
	compile_output: str = ""
	passed: int = 0
	failed: int = 0
	errors: int = 0
	skipped: int = 0
	total: int = 0
	duration_s: float = 0.0
	output: str = ""
	cases: list[TestCaseResult] = []
	coverage: Coverage = Coverage()
	timed_out: bool = False

	@property
	def all_green(self) -> bool:
	return self.compiled and not self.timed_out and self.total > 0 and self.failed == 0 and self.errors == 0

	def failing_names(self) -> set[str]:
	return {c.name for c in self.cases if c.status in ("failed", "error")}


	class Mutant(BaseModel):
	id: int
	line: int
	operator: str
	original: str
	mutated: str
	status: Literal["killed", "survived", "invalid", "timeout", "pending"] = "pending"
	note: str = ""
	full_line: str = Field(default="", exclude=True)


	class MutationReport(BaseModel):
	total: int = 0
	killed: int = 0
	survived: int = 0
	invalid: int = 0
	timeout: int = 0
	score: Optional[float] = None # killed / (killed + survived + timeout)
	mutants: list[Mutant] = []


	class IterationRecord(BaseModel):
	iteration: int
	compiled: bool
	passed: int
	failed: int
	total: int
	coverage: Optional[float]
	action: str = ""


	class Analysis(BaseModel):
	summary: str = ""
	explanation: str = ""
	units: list[dict[str, Any]] = []
	edge_cases: list[str] = []
	test_plan: list[str] = []
	risks: list[str] = []


	class JobResult(BaseModel):
	analysis: Analysis = Analysis()
	test_code: str = ""
	test_filename: str = ""
	final_run: Optional[RunResult] = None
	mutation: Optional[MutationReport] = None
	# Hedefli mutasyon turundan önceki ölçüm (tur yapılmadıysa None) ve tur özeti
	mutation_before: Optional[MutationReport] = None
	mutation_round: dict[str, Any] = {}
	iterations: list[IterationRecord] = []
	suspected_bugs: list[dict[str, str]] = []
	pruned_tests: int = 0 # tur sonunda düzeltilemeyip çıkarılan test sayısı
	rejected_bug_reports: list[dict[str, str]] = [] # hakemin reddettiği "kodda hata var" bildirimleri
	llm_calls: int = 0
	tool_calls: int = 0
	duration_s: float = 0.0
	stop_reason: str = ""
	stop_reason_i18n: dict[str, str] = {} # {"tr": ..., "en": ...}
	# Kod yorumunun sonradan üretilen çevirileri: {"en": {"analysis": {...}, "bug_reasons": [...], ...}}
	translations: dict[str, dict[str, Any]] = {}
	# Modelin çalıştığı donanım: {"model", "gpu_percent", "vram_gb", "size_gb", "tokens_per_s"}
	compute: dict[str, Any] = {}


	class Event(BaseModel):
	seq: int
	ts: float
	type: str # phase \| thought \| tool_call \| tool_result \| llm \| status \| result \| error \| done
	phase: Optional[str] = None
	title: str = ""
	data: dict[str, Any] = {}


	class JobState(BaseModel):
	id: str
	status: Literal["queued", "running", "done", "failed", "cancelled"] = "queued"
	created_at: float
	request: JobRequest
	events: list[Event] = []
	result: Optional[JobResult] = None
	error: Optional[str] = None
	source_filename: str = ""