feat(M9): 新增 50+ 子分纯函数 + pattern PG 只读叶子判定 client

- fifty_plus.py: 作者画像 → 50+ 子分(band 41+, TGI 6成+占比4成, TGI 200 满分)。
- pattern_pg.py: pg8000 只读连 open_aigc, 判 category 是否分类树叶子(Gate 1)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

content_agent/business_modules/content_discovery/fifty_plus.py

@@ -0,0 +1,67 @@
+"""M9A:抖音 50+ 受众子分(作者粉丝画像 → 0-100)。
+
+口径(已拍板):band=41+(41-50 + 50-=50 岁及以上);
+TGI_score=clamp(bandTGI/2,0,100)(TGI 200 满分);percent_score=clamp(band占比,0,100);
+50+子分=0.6·TGI_score+0.4·percent_score。TGI=画像 preference,占比=画像 percentage,同一接口。
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+SCHEMA_VERSION = "v4_fifty_plus.v1"
+BAND_KEYS = ("41-50", "50-")  # 50- = 50 岁及以上
+
+
+def fifty_plus_score(fans_age_data: dict[str, Any] | None) -> dict[str, Any]:
+    """输入 = 画像 fans.age.data(逐桶 {percentage, preference});输出 50+ 子分 dict。
+
+    取不到任一有效 band 桶(缺桶 / 缺 percentage / 缺 preference)→ status="incomplete"。
+    """
+    buckets = fans_age_data if isinstance(fans_age_data, dict) else {}
+    pairs: list[tuple[float, float]] = []
+    for key in BAND_KEYS:
+        bucket = buckets.get(key)
+        if not isinstance(bucket, dict):
+            continue
+        pct = _to_float(bucket.get("percentage"))
+        tgi = _to_float(bucket.get("preference"))
+        if pct is None or tgi is None:
+            continue
+        pairs.append((pct, tgi))
+
+    if not pairs:
+        return {"schema_version": SCHEMA_VERSION, "status": "incomplete", "band": "41+"}
+
+    band_percent = sum(pct for pct, _ in pairs)
+    weight = band_percent if band_percent > 0 else float(len(pairs))
+    band_tgi = sum(pct * tgi for pct, tgi in pairs) / weight
+    tgi_score = _clamp(band_tgi / 2)
+    percent_score = _clamp(band_percent)
+    score = round(0.6 * tgi_score + 0.4 * percent_score, 2)
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "status": "ok",
+        "score": score,
+        "band": "41+",
+        "band_tgi": round(band_tgi, 2),
+        "band_percent": round(band_percent, 2),
+        "tgi_score": round(tgi_score, 2),
+        "percent_score": round(percent_score, 2),
+    }
+
+
+def _to_float(value: Any) -> float | None:
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str):
+        text = value.strip().rstrip("%").strip()
+        try:
+            return float(text)
+        except ValueError:
+            return None
+    return None
+
+
+def _clamp(value: float, low: float = 0.0, high: float = 100.0) -> float:
+    return max(low, min(high, value))

content_agent/integrations/pattern_pg.py

@@ -0,0 +1,78 @@
+"""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()