server_init

app.py

@@ -0,0 +1,37 @@
+import logging
+from quart_cors import cors
+from quart import Quart
+
+from src.core.bootstrap import AppContext
+from src.core.dependency import ServerContainer
+from src.core.api.v1.routes import server_routes
+
+logging.basicConfig(level=logging.INFO)
+
+app = Quart(__name__)
+app = cors(app, allow_origin="*")
+
+server_container = ServerContainer()
+ctx = AppContext(server_container)
+
+config = server_container.config()
+log_service = server_container.log_service()
+mysql_pool = server_container.mysql_pool()
+
+
+routes = server_routes(mysql_pool, log_service, config)
+app.register_blueprint(routes)
+
+
+@app.before_serving
+async def startup():
+    logging.info("Starting application...")
+    await ctx.start_up()
+    logging.info("Application started successfully")
+
+
+@app.after_serving
+async def shutdown():
+    logging.info("Shutting down application...")
+    await ctx.shutdown()
+    logging.info("Application shutdown successfully")

requirements.txt

@@ -0,0 +1,38 @@
+# ASGI / Web
+hypercorn>=0.16.0
+quart>=0.19.0
+
+# Config & validation
+pydantic>=2.0.0
+pydantic-settings>=2.0.0
+
+# Database
+sqlalchemy[asyncio]>=2.0.0
+aiomysql>=0.2.0
+
+# Cache & search
+redis>=5.0.0
+elasticsearch>=8.0.0
+pymilvus>=2.3.0
+
+# Auth
+PyJWT>=2.8.0
+
+# LLM (interaction)
+langchain>=0.3.0
+langchain-core>=0.3.0
+langchain-openai>=0.2.0
+
+# LLM
+openai>=1.0.0
+
+# LangChain
+langchain>=0.3.0
+langchain-core>=0.3.0
+langchain-openai>=0.2.0
+langchain-community>=0.3.0
+
+# aliyun log
+aliyun-log-python-sdk
+aliyun-python-sdk-core
+aliyun-python-sdk-kms

src/config/__init__.py

@@ -0,0 +1 @@
+from .agent_config import LongArticlesSearchAgentConfig

src/config/agent_config.py

@@ -0,0 +1,31 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from src.config.api import *
+from src.config.aliyun import *
+from src.config.database import *
+
+
+class LongArticlesSearchAgentConfig(BaseSettings):
+    """智能体配置"""
+
+    # ============ 应用基础配置 ============
+    app_name: str = Field(default="LongArticleSearchAgent", description="应用名称")
+    environment: str = Field(
+        default="development", description="运行环境: development/pre/production"
+    )
+    debug: bool = Field(default=False, description="调试模式")
+
+    # ============ 数据库配置 ============
+    search_agent_db: SearchAgentMySQLConfig = Field(default_factory=SearchAgentMySQLConfig)
+
+    # ============ 外部服务配置 ============
+    deepseek: DeepSeekConfig = Field(default_factory=DeepSeekConfig)
+
+    aliyun_log: AliyunLogConfig = Field(default_factory=AliyunLogConfig)
+    # elasticsearch: ElasticsearchConfig = Field(default_factory=ElasticsearchConfig)
+    # apollo: ApolloConfig = Field(default_factory=ApolloConfig)
+
+    model_config = SettingsConfigDict(
+        env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore"
+    )

src/config/aliyun/__init__.py

@@ -0,0 +1 @@
+from .log import AliyunLogConfig

src/config/aliyun/log.py

@@ -0,0 +1,25 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class AliyunLogConfig(BaseSettings):
+    """阿里云日志配置"""
+
+    endpoint: str = "cn-hangzhou.log.aliyuncs.com"
+    access_key_id: str = "LTAIP6x1l3DXfSxm"
+    access_key_secret: str = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
+    project: str = "changwen-alg"
+    logstore: str = "long_articles_job"
+
+    model_config = SettingsConfigDict(
+        env_prefix="ALIYUN_LOG_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+    def to_dict(self) -> dict:
+        """转换为字典格式，用于兼容旧代码"""
+        return {
+            "endpoint": self.endpoint,
+            "access_key_id": self.access_key_id,
+            "access_key_secret": self.access_key_secret,
+            "project": self.project,
+            "logstore": self.logstore,
+        }

src/config/api/__init__.py

@@ -0,0 +1 @@
+from .deepseek import DeepSeekConfig

src/config/api/deepseek.py

@@ -0,0 +1,26 @@
+"""DEEPSEEK 配置"""
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class DeepSeekConfig(BaseSettings):
+    api_key: str = Field(
+        default="", description="DeepSeek API Key（环境变量 DEEP_SEEK_API_KEY）"
+    )
+    model: str = Field(default="deepseek-chat", description="模型，如V3, R1")
+    reason_model: str = Field(default="", description="推理模型")
+    base_url: str = Field(
+        default="https://api.deepseek.com", description="base_url链接"
+    )
+
+    model_config = SettingsConfigDict(
+        env_prefix="DEEP_SEEK_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+
+__all__ = ["DeepSeekConfig"]

src/config/database/__init__.py

@@ -0,0 +1,4 @@
+from .mysql_config import SearchAgentMySQLConfig
+
+
+__all__ = ["SearchAgentMySQLConfig"]

src/config/database/mysql_config.py

@@ -0,0 +1,52 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class MySQLConfig(BaseSettings):
+    """数据库配置基类"""
+
+    host: str
+    port: int = 3306
+    user: str
+    password: str
+    db: str
+    charset: str = "utf8mb4"
+    minsize: int = 5
+    maxsize: int = 20
+
+    model_config = SettingsConfigDict(
+        env_prefix="", case_sensitive=False, extra="ignore"
+    )
+
+    def to_dict(self) -> dict:
+        """转换为字典格式，用于兼容旧代码"""
+        return {
+            "host": self.host,
+            "port": self.port,
+            "user": self.user,
+            "password": self.password,
+            "db": self.db,
+            "charset": self.charset,
+            "minsize": self.minsize,
+            "maxsize": self.maxsize,
+        }
+
+
+class SearchAgentMySQLConfig(MySQLConfig):
+    host: str = "localhost"
+    user: str = "root"
+    db: str = "better_me"
+    password: str = "ljh000118"
+
+    model_config = SettingsConfigDict(
+        env_prefix="SEARCH_AGENT_DB_",
+        env_file=".env",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    def async_sqlalchemy_url(self) -> str:
+        """SQLAlchemy 异步 DSN（aiomysql 驱动），供 base.py 与 models 使用。"""
+        from urllib.parse import quote_plus
+
+        pw = quote_plus(self.password)
+        return f"mysql+aiomysql://{self.user}:{pw}@{self.host}:{self.port}/{self.db}?charset={self.charset}"

src/core/api/v1/endpoints/__init__.py

@@ -0,0 +1 @@
+from .health import create_health_bp

src/core/api/v1/endpoints/health.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from quart import Blueprint, jsonify
+
+
+def create_health_bp() -> Blueprint:
+    bp = Blueprint("health", __name__)
+
+    @bp.route("/health", methods=["GET"])
+    async def health():
+        return jsonify(
+            {"code": 0, "message": "success", "data": {"message": "hello world"}}
+        )
+
+    return bp

src/core/api/v1/routes/__init__.py

@@ -0,0 +1,4 @@
+from .route import server_routes
+
+
+__all__ = ["server_routes"]

src/core/api/v1/routes/route.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from quart import Blueprint
+
+from src.core.api.v1.utils import ApiDependencies
+from src.core.api.v1.endpoints import create_health_bp
+
+from src.config import LongArticlesSearchAgentConfig
+from src.infra.database import AsyncMySQLPool
+from src.infra.trace import LogService
+
+
+def register_v1_blueprints(deps: ApiDependencies) -> Blueprint:
+    """
+    v1 路由统一注册入口（按领域拆分）。
+
+    - /api/get_cover
+    - /api/run_task
+    - /api/tasks
+    - /api/save_token
+    - /api/health
+    """
+    api = Blueprint("api", __name__, url_prefix="/api")
+
+    api.register_blueprint(create_health_bp())
+
+    return api
+
+
+def server_routes(
+    pools: AsyncMySQLPool, log_service: LogService, config: LongArticlesSearchAgentConfig
+) -> Blueprint:
+    """
+    兼容旧入口：保留 server_routes 签名，内部转为新的 deps + 统一注册。
+    """
+    deps = ApiDependencies(db=pools, log=log_service, config=config)
+    return register_v1_blueprints(deps)

src/core/api/v1/utils/__init__.py

@@ -0,0 +1 @@
+from .deps import ApiDependencies

src/core/api/v1/utils/deps.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from src.config import LongArticlesSearchAgentConfig
+from src.infra.database import AsyncMySQLPool
+from src.infra.trace import LogService
+
+
+@dataclass(frozen=True)
+class ApiDependencies:
+    """API 层依赖容器：统一管理 db/log/config 等依赖。"""
+
+    db: AsyncMySQLPool
+    log: LogService
+    config: LongArticlesSearchAgentConfig

src/core/bootstrap/__init__.py

@@ -0,0 +1 @@
+from .resource_manager import AppContext

src/core/bootstrap/resource_manager.py

@@ -0,0 +1,35 @@
+import logging
+from src.core.dependency import ServerContainer
+
+logger = logging.getLogger(__name__)
+
+
+class AppContext:
+    def __init__(self, container: ServerContainer):
+        self.container = container
+
+    async def start_up(self):
+        logger.info("初始化数据库连接池")
+        mysql = self.container.mysql_pool()
+        await mysql.init_pools()
+        logger.info("Mysql pools init successfully")
+
+        logger.info("初始化日志服务")
+        log_service = self.container.log_service()
+        await log_service.start()
+        logger.info("aliyun log service init successfully")
+
+    async def shutdown(self):
+        logger.info("关闭数据库连接池")
+        mysql = self.container.mysql_pool()
+        await mysql.close_pools()
+        logger.info("应用资源已释放")
+        logger.info("关闭日志服务")
+        log_service = self.container.log_service()
+        await log_service.stop()
+        logger.info("aliyun log service stopped")
+
+
+__all__ = [
+    "AppContext",
+]

src/core/dependency/__init__.py

@@ -0,0 +1,6 @@
+from .dependencies import ServerContainer
+
+
+__all__ = [
+    "ServerContainer",
+]

src/core/dependency/dependencies.py

@@ -0,0 +1,21 @@
+from dependency_injector import containers, providers
+
+from src.config import LongArticlesSearchAgentConfig
+from src.infra.database import AsyncMySQLPool
+from src.infra.trace import LogService
+
+
+class ServerContainer(containers.DeclarativeContainer):
+    # config
+    config = providers.Singleton(LongArticlesSearchAgentConfig)
+
+    # 阿里云日志
+    log_service = providers.Singleton(LogService, log_config=config.provided.aliyun_log)
+
+    # MySQL
+    mysql_pool = providers.Singleton(AsyncMySQLPool, config=config)
+
+
+__all__ = [
+    "ServerContainer",
+]

src/infra/database/__init__.py

@@ -0,0 +1,6 @@
+from .mysql import AsyncMySQLPool
+
+
+__all__ = [
+    "AsyncMySQLPool",
+]

src/infra/database/mysql/__init__.py

@@ -0,0 +1,4 @@
+from .async_mysql_pool import AsyncMySQLPool
+
+
+__all__ = ["AsyncMySQLPool"]

src/infra/database/mysql/async_mysql_pool.py

@@ -0,0 +1,118 @@
+import logging
+
+from aiomysql import create_pool
+from aiomysql.cursors import DictCursor
+
+from src.config import LongArticlesSearchAgentConfig
+from src.infra.trace import LogService
+
+
+logging.basicConfig(level=logging.INFO)
+
+
+class AsyncMySQLPool(LogService):
+    def __init__(self, config: LongArticlesSearchAgentConfig):
+        super().__init__(config.aliyun_log)
+        self.database_mapper = {
+            "search_agent": config.search_agent_db,
+        }
+        self.pools = {}
+
+    async def init_pools(self):
+        # 从配置获取数据库配置，也可以直接在这里配置
+        for db_name, config in self.database_mapper.items():
+            try:
+                pool = await create_pool(
+                    host=config.host,
+                    port=config.port,
+                    user=config.user,
+                    password=config.password,
+                    db=config.db,
+                    minsize=config.minsize,
+                    maxsize=config.maxsize,
+                    cursorclass=DictCursor,
+                    autocommit=True,
+                )
+                self.pools[db_name] = pool
+                logging.info(f"{db_name} MYSQL连接池 created successfully")
+
+            except Exception as e:
+                await self.log(
+                    contents={
+                        "db_name": db_name,
+                        "error": str(e),
+                        "message": f"Failed to create pool for {db_name}",
+                    }
+                )
+                self.pools[db_name] = None
+
+    async def close_pools(self):
+        for name, pool in self.pools.items():
+            if pool:
+                pool.close()
+                await pool.wait_closed()
+                logging.info(f"{name} MYSQL连接池 closed successfully")
+
+    async def async_fetch(
+        self, query, db_name="long_articles", params=None, cursor_type=DictCursor
+    ):
+        pool = self.pools[db_name]
+        if not pool:
+            await self.init_pools()
+        # fetch from db
+        try:
+            async with pool.acquire() as conn:
+                async with conn.cursor(cursor_type) as cursor:
+                    await cursor.execute(query, params)
+                    fetch_response = await cursor.fetchall()
+
+            return fetch_response
+        except Exception as e:
+            await self.log(
+                contents={
+                    "task": "async_fetch",
+                    "db_name": db_name,
+                    "error": str(e),
+                    "message": f"Failed to fetch data from {db_name}",
+                    "query": query,
+                    "params": params,
+                }
+            )
+            return None
+
+    async def async_save(
+        self, query, params, db_name="long_articles", batch: bool = False
+    ):
+        pool = self.pools[db_name]
+        if not pool:
+            await self.init_pools()
+
+        async with pool.acquire() as connection:
+            async with connection.cursor() as cursor:
+                try:
+                    if batch:
+                        await cursor.executemany(query, params)
+                    else:
+                        await cursor.execute(query, params)
+                    affected_rows = cursor.rowcount
+                    await connection.commit()
+                    return affected_rows
+                except Exception as e:
+                    await connection.rollback()
+                    await self.log(
+                        contents={
+                            "task": "async_save",
+                            "db_name": db_name,
+                            "error": str(e),
+                            "message": f"Failed to save data to {db_name}",
+                            "query": query,
+                            "params": params,
+                        }
+                    )
+                    raise e
+
+    def get_pool(self, db_name):
+        return self.pools.get(db_name)
+
+    def list_databases(self):
+        return list(self.database_mapper.keys())

src/infra/trace/__init__.py

@@ -0,0 +1,6 @@
+from .logging import LogService
+
+
+__all__ = [
+    "LogService",
+]

src/infra/trace/logging/__init__.py

@@ -0,0 +1 @@
+from .log_service import LogService

src/infra/trace/logging/log_service.py

@@ -0,0 +1,95 @@
+import asyncio
+import traceback
+import time, json
+import datetime
+import contextlib
+from typing import Optional
+
+from aliyun.log import LogClient, PutLogsRequest, LogItem
+from src.config.aliyun import AliyunLogConfig
+
+
+class LogService:
+    def __init__(self, log_config: AliyunLogConfig):
+        self.config = log_config
+
+        self.client: Optional[LogClient] = None
+        self.queue: Optional[asyncio.Queue] = None
+
+        self._worker_task: Optional[asyncio.Task] = None
+        self._running = False
+
+    async def start(self):
+        if self._running:
+            return
+
+        self.client = LogClient(
+            self.config.endpoint,
+            self.config.access_key_id,
+            self.config.access_key_secret,
+        )
+        self.queue = asyncio.Queue(maxsize=10000)
+
+        self._running = True
+        self._worker_task = asyncio.create_task(self._worker())
+
+    async def stop(self):
+        if not self._running:
+            return
+
+        self._running = False
+
+        if self._worker_task:
+            self._worker_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._worker_task
+
+        self._worker_task = None
+        self.queue = None
+        self.client = None
+
+    async def log(self, contents: dict):
+        if not self._running or self.queue is None:
+            return
+
+        try:
+            self.queue.put_nowait(contents)
+        except asyncio.QueueFull:
+            # 可以打 stderr / 统计丢日志数量
+            pass
+
+    async def _worker(self):
+        try:
+            while self._running:
+                contents = await self.queue.get()
+                try:
+                    await asyncio.to_thread(self._put_log, contents)
+                except Exception as e:
+                    print(f"[Log Error] {e}")
+                    print(traceback.format_exc())
+        except asyncio.CancelledError:
+            pass
+
+    def _put_log(self, contents: dict):
+        timestamp = int(time.time())
+        contents["datetime"] = datetime.datetime.now().isoformat()
+
+        safe_items = [
+            (
+                str(k),
+                json.dumps(v, ensure_ascii=False)
+                if isinstance(v, (dict, list))
+                else str(v),
+            )
+            for k, v in contents.items()
+        ]
+
+        log_item = LogItem(timestamp=timestamp, contents=safe_items)
+        req = PutLogsRequest(
+            self.config.project,
+            self.config.logstore,
+            topic="",
+            source="",
+            logitems=[log_item],
+        )
+        self.client.put_logs(req)