init

pyproject.toml

@@ -9,6 +9,7 @@ description = "Agentic 架构基础设施供给层 —— 数据库连接池、
 requires-python = ">=3.11"
 dependencies = [
     "aiomysql>=0.2.0",
+    "asyncpg>=0.29",
     "aiohttp>=3.9",
     "pydantic>=2.0",
     "pydantic-settings>=2.0",

src/__init__.py

@@ -60,8 +60,8 @@ from src.domains import (
     FetchDemands,
     EnqueueDemands,
     DemandQueueMapper,
-    Article,
-    ArticleService,
+    DemandSearchArticle,
+    ArticleFetchDetail,
     Account,
     AccountService,
 )
@@ -102,8 +102,8 @@ __all__ = [
     "FetchDemands",
     "EnqueueDemands",
     "DemandQueueMapper",
-    "Article",
-    "ArticleService",
+    "DemandSearchArticle",
+    "ArticleFetchDetail",
     "Account",
     "AccountService",
     # Server

src/config/_global_config.py

@@ -10,6 +10,7 @@ from .database import (
     PiaoquanCrawlerDatabaseConfig,
     GrowthDatabaseConfig,
     ContentDeconstructionSupplyConfig,
+    PgConfig,
 )
 from .aliyun import AliyunLogConfig, AliyunOssConfig
 from .xxljob import XxlJobConfig
@@ -56,6 +57,7 @@ class GlobalConfig(BaseSettings):
     content_decode_db: ContentDeconstructionSupplyConfig = Field(
         default_factory=ContentDeconstructionSupplyConfig
     )
+    pg: PgConfig = Field(default_factory=PgConfig)
 
     # ============ 阿里云服务配置 ============
     aliyun_log: AliyunLogConfig = Field(default_factory=AliyunLogConfig)
@@ -89,4 +91,5 @@ class GlobalConfig(BaseSettings):
             "aigc": self.aigc_db,
             "long_articles": self.long_articles_db,
             "piaoquan_crawler": self.piaoquan_crawler_db,
+            "pg": self.pg,
         }

src/config/database.py

@@ -131,11 +131,11 @@ class ContentDeconstructionSupplyConfig(DatabaseConfig):
 class PgConfig(BaseSettings):
     """PostgreSQL 连接配置"""
 
-    host: str = "127.0.0.1"
+    host: str = "pgm-bp1x72iry10srsc2.pg.rds.aliyuncs.com"
     port: int = 5432
-    user: str = "postgres"
-    password: str = ""
-    db: str = "postgres"
+    user: str = "vector"
+    password: str = "vector123456@"
+    db: str = "vector"
     minsize: int = 5
     maxsize: int = 20
     backend: str = "postgresql"

src/domains/__init__.py

@@ -5,26 +5,27 @@
   domains: 本项目专属的业务逻辑和数据模型，依赖 core 和 infra
 
 子域:
-  demand   — 需求获取与管理（外部需求源 → 统一需求模型）
-  article  — 文章抓取与处理（需求驱动 → 文章采集 → 内容结构化）
-  account  — 账号识别与管理（文章反推账号 → 账号画像 → 采集策略）
+  fetch_demand   — 需求获取与管理（外部需求源 → 统一需求模型）
+  demand_search_article  — 文章抓取与处理（需求驱动 → 文章采集 → 内容结构化）
+  demand_search_account  — 账号识别与管理（文章反推账号 → 账号画像 → 采集策略）
 
-数据流: demand → article → account（下游消费上游产出）
+数据流: fetch_demand → demand_search_article → demand_search_account（下游消费上游产出）
 """
 
-from src.domains.demand import FetchDemands, EnqueueDemands, DemandQueueMapper
-from src.domains.article import Article, ArticleService
-from src.domains.account import Account, AccountService
+from src.domains.fetch_demand import FetchDemands, EnqueueDemands, DemandQueueMapper, VideoDeconstructMapper
+from src.domains.demand_search_article import DemandSearchArticle, ArticleFetchDetail
+from src.domains.demand_search_account import Account, AccountService
 
 __all__ = [
-    # Demand
+    # fetch_demand
     "FetchDemands",
     "EnqueueDemands",
     "DemandQueueMapper",
-    # Article
-    "Article",
-    "ArticleService",
-    # Account
+    "VideoDeconstructMapper",
+    # demand_search_article
+    "DemandSearchArticle",
+    "ArticleFetchDetail",
+    # demand_search_account
     "Account",
     "AccountService",
 ]

src/domains/article/__init__.py

@@ -1,16 +0,0 @@
-"""文章领域 —— 基于需求驱动文章抓取与内容结构化
-
-核心流程:
-  1. 接收 demand 领域产出的 Demand
-  2. 通过爬虫（微信公众号等）采集文章
-  3. 内容清洗、去重、结构化存储
-  4. 产出文章中的账号信息，供 account 领域消费
-"""
-
-from src.domains.article.models import Article
-from src.domains.article.service import ArticleService
-
-__all__ = [
-    "Article",
-    "ArticleService",
-]

src/domains/article/models.py

@@ -1,20 +0,0 @@
-"""文章领域模型"""
-
-from dataclasses import dataclass, field
-from datetime import datetime
-
-
-@dataclass
-class Article:
-    """抓取到的文章统一模型"""
-    title: str
-    url: str
-    content: str = ""
-    summary: str = ""
-    author: str = ""                                # 作者名
-    account_id: str = ""                            # 公众号 ID（供 account 领域消费）
-    account_name: str = ""                          # 公众号名称
-    publish_time: datetime | None = None
-    source_platform: str = "wechat"                 # 来源平台: wechat / toutiao / ...
-    raw_metadata: dict = field(default_factory=dict)
-    crawled_at: datetime = field(default_factory=datetime.now)

src/domains/article/service.py

@@ -1,36 +0,0 @@
-"""文章抓取与处理服务"""
-
-import logging
-from typing import AsyncIterator
-
-from src.domains.article.models import Article
-
-logger = logging.getLogger(__name__)
-
-
-class ArticleService:
-    """文章抓取服务
-
-    职责: 接收 demand dict（来自 FetchDemands.deal()），驱动爬虫采集文章，产出 Article。
-    不负责: 需求的管理（由 demand 领域负责）、账号的后续管理（由 account 领域负责）。
-
-    demand dict keys: search_terms, dt, channel_name, channel_level3,
-                      config_code, point_type, category_name, demand_type,
-                      demand_topic, match_video_id, experiment_id, match_score
-    """
-
-    async def crawl_by_demand(self, demand: dict, max_count: int = 20) -> list[Article]:
-        """根据需求关键词抓取文章"""
-        raise NotImplementedError
-
-    async def crawl_by_account(self, account_id: str, max_count: int = 10) -> list[Article]:
-        """根据账号 ID 抓取该账号的历史文章"""
-        raise NotImplementedError
-
-    async def deduplicate(self, articles: list[Article]) -> list[Article]:
-        """按 URL / 内容指纹去重"""
-        raise NotImplementedError
-
-    async def crawl_and_deduplicate(self, demand: dict) -> AsyncIterator[Article]:
-        """端到端：抓取 + 去重 + 逐条产出"""
-        raise NotImplementedError

src/domains/demand/__init__.py

@@ -1,19 +0,0 @@
-"""需求域 —— 从 ODPS 提取需求 → 按账号聚合 → 入搜索队列
-
-核心流程:
-  1. FetchDemands     — ODPS 查询 → 搜索词提取 → 按账号聚合
-  2. EnqueueDemands   — 聚合结果写入 MySQL demand_search_queue
-  3. DemandQueueMapper — 队列表纯 DB 读写（供 article 域消费时复用）
-"""
-
-from ._const import DemandConst
-from .fetch_demands import FetchDemands
-from .enqueue_demands import EnqueueDemands
-from ._mapper import DemandQueueMapper
-
-__all__ = [
-    "DemandConst",
-    "FetchDemands",
-    "EnqueueDemands",
-    "DemandQueueMapper",
-]

src/domains/demand/enqueue_demands.py

@@ -1,78 +0,0 @@
-"""需求入队服务 —— 将聚合后的需求写入 MySQL 搜索队列"""
-
-import logging
-from typing import Dict, List
-
-from src.infra.database.mysql.manager import MysqlManager
-
-from ._const import DemandConst
-from ._mapper import DemandQueueMapper
-
-logger = logging.getLogger(__name__)
-
-
-class EnqueueDemands(DemandConst):
-    """将 demand dict 列表转为队列条目并批量写入 MySQL
-
-    使用方式:
-        mysql = db.mysql_manager("long_articles")
-        enqueuer = EnqueueDemands(mysql)
-        n = await enqueuer.deal(demands)
-    """
-
-    def __init__(self, pool: MysqlManager):
-        self.pool = pool
-        self.mapper = DemandQueueMapper(self.pool)
-
-    # ═══════════════════════════════════════════════════════════════
-    # 入口
-    # ═══════════════════════════════════════════════════════════════
-
-    async def deal(self, demands: List[Dict]) -> int:
-        """批量写入搜索队列，返回总写入行数"""
-        if not demands:
-            logger.info("无需求待入队")
-            return 0
-        items = [self._to_queue_item(d) for d in demands]
-        n = await self.mapper.insert_batch(items)
-        logger.info("入队完成: %d 条写入 demand_search_queue", n)
-        return n
-
-    # ═══════════════════════════════════════════════════════════════
-    # 内部
-    # ═══════════════════════════════════════════════════════════════
-
-    @staticmethod
-    def _to_queue_item(demand: Dict) -> Dict:
-        """demand dict → 队列表 INSERT 参数 dict
-
-        DDL 列映射:
-          - 搜索词三列: standard_element, match_text, match_generalized_element
-          - 高频过滤列: channel_name, category_name, match_config_code（拆列 + 索引）
-          - 余下的 point_type, demand_type, demand_topic, demand_id 等归入 demand_info JSON
-        """
-        return {
-            "account": demand["channel_level3"],
-            "dt": demand["dt"],
-            "match_video_id": demand["match_video_id"],
-            # 高频过滤列（从 ODPS 原始值直接写入，支持按渠道/分类/配置码查询）
-            "channel_name": demand.get("channel_name", ""),
-            "category_name": demand.get("category_name", ""),
-            "match_config_code": demand.get("config_code", ""),
-            "match_score": demand["match_score"],
-            # 搜索词原始列（保留 ODPS 三个源列的原始值）
-            "standard_element": demand.get("standard_element", ""),
-            "match_text": demand.get("match_text", ""),
-            "match_generalized_element": demand.get("match_generalized_element", ""),
-            # demand_info JSON（未拆列的剩余字段）
-            "demand_info": {
-                "point_type": demand.get("point_type", ""),
-                "demand_type": demand.get("demand_type", ""),
-                "demand_topic": demand.get("demand_topic", ""),
-                "demand_id": demand.get("demand_id", ""),
-            },
-            "experiment_id": demand.get("experiment_id", ""),
-        }
-
-
-__all__ = ["EnqueueDemands"]

src/domains/demand/fetch_demands.py

@@ -1,256 +0,0 @@
-"""从 ODPS 拉取需求 → 提取搜索词 → 按账号聚合"""
-
-import asyncio
-import logging
-from collections import defaultdict
-from typing import Dict, List
-
-from src.config.odps import OdpsConfig
-from src.infra.external.odps import fetch_from_odps
-
-from ._const import DemandConst
-
-logger = logging.getLogger(__name__)
-
-# ODPS SELECT 列清单
-_SELECT_COLUMNS = [
-    "id",                         # 表自增主键
-    "dt",                         # 数据日期(yyyyMMdd)
-    "channel_name",               # 渠道类型
-    "channel_level3",             # 账号名称（聚合维度）
-    "match_config_code",          # 匹配配置编码
-    "standard_element",           # 标准化元素（搜索词，优先级最高）
-    "match_text",                 # 匹配文本（搜索词，优先级次之）
-    "match_generalized_element",  # 泛化元素（搜索词，优先级最低）
-    "point_type",                 # 点类型
-    "element_dimension",          # 元素维度
-    "category_name",              # 分类名称
-    "demand_type",                # 需求类型
-    "demand_content_topic",       # 需求内容选题
-    "demand_id",                  # 需求ID
-    "match_video_id",             # 匹配视频ID
-    "match_score",                # 匹配得分
-    "demand_strategy",            # 需求策略
-    "experiment_id",              # 实验批次ID
-]
-_SELECT_CLAUSE = ", ".join(_SELECT_COLUMNS)
-
-
-class FetchDemands(DemandConst):
-    """从 ODPS 拉取需求并聚合
-
-    使用方式:
-        config = OdpsConfig()
-        fetcher = FetchDemands(config)
-        demands = await fetcher.deal("20260701")
-    """
-
-    def __init__(self, config: OdpsConfig):
-        self._odps_config = config
-
-    # ═══════════════════════════════════════════════════════════════
-    # 入口
-    # ═══════════════════════════════════════════════════════════════
-
-    async def deal(self, dt: str) -> List[Dict]:
-        """查询指定日期的需求，提取搜索词，按账号聚合 top N"""
-        rows = await self._query_odps(dt)
-        demands = [self._row_to_demand(r) for r in rows]
-        # 两步过滤: 空账号 → 非白名单账号
-        valid, dropped_empty = self._filter_empty_account(demands)
-        valid_accounts = await self._load_valid_accounts()
-        filtered, dropped_unknown = self._filter_unknown_accounts(valid, valid_accounts)
-        total_dropped = dropped_empty + dropped_unknown
-        if total_dropped:
-            logger.warning(
-                "丢弃 %d 行 (空账号 %d + 非白名单账号 %d), dt=%s",
-                total_dropped, dropped_empty, dropped_unknown, dt,
-            )
-        logger.info(
-            "ODPS 查询完成: dt=%s, 原始 %d 行 → 有效 %d 行, 丢弃 %d 行",
-            dt, len(demands), len(filtered), total_dropped,
-        )
-        aggregated = self._aggregate(filtered)
-        return aggregated
-
-    # ═══════════════════════════════════════════════════════════════
-    # ODPS 查询
-    # ═══════════════════════════════════════════════════════════════
-
-    async def _query_odps(self, dt: str) -> List[Dict]:
-        """执行 ODPS SQL，返回原始 dict 列表"""
-        sql = (
-            f"SELECT {_SELECT_CLAUSE} "
-            f"FROM public_channel_demand_match_result "
-            f"WHERE dt = '{dt}' "
-            f"  AND channel_name IN {self._format_tuple(self.CHANNEL_NAMES)} "
-            f"  AND match_status = {self.MATCH_STATUS} "
-            f"  AND match_config_code IN {self._format_tuple(self.MATCH_CONFIG_CODES)} "
-            f"  AND element_dimension = '{self.ELEMENT_DIMENSION}' "
-            f"  AND match_score > 0.1 "
-            f"ORDER BY match_score DESC"
-        )
-        logger.info("ODPS 查询需求, dt=%s", dt)
-        return await asyncio.to_thread(fetch_from_odps, sql, self._odps_config)
-
-    async def _load_valid_accounts(self) -> set[str]:
-        """从两张账号基础表加载白名单，仅取 account_status='开' 的账号
-
-        两张表:
-          - feishu_wechat_mp_account_base  (公众号投流)
-          - feishu_wechat_fwh_account_base (服务号投流)
-        """
-        valid: set[str] = set()
-        for table in self.VALID_ACCOUNT_TABLES.values():
-            sql = (
-                f"SELECT account_name "
-                f"FROM loghubods.{table} "
-                f"WHERE account_status = '开'"
-            )
-            rows = await asyncio.to_thread(fetch_from_odps, sql, self._odps_config)
-            names = {r["account_name"] for r in rows if r.get("account_name")}
-            valid.update(names)
-            logger.info("白名单加载: %s → %d 个有效账号", table, len(names))
-        logger.info("白名单合计: %d 个有效账号", len(valid))
-        return valid
-
-    # ═══════════════════════════════════════════════════════════════
-    # 行映射
-    # ═══════════════════════════════════════════════════════════════
-
-    @staticmethod
-    def _row_to_demand(row: Dict) -> Dict:
-        """ODPS row → 需求 dict
-
-        search_terms: 去重后的搜索词列表，供下游 ArticleService 消费
-        standard_element / match_text / match_generalized_element:
-            原始列值，供 mapper 写入 demand_search_queue 对应列
-        """
-        return {
-            # 搜索词（两套表示，用途不同）
-            "search_terms": FetchDemands._extract_search_terms(row),
-            "standard_element": row.get("standard_element") or "",
-            "match_text": row.get("match_text") or "",
-            "match_generalized_element": row.get("match_generalized_element") or "",
-            # 业务维度
-            "dt": row.get("dt") or "",
-            "channel_name": row.get("channel_name") or "",
-            "channel_level3": row.get("channel_level3") or "",
-            "config_code": row.get("match_config_code") or "",
-            "point_type": row.get("point_type") or "",
-            "category_name": row.get("category_name") or "",
-            "demand_type": row.get("demand_type") or "",
-            "demand_topic": row.get("demand_content_topic") or "",
-            "demand_id": row.get("demand_id") or "",
-            # 匹配信息
-            "match_video_id": int(row.get("match_video_id") or 0),
-            "experiment_id": row.get("experiment_id") or "",
-            "match_score": float(row.get("match_score") or 0),
-            # 元数据
-            "row_id": row.get("id") or 0,
-        }
-
-    @staticmethod
-    def _filter_empty_account(demands: List[Dict]) -> tuple[List[Dict], int]:
-        """过滤 channel_level3 为空的无效行
-
-        Returns:
-            (valid_demands, dropped_count)
-        """
-        valid = [d for d in demands if d["channel_level3"]]
-        return valid, len(demands) - len(valid)
-
-    @staticmethod
-    def _filter_unknown_accounts(
-        demands: List[Dict],
-        valid_accounts: set[str],
-    ) -> tuple[List[Dict], int]:
-        """过滤不在白名单中的账号
-
-        Returns:
-            (valid_demands, dropped_count)
-        """
-        if not valid_accounts:
-            return demands, 0
-        filtered: list[dict] = []
-        dropped: list[str] = []
-        for d in demands:
-            if d["channel_level3"] in valid_accounts:
-                filtered.append(d)
-            else:
-                dropped.append(d["channel_level3"])
-        if dropped:
-            unique_dropped = set(dropped)
-            logger.warning(
-                "非白名单账号 (共 %d 个, %d 行): %s",
-                len(unique_dropped), len(dropped), sorted(unique_dropped),
-            )
-        return filtered, len(dropped)
-
-    @staticmethod
-    def _extract_search_terms(row: Dict) -> List[str]:
-        """从三列搜索词中提取、去重（case-insensitive），保持优先级顺序"""
-        seen: set[str] = set()
-        terms: list[str] = []
-        for col in DemandConst.SearchTermColumn.all_columns():
-            val = row.get(col)
-            if not val:
-                continue
-            text = str(val).strip()
-            if not text:
-                continue
-            key = text.lower()
-            if key not in seen:
-                seen.add(key)
-                terms.append(text)
-        return terms
-
-    # ═══════════════════════════════════════════════════════════════
-    # 聚合
-    # ═══════════════════════════════════════════════════════════════
-
-    def _aggregate(
-        self,
-        demands: List[Dict],
-        max_per_account: int | None = None,
-    ) -> List[Dict]:
-        """按 channel_level3 分组，每组取 match_score 最高的 top N
-
-        入参已按 match_score DESC 排序，直接取每组头 N 个即可。
-        """
-        limit = max_per_account or self.MAX_PER_ACCOUNT_PER_DAY
-        buckets: dict[str, list[dict]] = defaultdict(list)
-        for d in demands:
-            key = d["channel_level3"]
-            if len(buckets[key]) < limit:
-                buckets[key].append(d)
-
-        result: list[dict] = []
-        for account, items in sorted(buckets.items()):
-            result.extend(items)
-            logger.info(
-                "聚合账号 %s: 保留 %d/%d 条 (top score=%.4f)",
-                account,
-                len(items),
-                sum(1 for d in demands if d["channel_level3"] == account),
-                items[0]["match_score"] if items else 0,
-            )
-
-        logger.info(
-            "聚合完成: %d 条 → %d 条 (%d 个账号, max=%d/账号)",
-            len(demands), len(result), len(buckets), limit,
-        )
-        return result
-
-    # ═══════════════════════════════════════════════════════════════
-    # 工具
-    # ═══════════════════════════════════════════════════════════════
-
-    @staticmethod
-    def _format_tuple(values: tuple) -> str:
-        """('a', 'b') → "('a', 'b')" """
-        quoted = ", ".join(f"'{v}'" for v in values)
-        return f"({quoted})"
-
-
-__all__ = ["FetchDemands"]

src/domains/account/__init__.py → src/domains/demand_search_account/__init__.py

@@ -1,14 +1,14 @@
 """账号领域 —— 从抓取到的文章中识别和管理内容账号
 
 核心流程:
-  1. 接收 article 领域产出中的账号信息
+  1. 接收 demand_search_article 领域产出中的账号信息
   2. 账号去重、画像构建（领域、活跃度、质量评分）
   3. 输出采集策略建议（哪些账号值得持续跟踪）
-  4. 反哺 demand 领域：高质量账号可作为新的需求来源
+  4. 反哺 fetch_demand 领域：高质量账号可作为新的需求来源
 """
 
-from src.domains.account.models import Account
-from src.domains.account.service import AccountService
+from src.domains.demand_search_account.models import Account
+from src.domains.demand_search_account.service import AccountService
 
 __all__ = [
     "Account",

src/domains/account/service.py → src/domains/demand_search_account/service.py

@@ -1,9 +1,24 @@
 """账号管理服务"""
 
 import logging
+from dataclasses import dataclass, field
+from datetime import datetime
 
-from src.domains.article.models import Article
-from src.domains.account.models import Account
+from src.domains.demand_search_account.models import Account
+
+
+@dataclass
+class Article:
+    """文章内存对象 —— 从 demand_search_article_detail 结构化后供 account 域消费"""
+    channel_content_id: str
+    url: str
+    title: str = ""
+    body_text: str = ""
+    channel_account_id: str = ""
+    channel_account_name: str = ""
+    publish_at: datetime | None = None
+    source_platform: str = "wechat"
+    extra: dict = field(default_factory=dict)
 
 logger = logging.getLogger(__name__)
 
@@ -12,7 +27,7 @@ class AccountService:
     """账号管理服务
 
     职责: 从 Article 中提取账号信息，去重合并，构建账号画像，输出采集策略。
-    不负责: 文章的抓取（由 article 领域负责）。
+    不负责: 文章的抓取（由 demand_search_article 领域负责）。
     """
 
     async def extract_account_from_article(self, article: Article) -> Account | None:

src/domains/demand_search_article/__init__.py

@@ -0,0 +1,29 @@
+"""文章搜索领域 —— 基于需求驱动文章搜索与内容结构化
+
+Phase 1 — 搜索:
+  1. 从 demand_search_queue 取搜索词
+  2. 调微信搜索 API
+  3. 写入 demand_search_article_relation (status=0)
+
+Phase 2 — 拉详情:
+  1. 从 demand_search_article_relation(status=0) 取待拉项
+  2. 调微信详情 API
+  3. 写入 demand_search_article_detail (channel_content_id 去重)
+  4. 回填 relation 状态
+"""
+
+from src.domains.demand_search_article._const import DemandSearchArticleConst
+from src.domains.demand_search_article._mapper import (
+    ArticleSearchRelationMapper,
+    ArticleDetailMapper,
+)
+from src.domains.demand_search_article.article_search import DemandSearchArticle
+from src.domains.demand_search_article.article_fetch_detail import ArticleFetchDetail
+
+__all__ = [
+    "DemandSearchArticleConst",
+    "ArticleSearchRelationMapper",
+    "ArticleDetailMapper",
+    "DemandSearchArticle",
+    "ArticleFetchDetail",
+]

src/domains/fetch_demand/__init__.py

@@ -0,0 +1,20 @@
+"""需求域 —— 从 ODPS 提取白名单账号匹配视频 → 解构 → 搜索词入队
+
+核心流程:
+  1. FetchDemands     — 白名单加载 → 匹配视频查询 → rov 去重 → 视频解构 → 搜索词展开
+  2. EnqueueDemands   — 队列条目批量写入 MySQL demand_search_queue
+  3. DemandQueueMapper — 队列表纯 DB 读写（供 demand_search_article 域消费时复用）
+"""
+
+from ._const import DemandConst
+from .fetch_demands import FetchDemands
+from .enqueue_demands import EnqueueDemands
+from ._mapper import DemandQueueMapper, VideoDeconstructMapper
+
+__all__ = [
+    "DemandConst",
+    "FetchDemands",
+    "EnqueueDemands",
+    "DemandQueueMapper",
+    "VideoDeconstructMapper",
+]

src/domains/demand/_const.py → src/domains/fetch_demand/_const.py

@@ -5,7 +5,7 @@ class DemandConst:
     """需求域所有常量的聚合根，服务类继承此类即可通过 self. 访问"""
 
     # ═══════════════════════════════════════════════════════════════
-    # ODPS 查询过滤条件
+    # ODPS / 数据源
     # ═══════════════════════════════════════════════════════════════
 
     DEFAULT_DT = "20260501"
@@ -18,41 +18,25 @@ class DemandConst:
         "服务号投流": "feishu_wechat_fwh_account_base",
     }
 
-    MATCH_STATUS = 1  # 1 = 已匹配
-
-    MATCH_CONFIG_CODES = (
-        "VIDEO_TITLE",
-        "VIDEO_INSPIRATION",
-        "VIDEO_KEYPOINT",
-    )
-
-    ELEMENT_DIMENSION = "实质"
-
-    # ═══════════════════════════════════════════════════════════════
-    # ODPS 查询列
-    # ═══════════════════════════════════════════════════════════════
-
-    class SearchTermColumn:
-        """搜索词来源列（按优先级降序）"""
-        STANDARD_ELEMENT = "standard_element"
-        MATCH_TEXT = "match_text"
-        MATCH_GENERALIZED_ELEMENT = "match_generalized_element"
-
-        @classmethod
-        def all_columns(cls) -> tuple:
-            return cls.STANDARD_ELEMENT, cls.MATCH_TEXT, cls.MATCH_GENERALIZED_ELEMENT
+    # 匹配视频来源表
+    MATCH_RESULT_TABLE = "public_channel_demand_match_result"
 
     # ═══════════════════════════════════════════════════════════════
-    # 聚合配置
+    # 搜索词类型（video 解构驱动）
     # ═══════════════════════════════════════════════════════════════
 
-    MAX_PER_ACCOUNT_PER_DAY = 2000
+    class KeyType:
+        INSPIRATION_SUBSTANCE = "INSPIRATION_SUBSTANCE"
+        VIDEO_INSPIRATION =    "VIDEO_INSPIRATION"
+        VIDEO_KEYPOINT =       "VIDEO_KEYPOINT"
+        VIDEO_TITLE =          "VIDEO_TITLE"
 
     # ═══════════════════════════════════════════════════════════════
-    # 搜索队列表
+    # 队列
     # ═══════════════════════════════════════════════════════════════
 
     QUEUE_TABLE = "demand_search_queue"
+    DRIVE_TYPE_VIDEO = "VIDEO"
 
     class QueueStatus:
         INIT = 0        # 初始，待处理
@@ -61,4 +45,4 @@ class DemandConst:
         FAIL = 99       # 处理失败
 
 
-__all__ = ["DemandConst"]
+__all__ = ["DemandConst"]

src/domains/demand/_mapper.py → src/domains/fetch_demand/_mapper.py

@@ -1,9 +1,9 @@
-"""需求搜索队列 MySQL 读写 —— 仅负责 DB I/O，不含业务逻辑"""
+"""需求搜索队列 MySQL 读写 + 视频搜索词 PG 查询 —— 仅负责 DB I/O，不含业务逻辑"""
 
-import json
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 from src.infra.database.mysql.manager import MysqlManager
+from src.infra.database.postgresql.manager import PgManager
 
 from ._const import DemandConst
 
@@ -26,48 +26,39 @@ class DemandQueueMapper(DemandConst):
         """批量写入队列，返回受影响行数
 
         items 每条须包含:
-          account, dt, match_video_id, channel_name, category_name,
-          match_config_code, match_score, standard_element, match_text,
-          match_generalized_element, demand_info (dict), experiment_id
+          account, search_key, dt, video_id, channel_name,
+          key_type, experiment_id, drive_type
         """
         if not items:
             return 0
         query = f"""
             INSERT IGNORE INTO {self.QUEUE_TABLE}
-                (account, dt, match_video_id,
-                 channel_name, category_name, match_config_code,
-                 match_score,
-                 standard_element, match_text, match_generalized_element,
-                 demand_info, experiment_id, status)
+                (account, search_key, dt, video_id,
+                 channel_name, key_type, experiment_id,
+                 status, drive_type)
             VALUES
-                (%s, %s, %s,
+                (%s, %s, %s, %s,
                  %s, %s, %s,
-                 %s,
-                 %s, %s, %s,
-                 %s, %s, %s)
+                 %s, %s)
         """
         params = [
             (
                 it["account"],
+                it["search_key"],
                 it["dt"],
-                it["match_video_id"],
+                it["video_id"],
                 it["channel_name"],
-                it["category_name"],
-                it["match_config_code"],
-                it["match_score"],
-                it["standard_element"],
-                it["match_text"],
-                it["match_generalized_element"],
-                json.dumps(it["demand_info"], ensure_ascii=False),
+                it["key_type"],
                 it["experiment_id"],
                 self.QueueStatus.INIT,
+                it["drive_type"],
             )
             for it in items
         ]
         return await self.pool.save(query=query, params=params, batch=True)
 
     # ═══════════════════════════════════════════════════════════════
-    # 读取
+    # 读取（供下游 consumer 使用）
     # ═══════════════════════════════════════════════════════════════
 
     async def fetch_pending(
@@ -77,12 +68,12 @@ class DemandQueueMapper(DemandConst):
         limit: int = 500,
         offset: int = 0,
     ) -> List[Dict]:
-        """查询待搜索的队列项，按 match_score 降序"""
+        """查询待搜索的队列项（按 dt）"""
         query = f"""
             SELECT *
             FROM {self.QUEUE_TABLE}
             WHERE dt = %s AND status = %s
-            ORDER BY match_score DESC
+            ORDER BY id ASC
             LIMIT %s OFFSET %s
         """
         return await self.pool.fetch(
@@ -115,6 +106,21 @@ class DemandQueueMapper(DemandConst):
             params=(self.QueueStatus.PROCESSING, item_id, self.QueueStatus.INIT),
         )
 
+    async def mark_processing_batch(self, ids: List[int]) -> int:
+        """批量标记为处理中"""
+        if not ids:
+            return 0
+        placeholders = ",".join(["%s"] * len(ids))
+        query = f"""
+            UPDATE {self.QUEUE_TABLE}
+            SET status = %s
+            WHERE id IN ({placeholders})
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.QueueStatus.PROCESSING, *ids),
+        )
+
     async def mark_done(self, item_id: int) -> int:
         query = f"""
             UPDATE {self.QUEUE_TABLE}
@@ -137,4 +143,41 @@ class DemandQueueMapper(DemandConst):
         )
 
 
-__all__ = ["DemandQueueMapper"]
+class VideoDeconstructMapper:
+    """video_vectors 表的数据访问层（PostgreSQL）
+
+    职责: 按 video_id 查询搜索词，每行 (config_code, text) 为一个搜索词。
+    config_code 直接对应 DemandConst.KeyType。
+    """
+
+    COLUMNS = ("video_id", "config_code", "text")
+
+    def __init__(self, pg: PgManager):
+        self.pg = pg
+
+    async def find_by_video_id(self, video_id: int) -> Optional[list[Dict]]:
+        """按 video_id 查询所有搜索词
+
+        Returns:
+            [
+                {"config_code": "VIDEO_KEYPOINT", "text": "关键点文本"},
+                {"config_code": "VIDEO_INSPIRATION", "text": "灵感点文本"},
+                ...
+            ]
+            无结果返回空 list（非 None，统一语义）。
+        """
+        cols = ", ".join(self.COLUMNS)
+        rows = await self.pg.fetch(
+            f"SELECT {cols} FROM video_vectors WHERE video_id = $1",
+            params=(video_id,),
+        )
+        result: list[Dict] = []
+        for row in rows:
+            config_code = (row.get("config_code") or "").strip()
+            text_val = (row.get("text") or "").strip()
+            if config_code and text_val:
+                result.append({"config_code": config_code, "text": text_val})
+        return result
+
+
+__all__ = ["DemandQueueMapper", "VideoDeconstructMapper"]

src/domains/fetch_demand/enqueue_demands.py

@@ -0,0 +1,52 @@
+"""需求入队服务 —— 将 FetchDemands 产出的队列条目批量写入 MySQL"""
+
+import logging
+from typing import Dict, List
+
+from src.infra.database.mysql.manager import MysqlManager
+
+from ._const import DemandConst
+from ._mapper import DemandQueueMapper
+
+logger = logging.getLogger(__name__)
+
+
+class EnqueueDemands(DemandConst):
+    """将 fetch_demand 产出的队列条目批量写入 demand_search_queue
+
+    使用方式:
+        mysql = db.mysql_manager("long_articles")
+        enqueuer = EnqueueDemands(mysql)
+        n = await enqueuer.deal(items)
+    """
+
+    def __init__(self, pool: MysqlManager):
+        self.pool = pool
+        self.mapper = DemandQueueMapper(self.pool)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 入口
+    # ═══════════════════════════════════════════════════════════════
+
+    async def deal(self, items: List[Dict]) -> int:
+        """批量写入搜索队列，返回总写入行数
+
+        items 每条须包含:
+          account, search_key, dt, video_id, channel_name,
+          key_type, experiment_id, drive_type
+        """
+        if not items:
+            logger.info("无队列项待入队")
+            return 0
+        n = await self.mapper.insert_batch(items)
+        # 统计概要
+        accounts = len(set(it["account"] for it in items))
+        videos = len(set(it["video_id"] for it in items))
+        logger.info(
+            "入队完成: %d 条, %d 个账号, %d 个视频 → demand_search_queue",
+            n, accounts, videos,
+        )
+        return n
+
+
+__all__ = ["EnqueueDemands"]

src/domains/fetch_demand/fetch_demands.py

@@ -0,0 +1,257 @@
+"""从 ODPS 拉取白名单账号的匹配视频 → 从 video_vectors 取搜索词 → 入队
+
+核心流程:
+  1. 加载白名单账号（两张 ODPS 账号基础表）
+  2. 查询白名单账号的匹配视频（按 rov 去重：同一账号+视频取 rov 最高者）
+  3. 逐视频从 PG video_vectors 查询搜索词（config_code → KeyType, text → search_key）
+  4. 将搜索词展开为队列条目（每条一个搜索词）
+"""
+
+import asyncio
+import logging
+from typing import Dict, List, Set
+
+from src.config.odps import OdpsConfig
+from src.infra.external.odps import fetch_from_odps
+from src.infra.database.postgresql.manager import PgManager
+
+from ._const import DemandConst
+from ._mapper import VideoDeconstructMapper
+
+logger = logging.getLogger(__name__)
+
+# ODPS SELECT 列清单（匹配结果表）
+_SELECT_COLUMNS = [
+    "dt",
+    "channel_name",
+    "channel_level3",
+    "match_video_id",
+    "experiment_id",
+    "rov",
+    "category_name",
+]
+_SELECT_CLAUSE = ", ".join(_SELECT_COLUMNS)
+
+
+class FetchDemands(DemandConst):
+    """从 ODPS 拉取白名单账号匹配视频并生成搜索队列条目
+
+    使用方式:
+        config = OdpsConfig()
+        fetcher = FetchDemands(config)
+        items = await fetcher.deal("20260701")
+    """
+
+    def __init__(self, config: OdpsConfig, pg: PgManager):
+        self._odps_config = config
+        self._deconstruct_mapper = VideoDeconstructMapper(pg)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 入口
+    # ═══════════════════════════════════════════════════════════════
+
+    async def deal(self, dt: str) -> List[Dict]:
+        """主流程: 加载白名单 → 查匹配视频 → 去重 → 解构 → 展开搜索词"""
+        # 1. 加载白名单账号
+        valid_accounts = await self._load_valid_accounts()
+        if not valid_accounts:
+            logger.warning("白名单为空, dt=%s, 终止", dt)
+            return []
+
+        # 2. 查询匹配视频（ODPS）
+        rows = await self._query_matched_videos(dt)
+        logger.info("ODPS 原始匹配结果: dt=%s, %d 行", dt, len(rows))
+
+        # 3. 过滤非白名单账号
+        filtered, dropped = self._filter_by_accounts(rows, valid_accounts)
+        if dropped:
+            logger.warning("丢弃 %d 行非白名单账号, dt=%s", dropped, dt)
+        logger.info("白名单过滤后: %d 行", len(filtered))
+
+        # 4. 按 (账号, 视频) 去重，取 rov 最高
+        deduped = self._dedup_by_rov(filtered)
+        logger.info("rov 去重后: %d 条 (账号+视频)", len(deduped))
+
+        # 5. 获取视频解构 + 展开搜索词
+        items = await self._explode_search_items(deduped)
+        logger.info("搜索词展开: %d 条队列项", len(items))
+        return items
+
+    # ═══════════════════════════════════════════════════════════════
+    # 1. 白名单账号加载
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _load_valid_accounts(self) -> Set[str]:
+        """从两张账号基础表加载白名单，仅取 account_status='开' 的账号"""
+        valid: Set[str] = set()
+        for table in self.VALID_ACCOUNT_TABLES.values():
+            sql = (
+                f"SELECT account_name "
+                f"FROM loghubods.{table} "
+                f"WHERE account_status = '开'"
+            )
+            rows = await asyncio.to_thread(fetch_from_odps, sql, self._odps_config)
+            names = {r["account_name"] for r in rows if r.get("account_name")}
+            valid.update(names)
+            logger.info("白名单加载: %s → %d 个有效账号", table, len(names))
+        logger.info("白名单合计: %d 个有效账号", len(valid))
+        return valid
+
+    # ═══════════════════════════════════════════════════════════════
+    # 2. 匹配视频查询
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _query_matched_videos(self, dt: str) -> List[Dict]:
+        """查询匹配结果表中指定日期的匹配视频"""
+        sql = (
+            f"SELECT {_SELECT_CLAUSE} "
+            f"FROM {self.MATCH_RESULT_TABLE} "
+            f"WHERE dt = '{dt}' "
+            f"  AND channel_name IN {self._format_tuple(self.CHANNEL_NAMES)} "
+            f"  AND match_score > 0.1 "
+            f"ORDER BY match_score DESC"
+        )
+        logger.info("ODPS 查询匹配视频, dt=%s", dt)
+        return await asyncio.to_thread(fetch_from_odps, sql, self._odps_config)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 3. 白名单过滤
+    # ═══════════════════════════════════════════════════════════════
+
+    @staticmethod
+    def _filter_by_accounts(
+        rows: List[Dict],
+        valid_accounts: Set[str],
+    ) -> tuple:
+        """过滤不在白名单中的账号
+
+        Returns:
+            (valid_rows, dropped_count)
+        """
+        if not valid_accounts:
+            return rows, 0
+        filtered: List[Dict] = []
+        dropped_names: Set[str] = set()
+        for r in rows:
+            account = r.get("channel_level3") or ""
+            if account in valid_accounts:
+                filtered.append(r)
+            else:
+                dropped_names.add(account)
+        if dropped_names:
+            logger.warning("非白名单账号: %s", sorted(dropped_names))
+        return filtered, len(rows) - len(filtered)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 4. rov 去重
+    # ═══════════════════════════════════════════════════════════════
+
+    @staticmethod
+    def _dedup_by_rov(rows: List[Dict]) -> List[Dict]:
+        """按 (channel_level3, match_video_id) 分组，保留 rov 最高的行
+
+        同一账号的同一视频可能有多个 experiment_id，
+        取 rov 最高者作为该视频的代表 experiment_id。
+        """
+        groups: Dict[tuple, Dict] = {}
+        for r in rows:
+            key = (r.get("channel_level3") or "", r.get("match_video_id") or 0)
+            rov = float(r.get("rov") or 0)
+            if key not in groups or rov > float(groups[key].get("rov") or 0):
+                groups[key] = r
+        result = list(groups.values())
+        logger.info(
+            "rov 去重: %d 行 → %d 条 (按 账号+视频 分组)",
+            len(rows), len(result),
+        )
+        return result
+
+    # ═══════════════════════════════════════════════════════════════
+    # 5. 视频解构 + 搜索词展开
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _explode_search_items(self, rows: List[Dict]) -> List[Dict]:
+        """对每条去重后的视频记录，获取解构结果并展开为搜索队列条目
+
+        一条视频可能产生多条队列项（每个搜索词一条）。
+        """
+        items: List[Dict] = []
+        for r in rows:
+            video_id = int(r.get("match_video_id") or 0)
+            if not video_id:
+                continue
+            decomp = await self._get_video_decomposition(video_id)
+            if not decomp:
+                logger.warning("视频 %d 解构结果为空，跳过", video_id)
+                continue
+            exploded = self._build_items_from_decomp(r, decomp)
+            items.extend(exploded)
+        return items
+
+    @staticmethod
+    def _build_items_from_decomp(
+        row: Dict,
+        vectors: List[Dict],
+    ) -> List[Dict]:
+        """将一条视频记录 + video_vectors 行展开为多条搜索队列条目
+
+        vectors 结构（每条来自 video_vectors 一行）:
+          [
+            {"config_code": "VIDEO_KEYPOINT", "text": "关键点文本"},
+            {"config_code": "VIDEO_INSPIRATION", "text": "灵感点文本"},
+            ...
+          ]
+
+        config_code 直接用作 key_type，text 用作 search_key。
+        """
+        base = {
+            "account": row.get("channel_level3") or "",
+            "dt": row.get("dt") or "",
+            "video_id": int(row.get("match_video_id") or 0),
+            "channel_name": row.get("channel_name") or "",
+            "experiment_id": row.get("experiment_id") or "",
+            "drive_type": DemandConst.DRIVE_TYPE_VIDEO,
+        }
+
+        items: List[Dict] = []
+        seen: Set[str] = set()
+
+        for v in vectors:
+            text_val = (v.get("text") or "").strip()
+            if not text_val:
+                continue
+            key = text_val.lower()
+            if key in seen:
+                continue
+            seen.add(key)
+            items.append({
+                **base,
+                "search_key": text_val,
+                "key_type": v.get("config_code") or "",
+            })
+
+        return items
+
+    # ═══════════════════════════════════════════════════════════════
+    # 视频搜索词（PG: video_vectors）
+    # ═══════════════════════════════════════════════════════════════
+
+    async def _get_video_decomposition(self, video_id: int) -> list[Dict]:
+        """从 PostgreSQL video_vectors 表查询搜索词，返回 [{config_code, text}, ...]"""
+        vectors = await self._deconstruct_mapper.find_by_video_id(video_id)
+        if not vectors:
+            logger.debug("视频 %d 在 video_vectors 中无记录", video_id)
+        return vectors
+
+    # ═══════════════════════════════════════════════════════════════
+    # 工具
+    # ═══════════════════════════════════════════════════════════════
+
+    @staticmethod
+    def _format_tuple(values: tuple) -> str:
+        """('a', 'b') → "('a', 'b')" """
+        quoted = ", ".join(f"'{v}'" for v in values)
+        return f"({quoted})"
+
+
+__all__ = ["FetchDemands"]

src/domains/fetch_demand_videos/__init__.py

@@ -0,0 +1,6 @@
+"""
+基于人群需求匹配到的视频
+取 video_id、 video_title
+取 解构结果（关键点、灵感点）
+存 账号信息、 dt、 搜索词、 video_id、 experiment_id 、config_code
+"""

src/handlers/demand_enqueue.py

@@ -17,7 +17,7 @@ from typing import TYPE_CHECKING
 
 from src.config.odps import OdpsConfig
 from src.infra.xxl_jobs import xxl_job
-from src.domains.demand import FetchDemands, EnqueueDemands
+from src.domains.fetch_demand import FetchDemands, EnqueueDemands
 
 if TYPE_CHECKING:
     from src.infra.database import DatabaseManager
@@ -44,14 +44,15 @@ def _parse_dt(param: str) -> str:
 
 
 async def _do_enqueue(dt: str) -> None:
-    """后台执行: ODPS 拉取 → 过滤 → 聚合 → MySQL 入队"""
+    """后台执行: ODPS 拉取 → 白名单过滤 → rov 去重 → 视频解构 → MySQL 入队"""
     logger.info("demandEnqueue background task started, dt=%s", dt)
 
     try:
-        fetcher = FetchDemands(OdpsConfig())
-        demands = await fetcher.deal(dt)
-        if not demands:
-            logger.info("无需求数据, dt=%s", dt)
+        pg = _db_ref.pg_manager("pg")
+        fetcher = FetchDemands(OdpsConfig(), pg)
+        items = await fetcher.deal(dt)
+        if not items:
+            logger.info("无队列项生成, dt=%s", dt)
             return
     except Exception:
         logger.exception("ODPS 查询失败, dt=%s", dt)
@@ -60,16 +61,15 @@ async def _do_enqueue(dt: str) -> None:
     try:
         mysql = _db_ref.mysql_manager("long_articles")
         enqueuer = EnqueueDemands(mysql)
-        n = await enqueuer.deal(demands)
+        n = await enqueuer.deal(items)
     except Exception:
         logger.exception("MySQL 入队失败, dt=%s", dt)
         return
 
-    accounts = len(set(d["channel_level3"] for d in demands))
-    total_terms = sum(len(d["search_terms"]) for d in demands)
+    accounts = len(set(it["account"] for it in items))
     logger.info(
-        "demandEnqueue done: enqueued %d demands (%d accounts, %d terms) for dt=%s",
-        n, accounts, total_terms, dt,
+        "demandEnqueue done: enqueued %d items (%d accounts) for dt=%s",
+        n, accounts, dt,
     )
 
 

src/infra/database/manager.py

@@ -99,7 +99,7 @@ class DatabaseManager:
         from .mysql.manager import MysqlManager
         return MysqlManager(self._get_sql(db_name))
 
-    def pg_manager(self, db_name: str = "default"):
+    def pg_manager(self, db_name: str = "vector"):
         """获取 PostgreSQL 专用门面"""
         from .postgresql.manager import PgManager
         return PgManager(self._get_sql(db_name))

src/infra/database/postgresql/backend.py

@@ -1,11 +1,10 @@
-"""PostgreSQL 异步后端 —— 基于 asyncpg 连接池
-
-状态：未实现。init() 会抛出 NotImplementedError。
-"""
+"""PostgreSQL 异步后端 —— 基于 asyncpg 连接池"""
 
 import logging
 from typing import Any, Optional
 
+import asyncpg
+
 from ..ports import SqlBackend
 
 logger = logging.getLogger(__name__)
@@ -16,38 +15,66 @@ class PgBackend(SqlBackend):
 
     def __init__(self, name: str, config: Any):
         super().__init__(name, config)
-        self._pool = None
+        self._pool: Optional[asyncpg.Pool] = None
 
     # ==================== 生命周期 ====================
 
     async def init(self) -> None:
-        raise NotImplementedError(
-            "PgBackend 尚未实现。需要安装 asyncpg 并完成连接池初始化逻辑。"
+        if self._initialized:
+            return
+        self._pool = await asyncpg.create_pool(
+            host=self.config.host,
+            port=self.config.port,
+            user=self.config.user,
+            password=self.config.password,
+            database=self.config.db,
+            min_size=getattr(self.config, "minsize", 5),
+            max_size=getattr(self.config, "maxsize", 20),
         )
+        self._initialized = True
+        logger.info("PgBackend [%s] 连接池创建成功", self.name)
 
     async def close(self) -> None:
-        if self._pool is not None:
-            self._pool = None
+        if self._pool is None:
+            return
+        await self._pool.close()
+        self._pool = None
         self._initialized = False
+        logger.info("PgBackend [%s] 连接池已关闭", self.name)
 
     async def health(self) -> bool:
-        return False
+        try:
+            await self.fetch_one("SELECT 1")
+            return True
+        except Exception:
+            return False
 
     # ==================== SQL 操作 ====================
 
+    @staticmethod
+    def _row_to_dict(row: asyncpg.Record) -> dict:
+        """将 asyncpg.Record 转为普通 dict"""
+        return dict(row)
+
     async def fetch(
         self,
         query: str,
         params: Optional[tuple] = None,
     ) -> list[dict]:
-        raise NotImplementedError("PgBackend 尚未实现")
+        self._ensure_ready()
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch(query, *params if params else [])
+            return [self._row_to_dict(r) for r in rows]
 
     async def fetch_one(
         self,
         query: str,
         params: Optional[tuple] = None,
     ) -> Optional[dict]:
-        raise NotImplementedError("PgBackend 尚未实现")
+        self._ensure_ready()
+        async with self._pool.acquire() as conn:
+            row = await conn.fetchrow(query, *params if params else [])
+            return self._row_to_dict(row) if row else None
 
     async def execute(
         self,
@@ -56,4 +83,19 @@ class PgBackend(SqlBackend):
         *,
         batch: bool = False,
     ) -> int:
-        raise NotImplementedError("PgBackend 尚未实现")
+        self._ensure_ready()
+        async with self._pool.acquire() as conn:
+            async with conn.transaction():
+                if batch:
+                    result = await conn.executemany(query, params)
+                else:
+                    result = await conn.execute(query, *params if params else [])
+            # asyncpg execute 返回 'INSERT 0 1' 格式字符串，提取行数
+            affected = int(result.split()[-1]) if result else 0
+            return affected
+
+    # ==================== 内部 ====================
+
+    def _ensure_ready(self):
+        if not self._initialized or self._pool is None:
+            raise RuntimeError(f"PgBackend [{self.name}] 未初始化，请先调用 init()")