"""Deterministic governance agents composed with the LambdAgent runtime.

These agents deliberately use parsers and rules for facts that can be known
from DDL/OpenAPI/inventory input.  Semantic mappings are emitted as pending
inferences.  The distinction is preserved when results are persisted.
"""
from __future__ import annotations

import hashlib
import json
import re
from dataclasses import asdict
from typing import Any, Dict, List

import yaml

from lambdagent import Context, Guard, Par, Tool

from agentpaas.db.models import gen_id, now_utc
from .service import (
    GovernanceError,
    create_assertion,
    create_evidence,
    overview,
    upsert_element,
    validate_project,
)


SOURCE_TYPES = ("ddl", "openapi", "inventory")
_SENSITIVE_TOKENS = (
    "mobile", "phone", "email", "id_card", "idcard", "identity", "password",
    "secret", "token", "bank", "address", "姓名", "手机号", "身份证", "邮箱",
)


def _safe_name(value: str) -> str:
    value = re.sub(r"[^A-Za-z0-9_\-]+", "-", value.strip())
    return value.strip("-") or "unnamed"


def _pascal(value: str) -> str:
    words = re.split(r"[^A-Za-z0-9\u4e00-\u9fff]+|_+", value)
    return "".join(word[:1].upper() + word[1:] for word in words if word) or "Unnamed"


def _is_sensitive(name: str) -> bool:
    lowered = name.lower()
    return any(token in lowered for token in _SENSITIVE_TOKENS)


def _split_sql_columns(body: str) -> List[str]:
    chunks: List[str] = []
    current: List[str] = []
    depth = 0
    quote = ""
    for char in body:
        if quote:
            current.append(char)
            if char == quote:
                quote = ""
            continue
        if char in ("'", '"', "`"):
            quote = char
            current.append(char)
        elif char == "(":
            depth += 1
            current.append(char)
        elif char == ")":
            depth = max(depth - 1, 0)
            current.append(char)
        elif char == "," and depth == 0:
            chunks.append("".join(current).strip())
            current = []
        else:
            current.append(char)
    if current:
        chunks.append("".join(current).strip())
    return chunks


def _parse_ddl(content: str) -> dict:
    tables = []
    pattern = re.compile(
        r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(?:[`\"]?([\w-]+)[`\"]?\.)?"
        r"[`\"]?([\w-]+)[`\"]?\s*\((.*?)\)\s*(?:ENGINE\b[^;]*)?;",
        re.IGNORECASE | re.DOTALL,
    )
    for match in pattern.finditer(content):
        schema, table_name, body = match.groups()
        columns = []
        for raw in _split_sql_columns(body):
            line = raw.strip()
            if not line or re.match(
                r"^(PRIMARY|FOREIGN|UNIQUE|CONSTRAINT|KEY|INDEX|CHECK)\b", line, re.IGNORECASE
            ):
                continue
            col = re.match(r"^[`\"]?([\w-]+)[`\"]?\s+([A-Za-z]+(?:\s*\([^)]*\))?)", line)
            if not col:
                continue
            name, data_type = col.groups()
            columns.append({
                "name": name,
                "data_type": re.sub(r"\s+", "", data_type).upper(),
                "nullable": not bool(re.search(r"\bNOT\s+NULL\b", line, re.IGNORECASE)),
                "primary_key": bool(re.search(r"\bPRIMARY\s+KEY\b", line, re.IGNORECASE)),
                "sensitive": _is_sensitive(name),
            })
        tables.append({"schema": schema or "default", "name": table_name, "columns": columns})
    if not tables:
        raise GovernanceError("未从 DDL 中识别到 CREATE TABLE 语句", code="NO_DDL_ASSETS")
    return {"tables": tables, "apis": [], "schemas": []}


def _parse_openapi(content: str) -> dict:
    try:
        doc = yaml.safe_load(content) or {}
    except yaml.YAMLError as exc:
        raise GovernanceError(f"OpenAPI 解析失败: {exc}", code="INVALID_OPENAPI") from exc
    if not isinstance(doc, dict) or not isinstance(doc.get("paths"), dict):
        raise GovernanceError("输入不是有效的 OpenAPI 文档：缺少 paths", code="INVALID_OPENAPI")
    apis = []
    for path, methods in doc["paths"].items():
        if not isinstance(methods, dict):
            continue
        for method, operation in methods.items():
            if method.lower() not in {"get", "post", "put", "patch", "delete", "options", "head"}:
                continue
            operation = operation if isinstance(operation, dict) else {}
            apis.append({
                "path": str(path), "method": method.upper(),
                "operation_id": operation.get("operationId") or f"{method}-{path}",
                "summary": operation.get("summary", ""),
            })
    schemas = []
    components = doc.get("components") if isinstance(doc.get("components"), dict) else {}
    for name, schema in (components.get("schemas") or {}).items():
        props = schema.get("properties", {}) if isinstance(schema, dict) else {}
        schemas.append({
            "name": name,
            "properties": [
                {"name": prop_name, "data_type": str((prop or {}).get("type", "object")), "sensitive": _is_sensitive(prop_name)}
                for prop_name, prop in props.items()
            ],
        })
    return {"tables": [], "apis": apis, "schemas": schemas}


def _parse_inventory(content: str) -> dict:
    try:
        doc = json.loads(content)
    except ValueError as exc:
        raise GovernanceError("inventory 必须是 JSON", code="INVALID_INVENTORY") from exc
    if not isinstance(doc, dict):
        raise GovernanceError("inventory 顶层必须是对象", code="INVALID_INVENTORY")
    return {
        "tables": doc.get("tables") or [],
        "apis": doc.get("apis") or [],
        "schemas": doc.get("schemas") or [],
    }


def normalize_source(payload: Dict[str, Any]) -> dict:
    source_type = str(payload.get("source_type", "")).lower()
    content = str(payload.get("content", ""))
    if source_type not in SOURCE_TYPES:
        raise GovernanceError(f"source_type 必须是 {', '.join(SOURCE_TYPES)}")
    if not content.strip():
        raise GovernanceError("content 不能为空")
    if len(content.encode("utf-8")) > 512 * 1024:
        raise GovernanceError("输入超过 512KB", code="SOURCE_TOO_LARGE", status=413)
    if source_type == "ddl":
        parsed = _parse_ddl(content)
    elif source_type == "openapi":
        parsed = _parse_openapi(content)
    else:
        parsed = _parse_inventory(content)
    return {
        "source_type": source_type,
        "source_name": str(payload.get("source_name") or f"inline.{source_type}"),
        "environment": str(payload.get("environment") or "development"),
        "content": content,
        "parsed": parsed,
    }


def discover_assets(normalized: dict) -> dict:
    elements: List[dict] = []
    assertions: List[dict] = []
    env = normalized["environment"]
    source = normalized["source_name"]

    for table in normalized["parsed"]["tables"]:
        table_name = str(table.get("name", "unnamed"))
        schema = str(table.get("schema") or "default")
        table_key = f"m1:table:{schema}:{table_name}"
        physical_key = f"m0:table:{schema}:{table_name}"
        elements.extend((
            {"local_id": table_key, "layer": "M1", "profile": "software", "kind": "Table",
             "name": f"{_pascal(table_name)}Table", "status": "candidate",
             "uri": f"urn:ontorefactor:model:table:{_safe_name(schema)}:{_safe_name(table_name)}",
             "attributes": {"schema": schema, "table": table_name}},
            {"local_id": physical_key, "layer": "M0", "profile": "software", "kind": "PhysicalTable",
             "name": f"{env}.{schema}.{table_name}", "status": "published",
             "uri": f"db://{_safe_name(env)}/{_safe_name(schema)}/{_safe_name(table_name)}",
             "attributes": {"environment": env, "schema": schema, "table": table_name}},
        ))
        assertions.append({"subject": physical_key, "predicate": "INSTANCE_OF", "object": table_key,
                           "assertion_type": "extracted", "confidence": 1.0, "review_status": "accepted"})
        for column in table.get("columns") or []:
            column_name = str(column.get("name", "unnamed"))
            column_key = f"m1:column:{schema}:{table_name}:{column_name}"
            physical_column_key = f"m0:column:{schema}:{table_name}:{column_name}"
            attrs = {
                "environment": env, "data_type": column.get("data_type", ""),
                "nullable": bool(column.get("nullable", True)),
                "primary_key": bool(column.get("primary_key", False)),
                "sensitive": bool(column.get("sensitive", _is_sensitive(column_name))),
            }
            elements.extend((
                {"local_id": column_key, "layer": "M1", "profile": "software", "kind": "Column",
                 "name": f"{_pascal(table_name)}{_pascal(column_name)}Column", "status": "candidate",
                 "uri": f"urn:ontorefactor:model:column:{_safe_name(schema)}:{_safe_name(table_name)}:{_safe_name(column_name)}",
                 "attributes": attrs},
                {"local_id": physical_column_key, "layer": "M0", "profile": "software", "kind": "PhysicalColumn",
                 "name": f"{env}.{schema}.{table_name}.{column_name}", "status": "published",
                 "uri": f"db://{_safe_name(env)}/{_safe_name(schema)}/{_safe_name(table_name)}/{_safe_name(column_name)}",
                 "attributes": attrs},
            ))
            assertions.extend((
                {"subject": table_key, "predicate": "CONTAINS", "object": column_key,
                 "assertion_type": "extracted", "confidence": 1.0, "review_status": "accepted"},
                {"subject": physical_key, "predicate": "CONTAINS", "object": physical_column_key,
                 "assertion_type": "extracted", "confidence": 1.0, "review_status": "accepted"},
                {"subject": physical_column_key, "predicate": "INSTANCE_OF", "object": column_key,
                 "assertion_type": "extracted", "confidence": 1.0, "review_status": "accepted"},
            ))

    for api in normalized["parsed"]["apis"]:
        operation_id = str(api.get("operation_id") or "operation")
        m1_key = f"m1:api:{operation_id}"
        m0_key = f"m0:api:{operation_id}"
        elements.extend((
            {"local_id": m1_key, "layer": "M1", "profile": "software", "kind": "ApiOperation",
             "name": _pascal(operation_id), "status": "candidate",
             "uri": f"urn:ontorefactor:model:api:{_safe_name(operation_id)}",
             "attributes": {"path": api.get("path"), "method": api.get("method")}},
            {"local_id": m0_key, "layer": "M0", "profile": "software", "kind": "ApiObservation",
             "name": f"{api.get('method')} {api.get('path')}", "status": "published",
             "uri": f"openapi://{_safe_name(source)}/{_safe_name(operation_id)}",
             "attributes": {"environment": env, "path": api.get("path"), "method": api.get("method")}},
        ))
        assertions.append({"subject": m0_key, "predicate": "INSTANCE_OF", "object": m1_key,
                           "assertion_type": "extracted", "confidence": 1.0, "review_status": "accepted"})

    for schema in normalized["parsed"]["schemas"]:
        schema_name = str(schema.get("name", "Schema"))
        schema_key = f"m1:schema:{schema_name}"
        elements.append({"local_id": schema_key, "layer": "M1", "profile": "software", "kind": "Schema",
                         "name": _pascal(schema_name), "status": "candidate",
                         "uri": f"urn:ontorefactor:model:schema:{_safe_name(schema_name)}",
                         "attributes": {"properties": len(schema.get("properties") or [])}})
        for prop in schema.get("properties") or []:
            prop_name = str(prop.get("name", "property"))
            prop_key = f"m1:schema-field:{schema_name}:{prop_name}"
            elements.append({"local_id": prop_key, "layer": "M1", "profile": "software", "kind": "Column",
                             "name": f"{_pascal(schema_name)}{_pascal(prop_name)}Field", "status": "candidate",
                             "uri": f"urn:ontorefactor:model:schema-field:{_safe_name(schema_name)}:{_safe_name(prop_name)}",
                             "attributes": {"data_type": prop.get("data_type"), "sensitive": prop.get("sensitive", False)}})
            assertions.append({"subject": schema_key, "predicate": "CONTAINS", "object": prop_key,
                               "assertion_type": "extracted", "confidence": 1.0, "review_status": "accepted"})
    return {"agent": "asset-discovery", "elements": elements, "assertions": assertions, "issues": []}


def infer_semantics(normalized: dict) -> dict:
    elements: List[dict] = []
    assertions: List[dict] = []
    for table in normalized["parsed"]["tables"]:
        table_name = str(table.get("name", "object"))
        stem = re.sub(r"^(t_|tbl_|dcp_)", "", table_name, flags=re.IGNORECASE)
        business_name = _pascal(stem)
        object_key = f"m1:business-object:{stem}"
        table_key = f"m1:table:{table.get('schema') or 'default'}:{table_name}"
        elements.append({
            "local_id": object_key, "layer": "M1", "profile": "business", "kind": "BusinessObject",
            "name": business_name, "description": f"由表名 {table_name} 推断的候选业务对象",
            "status": "candidate", "uri": f"urn:ontorefactor:model:business-object:{_safe_name(stem)}",
            "attributes": {"inference": "table-name", "source_table": table_name},
        })
        assertions.append({
            "subject": table_key, "predicate": "REPRESENTS", "object": object_key,
            "assertion_type": "inferred", "confidence": 0.72, "review_status": "pending",
        })
    return {"agent": "business-semantics", "elements": elements, "assertions": assertions, "issues": []}


def analyze_controls(normalized: dict) -> dict:
    issues: List[dict] = []
    for table in normalized["parsed"]["tables"]:
        table_name = str(table.get("name", "unnamed"))
        schema = str(table.get("schema") or "default")
        column_names = {str(column.get("name", "")).lower() for column in table.get("columns") or []}
        if ("tenant" in table_name.lower() or table_name.lower().startswith("dcp_")) and "tenant_id" not in column_names:
            issues.append({
                "code": "TENANT_ID_MISSING", "severity": "critical",
                "title": f"{table_name} 缺少 tenant_id",
                "description": "租户域表缺少租户隔离字段，必须由 Owner 确认或整改",
                "target_local_id": f"m1:table:{schema}:{table_name}",
            })
        for column in table.get("columns") or []:
            if column.get("sensitive") or _is_sensitive(str(column.get("name", ""))):
                issues.append({
                    "code": "SENSITIVE_FIELD_REVIEW", "severity": "high",
                    "title": f"敏感字段待策略确认：{table_name}.{column.get('name')}",
                    "description": "需要补充分类分级、访问策略、脱敏方式和责任人",
                    "target_local_id": f"m1:column:{schema}:{table_name}:{column.get('name')}",
                })
    for schema in normalized["parsed"]["schemas"]:
        for prop in schema.get("properties") or []:
            if prop.get("sensitive"):
                issues.append({
                    "code": "SENSITIVE_API_FIELD", "severity": "high",
                    "title": f"API Schema 敏感字段：{schema.get('name')}.{prop.get('name')}",
                    "description": "需要核对 API 消费者权限和响应脱敏策略",
                    "target_local_id": f"m1:schema-field:{schema.get('name')}:{prop.get('name')}",
                })
    return {"agent": "governance-controls", "elements": [], "assertions": [], "issues": issues}


def synthesize(branches: tuple) -> dict:
    elements: Dict[str, dict] = {}
    assertions: Dict[tuple, dict] = {}
    issues: Dict[tuple, dict] = {}
    agents = []
    for branch in branches:
        agents.append(branch["agent"])
        for element in branch["elements"]:
            elements[element["local_id"]] = element
        for assertion in branch["assertions"]:
            key = (assertion["subject"], assertion["predicate"], assertion["object"])
            assertions[key] = assertion
        for issue in branch["issues"]:
            issues[(issue["code"], issue["target_local_id"])] = issue
    return {
        "agents": agents,
        "elements": list(elements.values()),
        "assertions": list(assertions.values()),
        "issues": list(issues.values()),
        "summary": {
            "elements": len(elements), "assertions": len(assertions), "issues": len(issues),
        },
    }


def _valid_result(result: Any) -> bool:
    return isinstance(result, dict) and all(key in result for key in ("elements", "assertions", "issues", "summary"))


def build_governance_agent():
    """Build the typed execution graph used by the governance ingest API."""
    normalize = Tool("governance-source-normalizer", normalize_source)
    parallel_analysis = Par(
        Tool("asset-discovery-agent", discover_assets),
        Tool("business-semantics-agent", infer_semantics),
        Tool("governance-controls-agent", analyze_controls),
    )
    merge = Tool("governance-synthesis-agent", synthesize)
    return Guard(normalize >> parallel_analysis >> merge, validator=_valid_result, retry=0)


def execute_governance_agent(
    db, tenant_id: str, project_id: str, payload: Dict[str, Any], *, persist: bool = True,
) -> dict:
    """Execute the LambdAgent graph and optionally persist its proposals/facts."""
    run_id = gen_id("gar_")
    now = now_utc()
    raw = str(payload.get("content", ""))
    source_type = str(payload.get("source_type", ""))
    db.execute(
        "INSERT INTO governance_agent_runs "
        "(id,tenant_id,project_id,source_type,input_hash,status,result_json,trace_json,error,created_at) "
        "VALUES (?,?,?,?,?,'running','{}','[]','',?)",
        (run_id, tenant_id, project_id, source_type, hashlib.sha256(raw.encode("utf-8")).hexdigest(), now),
    )
    db.commit()
    ctx = Context(run_id=run_id)
    try:
        result = build_governance_agent().apply(payload, ctx)
        persisted = {"elements": 0, "assertions": 0, "issues": 0}
        if persist:
            evidence = create_evidence(db, tenant_id, project_id, {
                "kind": "SourceArtifact", "title": str(payload.get("source_name") or f"智能体输入 {source_type}"),
                "uri": f"evidence://agent-run/{run_id}", "content": raw,
                "source": str(payload.get("source_name") or "inline"),
                "metadata": {"agent_run_id": run_id, "source_type": source_type},
            })
            id_map: Dict[str, str] = {}
            for element in result["elements"]:
                saved = upsert_element(db, tenant_id, project_id, {
                    **element, "source": f"agent-run:{run_id}",
                })
                id_map[element["local_id"]] = saved["id"]
                persisted["elements"] += 1
            for assertion in result["assertions"]:
                if assertion["subject"] not in id_map or assertion["object"] not in id_map:
                    continue
                create_assertion(db, tenant_id, project_id, {
                    "subject_id": id_map[assertion["subject"]],
                    "predicate": assertion["predicate"],
                    "object_id": id_map[assertion["object"]],
                    "assertion_type": assertion["assertion_type"],
                    "confidence": assertion["confidence"],
                    "review_status": assertion["review_status"],
                    "evidence_ids": [evidence["id"]],
                    "generated_by": f"lambdagent:governance-ingest@1.0#{run_id}",
                })
                persisted["assertions"] += 1
            issue_now = now_utc()
            for issue in result["issues"]:
                target_id = id_map.get(issue["target_local_id"], "")
                db.execute(
                    "INSERT INTO governance_issues "
                    "(id,tenant_id,project_id,rule_id,target_id,assertion_id,severity,title,description,"
                    "status,evidence_ids,assigned_to,created_at,updated_at) "
                    "VALUES (?,?,?,?,?,'',?,?,?,'open',?,'',?,?)",
                    (
                        gen_id("gi_"), tenant_id, project_id, f"agent:{issue['code']}", target_id,
                        issue["severity"], issue["title"], issue["description"],
                        json.dumps([evidence["id"]]), issue_now, issue_now,
                    ),
                )
                persisted["issues"] += 1
            db.commit()
            validation = validate_project(db, tenant_id, project_id, persist=True)
        else:
            validation = None
        traces = [asdict(item) for item in ctx.trace]
        response = {
            "run_id": run_id, "status": "completed", "result": result,
            "persisted": persisted, "validation": validation,
            "trace": traces, "overview": overview(db, tenant_id, project_id) if persist else None,
        }
        db.execute(
            "UPDATE governance_agent_runs SET status='completed',result_json=?,trace_json=?,completed_at=? WHERE id=?",
            (json.dumps(response, ensure_ascii=False, default=str), json.dumps(traces, ensure_ascii=False, default=str), now_utc(), run_id),
        )
        db.commit()
        return response
    except Exception as exc:
        db.execute(
            "UPDATE governance_agent_runs SET status='failed',error=?,trace_json=?,completed_at=? WHERE id=?",
            (str(exc)[:1000], json.dumps([asdict(item) for item in ctx.trace], ensure_ascii=False, default=str), now_utc(), run_id),
        )
        db.commit()
        raise


def list_agent_runs(db, tenant_id: str, project_id: str, limit: int = 50) -> List[dict]:
    values = db.fetchall(
        "SELECT * FROM governance_agent_runs WHERE tenant_id=? AND project_id=? "
        "ORDER BY created_at DESC LIMIT ?",
        (tenant_id, project_id, min(max(limit, 1), 200)),
    )
    result = []
    for value in values:
        value["result_json"] = json.loads(value.get("result_json") or "{}")
        value["trace_json"] = json.loads(value.get("trace_json") or "[]")
        result.append(value)
    return result

