Skip to content

Refactor duplicated code into shared utilities #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
devin-ai-integration wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Refactor duplicated code into shared utilities #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 42 additions & 61 deletions src/analyzer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,27 @@ def _build_refutation_prompt(self, finding: Dict[str, Any], spec_text: str,
{claim}
"""

def _result_from_payload(self, payload: Dict[str, Any],
raw_response: Optional[str] = None) -> AnalysisResult:
"""Build an :class:`AnalysisResult` from a parsed LLM JSON payload."""
return AnalysisResult(
status=payload.get("status", "UNCERTAIN"),
confidence=payload.get("confidence", 0),
issues=payload.get("issues", []),
summary=payload.get("summary", ""),
raw_response=raw_response,
)

@staticmethod
def _error_result(provider: str, error: Exception) -> AnalysisResult:
"""Build the ERROR result returned when a backend call fails."""
return AnalysisResult(
status="ERROR",
confidence=0,
issues=[],
summary=f"{provider} analysis failed: {str(error)}",
)

def _parse_json_response(self, response_text: str) -> Dict[str, Any]:
"""Parse JSON from LLM response, handling markdown code blocks
and truncated output from the model."""
Expand Down Expand Up @@ -251,21 +272,10 @@ def analyze_compliance(self, spec_text: str, code_text: str,

result = self._parse_json_response(response.text)

return AnalysisResult(
status=result.get("status", "UNCERTAIN"),
confidence=result.get("confidence", 0),
issues=result.get("issues", []),
summary=result.get("summary", ""),
raw_response=response.text
)
return self._result_from_payload(result, response.text)

except Exception as e:
return AnalysisResult(
status="ERROR",
confidence=0,
issues=[],
summary=f"Gemini analysis failed: {str(e)}"
)
return self._error_result("Gemini", e)

def get_model_info(self) -> Dict[str, Any]:
"""Get information about the current model"""
Expand Down Expand Up @@ -318,21 +328,10 @@ def analyze_compliance(self, spec_text: str, code_text: str,
response_text = response.choices[0].message.content
result = self._parse_json_response(response_text)

return AnalysisResult(
status=result.get("status", "UNCERTAIN"),
confidence=result.get("confidence", 0),
issues=result.get("issues", []),
summary=result.get("summary", ""),
raw_response=response_text
)
return self._result_from_payload(result, response_text)

except Exception as e:
return AnalysisResult(
status="ERROR",
confidence=0,
issues=[],
summary=f"OpenAI analysis failed: {str(e)}"
)
return self._error_result("OpenAI", e)

def get_model_info(self) -> Dict[str, Any]:
"""Get information about the current model"""
Expand Down Expand Up @@ -453,21 +452,10 @@ def analyze_compliance(self, spec_text: str, code_text: str,
)
result = self._parse_json_response(response_text)

return AnalysisResult(
status=result.get("status", "UNCERTAIN"),
confidence=result.get("confidence", 0),
issues=result.get("issues", []),
summary=result.get("summary", ""),
raw_response=response_text
)
return self._result_from_payload(result, response_text)

except Exception as e:
return AnalysisResult(
status="ERROR",
confidence=0,
issues=[],
summary=f"Azure AI analysis failed: {str(e)}"
)
return self._error_result("Azure AI", e)

def get_model_info(self) -> Dict[str, Any]:
"""Get information about the current model"""
Expand All @@ -479,32 +467,25 @@ def get_model_info(self) -> Dict[str, Any]:
}


# Analyzer class and its required constructor arguments, per provider.
_PROVIDERS = {
"gemini": (GeminiAnalyzer, ["api_key"]),
"openai": (OpenAIAnalyzer, ["api_key"]),
"azure": (AzureAIAnalyzer, ["api_key", "endpoint", "model"]),
}


def get_analyzer(provider: str = "gemini", **kwargs) -> BaseAnalyzer:
"""Factory: return a GeminiAnalyzer, OpenAIAnalyzer, or AzureAIAnalyzer."""
provider = provider.lower()

if provider == "gemini":
required = ["api_key"]
for key in required:
if key not in kwargs:
raise ValueError(f"Missing required parameter: {key}")
return GeminiAnalyzer(**kwargs)

elif provider == "openai":
required = ["api_key"]
for key in required:
if key not in kwargs:
raise ValueError(f"Missing required parameter: {key}")
return OpenAIAnalyzer(**kwargs)

elif provider == "azure":
required = ["api_key", "endpoint", "model"]
for key in required:
if key not in kwargs:
raise ValueError(f"Missing required parameter: {key}")
return AzureAIAnalyzer(**kwargs)

else:
if provider not in _PROVIDERS:
raise ValueError(
f"Unknown provider: {provider}. Use 'gemini', 'openai', or 'azure'."
)

analyzer_cls, required = _PROVIDERS[provider]
for key in required:
if key not in kwargs:
raise ValueError(f"Missing required parameter: {key}")
return analyzer_cls(**kwargs)
120 changes: 54 additions & 66 deletions src/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,35 @@ def cli():
pass


def _print_info_panel(title: str, rows):
"""Render the banner plus a two-column key/value configuration panel."""
console.print(BANNER)
info_table = Table(show_header=False, box=None, padding=(0, 2))
info_table.add_column(style="bold white")
info_table.add_column(style="cyan")
for label, value in rows:
info_table.add_row(label, value)
console.print(Panel(info_table, title=f"[bold]{title}[/bold]", border_style="blue"))


def _abort_on_error(error: Exception, verbose: bool):
"""Print a command failure (with optional traceback) and abort."""
if RICH_AVAILABLE:
console.print(f"[red]Error:[/red] {str(error)}")
else:
click.echo(f"Error: {str(error)}", err=True)

if verbose:
import traceback
trace = traceback.format_exc()
if RICH_AVAILABLE:
console.print(f"[dim]{trace}[/dim]")
else:
click.echo(trace, err=True)

raise click.Abort()


def _analyze_one_file(analyzer, spec_text, file_path, code_content, context):
"""Analyze a single file — designed to run inside a thread pool."""
result = analyzer.analyze_compliance(spec_text, code_content, context)
Expand Down Expand Up @@ -180,16 +209,13 @@ def analyze(eip: int, client: str, provider: Optional[str], output: str,

# Banner + config summary
if RICH_AVAILABLE:
console.print(BANNER)
info_table = Table(show_header=False, box=None, padding=(0, 2))
info_table.add_column(style="bold white")
info_table.add_column(style="cyan")
info_table.add_row("EIP", str(eip))
info_table.add_row("Client", client)
info_table.add_row("Provider", llm_provider)
info_table.add_row("Output", output)
info_table.add_row("Verify", f"on · {verify_rounds} rounds" if verify else "off")
console.print(Panel(info_table, title="[bold]Configuration[/bold]", border_style="blue"))
_print_info_panel("Configuration", [
("EIP", str(eip)),
("Client", client),
("Provider", llm_provider),
("Output", output),
("Verify", f"on · {verify_rounds} rounds" if verify else "off"),
])
else:
click.echo("\n PRSpec - Ethereum Specification Compliance Checker\n")
click.echo(f" EIP: {eip} | Client: {client} | Provider: {llm_provider}")
Expand Down Expand Up @@ -250,17 +276,7 @@ def on_file_done(fname):
click.echo(f"\nReport saved to: {report_path}")

except Exception as e:
if RICH_AVAILABLE:
console.print(f"[red]Error:[/red] {str(e)}")
if verbose:
import traceback
console.print(f"[dim]{traceback.format_exc()}[/dim]")
else:
click.echo(f"Error: {str(e)}", err=True)
if verbose:
import traceback
click.echo(traceback.format_exc(), err=True)
raise click.Abort()
_abort_on_error(e, verbose)


@cli.command()
Expand All @@ -287,7 +303,7 @@ def diff(eip: int, clients: Optional[str], provider: Optional[str], output: str,
prspec diff --eip 1559
prspec diff --eip 4844 --clients go-ethereum,nethermind,besu --output html
"""
from .differential import ClientAnalysis, DifferentialEngine
from .differential import analyze_clients

try:
cfg = Config(config)
Expand Down Expand Up @@ -315,47 +331,29 @@ def diff(eip: int, clients: Optional[str], provider: Optional[str], output: str,
f"Usable: {usable or 'none'}."
)

rows = [
("EIP", str(eip)),
("Clients", ", ".join(usable)),
("Provider", llm_provider),
("Output", output),
]
if RICH_AVAILABLE:
console.print(BANNER)
info_table = Table(show_header=False, box=None, padding=(0, 2))
info_table.add_column(style="bold white")
info_table.add_column(style="cyan")
info_table.add_row("EIP", str(eip))
info_table.add_row("Clients", ", ".join(usable))
info_table.add_row("Provider", llm_provider)
info_table.add_row("Output", output)
if skipped:
info_table.add_row("Skipped", ", ".join(skipped))
console.print(Panel(info_table, title="[bold]Differential[/bold]", border_style="blue"))
rows.append(("Skipped", ", ".join(skipped)))
_print_info_panel("Differential", rows)
else:
click.echo(f"\n PRSpec differential — EIP-{eip} across {', '.join(usable)}\n")

# Analyze each client through the standard pipeline.
per_client = {}
last_analyzer = None
for client in usable:
def on_client_start(client):
if RICH_AVAILABLE:
console.print(f"[dim]Analyzing {client}...[/dim]")
results, analyzer = _run_analysis(
eip, client, cfg, llm_provider,
verify=verify, verify_rounds=verify_rounds,
)
per_client[client] = ClientAnalysis(
client=client,
language=CodeFetcher.client_language(client),
results=results,
)
last_analyzer = analyzer

# Build the differential.
engine = DifferentialEngine(focus_areas=cfg.get_eip_focus_areas(eip))
eip_title = SpecFetcher.get_eip_title(eip)
differential = engine.build(per_client, eip, eip_title, confirmed_only=verify)

if llm_synthesis and last_analyzer is not None:
differential.llm_synthesis = engine.synthesize(
last_analyzer, differential, per_client
)
differential = analyze_clients(
eip, usable, cfg, provider=llm_provider,
use_llm_synthesis=llm_synthesis,
verify=verify, verify_rounds=verify_rounds,
on_client_start=on_client_start,
)

# Report.
report_gen = ReportGenerator(cfg.output_config.get("directory", "output"))
Expand All @@ -371,17 +369,7 @@ def diff(eip: int, clients: Optional[str], provider: Optional[str], output: str,
except click.ClickException:
raise
except Exception as e:
if RICH_AVAILABLE:
console.print(f"[red]Error:[/red] {str(e)}")
if verbose:
import traceback
console.print(f"[dim]{traceback.format_exc()}[/dim]")
else:
click.echo(f"Error: {str(e)}", err=True)
if verbose:
import traceback
click.echo(traceback.format_exc(), err=True)
raise click.Abort()
_abort_on_error(e, verbose)


@cli.command()
Expand Down
Loading
Loading