2026 技术观察：企业智能搜索进入权限化阶段，多源文档、语义检索和访问控制成为新入口

import json
import math
from datetime import datetime
from collections import defaultdict


DOCUMENTS = [
    {
        "doc_id": "doc_001",
        "title": "订单系统接口说明",
        "content": "本文档描述订单创建、订单查询和订单状态回调接口。",
        "department": "技术部",
        "security_level": "internal",
        "roles": ["developer", "ops"]
    },
    {
        "doc_id": "doc_002",
        "title": "客户续费分析报告",
        "content": "报告分析客户续费率、流失原因和销售跟进策略。",
        "department": "销售部",
        "security_level": "confidential",
        "roles": ["sales_manager", "executive"]
    },
    {
        "doc_id": "doc_003",
        "title": "服务器故障处理手册",
        "content": "当服务器 CPU 异常、磁盘满或服务不可用时，可参考该手册处理。",
        "department": "运维部",
        "security_level": "internal",
        "roles": ["ops", "developer"]
    },
    {
        "doc_id": "doc_004",
        "title": "员工薪酬调整方案",
        "content": "本文档包含薪酬结构、调整规则和预算测算。",
        "department": "人事部",
        "security_level": "secret",
        "roles": ["hr_manager"]
    }
]

def simple_text_vector(text):
    keywords = [
        "订单", "接口", "客户", "续费", "服务器",
        "故障", "薪酬", "员工", "销售", "运维",
        "CPU", "磁盘", "预算", "流失"
    ]

    vector = []

    for keyword in keywords:
        vector.append(
            1 if keyword in text else 0
        )

    return vector

def cosine_similarity(vec_a, vec_b):
    dot = sum(
        a * b
        for a, b in zip(vec_a, vec_b)
    )

    norm_a = math.sqrt(
        sum(a * a for a in vec_a)
    )

    norm_b = math.sqrt(
        sum(b * b for b in vec_b)
    )

    if norm_a == 0 or norm_b == 0:
        return 0

    return round(
        dot / (norm_a * norm_b),
        3
    )

def has_document_permission(user, document):
    user_roles = set(
        user.get("roles", [])
    )

    doc_roles = set(
        document.get("roles", [])
    )

    if user_roles & doc_roles:
        return True

    if "admin" in user_roles:
        return True

    return False

def search_enterprise_documents(query, user, documents):
    query_vector = simple_text_vector(query)

    results = []

    for document in documents:
        if not has_document_permission(user, document):
            continue

        doc_text = document["title"] + " " + document["content"]
        doc_vector = simple_text_vector(doc_text)

        score = cosine_similarity(
            query_vector,
            doc_vector
        )

        if score > 0:
            results.append({
                "doc_id": document["doc_id"],
                "title": document["title"],
                "department": document["department"],
                "security_level": document["security_level"],
                "score": score,
                "content": document["content"]
            })

    results.sort(
        key=lambda item: item["score"],
        reverse=True
    )

    return results

SEARCH_LOGS = []


def build_search_summary(results):
    summaries = []

    for item in results[:3]:
        summaries.append({
            "doc_id": item["doc_id"],
            "title": item["title"],
            "summary": item["content"][:40] + "..."
        })

    return summaries


def write_search_audit_log(user, query, result_count):
    log = {
        "user_id": user["user_id"],
        "roles": user["roles"],
        "query": 30664.t.kuaisou.com 
        "result_count": result_count,
        "search_time": datetime.now().isoformat()
    }

    SEARCH_LOGS.append(log)

    return log

def run_enterprise_search():
    user = {
        "user_id": "user_001",
        "roles": ["ops"]
    }

    query = "服务器磁盘满了怎么处理"

    results = search_enterprise_documents(
        query,
        user,
        DOCUMENTS
    )

    summaries = build_search_summary(
        results
    )

    audit_log = write_search_audit_log(
        user,
        query,
        len(results)
    )

    report = {
        "report_name": "企业智能搜索结果报告",
        "query": query,
        "user": 30661.t.kuaisou.com 
        "results": results,
        "summaries": summaries,
        "audit_log": audit_log,
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_enterprise_search()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))