import os from datetime import datetime, timezone import requests import streamlit as st import yaml st.set_page_config(page_title="DevOps AI Assistant", page_icon="πŸš€") st.title("DevOps AI Assistant") st.markdown("Π’Π΅Π±-интСрфСйс для Π³Π΅Π½Π΅Ρ€Π°Ρ†ΠΈΠΈ ΠΈ ΠΏΡ€ΠΎΠ²Π΅Ρ€ΠΊΠΈ CI/CD ΠΊΠΎΠ½Ρ„ΠΈΠ³ΡƒΡ€Π°Ρ†ΠΈΠΉ.") def _env(name: str, default: str) -> str: value = os.getenv(name) return value if value else default default_backend_url = _env("BACKEND_URL", "http://backend:8000/generate") with st.sidebar: st.header("Настройки") api_url = st.text_input("Backend URL", default_backend_url) model = st.selectbox( "LLM МодСль", ["deepseek-r1:1.5b", "qwen2.5-coder:1.5b", "llama3.1:8b", "Arena Model"], ) platform = st.radio("ΠŸΠ»Π°Ρ‚Ρ„ΠΎΡ€ΠΌΠ°", ["PV DOT CI", "PV DOT CD", "DPM CI", "DPM CD"]) if "history" not in st.session_state: st.session_state.history = [] def _push_history(*, platform: str, model: str, prompt: str, status: str, config: str, error: str | None): st.session_state.history.insert( 0, { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"), "platform": platform, "model": model, "prompt": prompt, "status": status, "error": error, "config": config, }, ) st.session_state.history = st.session_state.history[:50] tab_overview, tab_generator, tab_validator, tab_history = st.tabs( ["ΠžΠ±Π·ΠΎΡ€", "Pipeline generator", "ΠŸΡ€ΠΎΠ²Π΅Ρ€ΠΊΠ° YAML", "Π˜ΡΡ‚ΠΎΡ€ΠΈΡ"] ) with tab_overview: st.subheader("Π§Ρ‚ΠΎ ΡƒΠΌΠ΅Π΅Ρ‚ сСрвис") st.markdown( "- **Pipeline generator**: Π³Π΅Π½Π΅Ρ€ΠΈΡ€ΡƒΠ΅Ρ‚ YAML Ρ‡Π΅Ρ€Π΅Π· backend (FastAPI β†’ Ollama).\n" "- **ΠŸΡ€ΠΎΠ²Π΅Ρ€ΠΊΠ° YAML**: локально провСряСт синтаксис YAML, Ρ‡Ρ‚ΠΎΠ±Ρ‹ быстро ΠΏΡ€Π°Π²ΠΈΡ‚ΡŒ ΠΊΠΎΠ½Ρ„ΠΈΠ³.\n" "- **Π˜ΡΡ‚ΠΎΡ€ΠΈΡ**: Ρ…Ρ€Π°Π½ΠΈΡ‚ послСдниС Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Ρ‹ Π² Ρ€Π°ΠΌΠΊΠ°Ρ… сСссии Π±Ρ€Π°ΡƒΠ·Π΅Ρ€Π°." ) st.divider() st.caption(f"Backend URL: {api_url}") with tab_generator: st.subheader("AI CI/CD Pipeline Generator") user_input = st.text_area( "ΠžΠΏΠΈΡˆΠΈΡ‚Π΅ ΠΏΠ°ΠΉΠΏΠ»Π°ΠΉΠ½ (Π½Π°ΠΏΡ€ΠΈΠΌΠ΅Ρ€: 'Python app with pytest and docker build')", height=150, key="generator_prompt", ) col_a, col_b = st.columns([1, 3]) with col_a: generate_clicked = st.button("Π‘Π³Π΅Π½Π΅Ρ€ΠΈΡ€ΠΎΠ²Π°Ρ‚ΡŒ ΠΊΠΎΠ½Ρ„ΠΈΠ³ΡƒΡ€Π°Ρ†ΠΈΡŽ", type="primary") with col_b: st.caption("Подсказка: Ρ‡Π΅ΠΌ ΠΊΠΎΠ½ΠΊΡ€Π΅Ρ‚Π½Π΅Π΅ трСбования, Ρ‚Π΅ΠΌ Π²Π°Π»ΠΈΠ΄Π½Π΅Π΅ YAML.") if generate_clicked: if user_input.strip(): with st.spinner("LLM Π΄ΡƒΠΌΠ°Π΅Ρ‚..."): try: res = requests.post( api_url, json={"description": user_input, "platform": platform, "model": model}, timeout=90, ) res.raise_for_status() data = res.json() status = data.get("status", "error") config = data.get("config", "") err = data.get("validation_error") _push_history( platform=platform, model=model, prompt=user_input, status=status, config=config, error=err, ) if status == "ok": st.success("ΠšΠΎΠ½Ρ„ΠΈΠ³ΡƒΡ€Π°Ρ†ΠΈΡ Π²Π°Π»ΠΈΠ΄Π½Π°.") else: st.error(f"Ошибка Π²Π°Π»ΠΈΠ΄Π°Ρ†ΠΈΠΈ: {err}") st.code(config, language="yaml") st.download_button("Π‘ΠΊΠ°Ρ‡Π°Ρ‚ΡŒ .yml", config, file_name="pipeline.yml") except Exception as e: st.error(f"Ошибка ΠΏΠΎΠ΄ΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΡ ΠΊ Π±Π΅ΠΊΠ΅Π½Π΄Ρƒ: {e}") else: st.warning("ΠŸΠΎΠΆΠ°Π»ΡƒΠΉΡΡ‚Π°, Π²Π²Π΅Π΄ΠΈΡ‚Π΅ описаниС.") with tab_validator: st.subheader("Π›ΠΎΠΊΠ°Π»ΡŒΠ½Π°Ρ ΠΏΡ€ΠΎΠ²Π΅Ρ€ΠΊΠ° YAML") yaml_text = st.text_area("Π’ΡΡ‚Π°Π²ΡŒΡ‚Π΅ YAML для ΠΏΡ€ΠΎΠ²Π΅Ρ€ΠΊΠΈ", height=240, key="validator_yaml") c1, c2 = st.columns([1, 1]) with c1: validate_clicked = st.button("ΠŸΡ€ΠΎΠ²Π΅Ρ€ΠΈΡ‚ΡŒ YAML") with c2: if st.button("Π’ΡΡ‚Π°Π²ΠΈΡ‚ΡŒ послСдний Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚", disabled=(len(st.session_state.history) == 0)): st.session_state.validator_yaml = (st.session_state.history[0].get("config") or "") st.rerun() if validate_clicked: if yaml_text.strip(): try: yaml.safe_load(yaml_text) st.success("YAML синтаксичСски Π²Π°Π»ΠΈΠ΄Π΅Π½.") except Exception as e: st.error(f"Ошибка YAML: {e}") else: st.warning("Π’ΡΡ‚Π°Π²ΡŒΡ‚Π΅ YAML.") with tab_history: st.subheader("Π˜ΡΡ‚ΠΎΡ€ΠΈΡ (послСдниС 50)") if not st.session_state.history: st.info("Пока пусто. Π‘Π³Π΅Π½Π΅Ρ€ΠΈΡ€ΡƒΠΉΡ‚Π΅ ΠΊΠΎΠ½Ρ„ΠΈΠ³ΡƒΡ€Π°Ρ†ΠΈΡŽ Π²ΠΎ Π²ΠΊΠ»Π°Π΄ΠΊΠ΅ Pipeline generator.") else: if st.button("ΠžΡ‡ΠΈΡΡ‚ΠΈΡ‚ΡŒ ΠΈΡΡ‚ΠΎΡ€ΠΈΡŽ"): st.session_state.history = [] st.rerun() for idx, item in enumerate(st.session_state.history): title = f"[{item['ts']}] {item['platform']} Β· {item['model']} Β· {item['status']}" with st.expander(title, expanded=(idx == 0)): st.caption(item.get("prompt", "")) if item.get("error"): st.error(f"Validation error: {item['error']}") st.code(item.get("config", ""), language="yaml") st.download_button( "Π‘ΠΊΠ°Ρ‡Π°Ρ‚ΡŒ .yml", item.get("config", ""), file_name=f"pipeline-{idx+1}.yml", key=f"download_{idx}", )