#!/usr/bin/env python3 """Route a 1C user question to docs RAG, current-config fact checks, or both.""" from __future__ import annotations import argparse import base64 import json import re import sys from pathlib import Path from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parent)) from resolve_1c_fact import resolve_from_route_index, split_fact_path # noqa: E402 from resolve_1c_object import canonical_kind, load_json # noqa: E402 ROOT = Path(__file__).resolve().parents[1] DEFAULT_INDEX = ROOT / "reports" / "1c-sql" / "upo" / "unified-object-route-index.json" KIND_WORDS = { "справочник": "Справочник", "справочника": "Справочник", "документ": "Документ", "документа": "Документ", "обработка": "Обработка", "обработки": "Обработка", "отчет": "Отчет", "отчета": "Отчет", "регистрсведений": "РегистрСведений", "регистр сведений": "РегистрСведений", "регистрнакопления": "РегистрНакопления", "регистр накопления": "РегистрНакопления", } EXPLICIT_PATH_RE = re.compile( r"\b(Справочник|Документ|Обработка|Отчет|РегистрСведений|РегистрНакопления|РегистрБухгалтерии|Перечисление|ОбщийМодуль)\." r"([A-Za-zА-Яа-яЁё_][A-Za-zА-Яа-яЁё0-9_]*)" r"(?:\.([A-Za-zА-Яа-яЁё_][A-Za-zА-Яа-яЁё0-9_]*))?" r"(?:\.([A-Za-zА-Яа-яЁё_][A-Za-zА-Яа-яЁё0-9_]*))?" ) IDENT_RE = r"([A-Za-zА-Яа-яЁё_][A-Za-zА-Яа-яЁё0-9_]{2,})" def decode_arg(value: str | None, encoded: str | None) -> str | None: if encoded: return base64.b64decode(encoded).decode("utf-8") return value def unique_fact_paths(paths: list[str]) -> list[str]: result = [] seen = set() for path in paths: key = path.casefold() if key in seen: continue seen.add(key) result.append(path) return result def detect_source_risk(text: str) -> list[dict[str, Any]]: lowered = text.casefold() risks = [] if any(marker in lowered for marker in ("пример rag", "metadata.example", "synthetic-example", "в примере")): risks.append( { "code": "example_is_not_current_fact", "severity": "warning", "message": "Question mentions an example source; it cannot confirm current-base facts.", } ) return risks def extract_explicit_paths(text: str) -> list[str]: paths = [] for match in EXPLICIT_PATH_RE.finditer(text): parts = [part for part in match.groups() if part] paths.append(".".join(parts)) return paths def extract_attribute_phrases(text: str) -> list[str]: paths = [] kind_pattern = "|".join(sorted((re.escape(key) for key in KIND_WORDS), key=len, reverse=True)) for match in re.finditer(rf"\bреквизит\s+{IDENT_RE}\s+у\s+({kind_pattern})\s+{IDENT_RE}", text, flags=re.IGNORECASE): member = match.group(1) kind_word = match.group(2).casefold() object_name = match.group(3) kind = KIND_WORDS.get(kind_word) if kind: paths.append(f"{kind}.{object_name}.{member}") for match in re.finditer(rf"\bу\s+({kind_pattern})\s+{IDENT_RE}[^.?!\n]{{0,80}}\bреквизит\s+{IDENT_RE}", text, flags=re.IGNORECASE): kind_word = match.group(1).casefold() object_name = match.group(2) member = match.group(3) kind = KIND_WORDS.get(kind_word) if kind: paths.append(f"{kind}.{object_name}.{member}") return paths def extract_fact_paths(text: str) -> list[str]: return unique_fact_paths(extract_explicit_paths(text) + extract_attribute_phrases(text)) def classify(text: str, fact_paths: list[str], risks: list[dict[str, Any]]) -> dict[str, Any]: lowered = text.casefold() docs_markers = ("что такое", "как ", "пример", "документац", "событие", "при открытии", "приоткрытии") current_markers = ("текущ", "в базе", "конфигурац", "есть ли", "реквизит", "форма", "команда", "модуль") needs_fact = bool(fact_paths) or ("реквизит" in lowered and any(word in lowered for word in KIND_WORDS)) needs_docs = any(marker in lowered for marker in docs_markers) if risks: needs_fact = True if needs_fact and needs_docs: route = "mixed_docs_and_current_config" elif needs_fact: route = "current_config_fact" elif needs_docs: route = "docs_rag" else: route = "needs_clarification" return { "route": route, "needs_docs_rag": needs_docs, "needs_current_config": needs_fact, "current_config_required_before_code": needs_fact, "safe_rag_scope": "official_1c_docs" if needs_docs else None, } def resolve_facts(index_path: Path | None, fact_paths: list[str], *, view: str) -> list[dict[str, Any]]: if not index_path or not index_path.exists(): return [ { "path": path, "status": "not_checked", "reason": "route index is not available", } for path in fact_paths ] index = load_json(index_path) results = [] for path in fact_paths: kind, name, table_section, member = split_fact_path(path) result = resolve_from_route_index( index, index_path=index_path, kind=kind, object_name=name, member=member, table_section=table_section, area="any", view=view, extension=None, ) results.append({"path": path, "status": "checked", "result": result}) return results def route_question(text: str, *, index_path: Path | None, view: str) -> dict[str, Any]: risks = detect_source_risk(text) fact_paths = extract_fact_paths(text) decision = classify(text, fact_paths, risks) fact_checks = resolve_facts(index_path, fact_paths, view=view) if fact_paths else [] return { "schema": "onec_question_route.v1", "question": text, "decision": decision, "source_risks": risks, "fact_paths": fact_paths, "fact_checks": fact_checks, "recommended_next": recommended_next(decision, fact_checks), } def recommended_next(decision: dict[str, Any], fact_checks: list[dict[str, Any]]) -> list[dict[str, str]]: steps = [] if decision.get("needs_current_config"): steps.append({"tool": "fact_resolver", "purpose": "confirm current configuration facts before code"}) if decision.get("needs_docs_rag"): steps.append({"tool": "docs_rag", "purpose": "retrieve official platform/documentation context"}) if fact_checks: missing = [row for row in fact_checks if not (((row.get("result") or {}).get("exists")))] if missing: steps.append({"tool": "ask_or_adjust_task", "purpose": "do not write code for unresolved facts"}) return steps def main() -> int: parser = argparse.ArgumentParser(description="Route a 1C question to docs RAG or current-config adapter tools.") parser.add_argument("--text") parser.add_argument("--text-b64") parser.add_argument("--index", type=Path, default=DEFAULT_INDEX) parser.add_argument("--view", choices=["effective", "base"], default="effective") parser.add_argument("--output", type=Path) args = parser.parse_args() text = decode_arg(args.text, args.text_b64) if not text: raise SystemExit("Use --text or --text-b64.") result = route_question(text, index_path=args.index, view=args.view) output = json.dumps(result, ensure_ascii=False, indent=2) + "\n" if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(output, encoding="utf-8") print(json.dumps({"output": str(args.output), "route": result["decision"]["route"], "fact_paths": len(result["fact_paths"])}, ensure_ascii=False)) else: print(output) return 0 if __name__ == "__main__": raise SystemExit(main())