3 Commits
0.4.1 ... 0.5.0

Author SHA1 Message Date
a357a307a7 release: version 0.5.0 🚀
All checks were successful
Upload Python Package / Create Release (push) Successful in 22s
Upload Python Package / deploy (push) Successful in 49s
2026-04-10 20:27:26 +02:00
af4247e657 feat(dashboard): expose repository urls refs NOISSUE 2026-04-10 20:27:08 +02:00
227ad1ad6f feat(factory): serve dashboard at root and create project repos refs NOISSUE 2026-04-10 20:23:07 +02:00
14 changed files with 298 additions and 156 deletions

1
.gitignore vendored
View File

@@ -1 +1,2 @@
sqlite.db sqlite.db
.nicegui/

View File

@@ -4,6 +4,14 @@ Changelog
(unreleased) (unreleased)
------------ ------------
- Feat(dashboard): expose repository urls refs NOISSUE. [Simon
Diesenreiter]
- Feat(factory): serve dashboard at root and create project repos refs
NOISSUE. [Simon Diesenreiter]
0.4.1 (2026-04-10)
------------------
- Fix(ci): pin docker api version for release builds refs NOISSUE. - Fix(ci): pin docker api version for release builds refs NOISSUE.
[Simon Diesenreiter] [Simon Diesenreiter]

View File

@@ -6,7 +6,7 @@ Automated software generation service powered by Ollama LLM. This service allows
- **Telegram Integration**: Receive software requests via Telegram bot - **Telegram Integration**: Receive software requests via Telegram bot
- **Ollama LLM**: Uses Ollama-hosted models for code generation - **Ollama LLM**: Uses Ollama-hosted models for code generation
- **Git Integration**: Automatically commits code to gitea - **Git Integration**: Creates a dedicated Gitea repository per generated project inside your organization
- **Pull Requests**: Creates PRs for user review before merging - **Pull Requests**: Creates PRs for user review before merging
- **Web UI**: Beautiful dashboard for monitoring project progress - **Web UI**: Beautiful dashboard for monitoring project progress
- **n8n Workflows**: Bridges Telegram with LLMs via n8n webhooks - **n8n Workflows**: Bridges Telegram with LLMs via n8n webhooks
@@ -49,9 +49,10 @@ OLLAMA_MODEL=llama3
# Gitea # Gitea
GITEA_URL=https://gitea.yourserver.com GITEA_URL=https://gitea.yourserver.com
GITEA_TOKEN= analyze your_gitea_api_token GITEA_TOKEN=your_gitea_api_token
GITEA_OWNER=ai-software-factory GITEA_OWNER=ai-software-factory
GITEA_REPO=ai-software-factory # Optional legacy fixed-repository mode. Leave empty to create one repo per project.
GITEA_REPO=
# n8n # n8n
N8N_WEBHOOK_URL=http://n8n.yourserver.com/webhook/telegram N8N_WEBHOOK_URL=http://n8n.yourserver.com/webhook/telegram
@@ -83,7 +84,7 @@ docker-compose up -d
2. **Monitor progress via Web UI:** 2. **Monitor progress via Web UI:**
Open `http://yourserver:8000` to see real-time progress Open `http://yourserver:8000/` to see the dashboard and `http://yourserver:8000/api` for API metadata
3. **Review PRs in Gitea:** 3. **Review PRs in Gitea:**
@@ -93,7 +94,8 @@ docker-compose up -d
| Endpoint | Method | Description | | Endpoint | Method | Description |
|------|------|-------| |------|------|-------|
| `/` | GET | API information | | `/` | GET | Dashboard |
| `/api` | GET | API information |
| `/health` | GET | Health check | | `/health` | GET | Health check |
| `/generate` | POST | Generate new software | | `/generate` | POST | Generate new software |
| `/status/{project_id}` | GET | Get project status | | `/status/{project_id}` | GET | Get project status |

View File

@@ -15,7 +15,7 @@ OLLAMA_MODEL=llama3
GITEA_URL=https://gitea.yourserver.com GITEA_URL=https://gitea.yourserver.com
GITEA_TOKEN=your_gitea_api_token GITEA_TOKEN=your_gitea_api_token
GITEA_OWNER=your_organization_name GITEA_OWNER=your_organization_name
GITEA_REPO= (optional - leave empty for any repo, or specify a default) GITEA_REPO= (optional legacy fixed repository mode; leave empty to create one repo per project)
# n8n # n8n
# n8n webhook for Telegram integration # n8n webhook for Telegram integration

View File

@@ -1 +1 @@
0.4.1 0.5.0

View File

@@ -1,11 +1,11 @@
"""AI Software Factory agents.""" """AI Software Factory agents."""
from agents.orchestrator import AgentOrchestrator from .orchestrator import AgentOrchestrator
from agents.git_manager import GitManager from .git_manager import GitManager
from agents.ui_manager import UIManager from .ui_manager import UIManager
from agents.telegram import TelegramHandler from .telegram import TelegramHandler
from agents.gitea import GiteaAPI from .gitea import GiteaAPI
from agents.database_manager import DatabaseManager from .database_manager import DatabaseManager
__all__ = [ __all__ = [
"AgentOrchestrator", "AgentOrchestrator",

View File

@@ -4,6 +4,7 @@ from sqlalchemy.orm import Session
from sqlalchemy import text from sqlalchemy import text
try: try:
from ..config import settings
from ..models import ( from ..models import (
AuditTrail, AuditTrail,
ProjectHistory, ProjectHistory,
@@ -17,6 +18,7 @@ try:
UserAction, UserAction,
) )
except ImportError: except ImportError:
from config import settings
from models import ( from models import (
AuditTrail, AuditTrail,
ProjectHistory, ProjectHistory,
@@ -307,6 +309,31 @@ class DatabaseManager:
self.db.refresh(pr) self.db.refresh(pr)
return pr return pr
def _get_latest_ui_snapshot_data(self, history_id: int) -> dict:
"""Return the latest stored UI snapshot payload for a project."""
snapshot = self.db.query(UISnapshot).filter(
UISnapshot.history_id == history_id
).order_by(UISnapshot.created_at.desc(), UISnapshot.id.desc()).first()
if not snapshot:
return {}
return self._normalize_metadata(snapshot.snapshot_data)
def _get_project_repository(self, history: ProjectHistory) -> dict | None:
"""Resolve repository metadata for a project."""
snapshot_data = self._get_latest_ui_snapshot_data(history.id)
repository = snapshot_data.get("repository")
if isinstance(repository, dict) and any(repository.values()):
return repository
if settings.gitea_owner and settings.gitea_repo and settings.gitea_url:
return {
"owner": settings.gitea_owner,
"name": settings.gitea_repo,
"url": f"{settings.gitea_url.rstrip('/')}/{settings.gitea_owner}/{settings.gitea_repo}",
"mode": "shared",
}
return None
def get_project_by_id(self, project_id: str) -> ProjectHistory | None: def get_project_by_id(self, project_id: str) -> ProjectHistory | None:
"""Get project by ID.""" """Get project by ID."""
return self.db.query(ProjectHistory).filter(ProjectHistory.project_id == project_id).first() return self.db.query(ProjectHistory).filter(ProjectHistory.project_id == project_id).first()
@@ -708,6 +735,7 @@ class DatabaseManager:
prompts = self.get_prompt_events(project_id=project_id) prompts = self.get_prompt_events(project_id=project_id)
code_changes = self.get_code_changes(project_id=project_id) code_changes = self.get_code_changes(project_id=project_id)
correlations = self.get_prompt_change_correlations(project_id=project_id) correlations = self.get_prompt_change_correlations(project_id=project_id)
repository = self._get_project_repository(history)
return { return {
"project": { "project": {
@@ -720,6 +748,7 @@ class DatabaseManager:
"message": history.message, "message": history.message,
"error_message": history.error_message, "error_message": history.error_message,
"current_step": history.current_step, "current_step": history.current_step,
"repository": repository,
"completed_at": history.completed_at.isoformat() if history.completed_at else None, "completed_at": history.completed_at.isoformat() if history.completed_at else None,
"created_at": history.started_at.isoformat() if history.started_at else None "created_at": history.started_at.isoformat() if history.started_at else None
}, },
@@ -759,6 +788,7 @@ class DatabaseManager:
"prompts": prompts, "prompts": prompts,
"code_changes": code_changes, "code_changes": code_changes,
"prompt_change_correlations": correlations, "prompt_change_correlations": correlations,
"repository": repository,
} }
def get_prompt_events(self, project_id: str | None = None, limit: int = 100) -> list[dict]: def get_prompt_events(self, project_id: str | None = None, limit: int = 100) -> list[dict]:
@@ -840,6 +870,13 @@ class DatabaseManager:
} }
def cleanup_audit_trail(self) -> None: def cleanup_audit_trail(self) -> None:
"""Clear all audit trail entries.""" """Clear audit-related test data across all related tables."""
self.db.query(PromptCodeLink).delete()
self.db.query(PullRequest).delete()
self.db.query(UISnapshot).delete()
self.db.query(UserAction).delete()
self.db.query(ProjectLog).delete()
self.db.query(AuditTrail).delete() self.db.query(AuditTrail).delete()
self.db.query(SystemLog).delete()
self.db.query(ProjectHistory).delete()
self.db.commit() self.db.commit()

View File

@@ -1,7 +1,6 @@
"""Gitea API integration for commits and PRs.""" """Gitea API integration for repository and pull request operations."""
import os import os
from typing import Optional
class GiteaAPI: class GiteaAPI:
@@ -14,7 +13,7 @@ class GiteaAPI:
self.repo = repo self.repo = repo
self.headers = { self.headers = {
"Authorization": f"token {token}", "Authorization": f"token {token}",
"Content-Type": "application/json" "Content-Type": "application/json",
} }
def get_config(self) -> dict: def get_config(self) -> dict:
@@ -23,60 +22,91 @@ class GiteaAPI:
token = os.getenv("GITEA_TOKEN", "") token = os.getenv("GITEA_TOKEN", "")
owner = os.getenv("GITEA_OWNER", "ai-test") owner = os.getenv("GITEA_OWNER", "ai-test")
repo = os.getenv("GITEA_REPO", "") repo = os.getenv("GITEA_REPO", "")
# Allow empty repo for any repo mode (org/repo pattern)
if not repo:
repo = "any-repo" # Use this as a placeholder for org/repo operations
# Check for repo suffix pattern (e.g., repo-* for multiple repos)
repo_suffix = os.getenv("GITEA_REPO_SUFFIX", "")
return { return {
"base_url": base_url.rstrip("/"), "base_url": base_url.rstrip("/"),
"token": token, "token": token,
"owner": owner, "owner": owner,
"repo": repo, "repo": repo,
"repo_suffix": repo_suffix, "supports_project_repos": not bool(repo),
"supports_any_repo": not repo or repo_suffix
} }
def get_auth_headers(self) -> dict: def get_auth_headers(self) -> dict:
"""Get authentication headers.""" """Get authentication headers."""
return { return {
"Authorization": f"token {self.token}", "Authorization": f"token {self.token}",
"Content-Type": "application/json" "Content-Type": "application/json",
} }
async def create_branch(self, branch: str, base: str = "main", owner: str | None = None, repo: str | None = None): def _api_url(self, path: str) -> str:
"""Create a new branch. """Build a Gitea API URL from a relative path."""
return f"{self.base_url}/api/v1/{path.lstrip('/')}"
Args:
branch: Branch name to create
base: Base branch to create from (default: "main")
owner: Organization/owner name (optional, falls back to configured owner)
repo: Repository name (optional, falls back to configured repo)
Returns:
API response or error message
"""
# Use provided owner/repo or fall back to configured values
_owner = owner or self.owner
_repo = repo or self.repo
url = f"{self.base_url}/repos/{_owner}/{_repo}/branches/{branch}"
payload = {"base": base}
async def _request(self, method: str, path: str, payload: dict | None = None) -> dict:
"""Perform a Gitea API request and normalize the response."""
try: try:
import aiohttp import aiohttp
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
async with session.post(url, headers=self.get_auth_headers(), json=payload) as resp: async with session.request(
if resp.status == 201: method,
self._api_url(path),
headers=self.get_auth_headers(),
json=payload,
) as resp:
if resp.status in (200, 201):
return await resp.json() return await resp.json()
else: return {"error": await resp.text(), "status_code": resp.status}
return {"error": await resp.text()}
except Exception as e: except Exception as e:
return {"error": str(e)} return {"error": str(e)}
def build_project_repo_name(self, project_id: str, project_name: str | None = None) -> str:
"""Build a repository name for a generated project."""
preferred = (project_name or project_id or "project").strip().lower().replace(" ", "-")
sanitized = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "-" for ch in preferred)
while "--" in sanitized:
sanitized = sanitized.replace("--", "-")
return sanitized.strip("-") or project_id
async def create_repo(
self,
repo_name: str,
owner: str | None = None,
description: str | None = None,
private: bool = False,
auto_init: bool = True,
) -> dict:
"""Create a repository inside the configured organization."""
_owner = owner or self.owner
if not _owner:
return {"error": "Owner or organization is required"}
payload = {
"name": repo_name,
"description": description or f"AI-generated project repository for {repo_name}",
"private": private,
"auto_init": auto_init,
"default_branch": "main",
}
result = await self._request("POST", f"orgs/{_owner}/repos", payload)
if result.get("status_code") == 409:
existing = await self.get_repo_info(owner=_owner, repo=repo_name)
if not existing.get("error"):
existing["status"] = "exists"
return existing
if not result.get("error"):
result.setdefault("status", "created")
return result
async def create_branch(self, branch: str, base: str = "main", owner: str | None = None, repo: str | None = None):
"""Create a new branch."""
_owner = owner or self.owner
_repo = repo or self.repo
return await self._request(
"POST",
f"repos/{_owner}/{_repo}/branches",
{"new_branch_name": branch, "old_ref_name": base},
)
async def create_pull_request( async def create_pull_request(
self, self,
title: str, title: str,
@@ -84,43 +114,18 @@ class GiteaAPI:
owner: str, owner: str,
repo: str, repo: str,
base: str = "main", base: str = "main",
head: str | None = None head: str | None = None,
) -> dict: ) -> dict:
"""Create a pull request. """Create a pull request."""
Args:
title: PR title
body: PR description
owner: Organization/owner name
repo: Repository name
base: Base branch (default: "main")
head: Head branch (optional, auto-generated if not provided)
Returns:
API response or error message
"""
_owner = owner or self.owner _owner = owner or self.owner
_repo = repo or self.repo _repo = repo or self.repo
url = f"{self.base_url}/repos/{_owner}/{_repo}/pulls"
payload = { payload = {
"title": title, "title": title,
"body": body, "body": body,
"base": {"branch": base}, "base": base,
"head": head or f"{_owner}-{_repo}-ai-gen-{hash(title) % 10000}" "head": head or f"{_owner}-{_repo}-ai-gen-{hash(title) % 10000}",
} }
return await self._request("POST", f"repos/{_owner}/{_repo}/pulls", payload)
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=self.get_auth_headers(), json=payload) as resp:
if resp.status == 201:
return await resp.json()
else:
return {"error": await resp.text()}
except Exception as e:
return {"error": str(e)}
async def push_commit( async def push_commit(
self, self,
@@ -128,25 +133,13 @@ class GiteaAPI:
files: list[dict], files: list[dict],
message: str, message: str,
owner: str | None = None, owner: str | None = None,
repo: str | None = None repo: str | None = None,
) -> dict: ) -> dict:
""" """Push files to a branch.
Push files to a branch.
In production, this would use gitea's API or git push. In production, this would use gitea's API or git push.
For now, we'll simulate the operation. For now, this remains simulated.
Args:
branch: Branch name
files: List of files to push
message: Commit message
owner: Organization/owner name (optional, falls back to configured owner)
repo: Repository name (optional, falls back to configured repo)
Returns:
Status response
""" """
# Use provided owner/repo or fall back to configured values
_owner = owner or self.owner _owner = owner or self.owner
_repo = repo or self.repo _repo = repo or self.repo
@@ -156,35 +149,15 @@ class GiteaAPI:
"message": message, "message": message,
"files": files, "files": files,
"owner": _owner, "owner": _owner,
"repo": _repo "repo": _repo,
} }
async def get_repo_info(self, owner: str | None = None, repo: str | None = None) -> dict: async def get_repo_info(self, owner: str | None = None, repo: str | None = None) -> dict:
"""Get repository information. """Get repository information."""
Args:
owner: Organization/owner name (optional, falls back to configured owner)
repo: Repository name (optional, falls back to configured repo)
Returns:
Repository info or error message
"""
# Use provided owner/repo or fall back to configured values
_owner = owner or self.owner _owner = owner or self.owner
_repo = repo or self.repo _repo = repo or self.repo
if not _repo: if not _repo:
return {"error": "Repository name required for org operations"} return {"error": "Repository name required for org operations"}
url = f"{self.base_url}/repos/{_owner}/{_repo}" return await self._request("GET", f"repos/{_owner}/{_repo}")
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=self.get_auth_headers()) as resp:
if resp.status == 200:
return await resp.json()
else:
return {"error": await resp.text()}
except Exception as e:
return {"error": str(e)}

View File

@@ -3,7 +3,6 @@
from __future__ import annotations from __future__ import annotations
import py_compile import py_compile
from pathlib import Path
from typing import Optional from typing import Optional
from datetime import datetime from datetime import datetime
@@ -51,18 +50,21 @@ class AgentOrchestrator:
self.prompt_text = prompt_text self.prompt_text = prompt_text
self.prompt_actor = prompt_actor self.prompt_actor = prompt_actor
self.changed_files: list[str] = [] self.changed_files: list[str] = []
self.project_root = settings.projects_root / project_id
self.prompt_audit = None
# Initialize agents
self.git_manager = GitManager(project_id)
self.ui_manager = UIManager(project_id)
self.gitea_api = GiteaAPI( self.gitea_api = GiteaAPI(
token=settings.GITEA_TOKEN, token=settings.GITEA_TOKEN,
base_url=settings.GITEA_URL, base_url=settings.GITEA_URL,
owner=settings.GITEA_OWNER, owner=settings.GITEA_OWNER,
repo=settings.GITEA_REPO or "" repo=settings.GITEA_REPO or ""
) )
self.project_root = settings.projects_root / project_id
self.prompt_audit = None
self.repo_name = settings.gitea_repo or self.gitea_api.build_project_repo_name(project_id, project_name)
self.repo_owner = settings.gitea_owner
self.repo_url = self._build_repo_url(self.repo_owner, self.repo_name)
# Initialize agents
self.git_manager = GitManager(project_id)
self.ui_manager = UIManager(project_id)
# Initialize database manager if db session provided # Initialize database manager if db session provided
self.db_manager = None self.db_manager = None
@@ -90,6 +92,53 @@ class AgentOrchestrator:
self.ui_manager.ui_data["project_root"] = str(self.project_root) self.ui_manager.ui_data["project_root"] = str(self.project_root)
self.ui_manager.ui_data["features"] = list(self.features) self.ui_manager.ui_data["features"] = list(self.features)
self.ui_manager.ui_data["tech_stack"] = list(self.tech_stack) self.ui_manager.ui_data["tech_stack"] = list(self.tech_stack)
self.ui_manager.ui_data["repository"] = {
"owner": self.repo_owner,
"name": self.repo_name,
"url": self.repo_url,
"mode": "project" if settings.use_project_repositories else "shared",
}
def _build_repo_url(self, owner: str | None, repo: str | None) -> str | None:
if not owner or not repo or not settings.gitea_url:
return None
return f"{settings.gitea_url.rstrip('/')}/{owner}/{repo}"
async def _ensure_remote_repository(self) -> None:
if not settings.use_project_repositories:
return
if not self.repo_owner or not settings.gitea_token or not settings.gitea_url:
return
repo_name = self.repo_name
result = await self.gitea_api.create_repo(
repo_name=repo_name,
owner=self.repo_owner,
description=f"AI-generated project for {self.project_name}",
)
if result.get("status") == "exists" and repo_name == self.gitea_api.build_project_repo_name(self.project_id, self.project_name):
repo_name = f"{repo_name}-{self.project_id.split('-')[-1]}"
result = await self.gitea_api.create_repo(
repo_name=repo_name,
owner=self.repo_owner,
description=f"AI-generated project for {self.project_name}",
)
self.repo_name = repo_name
self.ui_manager.ui_data["repository"]["name"] = repo_name
if self.db_manager:
self.db_manager.log_system_event(
component="gitea",
level="ERROR" if result.get("error") else "INFO",
message=(
f"Repository setup failed for {self.repo_owner}/{self.repo_name}: {result.get('error')}"
if result.get("error")
else f"Prepared repository {self.repo_owner}/{self.repo_name}"
),
)
self.ui_manager.ui_data["repository"]["status"] = result.get("status", "error" if result.get("error") else "ready")
if result.get("html_url"):
self.repo_url = result["html_url"]
self.ui_manager.ui_data["repository"]["url"] = self.repo_url
def _append_log(self, message: str) -> None: def _append_log(self, message: str) -> None:
timestamped = f"[{datetime.utcnow().isoformat()}] {message}" timestamped = f"[{datetime.utcnow().isoformat()}] {message}"
@@ -165,6 +214,8 @@ class AgentOrchestrator:
self._update_progress(5, "initializing", "Setting up project structure...") self._update_progress(5, "initializing", "Setting up project structure...")
self._append_log("Initializing project.") self._append_log("Initializing project.")
await self._ensure_remote_repository()
# Step 2: Create project structure (skip git operations) # Step 2: Create project structure (skip git operations)
self._update_progress(20, "project-structure", "Creating project files...") self._update_progress(20, "project-structure", "Creating project files...")
await self._create_project_structure() await self._create_project_structure()
@@ -201,6 +252,7 @@ class AgentOrchestrator:
"history_id": self.history.id if self.history else None, "history_id": self.history.id if self.history else None,
"project_root": str(self.project_root), "project_root": str(self.project_root),
"changed_files": list(dict.fromkeys(self.changed_files)), "changed_files": list(dict.fromkeys(self.changed_files)),
"repository": self.ui_manager.ui_data.get("repository"),
} }
except Exception as e: except Exception as e:
@@ -226,6 +278,7 @@ class AgentOrchestrator:
"history_id": self.history.id if self.history else None, "history_id": self.history.id if self.history else None,
"project_root": str(self.project_root), "project_root": str(self.project_root),
"changed_files": list(dict.fromkeys(self.changed_files)), "changed_files": list(dict.fromkeys(self.changed_files)),
"repository": self.ui_manager.ui_data.get("repository"),
} }
async def _create_project_structure(self) -> None: async def _create_project_structure(self) -> None:

View File

@@ -1,6 +1,7 @@
[alembic] [alembic]
script_location = alembic script_location = alembic
prepend_sys_path = . prepend_sys_path = .
path_separator = os
sqlalchemy.url = sqlite:////tmp/ai_software_factory_test.db sqlalchemy.url = sqlite:////tmp/ai_software_factory_test.db
[loggers] [loggers]

View File

@@ -4,12 +4,18 @@ import os
from typing import Optional from typing import Optional
from pathlib import Path from pathlib import Path
from pydantic import Field from pydantic import Field
from pydantic_settings import BaseSettings from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings): class Settings(BaseSettings):
"""Application settings loaded from environment variables.""" """Application settings loaded from environment variables."""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
# Server settings # Server settings
HOST: str = "0.0.0.0" HOST: str = "0.0.0.0"
PORT: int = 8000 PORT: int = 8000
@@ -23,7 +29,7 @@ class Settings(BaseSettings):
GITEA_URL: str = "https://gitea.yourserver.com" GITEA_URL: str = "https://gitea.yourserver.com"
GITEA_TOKEN: str = "" GITEA_TOKEN: str = ""
GITEA_OWNER: str = "ai-software-factory" GITEA_OWNER: str = "ai-software-factory"
GITEA_REPO: str = "ai-software-factory" GITEA_REPO: str = ""
# n8n settings # n8n settings
N8N_WEBHOOK_URL: str = "" N8N_WEBHOOK_URL: str = ""
@@ -105,6 +111,21 @@ class Settings(BaseSettings):
"""Get Gitea token with trimmed whitespace.""" """Get Gitea token with trimmed whitespace."""
return self.GITEA_TOKEN.strip() return self.GITEA_TOKEN.strip()
@property
def gitea_owner(self) -> str:
"""Get Gitea owner/organization with trimmed whitespace."""
return self.GITEA_OWNER.strip()
@property
def gitea_repo(self) -> str:
"""Get the optional fixed Gitea repository name with trimmed whitespace."""
return self.GITEA_REPO.strip()
@property
def use_project_repositories(self) -> bool:
"""Whether the service should create one repository per generated project."""
return not bool(self.gitea_repo)
@property @property
def n8n_webhook_url(self) -> str: def n8n_webhook_url(self) -> str:
"""Get n8n webhook URL with trimmed whitespace.""" """Get n8n webhook URL with trimmed whitespace."""
@@ -177,11 +198,5 @@ class Settings(BaseSettings):
"""Get test PostgreSQL database name.""" """Get test PostgreSQL database name."""
return self.POSTGRES_TEST_DB.strip() return self.POSTGRES_TEST_DB.strip()
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
extra = "ignore"
# Create instance for module-level access # Create instance for module-level access
settings = Settings() settings = Settings()

View File

@@ -27,6 +27,30 @@ def _resolve_n8n_api_url() -> str:
return '' return ''
def _render_repository_block(repository: dict | None) -> None:
"""Render repository details and URL when available."""
if not repository:
ui.label('Repository URL not available yet.').classes('factory-muted')
return
owner = repository.get('owner') or 'unknown-owner'
name = repository.get('name') or 'unknown-repo'
mode = repository.get('mode') or 'project'
status = repository.get('status')
repo_url = repository.get('url')
with ui.column().classes('gap-1'):
with ui.row().classes('items-center gap-2'):
ui.label(f'{owner}/{name}').style('font-weight: 700; color: #2f241d;')
ui.label(mode).classes('factory-chip')
if status:
ui.label(status).classes('factory-chip')
if repo_url:
ui.link(repo_url, repo_url, new_tab=True).classes('factory-code')
else:
ui.label('Repository URL not available yet.').classes('factory-muted')
def _load_dashboard_snapshot() -> dict: def _load_dashboard_snapshot() -> dict:
"""Load dashboard data from the database.""" """Load dashboard data from the database."""
db = get_db_sync() db = get_db_sync()
@@ -101,6 +125,14 @@ def create_dashboard():
projects = snapshot['projects'] projects = snapshot['projects']
correlations = snapshot['correlations'] correlations = snapshot['correlations']
system_logs = snapshot['system_logs'] system_logs = snapshot['system_logs']
project_repository_map = {
project_bundle['project']['project_id']: {
'project_name': project_bundle['project']['project_name'],
'repository': project_bundle.get('repository') or project_bundle['project'].get('repository'),
}
for project_bundle in projects
if project_bundle.get('project')
}
with ui.column().classes('factory-shell w-full gap-4 q-pa-lg'): with ui.column().classes('factory-shell w-full gap-4 q-pa-lg'):
with ui.card().classes('factory-panel w-full q-pa-lg'): with ui.card().classes('factory-panel w-full q-pa-lg'):
@@ -171,6 +203,10 @@ def create_dashboard():
project = project_bundle['project'] project = project_bundle['project']
with ui.expansion(f"{project['project_name']} · {project['status']}", icon='folder').classes('factory-panel w-full q-mb-md'): with ui.expansion(f"{project['project_name']} · {project['status']}", icon='folder').classes('factory-panel w-full q-mb-md'):
with ui.grid(columns=2).classes('w-full gap-4 q-pa-md'): with ui.grid(columns=2).classes('w-full gap-4 q-pa-md'):
with ui.card().classes('q-pa-md'):
ui.label('Repository').style('font-weight: 700; color: #3a281a;')
_render_repository_block(project_bundle.get('repository') or project.get('repository'))
with ui.card().classes('q-pa-md'): with ui.card().classes('q-pa-md'):
ui.label('Prompt').style('font-weight: 700; color: #3a281a;') ui.label('Prompt').style('font-weight: 700; color: #3a281a;')
prompts = project_bundle.get('prompts', []) prompts = project_bundle.get('prompts', [])
@@ -178,7 +214,7 @@ def create_dashboard():
prompt = prompts[0] prompt = prompts[0]
ui.markdown(f"**Requested features:** {', '.join(prompt['features']) or 'None'}") ui.markdown(f"**Requested features:** {', '.join(prompt['features']) or 'None'}")
ui.markdown(f"**Tech stack:** {', '.join(prompt['tech_stack']) or 'None'}") ui.markdown(f"**Tech stack:** {', '.join(prompt['tech_stack']) or 'None'}")
ui.element('div').classes('factory-code').set_text(prompt['prompt_text']) ui.label(prompt['prompt_text']).classes('factory-code')
else: else:
ui.label('No prompt recorded.').classes('factory-muted') ui.label('No prompt recorded.').classes('factory-muted')
@@ -219,9 +255,11 @@ def create_dashboard():
ui.label('Each prompt entry is linked to the generated files recorded after that prompt for the same project.').classes('factory-muted') ui.label('Each prompt entry is linked to the generated files recorded after that prompt for the same project.').classes('factory-muted')
if correlations: if correlations:
for correlation in correlations: for correlation in correlations:
correlation_project = project_repository_map.get(correlation['project_id'], {})
with ui.card().classes('q-pa-md q-mt-md'): with ui.card().classes('q-pa-md q-mt-md'):
ui.label(correlation['project_id']).style('font-size: 1rem; font-weight: 700; color: #2f241d;') ui.label(correlation_project.get('project_name') or correlation['project_id']).style('font-size: 1rem; font-weight: 700; color: #2f241d;')
ui.element('div').classes('factory-code q-mt-sm').set_text(correlation['prompt_text']) _render_repository_block(correlation_project.get('repository'))
ui.label(correlation['prompt_text']).classes('factory-code q-mt-sm')
if correlation['changes']: if correlation['changes']:
for change in correlation['changes']: for change in correlation['changes']:
ui.markdown( ui.markdown(

View File

@@ -5,6 +5,7 @@ The dashboard shown is from dashboard_ui.py with real-time database data.
""" """
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from nicegui import app, ui from nicegui import app, ui
@@ -22,14 +23,25 @@ def init(fastapi_app: FastAPI, storage_secret: str = 'Secr2t!') -> None:
storage_secret: Optional secret for persistent user storage. storage_secret: Optional secret for persistent user storage.
""" """
@ui.page('/show') def render_dashboard_page() -> None:
def show():
create_dashboard() create_dashboard()
# NOTE dark mode will be persistent for each user across tabs and server restarts # NOTE dark mode will be persistent for each user across tabs and server restarts
ui.dark_mode().bind_value(app.storage.user, 'dark_mode') ui.dark_mode().bind_value(app.storage.user, 'dark_mode')
ui.checkbox('dark mode').bind_value(app.storage.user, 'dark_mode') ui.checkbox('dark mode').bind_value(app.storage.user, 'dark_mode')
@ui.page('/')
def home() -> None:
render_dashboard_page()
@ui.page('/show')
def show() -> None:
render_dashboard_page()
@fastapi_app.get('/dashboard', include_in_schema=False)
def dashboard_redirect() -> RedirectResponse:
return RedirectResponse(url='/', status_code=307)
ui.run_with( ui.run_with(
fastapi_app, fastapi_app,
storage_secret=storage_secret, # NOTE setting a secret is optional but allows for persistent storage per user storage_secret=storage_secret, # NOTE setting a secret is optional but allows for persistent storage per user

View File

@@ -6,7 +6,7 @@ This application uses FastAPI to:
2. Host NiceGUI frontend via ui.run_with() 2. Host NiceGUI frontend via ui.run_with()
The NiceGUI frontend provides: The NiceGUI frontend provides:
1. Interactive dashboard at /show 1. Interactive dashboard at /
2. Real-time data visualization 2. Real-time data visualization
3. Audit trail display 3. Audit trail display
""" """
@@ -149,14 +149,15 @@ def _resolve_n8n_api_url(explicit_url: str | None = None) -> str:
return "" return ""
@app.get('/') @app.get('/api')
def read_root(): def read_api_info():
"""Root endpoint that returns service metadata.""" """Return service metadata for API clients."""
return { return {
'service': 'AI Software Factory', 'service': 'AI Software Factory',
'version': __version__, 'version': __version__,
'endpoints': [ 'endpoints': [
'/', '/',
'/api',
'/health', '/health',
'/generate', '/generate',
'/projects', '/projects',
@@ -217,6 +218,7 @@ async def generate_software(request: SoftwareRequest, db: DbSession):
response_data['tech_stack'] = request.tech_stack response_data['tech_stack'] = request.tech_stack
response_data['project_root'] = result.get('project_root', str(_project_root(project_id))) response_data['project_root'] = result.get('project_root', str(_project_root(project_id)))
response_data['changed_files'] = result.get('changed_files', []) response_data['changed_files'] = result.get('changed_files', [])
response_data['repository'] = result.get('repository')
return {'status': result['status'], 'data': response_data} return {'status': result['status'], 'data': response_data}