from __future__ import annotations import hashlib import json from typing import Any from core_db import connect from core_schema import utc_now from secret_store import load_secret DEFAULT_MODEL = "gpt-5.6-sol" def _client(): try: from openai import OpenAI except ImportError as exc: raise RuntimeError("Не установлен пакет openai.") from exc secret = load_secret("openai") key = str(secret.get("api_key") or "").strip() if not key: raise RuntimeError("API key OpenAI не настроен.") return OpenAI(api_key=key), str(secret.get("model") or DEFAULT_MODEL) def _run_start(purpose: str, model: str, input_text: str) -> int: digest = hashlib.sha256(input_text.encode("utf-8")).hexdigest() with connect() as con: cur = con.execute( """ INSERT INTO ai_runs(purpose,model,created_at,status,input_hash) VALUES(?,?,?,?,?) """, (purpose, model, utc_now(), "running", digest), ) con.commit() return int(cur.lastrowid) def _usage_meta(response: Any, model: str) -> dict[str, Any]: meta = { "response_id": getattr(response, "id", None), "model": getattr(response, "model", model), } usage = getattr(response, "usage", None) if usage is not None: for key in ("input_tokens", "output_tokens", "total_tokens"): value = getattr(usage, key, None) if value is not None: meta[key] = value return meta def _run_ok(run_id: int, meta: dict[str, Any]) -> None: with connect() as con: con.execute( """ UPDATE ai_runs SET status='ok',result_meta_json=?,error=NULL WHERE id=? """, (json.dumps(meta, ensure_ascii=False), run_id), ) con.execute( """ UPDATE connectors SET status='ok',last_error=NULL WHERE connector_key='openai' """ ) con.commit() def _run_error(run_id: int, exc: Exception) -> None: with connect() as con: con.execute( "UPDATE ai_runs SET status='error',error=? WHERE id=?", (str(exc)[:3000], run_id), ) con.execute( """ UPDATE connectors SET status='error',last_error=? WHERE connector_key='openai' """, (str(exc)[:2000],), ) con.commit() def test_connection() -> dict[str, Any]: client, model = _client() # Проверяем доступ к выбранной модели без генерации ответа. obj = client.models.retrieve(model) now = utc_now() with connect() as con: con.execute( """ UPDATE connectors SET status='ok',last_test_at=?,last_error=NULL,meta_json=? WHERE connector_key='openai' """, ( now, json.dumps({"model": model}, ensure_ascii=False), ), ) con.commit() return {"ok": True, "model": getattr(obj, "id", model)} def analyze_text( *, purpose: str, text: str, instructions: str | None = None, model: str | None = None, ) -> dict[str, Any]: if not text.strip(): raise RuntimeError("Пустой текст для анализа.") client, configured_model = _client() chosen = model or configured_model prompt = instructions or ( "Проанализируй материал строго по предоставленному тексту. " "Не восстанавливай отсутствующие нормы и явно отмечай пробелы источника." ) run_id = _run_start(purpose, chosen, text) try: response = client.responses.create( model=chosen, instructions=prompt, input=text, store=False, ) output = response.output_text meta = _usage_meta(response, chosen) _run_ok(run_id, meta) return { "ok": True, "text": output, "meta": meta, "ai_run_id": run_id, } except Exception as exc: _run_error(run_id, exc) raise def structured_response( *, purpose: str, text: str, instructions: str, schema_name: str, schema: dict[str, Any], model: str | None = None, ) -> dict[str, Any]: if not text.strip(): raise RuntimeError("Пустой текст для анализа.") client, configured_model = _client() chosen = model or configured_model run_id = _run_start(purpose, chosen, text) try: response = client.responses.create( model=chosen, instructions=instructions, input=text, store=False, text={ "format": { "type": "json_schema", "name": schema_name, "schema": schema, "strict": True, } }, ) raw = response.output_text result = json.loads(raw) meta = _usage_meta(response, chosen) _run_ok(run_id, meta) return { "ok": True, "data": result, "meta": meta, "ai_run_id": run_id, } except Exception as exc: _run_error(run_id, exc) raise LEGAL_DOCUMENT_SCHEMA: dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "document": { "type": "object", "additionalProperties": False, "properties": { "code": {"type": "string"}, "title": {"type": "string"}, "document_type": {"type": "string"}, "revision": {"type": "string"}, }, "required": ["code", "title", "document_type", "revision"], }, "completeness": { "type": "object", "additionalProperties": False, "properties": { "status": { "type": "string", "enum": ["complete", "possibly_incomplete", "incomplete"], }, "score": {"type": "number"}, "reasons": { "type": "array", "items": {"type": "string"}, }, "last_visible_section": {"type": "string"}, }, "required": [ "status", "score", "reasons", "last_visible_section", ], }, "rules": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "rule_type": { "type": "string", "enum": [ "scope", "exclusion", "definition", "conformity", "requirement", "reference", ], }, "legal_basis": {"type": "string"}, "statement": {"type": "string"}, "conditions": { "type": "array", "items": {"type": "string"}, }, "outcomes": { "type": "array", "items": {"type": "string"}, }, "confidence": {"type": "number"}, }, "required": [ "rule_type", "legal_basis", "statement", "conditions", "outcomes", "confidence", ], }, }, }, "required": ["document", "completeness", "rules"], } PRODUCT_IDENTITY_SCHEMA: dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "identity": { "type": "object", "additionalProperties": False, "properties": { "product_name": {"type": "string"}, "product_type": {"type": "string"}, "purpose": {"type": "string"}, "confidence": {"type": "number"}, }, "required": [ "product_name", "product_type", "purpose", "confidence", ], }, "features": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "name": {"type": "string"}, "value": {"type": "string"}, "source": {"type": "string"}, }, "required": ["name", "value", "source"], }, }, "candidate_regulations": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "code": {"type": "string"}, "reason": {"type": "string"}, "confidence": {"type": "number"}, }, "required": ["code", "reason", "confidence"], }, }, "missing_questions": { "type": "array", "items": {"type": "string"}, }, }, "required": [ "identity", "features", "candidate_regulations", "missing_questions", ], } def extract_legal_document( *, document_code: str, title: str, text: str, model: str | None = None, ) -> dict[str, Any]: instructions = ( "Ты анализируешь нормативный документ для внутренней системы ORIGUS. " "Работай ТОЛЬКО с переданным текстом. Не дополняй отсутствующие статьи " "из памяти или внешних знаний. Сначала оцени полноту источника. " "Если документ обрывается, отсутствуют ожидаемые продолжения, приложения " "или структура выглядит неполной — пометь possibly_incomplete/incomplete. " "Извлекай только юридически значимые правила: область применения, " "исключения, определения, подтверждение соответствия, обязательные " "требования и нормативные ссылки. Все извлечённые правила требуют " "проверки специалистом и не являются опубликованными правилами системы." ) input_text = ( f"Код документа: {document_code}\n" f"Заголовок источника: {title}\n\n" f"ТЕКСТ ИСТОЧНИКА:\n{text}" ) return structured_response( purpose="legal_document_extract", text=input_text, instructions=instructions, schema_name="origus_legal_document", schema=LEGAL_DOCUMENT_SCHEMA, model=model, ) def identify_product( *, name: str, purpose: str, description: str, characteristics: dict[str, Any], model: str | None = None, ) -> dict[str, Any]: payload = { "name": name, "purpose": purpose, "description": description, "characteristics": characteristics, } instructions = ( "Ты выполняешь только идентификацию продукции для ORIGUS. " "Не выноси окончательное юридическое решение о применимости технических " "регламентов и не назначай сертификат/декларацию. " "Выдели фактические признаки продукции из введённых данных, предложи " "кандидатные технические регламенты для дальнейшей проверки rules engine " "и сформулируй недостающие вопросы. Не придумывай характеристики." ) return structured_response( purpose="product_identification", text=json.dumps(payload, ensure_ascii=False, indent=2), instructions=instructions, schema_name="origus_product_identity", schema=PRODUCT_IDENTITY_SCHEMA, model=model, )