"""M9C:只读直连 pattern PG(open_aigc),判需求是否含分类树终端元素(叶子)。 叶子 = pattern_mining_category 中某 id 没有任何行 parent_id=该 id(同 execution)。 Gate 1:需求 itemset_items 的 category_id 里至少有一个叶子 → 放行;否则不做。 """ from __future__ import annotations from typing import Any from content_agent.errors import ContentAgentError, ErrorCode from content_agent.integrations.crawapi_http import _env, _load_env_file _LEAF_SQL = ( "SELECT 1 FROM pattern_mining_category c " "WHERE c.execution_id = %s AND c.id = ANY(%s) " "AND NOT EXISTS (SELECT 1 FROM pattern_mining_category ch " "WHERE ch.execution_id = c.execution_id AND ch.parent_id = c.id) LIMIT 1" ) class PatternPgClient: def __init__( self, *, host: str, port: int, user: str, password: str, database: str, timeout_seconds: float = 10.0, ) -> None: self.host = host self.port = port self.user = user self.password = password self.database = database self.timeout_seconds = timeout_seconds @classmethod def from_env(cls, env_path: str = ".env") -> "PatternPgClient": env = _load_env_file(env_path) return cls( host=_env("OPEN_AIGC_PG_HOST", env, required=True), port=int(_env("OPEN_AIGC_PG_PORT", env, default="5432")), user=_env("OPEN_AIGC_PG_USER", env, required=True), password=_env("OPEN_AIGC_PG_PASSWORD", env, required=True), database=_env("OPEN_AIGC_PG_DB_NAME", env, default="open_aigc"), ) def has_terminal_element(self, execution_id: int, category_ids: list[int]) -> bool: ids = [int(c) for c in category_ids if c is not None] if not ids: return True # 无可判定 category → 不阻断(交给后续判定) import pg8000.dbapi # 惰性导入:mock 测试用 fake client,不触发依赖 try: conn = pg8000.dbapi.connect( host=self.host, port=self.port, user=self.user, password=self.password, database=self.database, ssl_context=None, # 服务器拒绝 SSL,必须明确关闭 timeout=self.timeout_seconds, ) except Exception as exc: # PG 不可达:明确报错,不静默放过 raise ContentAgentError( ErrorCode.DB_CONNECTION_FAILED, "pattern pg unreachable for gate1 terminal-element check", {"exception_type": type(exc).__name__}, ) from exc try: cur = conn.cursor() cur.execute(_LEAF_SQL, (int(execution_id), ids)) return cur.fetchone() is not None finally: conn.close()