125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
"""Search profile resolver for general-purpose AIOS Search Gateway.
|
|
|
|
The Search Gateway is a shared platform capability. Groups and users may customize
|
|
provider preference, query expansion hints, trusted sources and domain profiles on
|
|
top of the shared base profile. Customization is intentionally layered and cannot
|
|
override hard safety, authorization or audit rules.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
HARD_BLOCKED_BEHAVIORS = {
|
|
"bypass_login",
|
|
"bypass_captcha",
|
|
"scrape_private_student_grades",
|
|
"scrape_student_private_data",
|
|
"bulk_scrape_protected_platforms",
|
|
"collect_data_without_authorization",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SearchProfile:
|
|
profile_id: str
|
|
scope: str
|
|
provider_order: list[str] = field(default_factory=lambda: [
|
|
"cache",
|
|
"official_site",
|
|
"general_web",
|
|
"vertical_source",
|
|
"browser_fetch",
|
|
"user_authorized_file",
|
|
"internal_archive",
|
|
])
|
|
trusted_domains: list[str] = field(default_factory=list)
|
|
query_expansion_hints: list[str] = field(default_factory=list)
|
|
domain_profiles: list[str] = field(default_factory=list)
|
|
blocked_behaviors: set[str] = field(default_factory=lambda: set(HARD_BLOCKED_BEHAVIORS))
|
|
audit_required: bool = True
|
|
fallback_required: bool = True
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["blocked_behaviors"] = sorted(self.blocked_behaviors)
|
|
return data
|
|
|
|
|
|
class SearchProfileResolver:
|
|
"""Resolves base -> group -> user -> session search preferences."""
|
|
|
|
def base_profile(self) -> SearchProfile:
|
|
return SearchProfile(profile_id="base", scope="global")
|
|
|
|
def resolve(
|
|
self,
|
|
group_profile: dict[str, Any] | None = None,
|
|
user_profile: dict[str, Any] | None = None,
|
|
session_profile: dict[str, Any] | None = None,
|
|
) -> SearchProfile:
|
|
profile = self.base_profile().to_dict()
|
|
for overlay in (group_profile, user_profile, session_profile):
|
|
if overlay:
|
|
profile = self._merge(profile, overlay)
|
|
profile["blocked_behaviors"] = set(HARD_BLOCKED_BEHAVIORS)
|
|
profile["audit_required"] = True
|
|
profile["fallback_required"] = True
|
|
return SearchProfile(**profile)
|
|
|
|
def _merge(self, base: dict[str, Any], overlay: dict[str, Any]) -> dict[str, Any]:
|
|
merged = dict(base)
|
|
if overlay.get("provider_order"):
|
|
merged["provider_order"] = list(overlay["provider_order"])
|
|
for list_key in ("trusted_domains", "query_expansion_hints", "domain_profiles"):
|
|
merged[list_key] = self._dedupe(list(merged.get(list_key, [])) + list(overlay.get(list_key, [])))
|
|
merged["profile_id"] = overlay.get("profile_id", merged.get("profile_id", "custom"))
|
|
merged["scope"] = overlay.get("scope", merged.get("scope", "custom"))
|
|
return merged
|
|
|
|
def _dedupe(self, items: list[str]) -> list[str]:
|
|
return list(dict.fromkeys([item for item in items if item]))
|
|
|
|
|
|
class DomainQueryProfileRegistry:
|
|
"""Optional domain-specific expansions for the general Search Gateway."""
|
|
|
|
DEFAULT_PROFILES = {
|
|
"education_exam_data": [
|
|
"学校 年级 科目 试卷 答案 成绩统计",
|
|
"联考 月考 期中 期末 质量检测",
|
|
],
|
|
"policy_regulation": [
|
|
"官方 通知 政策 解读 实施细则",
|
|
"site:gov.cn OR site:edu.cn",
|
|
],
|
|
"software_library": [
|
|
"release notes changelog migration guide",
|
|
"official documentation GitHub issues",
|
|
],
|
|
"product_research": [
|
|
"specs review comparison price availability",
|
|
"official page manual support",
|
|
],
|
|
"company_research": [
|
|
"official announcement press release filings",
|
|
"leadership product financial report",
|
|
],
|
|
}
|
|
|
|
def hints_for(self, domain_profiles: list[str]) -> list[str]:
|
|
hints: list[str] = []
|
|
for profile in domain_profiles:
|
|
hints.extend(self.DEFAULT_PROFILES.get(profile, []))
|
|
return list(dict.fromkeys(hints))
|
|
|
|
|
|
__all__ = [
|
|
"HARD_BLOCKED_BEHAVIORS",
|
|
"SearchProfile",
|
|
"SearchProfileResolver",
|
|
"DomainQueryProfileRegistry",
|
|
]
|