init

.gitignore

@@ -10,3 +10,6 @@ build/
 .vscode/
 .claude
 CLAUDE.md
+/scripts
+.env
+/docs

src/__init__.py

@@ -27,6 +27,7 @@ from src.config import (
 # 基础设施
 from src.infra import (
     DatabaseManager,
+    MysqlManager,
     create_backend,
     MySQLBackend,
     LogService,
@@ -54,6 +55,17 @@ from src.core import (
     VectorMemory,
 )
 
+# 业务领域
+from src.domains import (
+    FetchDemands,
+    EnqueueDemands,
+    DemandQueueMapper,
+    Article,
+    ArticleService,
+    Account,
+    AccountService,
+)
+
 # 服务
 from src.server import create_app
 
@@ -63,6 +75,7 @@ __all__ = [
 
     # Infra
     "DatabaseManager",
+    "MysqlManager",
     "create_backend",
     "MySQLBackend",
     "LogService",
@@ -85,6 +98,14 @@ __all__ = [
     "Memory",
     "ConversationMemory",
     "VectorMemory",
+    # Domains
+    "FetchDemands",
+    "EnqueueDemands",
+    "DemandQueueMapper",
+    "Article",
+    "ArticleService",
+    "Account",
+    "AccountService",
     # Server
     "create_app",
 ]

src/config/_global_config.py

@@ -14,6 +14,7 @@ from .database import (
 from .aliyun import AliyunLogConfig, AliyunOssConfig
 from .xxljob import XxlJobConfig
 from .apollo import ApolloConfig
+from .odps import OdpsConfig
 from .volcengine import VolcengineConfig
 
 
@@ -66,9 +67,12 @@ class GlobalConfig(BaseSettings):
     # 火山引擎
     volcengine: VolcengineConfig = Field(default_factory=VolcengineConfig)
 
-    # apollo
+    # Apollo
     apollo: ApolloConfig = Field(default_factory=ApolloConfig)
 
+    # ODPS
+    odps: OdpsConfig = Field(default_factory=OdpsConfig)
+
 
     model_config = SettingsConfigDict(
         env_file=".env",

src/config/database.py

@@ -11,7 +11,7 @@ class DatabaseConfig(BaseSettings):
     """数据库连接配置基类
 
     通过 backend 字段指定存储引擎类型：
-      - "mysql" / ""  → MySQLBackend
+      - "mysql"       → MySQLBackend
       - "postgresql"  → PgBackend
       - "milvus"      → MilvusBackend
       - "redis"       → RedisBackend
@@ -25,7 +25,7 @@ class DatabaseConfig(BaseSettings):
     charset: str = "utf8mb4"
     minsize: int = 5
     maxsize: int = 20
-    backend: str = ""
+    backend: str = "mysql"
 
     model_config = SettingsConfigDict(
         env_prefix="", case_sensitive=False, extra="ignore"

src/config/odps.py

@@ -0,0 +1,16 @@
+"""ODPS (MaxCompute) 配置"""
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class OdpsConfig(BaseSettings):
+    """阿里云 ODPS (MaxCompute) 配置"""
+
+    access_id: str = "LTAIWYUujJAm7CbH"
+    secret_access_key: str = "RfSjdiWwED1sGFlsjXv0DlfTnZTG1P"
+    endpoint: str = "http://service.cn.maxcompute.aliyun.com/api"
+    project: str = "loghubods"
+
+    model_config = SettingsConfigDict(
+        env_prefix="ODPS_", env_file=".env", case_sensitive=False, extra="ignore"
+    )

src/domains/__init__.py

@@ -0,0 +1,30 @@
+"""业务领域层 —— 领域模型 + 业务服务
+
+与 core 的区别:
+  core:  框架抽象（Agent/Tool/Memory），不包含具体业务知识，可跨项目复用
+  domains: 本项目专属的业务逻辑和数据模型，依赖 core 和 infra
+
+子域:
+  demand   — 需求获取与管理（外部需求源 → 统一需求模型）
+  article  — 文章抓取与处理（需求驱动 → 文章采集 → 内容结构化）
+  account  — 账号识别与管理（文章反推账号 → 账号画像 → 采集策略）
+
+数据流: demand → article → account（下游消费上游产出）
+"""
+
+from src.domains.demand import FetchDemands, EnqueueDemands, DemandQueueMapper
+from src.domains.article import Article, ArticleService
+from src.domains.account import Account, AccountService
+
+__all__ = [
+    # Demand
+    "FetchDemands",
+    "EnqueueDemands",
+    "DemandQueueMapper",
+    # Article
+    "Article",
+    "ArticleService",
+    # Account
+    "Account",
+    "AccountService",
+]

src/domains/account/__init__.py

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

src/domains/account/models.py

@@ -0,0 +1,23 @@
+"""账号领域模型"""
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass
+class Account:
+    """内容账号模型
+
+    从文章元信息中反推识别，逐步构建画像。
+    """
+    account_id: str                                 # 平台唯一 ID（如公众号 fakeid）
+    account_name: str                               # 账号名称
+    platform: str = "wechat"                        # 所属平台
+    description: str = ""                           # 账号简介
+    domain_tags: list[str] = field(default_factory=list)   # 领域标签
+    article_count_crawled: int = 0                  # 已抓取文章数
+    quality_score: float = 0.0                      # 质量评分 (0-100)
+    is_active: bool = True                          # 是否活跃采集
+    first_seen_at: datetime | None = None
+    last_crawled_at: datetime | None = None
+    extra: dict = field(default_factory=dict)

src/domains/account/service.py

@@ -0,0 +1,34 @@
+"""账号管理服务"""
+
+import logging
+
+from src.domains.article.models import Article
+from src.domains.account.models import Account
+
+logger = logging.getLogger(__name__)
+
+
+class AccountService:
+    """账号管理服务
+
+    职责: 从 Article 中提取账号信息，去重合并，构建账号画像，输出采集策略。
+    不负责: 文章的抓取（由 article 领域负责）。
+    """
+
+    async def extract_account_from_article(self, article: Article) -> Account | None:
+        """从单篇文章提取账号信息"""
+        raise NotImplementedError
+
+    async def upsert_account(self, account: Account) -> Account:
+        """新建或更新账号信息（去重合并）"""
+        raise NotImplementedError
+
+    async def build_profile(self, account_id: str) -> Account:
+        """构建/更新账号画像（领域标签、质量评分等）"""
+        raise NotImplementedError
+
+    async def list_active_accounts(
+        self, platform: str = "wechat", min_score: float = 0.0
+    ) -> list[Account]:
+        """列出活跃采集的账号列表，按质量评分降序"""
+        raise NotImplementedError

src/domains/article/__init__.py

@@ -0,0 +1,16 @@
+"""文章领域 —— 基于需求驱动文章抓取与内容结构化
+
+核心流程:
+  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

@@ -0,0 +1,20 @@
+"""文章领域模型"""
+
+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

@@ -0,0 +1,36 @@
+"""文章抓取与处理服务"""
+
+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

@@ -0,0 +1,19 @@
+"""需求域 —— 从 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/_const.py

@@ -0,0 +1,64 @@
+"""需求域常量"""
+
+
+class DemandConst:
+    """需求域所有常量的聚合根，服务类继承此类即可通过 self. 访问"""
+
+    # ═══════════════════════════════════════════════════════════════
+    # ODPS 查询过滤条件
+    # ═══════════════════════════════════════════════════════════════
+
+    DEFAULT_DT = "20260501"
+
+    CHANNEL_NAMES = ("公众号投流-稳定", "服务号投流")
+
+    # 账号白名单表：channel_name → ODPS 账号基础表
+    VALID_ACCOUNT_TABLES = {
+        "公众号投流-稳定": "feishu_wechat_mp_account_base",
+        "服务号投流": "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
+
+    # ═══════════════════════════════════════════════════════════════
+    # 聚合配置
+    # ═══════════════════════════════════════════════════════════════
+
+    MAX_PER_ACCOUNT_PER_DAY = 2000
+
+    # ═══════════════════════════════════════════════════════════════
+    # 搜索队列表
+    # ═══════════════════════════════════════════════════════════════
+
+    QUEUE_TABLE = "demand_search_queue"
+
+    class QueueStatus:
+        INIT = 0        # 初始，待处理
+        PROCESSING = 1  # 处理中
+        SUCCESS = 2     # 处理成功
+        FAIL = 99       # 处理失败
+
+
+__all__ = ["DemandConst"]

src/domains/demand/_mapper.py

@@ -0,0 +1,140 @@
+"""需求搜索队列 MySQL 读写 —— 仅负责 DB I/O，不含业务逻辑"""
+
+import json
+from typing import Dict, List
+
+from src.infra.database.mysql.manager import MysqlManager
+
+from ._const import DemandConst
+
+
+class DemandQueueMapper(DemandConst):
+    """demand_search_queue 表的数据访问层
+
+    职责: SQL 拼接 + 参数绑定 + 返回原生 dict。
+    不负责: 业务校验、格式转换（由上层 service 负责）。
+    """
+
+    def __init__(self, pool: MysqlManager):
+        self.pool = pool
+
+    # ═══════════════════════════════════════════════════════════════
+    # 写入
+    # ═══════════════════════════════════════════════════════════════
+
+    async def insert_batch(self, items: List[Dict]) -> int:
+        """批量写入队列，返回受影响行数
+
+        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
+        """
+        if not items:
+            return 0
+        query = f"""
+            INSERT 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)
+            VALUES
+                (%s, %s, %s,
+                 %s, %s, %s,
+                 %s,
+                 %s, %s, %s,
+                 %s, %s, %s)
+        """
+        params = [
+            (
+                it["account"],
+                it["dt"],
+                it["match_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["experiment_id"],
+                self.QueueStatus.INIT,
+            )
+            for it in items
+        ]
+        return await self.pool.save(query=query, params=params, batch=True)
+
+    # ═══════════════════════════════════════════════════════════════
+    # 读取
+    # ═══════════════════════════════════════════════════════════════
+
+    async def fetch_pending(
+        self,
+        dt: str,
+        *,
+        limit: int = 500,
+        offset: int = 0,
+    ) -> List[Dict]:
+        """查询待搜索的队列项，按 match_score 降序"""
+        query = f"""
+            SELECT *
+            FROM {self.QUEUE_TABLE}
+            WHERE dt = %s AND status = %s
+            ORDER BY match_score DESC
+            LIMIT %s OFFSET %s
+        """
+        return await self.pool.fetch(
+            query=query,
+            params=(dt, self.QueueStatus.INIT, limit, offset),
+        )
+
+    async def count_by_status(self, dt: str, status: int) -> int:
+        """统计指定状态的队列项数量"""
+        query = f"""
+            SELECT COUNT(*) AS cnt
+            FROM {self.QUEUE_TABLE}
+            WHERE dt = %s AND status = %s
+        """
+        row = await self.pool.fetch_one(query=query, params=(dt, status))
+        return row["cnt"] if row else 0
+
+    # ═══════════════════════════════════════════════════════════════
+    # 状态更新
+    # ═══════════════════════════════════════════════════════════════
+
+    async def mark_processing(self, item_id: int) -> int:
+        query = f"""
+            UPDATE {self.QUEUE_TABLE}
+            SET status = %s
+            WHERE id = %s AND status = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.QueueStatus.PROCESSING, item_id, self.QueueStatus.INIT),
+        )
+
+    async def mark_done(self, item_id: int) -> int:
+        query = f"""
+            UPDATE {self.QUEUE_TABLE}
+            SET status = %s
+            WHERE id = %s
+        """
+        return await self.pool.save(
+            query=query, params=(self.QueueStatus.SUCCESS, item_id),
+        )
+
+    async def mark_failed(self, item_id: int, reason: str = "") -> int:
+        query = f"""
+            UPDATE {self.QUEUE_TABLE}
+            SET status = %s, fail_reason = %s
+            WHERE id = %s
+        """
+        return await self.pool.save(
+            query=query,
+            params=(self.QueueStatus.FAIL, reason[:512], item_id),
+        )
+
+
+__all__ = ["DemandQueueMapper"]

src/domains/demand/enqueue_demands.py

@@ -0,0 +1,78 @@
+"""需求入队服务 —— 将聚合后的需求写入 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

@@ -0,0 +1,256 @@
+"""从 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/handlers/demand_enqueue.py

@@ -0,0 +1,75 @@
+"""需求入队 XXL-JOB Handler —— ODPS 拉取 → 聚合 → 写入 MySQL
+
+XXL-JOB Admin 配置:
+    执行器: long-articles-agentic-src
+    处理器: demandEnqueue
+    参数:   dt=yyyyMMdd (可选，不传则默认前天)
+"""
+
+import asyncio
+import logging
+from datetime import datetime, timedelta
+
+from src.config.odps import OdpsConfig
+from src.config.database import LongArticlesDatabaseConfig
+from src.infra.database import create_backend, MysqlManager
+from src.infra.xxl_jobs import xxl_job
+from src.domains.demand import FetchDemands, EnqueueDemands
+
+logger = logging.getLogger(__name__)
+
+
+def _parse_dt(param: str) -> str:
+    """从 XXL-JOB 参数中提取 dt，默认前天"""
+    if param and param.strip():
+        for part in param.strip().split():
+            if part.startswith("dt="):
+                return part[3:]
+        # 如果没有 dt= 前缀，直接把参数当日期字符串
+        return param.strip()
+    return (datetime.now() - timedelta(days=2)).strftime("%Y%m%d")
+
+
+@xxl_job("demandEnqueue")
+async def demand_enqueue(param: str) -> dict:
+    """需求入队 handler
+
+    param: executorParams，格式 "dt=20260629" 或 "20260629" 或空
+    """
+    dt = _parse_dt(param)
+    logger.info("demandEnqueue started, dt=%s", dt)
+
+    # ── 1. ODPS 拉取 ────────────────────────────────
+    odps_config = OdpsConfig()
+    fetcher = FetchDemands(odps_config)
+    try:
+        demands = await fetcher.deal(dt)
+    except Exception:
+        logger.exception("ODPS 查询失败, dt=%s", dt)
+        return {"code": 500, "msg": f"ODPS query failed, dt={dt}"}
+
+    if not demands:
+        logger.info("无需求数据, dt=%s", dt)
+        return {"code": 200, "msg": f"no demands found for dt={dt}"}
+
+    # ── 2. MySQL 入队 ──────────────────────────────
+    db_config = LongArticlesDatabaseConfig()
+    backend = create_backend("long_articles", db_config)
+    try:
+        await backend.init()
+        mysql = MysqlManager(backend)
+        enqueuer = EnqueueDemands(mysql)
+        n = await enqueuer.deal(demands)
+    except Exception:
+        logger.exception("MySQL 入队失败, dt=%s", dt)
+        return {"code": 500, "msg": f"MySQL enqueue failed, dt={dt}"}
+    finally:
+        await backend.close()
+
+    accounts = len(set(d["channel_level3"] for d in demands))
+    total_terms = sum(len(d["search_terms"]) for d in demands)
+    msg = (
+        f"enqueued {n} demands ({accounts} accounts, {total_terms} terms) for dt={dt}"
+    )
+    logger.info("demandEnqueue done: %s", msg)
+    return {"code": 200, "msg": msg}

src/infra/__init__.py

@@ -11,6 +11,9 @@ from src.infra.database import (
     MilvusBackend,
     RedisBackend,
     DatabaseManager,
+    MysqlManager,
+    PgManager,
+    RedisManager,
     create_backend,
 )
 from src.infra.observability import (
@@ -43,7 +46,7 @@ from src.infra.external import (
     AsyncApolloApi,
     FeishuBotApi,
     FeishuSheetApi,
-    OdpsService,
+    fetch_from_odps,
     fetch_deepseek_completion,
 )
 from src.infra.spider import (
@@ -65,6 +68,9 @@ __all__ = [
     "MilvusBackend",
     "RedisBackend",
     "DatabaseManager",
+    "MysqlManager",
+    "PgManager",
+    "RedisManager",
     "create_backend",
     # 可观测性
     "LogCategory",
@@ -92,7 +98,7 @@ __all__ = [
     "AsyncApolloApi",
     "FeishuBotApi",
     "FeishuSheetApi",
-    "OdpsService",
+    "fetch_from_odps",
     "fetch_deepseek_completion",
     # 爬虫
     "get_article_detail",

src/infra/database/__init__.py

@@ -1,4 +1,11 @@
-"""数据库设施层 —— 抽象接口、具体后端、门面管理器"""
+"""数据库设施层 —— 抽象接口、门面管理器、各存储引擎子包
+
+子包:
+  mysql/      MySQL  (backend + manager)
+  postgresql/ PostgreSQL (backend + manager)
+  redis/      Redis (backend + manager)
+  milvus/     Milvus (backend)
+"""
 
 from src.infra.database.ports import (
     BackendType,
@@ -7,10 +14,10 @@ from src.infra.database.ports import (
     VectorBackend,
     KvBackend,
 )
-from src.infra.database.mysql import MySQLBackend
-from src.infra.database.postgresql import PgBackend
+from src.infra.database.mysql import MySQLBackend, MysqlManager
+from src.infra.database.postgresql import PgBackend, PgManager
 from src.infra.database.milvus import MilvusBackend
-from src.infra.database.redis import RedisBackend
+from src.infra.database.redis import RedisBackend, RedisManager
 from src.infra.database.manager import DatabaseManager, create_backend
 
 __all__ = [
@@ -24,5 +31,8 @@ __all__ = [
     "MilvusBackend",
     "RedisBackend",
     "DatabaseManager",
+    "MysqlManager",
+    "PgManager",
+    "RedisManager",
     "create_backend",
 ]

src/infra/database/manager.py

@@ -1,9 +1,7 @@
-"""数据库门面管理器 —— 依赖反转，接收已创建的 Backend 实例
+"""数据库管理器 —— 连接封装 + 生命周期 + 后端路由
 
-与旧版 DatabaseManager 的核心区别：
-  - 不再通过 config dict 内部创建后端，而是接收 Dict[str, BaseBackend]
-  - 后端创建逻辑剥离为 create_backend() 工厂函数，由调用方（DI 容器 / app 工厂）负责
-  - Manager 只关心生命周期编排与统一访问路由
+不包含具体数据操作逻辑（操作在各子包 backend.py 实现类内部）。
+子门面 MysqlManager / PgManager / RedisManager 在各子包 manager.py 中。
 """
 
 import logging
@@ -14,10 +12,6 @@ from .ports import (
     BaseBackend,
     SqlBackend,
 )
-from .mysql import MySQLBackend
-from .postgresql import PgBackend
-from .milvus import MilvusBackend
-from .redis import RedisBackend
 
 logger = logging.getLogger(__name__)
 
@@ -26,24 +20,29 @@ logger = logging.getLogger(__name__)
 
 
 def _detect_backend_type(config: Any) -> BackendType:
-    """从配置推断后端类型：优先读取显式 backend 字段，否则按类名匹配"""
-    if hasattr(config, "backend") and config.backend:
-        try:
-            return BackendType(config.backend)
-        except ValueError:
-            pass
-    type_name = type(config).__name__.lower()
-    for bt in BackendType:
-        if bt.value in type_name:
-            return bt
-    return BackendType.MYSQL
+    """从配置读取后端类型。要求显式声明 backend 字段。"""
+    backend = getattr(config, "backend", None)
+    if not backend:
+        raise ValueError(
+            f"配置 {type(config).__name__} 缺少 backend 字段，"
+            "请显式设置为 mysql/postgresql/milvus/redis"
+        )
+    try:
+        return BackendType(str(backend).lower())
+    except ValueError as exc:
+        allowed = ", ".join(bt.value for bt in BackendType)
+        raise ValueError(
+            f"配置 {type(config).__name__} 的 backend={backend!r} 非法，可选: {allowed}"
+        ) from exc
 
 
 def create_backend(name: str, config: Any) -> BaseBackend:
-    """工厂函数：根据配置创建对应后端实例
+    """工厂函数：根据配置创建对应后端实例"""
+    from .mysql.backend import MySQLBackend
+    from .postgresql.backend import PgBackend
+    from .milvus.backend import MilvusBackend
+    from .redis.backend import RedisBackend
 
-    由调用方（app 工厂 / DI 容器）使用，创建实例后传给 DatabaseManager。
-    """
     bt = _detect_backend_type(config)
     if bt == BackendType.MYSQL:
         return MySQLBackend(name, config)
@@ -56,102 +55,108 @@ def create_backend(name: str, config: Any) -> BaseBackend:
     raise ValueError(f"未知的存储后端类型: {bt}")
 
 
-# ==================== 门面管理器 ====================
+# ==================== 管理器 ====================
 
 
 class DatabaseManager:
-    """多后端数据库门面
+    """数据库连接管理器 —— 封装多后端实例的生命周期、路由、子门面工厂
 
-    接收已创建的后端实例，负责：
-      - 统一生命周期管理 (init_pools / close_pools)
-      - 向后兼容 SQL 接口
-      - 类型安全的后端直取访问器
+    职责: 连接池初始化/关闭、后端直取、子门面创建
+    不负责: 具体 SQL/KV 操作（操作在各子门面及后端实现类内部）
 
-    使用方式：
-        backends = {
-            "aigc": MySQLBackend("aigc", AigcDatabaseConfig()),
-        }
+    使用方式:
+        backends = {"aigc": MySQLBackend("aigc", config)}
         db = DatabaseManager(backends)
         await db.init_pools()
-        rows = await db.async_fetch("SELECT * FROM t")
+
+        # 通过子门面操作
+        mysql = db.mysql_manager("long_articles")
+        rows = await mysql.fetch("SELECT * FROM t")
     """
 
     def __init__(
         self,
         backends: Dict[str, BaseBackend],
         *,
-        default_db: str = "default",
+        default_db: Optional[str] = None,
     ):
         if not backends:
             raise ValueError("DatabaseManager 至少需要一个后端")
         self._backends = backends
-        self._default_db = default_db
+        if default_db is None:
+            self._default_db = next(iter(backends))
+        else:
+            if default_db not in backends:
+                raise ValueError(
+                    f"default_db={default_db!r} 未注册，可用: {list(backends.keys())}"
+                )
+            self._default_db = default_db
+
+    # ==================== 子门面工厂 ====================
+
+    def mysql_manager(self, db_name: str = "long_articles"):
+        """获取 MySQL 专用门面"""
+        from .mysql.manager import MysqlManager
+        return MysqlManager(self._get_sql(db_name))
+
+    def pg_manager(self, db_name: str = "default"):
+        """获取 PostgreSQL 专用门面"""
+        from .postgresql.manager import PgManager
+        return PgManager(self._get_sql(db_name))
+
+    def redis_manager(self, db_name: str = "default"):
+        """获取 Redis 专用门面"""
+        from .redis.backend import RedisBackend
+        from .redis.manager import RedisManager
+        return RedisManager(self._get_typed(db_name, RedisBackend))
 
     # ==================== 生命周期 ====================
 
     async def init_pools(self) -> None:
-        for name, backend in self._backends.items():
-            await backend.init()
-            logger.info("存储后端 [%s] 初始化完成 → %s", name, type(backend).__name__)
+        initialized: list[tuple[str, BaseBackend]] = []
+        try:
+            for name, backend in self._backends.items():
+                await backend.init()
+                initialized.append((name, backend))
+                logger.info("存储后端 [%s] 初始化完成 → %s", name, type(backend).__name__)
+        except Exception as exc:
+            logger.exception("存储后端初始化失败，开始回滚已初始化连接: %s", exc)
+            for name, backend in reversed(initialized):
+                try:
+                    await backend.close()
+                    logger.info("存储后端 [%s] 回滚关闭完成", name)
+                except Exception:
+                    logger.exception("存储后端 [%s] 回滚关闭失败", name)
+            raise
 
     async def close_pools(self) -> None:
         for name, backend in self._backends.items():
             await backend.close()
             logger.info("存储后端 [%s] 已关闭", name)
 
-    # ==================== 向后兼容 SQL 接口 ====================
-
-    async def async_fetch(
-        self,
-        query: str,
-        *,
-        params: Optional[tuple] = None,
-        db_name: Optional[str] = None,
-    ) -> List[Dict[str, Any]]:
-        backend = self._get_sql(db_name)
-        return await backend.fetch(query, params=params)
-
-    async def async_fetch_one(
-        self,
-        query: str,
-        *,
-        params: Optional[tuple] = None,
-        db_name: Optional[str] = None,
-    ) -> Optional[Dict[str, Any]]:
-        backend = self._get_sql(db_name)
-        return await backend.fetch_one(query, params=params)
-
-    async def async_save(
-        self,
-        query: str,
-        *,
-        params: Any = None,
-        batch: bool = False,
-        db_name: Optional[str] = None,
-    ) -> int:
-        backend = self._get_sql(db_name)
-        return await backend.execute(query, params=params, batch=batch)
-
-    # ==================== 类型安全的后端直取 ====================
+    # ==================== 后端直取 ====================
 
-    def mysql(self, name: Optional[str] = None) -> MySQLBackend:
+    def mysql(self, name: Optional[str] = None):
+        from .mysql.backend import MySQLBackend
         return self._get_typed(name or self._default_db, MySQLBackend)
 
-    def pg(self, name: str) -> PgBackend:
+    def pg(self, name: str):
+        from .postgresql.backend import PgBackend
         return self._get_typed(name, PgBackend)
 
-    def milvus(self, name: str) -> MilvusBackend:
+    def milvus(self, name: str):
+        from .milvus.backend import MilvusBackend
         return self._get_typed(name, MilvusBackend)
 
-    def redis(self, name: str) -> RedisBackend:
+    def redis(self, name: str):
+        from .redis.backend import RedisBackend
         return self._get_typed(name, RedisBackend)
 
     # ==================== 辅助 ====================
 
     def get_pool(self, db_name: str):
-        """向后兼容：返回 MySQL 连接池对象"""
-        backend = self.mysql(db_name)
-        return backend._pool
+        """向后兼容：返回 MySQL 连接池对象（不建议新增调用）"""
+        return self.mysql(db_name)._pool
 
     def list_databases(self) -> List[str]:
         return list(self._backends.keys())

src/infra/database/milvus/__init__.py

@@ -0,0 +1,5 @@
+"""Milvus 子包 —— 向量数据库后端"""
+
+from .backend import MilvusBackend
+
+__all__ = ["MilvusBackend"]

src/infra/database/milvus.py → src/infra/database/milvus/backend.py

@@ -6,7 +6,7 @@
 import logging
 from typing import Any, Optional
 
-from .ports import VectorBackend
+from ..ports import VectorBackend
 
 logger = logging.getLogger(__name__)
 

src/infra/database/mysql/__init__.py

@@ -0,0 +1,6 @@
+"""MySQL 子包 —— 连接池后端 + 专用门面"""
+
+from .backend import MySQLBackend
+from .manager import MysqlManager
+
+__all__ = ["MySQLBackend", "MysqlManager"]

src/infra/database/mysql.py → src/infra/database/mysql/backend.py

@@ -5,7 +5,7 @@ from typing import Any, Optional
 
 from aiomysql import create_pool, DictCursor
 
-from .ports import SqlBackend
+from ..ports import SqlBackend
 
 logger = logging.getLogger(__name__)
 
@@ -32,7 +32,8 @@ class MySQLBackend(SqlBackend):
             minsize=getattr(self.config, "minsize", 5),
             maxsize=getattr(self.config, "maxsize", 20),
             cursorclass=DictCursor,
-            autocommit=True,
+            # 使用显式事务提交，避免隐式自动提交导致的语义不一致。
+            autocommit=False,
         )
         self._initialized = True
         logger.info("MySQL backend [%s] 连接池创建成功", self.name)

src/infra/database/mysql/manager.py

@@ -0,0 +1,49 @@
+"""MySQL 专用访问门面"""
+
+from typing import Any, Dict, List, Optional
+
+from ..ports import SqlBackend
+
+
+class MysqlManager:
+    """MySQL 专用访问门面 —— 封装 SqlBackend，domain 层直接依赖此类
+
+    使用方式:
+        db = DatabaseManager(backends)
+        mysql = db.mysql_manager("long_articles")
+        rows = await mysql.fetch("SELECT * FROM t LIMIT %s", params=(10,))
+    """
+
+    def __init__(self, backend: SqlBackend):
+        self._backend = backend
+
+    async def fetch(
+        self,
+        query: str,
+        *,
+        params: Optional[tuple] = None,
+    ) -> List[Dict[str, Any]]:
+        """SELECT 多行"""
+        return await self._backend.fetch(query, params=params)
+
+    async def fetch_one(
+        self,
+        query: str,
+        *,
+        params: Optional[tuple] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """SELECT 单行"""
+        return await self._backend.fetch_one(query, params=params)
+
+    async def save(
+        self,
+        query: str,
+        *,
+        params: Any = None,
+        batch: bool = False,
+    ) -> int:
+        """INSERT / UPDATE / DELETE，返回受影响行数"""
+        return await self._backend.execute(query, params=params, batch=batch)
+
+
+__all__ = ["MysqlManager"]

src/infra/database/postgresql/__init__.py

@@ -0,0 +1,6 @@
+"""PostgreSQL 子包 —— 连接池后端 + 专用门面"""
+
+from .backend import PgBackend
+from .manager import PgManager
+
+__all__ = ["PgBackend", "PgManager"]

src/infra/database/postgresql.py → src/infra/database/postgresql/backend.py

@@ -6,7 +6,7 @@
 import logging
 from typing import Any, Optional
 
-from .ports import SqlBackend
+from ..ports import SqlBackend
 
 logger = logging.getLogger(__name__)
 

src/infra/database/postgresql/manager.py

@@ -0,0 +1,46 @@
+"""PostgreSQL 专用访问门面"""
+
+from typing import Any, Dict, List, Optional
+
+from ..ports import SqlBackend
+
+
+class PgManager:
+    """PostgreSQL 专用访问门面
+
+    使用方式:
+        db = DatabaseManager(backends)
+        pg = db.pg_manager("my_pg")
+        rows = await pg.fetch("SELECT * FROM t LIMIT %s", params=(10,))
+    """
+
+    def __init__(self, backend: SqlBackend):
+        self._backend = backend
+
+    async def fetch(
+        self,
+        query: str,
+        *,
+        params: Optional[tuple] = None,
+    ) -> List[Dict[str, Any]]:
+        return await self._backend.fetch(query, params=params)
+
+    async def fetch_one(
+        self,
+        query: str,
+        *,
+        params: Optional[tuple] = None,
+    ) -> Optional[Dict[str, Any]]:
+        return await self._backend.fetch_one(query, params=params)
+
+    async def save(
+        self,
+        query: str,
+        *,
+        params: Any = None,
+        batch: bool = False,
+    ) -> int:
+        return await self._backend.execute(query, params=params, batch=batch)
+
+
+__all__ = ["PgManager"]

src/infra/database/redis/__init__.py

@@ -0,0 +1,6 @@
+"""Redis 子包 —— 后端 + 专用门面"""
+
+from .backend import RedisBackend
+from .manager import RedisManager
+
+__all__ = ["RedisBackend", "RedisManager"]

src/infra/database/redis.py → src/infra/database/redis/backend.py

@@ -6,7 +6,7 @@
 import logging
 from typing import Any, Optional
 
-from .ports import KvBackend
+from ..ports import KvBackend
 
 logger = logging.getLogger(__name__)
 

src/infra/database/redis/manager.py

@@ -0,0 +1,45 @@
+"""Redis 专用 KV 存储门面"""
+
+from typing import Optional
+
+from ..ports import KvBackend
+
+
+class RedisManager:
+    """Redis 专用 KV 存储门面
+
+    使用方式:
+        db = DatabaseManager(backends)
+        redis = db.redis_manager("cache")
+        await redis.set("key", "value", ttl=3600)
+    """
+
+    def __init__(self, backend: KvBackend):
+        self._backend = backend
+
+    async def get(self, key: str) -> Optional[str]:
+        return await self._backend.get(key)
+
+    async def set(
+        self,
+        key: str,
+        value: str,
+        *,
+        ttl: Optional[int] = None,
+    ) -> bool:
+        return await self._backend.set(key, value, ttl=ttl)
+
+    async def delete(self, *keys: str) -> int:
+        return await self._backend.delete(*keys)
+
+    async def exists(self, *keys: str) -> int:
+        return await self._backend.exists(*keys)
+
+    async def expire(self, key: str, ttl: int) -> bool:
+        return await self._backend.expire(key, ttl)
+
+    async def ttl(self, key: str) -> int:
+        return await self._backend.ttl(key)
+
+
+__all__ = ["RedisManager"]

src/infra/external/__init__.py

@@ -1,7 +1,7 @@
 """外部服务集成 —— 第三方 API 客户端"""
 from .apollo import AsyncApolloApi
 from .feishu import FeishuBotApi, FeishuSheetApi
-from .odps import OdpsService
+from .odps import fetch_from_odps
 from .volcengine import fetch_deepseek_completion
 
 
@@ -9,6 +9,6 @@ __all__ = [
     "AsyncApolloApi",
     "FeishuBotApi",
     "FeishuSheetApi",
-    "OdpsService",
+    "fetch_from_odps",
     "fetch_deepseek_completion",
 ]

src/infra/external/odps.py

@@ -1,38 +1,24 @@
-import asyncio
 from odps import ODPS
 
+from src.config.odps import OdpsConfig
 
-class OdpsService:
-    def __init__(self, access_id, secret_access_key, project, endpoint):
-        self.odps_client = ODPS(access_id, secret_access_key, project, endpoint)
 
-    async def execute_odps_query(self, query: str) -> bool:
-        loop = asyncio.get_running_loop()
+def fetch_from_odps(query: str, config: OdpsConfig) -> list[dict]:
+    """执行 ODPS SQL 查询，返回 dict 列表
 
-        def _execute():
-            instance = self.odps_client.execute_sql(
-                query, hints={"odps.sql.submit.mode": "script"}
-            )
-            instance.wait_for_success()
-
-        try:
-            await loop.run_in_executor(None, _execute)
-            return True
-        except Exception as e:
-            print(f"[ODPS ERROR] {e}")
-            return False
-
-    async def read_from_odps(self, query: str):
-        loop = asyncio.get_running_loop()
-
-        def _read():
-            with self.odps_client.execute_sql(query).open_reader() as reader:
-                if reader:
-                    return [item for item in reader]
-                return []
-
-        try:
-            return await loop.run_in_executor(None, _read)
-        except Exception as e:
-            print(f"[ODPS READ ERROR] {e}")
+    在基础设施边界将 ODPS Record 转为 dict，避免专有类型泄漏到领域层。
+    """
+    client = ODPS(
+        access_id=config.access_id,
+        secret_access_key=config.secret_access_key,
+        endpoint=config.endpoint,
+        project=config.project,
+    )
+    with client.execute_sql(query).open_reader() as reader:
+        if not reader:
+            return []
+        records = [r for r in reader]
+        if not records:
             return []
+        col_names = sorted(records[0]._name_indexes, key=records[0]._name_indexes.get)
+        return [dict(zip(col_names, r.values)) for r in records]

src/infra/spider/wechat/gzh.py

@@ -148,4 +148,4 @@ async def weixin_search(keyword: str, page: str = "1") -> dict | None:
             "weixin_search API请求失败: %s, keyword=%s, page=%s", e, keyword, page
         )
         return None
-    return response
+    return response