diff --git a/docs/superpowers/plans/2026-05-09-fastapi-docker-migration.md b/docs/superpowers/plans/2026-05-09-fastapi-docker-migration.md new file mode 100644 index 0000000..db252be --- /dev/null +++ b/docs/superpowers/plans/2026-05-09-fastapi-docker-migration.md @@ -0,0 +1,2760 @@ +# FastAPI + Docker 迁移实施计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 将广西政府采购网公告监控系统从 Flask CLI 架构迁移到 FastAPI + Docker，含 Spider 模块化、SQLAlchemy ORM、APScheduler 和异步改造。 + +**Architecture:** 单体 FastAPI 应用，Spider 基类 + Pipeline 策略模式统一爬虫管理，PostgreSQL + asyncpg 异步数据库，APScheduler 内置定时任务，Docker 单容器部署连接外部 PostgreSQL。 + +**Tech Stack:** Python 3.12, FastAPI, SQLAlchemy 2.0 (async), asyncpg, httpx, APScheduler, pydantic-settings, Alembic, pytest, ruff, Docker, BeautifulSoup4, pycryptodome + +--- + +### Task 1: 创建项目骨架 + +**Files:** +- Create: `pyproject.toml` +- Create: `app/__init__.py` +- Create: `app/config.py` +- Create: `app/main.py` +- Create: `.env.example` +- Create: `.gitignore` + +- [ ] **Step 1: 创建 pyproject.toml** + +```toml +[project] +name = "gx-gp-notify" +version = "2.0.0" +description = "广西政府采购网公告监控系统" +requires-python = ">=3.12" +dependencies = [ + "fastapi>=0.115.0", + "uvicorn[standard]>=0.30.0", + "sqlalchemy[asyncio]>=2.0.30", + "asyncpg>=0.29.0", + "alembic>=1.13.0", + "httpx>=0.27.0", + "apscheduler>=3.10.0", + "pydantic-settings>=2.3.0", + "beautifulsoup4>=4.12.0", + "lxml>=5.2.0", + "pycryptodome>=3.20.0", + "python-dateutil>=2.9.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.2.0", + "pytest-asyncio>=0.23.0", + "pytest-cov>=5.0.0", + "httpx>=0.27.0", + "ruff>=0.4.0", +] + +[tool.ruff] +target-version = "py312" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I", "N", "W", "UP"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] +``` + +- [ ] **Step 2: 创建 .env.example** + +```bash +# 应用 +DEBUG=false +LOG_LEVEL=INFO + +# 数据库 +DATABASE_URL=postgresql+asyncpg://gx-gp-notify:password@10.10.10.14:5432/gx-gp-notify + +# 爬虫 +CRAWLER_BASE_URL=https://zfcg.gxzf.gov.cn +CRAWLER_KEYWORDS=["大化"] +CRAWLER_MAX_PAGES=10 +CRAWLER_TIMEOUT=30 + +# 企业微信 +WECHAT_ENABLED=true +WECHAT_CORP_ID=ww69e8e44636f47780 +WECHAT_AGENT_ID=1000007 +WECHAT_SECRET= +WECHAT_TOKEN= +WECHAT_ENCODING_AES_KEY= +WECHAT_PORT=18001 +WECHAT_HOST=0.0.0.0 + +# 定时任务 +SCHEDULER_ENABLED=true +SCHEDULER_CRON=0 8,14,18 * * * + +# Markdown +MARKDOWN_ENABLED=true +MARKDOWN_OUTPUT_FILE=onu.md +``` + +- [ ] **Step 3: 创建 .gitignore** + +```gitignore +.env +logs/ +*.log +__pycache__/ +*.pyc +.venv/ +.ruff_cache/ +.pytest_cache/ +*.egg-info/ +dist/ +``` + +- [ ] **Step 4: 创建 app/__init__.py** (空文件) + +- [ ] **Step 5: 创建 app/config.py** + +```python +from pydantic_settings import BaseSettings, SettingsConfigDict +from typing import List + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") + + # 应用 + debug: bool = False + log_level: str = "INFO" + + # 数据库 + database_url: str = "postgresql+asyncpg://gx-gp-notify:password@localhost:5432/gx-gp-notify" + + # 爬虫 + crawler_base_url: str = "https://zfcg.gxzf.gov.cn" + crawler_keywords: List[str] = ["大化"] + crawler_max_pages: int = 10 + crawler_timeout: int = 30 + crawler_page_size: int = 100 + + # 企业微信 + wechat_enabled: bool = True + wechat_corp_id: str = "" + wechat_agent_id: str = "" + wechat_secret: str = "" + wechat_token: str = "" + wechat_encoding_aes_key: str = "" + wechat_port: int = 18001 + wechat_host: str = "0.0.0.0" + + # 定时任务 + scheduler_enabled: bool = True + scheduler_cron: str = "0 8,14,18 * * *" + + # Markdown + markdown_enabled: bool = True + markdown_output_file: str = "onu.md" + + # 公告来源（JSON 字符串，从环境变量读取） + announcement_sources: str = '{"ZcyAnnouncement1":{"category_id":66485,"name":"采购公告","type":"purchase"},"ZcyAnnouncement2":{"category_id":66485,"name":"结果公告","type":"result"},"ZcyAnnouncement3":{"category_id":66485,"name":"合同公告","type":"contract"},"ZcyAnnouncement4":{"category_id":66485,"name":"更正公告","type":"correction"},"ZcyAnnouncement5":{"category_id":66485,"name":"招标文件预公示","type":"pre_announcement"},"ZcyAnnouncement6":{"category_id":66485,"name":"单一来源公示","type":"single_source"},"ZcyAnnouncement7":{"category_id":66485,"name":"电子卖场公示","type":"electronic_market"},"ZcyAnnouncement10":{"category_id":66485,"name":"履约验收公示","type":"acceptance"},"ZcyAnnouncement11":{"category_id":66485,"name":"工程类公告","type":"engineering"},"ZcyAnnouncement20":{"category_id":66485,"name":"框架协议征集公告","type":"framework_agreement"},"ZcyAnnouncement21":{"category_id":66485,"name":"框架协议入围结果公告","type":"framework_result"},"ZcyAnnouncement23":{"category_id":66485,"name":"框架协议成交结果汇总公告","type":"framework_summary"},"61-266648":{"category_id":66485,"name":"采购意向公开","type":"intention"}}' + + +settings = Settings() +``` + +- [ ] **Step 6: 创建 app/main.py** + +```python +import logging +from contextlib import asynccontextmanager +from fastapi import FastAPI +from app.config import settings + + +@asynccontextmanager +async def lifespan(app: FastAPI): + # 启动时初始化 + logging.basicConfig(level=settings.log_level) + yield + # 关闭时清理 + + +app = FastAPI( + title="广西政府采购网公告监控系统", + version="2.0.0", + lifespan=lifespan, + docs_url="/docs" if settings.debug else None, + redoc_url=None, +) + + +@app.get("/health") +async def health(): + return {"status": "ok"} +``` + +- [ ] **Step 7: 安装依赖并验证** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +pip install -e ".[dev]" +uvicorn app.main:app --host 0.0.0.0 --port 8000 & +sleep 2 +curl http://localhost:8000/health +# Expected: {"status":"ok"} +kill %1 +``` + +- [ ] **Step 8: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add pyproject.toml app/__init__.py app/config.py app/main.py .env.example .gitignore +git commit -m "feat: 创建 FastAPI 项目骨架（config + main + 依赖管理）" +``` + +--- + +### Task 2: 数据库模型 + Alembic + +**Files:** +- Create: `app/models/__init__.py` +- Create: `app/models/announcement.py` +- Create: `app/models/schemas.py` +- Create: `alembic.ini` +- Create: `alembic/env.py` +- Create: `alembic/script.py.mako` +- Modify: `app/main.py` — 添加数据库引擎 + +- [ ] **Step 1: 创建 app/models/__init__.py** (空文件) + +- [ ] **Step 2: 创建 app/models/announcement.py** + +```python +import hashlib +from datetime import datetime, date +from sqlalchemy import String, Boolean, DateTime, Integer, Text, func +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +class Base(DeclarativeBase): + pass + + +class Announcement(Base): + __tablename__ = "announcements" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + title: Mapped[str] = mapped_column(String(500), nullable=False) + publish_date: Mapped[datetime] = mapped_column(DateTime, nullable=False) + purchase_name: Mapped[str] = mapped_column(String(200), default="") + content_url: Mapped[str] = mapped_column(Text, default="") + source_code: Mapped[str] = mapped_column(String(50), nullable=False) + source_name: Mapped[str] = mapped_column(String(100), nullable=False) + announcement_type: Mapped[str] = mapped_column(String(50), default="purchase") + content_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) + crawl_mode: Mapped[str] = mapped_column(String(20), default="auto") + is_new: Mapped[bool] = mapped_column(Boolean, default=True) + is_sent: Mapped[bool] = mapped_column(Boolean, default=False) + keyword_matched: Mapped[bool] = mapped_column(Boolean, default=False) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now()) + + @staticmethod + def generate_hash(title: str, publish_date: str, purchase_name: str, + content_url: str, source_code: str) -> str: + content = f"{title}|{publish_date}|{purchase_name}|{content_url}|{source_code}" + return hashlib.sha256(content.encode("utf-8")).hexdigest() + + @staticmethod + def source_map() -> dict: + import json + from app.config import settings + return json.loads(settings.announcement_sources) +``` + +- [ ] **Step 3: 创建 app/models/schemas.py** + +```python +from datetime import datetime +from typing import Optional +from pydantic import BaseModel + + +class AnnouncementResponse(BaseModel): + id: int + title: str + publish_date: datetime + purchase_name: str + content_url: str + source_code: str + source_name: str + announcement_type: str + crawl_mode: str + is_new: bool + is_sent: bool + keyword_matched: bool + created_at: datetime + + model_config = {"from_attributes": True} + + +class AnnouncementListResponse(BaseModel): + total: int + page: int + page_size: int + items: list[AnnouncementResponse] + + +class CrawlTriggerRequest(BaseModel): + keywords: Optional[list[str]] = None + sources: Optional[list[str]] = None + manual: bool = False + + +class CrawlStatusResponse(BaseModel): + running: bool + last_crawl_time: Optional[datetime] = None + total_sources: int + + +class SourceInfo(BaseModel): + code: str + name: str + type: str + + +class SourcesResponse(BaseModel): + sources: list[SourceInfo] + + +class JobResponse(BaseModel): + id: str + name: str + next_run_time: Optional[str] = None +``` + +- [ ] **Step 4: 初始化 Alembic** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +pip install alembic +alembic init alembic +``` + +- [ ] **Step 5: 修改 alembic/env.py** + +Read the generated file, replace its content with: + +```python +from alembic import context +from sqlalchemy import engine_from_config, pool +from app.models.announcement import Base + +config = context.config +config.set_main_option("sqlalchemy.url", "postgresql+asyncpg://placeholder:placeholder@localhost:5432/placeholder") + +target_metadata = Base.metadata + + +def run_migrations_offline(): + url = config.get_main_option("sqlalchemy.url") + context.configure(url=url, target_metadata=target_metadata, literal_binds=True) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online(): + from app.config import settings + connectable = engine_from_config( + {"sqlalchemy.url": settings.database_url}, + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() +``` + +- [ ] **Step 6: 生成初始迁移** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +DATABASE_URL=postgresql+asyncpg://gx-gp-notify:MA6RBX4F6Bd5DGmw@10.10.10.14:5432/gx-gp-notify \ + alembic revision --autogenerate -m "create_announcements_table" +# Expected: Generating .../alembic/versions/xxxx_create_announcements_table.py ... done +``` + +- [ ] **Step 7: 运行迁移** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +DATABASE_URL=postgresql+asyncpg://gx-gp-notify:MA6RBX4F6Bd5DGmw@10.10.10.14:5432/gx-gp-notify \ + alembic upgrade head +# Expected: Running upgrade ... -> xxxx, create announcements table +``` + +- [ ] **Step 8: 更新 app/main.py 添加数据库初始化** + +```python +import logging +from contextlib import asynccontextmanager +from fastapi import FastAPI +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession +from app.config import settings +from app.models.announcement import Base + +engine = create_async_engine(settings.database_url, echo=settings.debug) +async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + +async def get_db() -> AsyncSession: + async with async_session() as session: + yield session + + +@asynccontextmanager +async def lifespan(app: FastAPI): + logging.basicConfig( + level=getattr(logging, settings.log_level), + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + ) + yield + await engine.dispose() + + +app = FastAPI( + title="广西政府采购网公告监控系统", + version="2.0.0", + lifespan=lifespan, + docs_url="/docs" if settings.debug else None, + redoc_url=None, +) + + +@app.get("/health") +async def health(): + return {"status": "ok"} +``` + +- [ ] **Step 9: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/models/ alembic/ alembic.ini app/main.py +git commit -m "feat: 添加 SQLAlchemy 模型 + Alembic 数据库迁移" +``` + +--- + +### Task 3: Spider 基类 + Pipeline 配置 + +**Files:** +- Create: `app/crawler/__init__.py` +- Create: `app/crawler/base.py` +- Test: `tests/test_crawler/test_base.py` +- Create: `tests/__init__.py` +- Create: `tests/test_crawler/__init__.py` +- Create: `tests/conftest.py` + +- [ ] **Step 1: 创建 app/crawler/__init__.py** (空文件) + +- [ ] **Step 2: 创建 app/crawler/base.py** + +```python +import hashlib +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from datetime import datetime +from typing import List, Optional + + +@dataclass +class CrawlResult: + source_code: str + source_name: str + total_count: int = 0 + new_count: int = 0 + announcements: list = field(default_factory=list) + error_message: Optional[str] = None + crawled_at: datetime = field(default_factory=datetime.now) + duration: float = 0.0 + + @property + def success(self) -> bool: + return self.error_message is None + + +@dataclass +class PipelineConfig: + filter_enabled: bool = True + keywords: List[str] = field(default_factory=list) + dedup_enabled: bool = True + notify_mode: str = "filtered" + mark_sent: bool = False + + +@dataclass +class PipelineResult: + stored: int = 0 + filtered: int = 0 + notified: int = 0 + markdown_generated: bool = False + + +class BaseSpider(ABC): + name: str + source_code: str + source_name: str + + @abstractmethod + async def crawl(self) -> CrawlResult: + ... + + def get_pipeline_config(self) -> PipelineConfig: + return PipelineConfig() + + @staticmethod + def generate_content_hash(title: str, publish_date: str, purchase_name: str, + content_url: str, source_code: str) -> str: + content = f"{title}|{publish_date}|{purchase_name}|{content_url}|{source_code}" + return hashlib.sha256(content.encode("utf-8")).hexdigest() +``` + +- [ ] **Step 3: 创建 tests/__init__.py, tests/test_crawler/__init__.py** (空文件) + +- [ ] **Step 4: 创建 tests/conftest.py** + +```python +import pytest + + +@pytest.fixture +def sample_announcement_data(): + return { + "title": "测试公告标题", + "publish_date": "2026-05-09", + "purchase_name": "测试单位", + "content_url": "https://example.com/detail/123", + "source_code": "ZcyAnnouncement1", + "source_name": "采购公告", + } +``` + +- [ ] **Step 5: 创建 tests/test_crawler/test_base.py** + +```python +import pytest +from datetime import datetime +from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig, PipelineResult + + +class FakeSpider(BaseSpider): + name = "test_spider" + source_code = "test_source" + source_name = "测试来源" + + async def crawl(self) -> CrawlResult: + return CrawlResult( + source_code=self.source_code, + source_name=self.source_name, + total_count=5, + new_count=3, + crawled_at=datetime.now(), + ) + + def get_pipeline_config(self) -> PipelineConfig: + return PipelineConfig( + filter_enabled=True, + keywords=["测试"], + dedup_enabled=True, + notify_mode="filtered", + mark_sent=False, + ) + + +@pytest.mark.asyncio +async def test_base_spider_crawl(): + spider = FakeSpider() + result = await spider.crawl() + assert result.source_code == "test_source" + assert result.total_count == 5 + assert result.new_count == 3 + assert result.success is True + + +def test_pipeline_config_defaults(): + config = PipelineConfig() + assert config.filter_enabled is True + assert config.keywords == [] + assert config.dedup_enabled is True + assert config.notify_mode == "filtered" + assert config.mark_sent is False + + +def test_pipeline_result_defaults(): + result = PipelineResult() + assert result.stored == 0 + assert result.filtered == 0 + assert result.notified == 0 + assert result.markdown_generated is False + + +def test_generate_content_hash(): + h = BaseSpider.generate_content_hash( + "title", "2026-05-09", "unit", "https://x.com", "source" + ) + assert len(h) == 64 + assert all(c in "0123456789abcdef" for c in h) + + +def test_generate_content_hash_deterministic(): + h1 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s") + h2 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s") + assert h1 == h2 + + +def test_crawl_result_failure(): + result = CrawlResult( + source_code="s", source_name="n", + error_message="连接超时" + ) + assert result.success is False +``` + +- [ ] **Step 6: 运行测试** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_crawler/test_base.py -v +# Expected: 6 passed +``` + +- [ ] **Step 7: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/crawler/ tests/ +git commit -m "feat: 添加 Spider 基类 + Pipeline 配置 + 测试" +``` + +--- + +### Task 4: Parssers 模块 + +**Files:** +- Create: `app/crawler/parsers.py` +- Test: `tests/test_crawler/test_parsers.py` + +- [ ] **Step 1: 创建测试 tests/test_crawler/test_parsers.py** + +```python +import json +from datetime import datetime +from app.crawler.parsers import ( + parse_gxgp_api_response, + parse_dahuagov_html, + extract_pagination, +) + + +def make_api_response(records_data): + return { + "success": True, + "result": { + "data": { + "data": records_data, + "total": len(records_data), + "pageNo": 1, + "pageSize": 100, + "pages": 1, + "empty": len(records_data) == 0, + "hasNext": False, + "hasPrevious": False, + } + }, + } + + +def test_parse_gxgp_single_record(): + response = make_api_response([ + { + "title": "测试采购公告", + "publishDate": 1746720000000, + "purchaseName": "测试采购单位", + "articleId": 12345, + } + ]) + crawled_at = datetime(2026, 5, 9, 10, 0, 0) + results = parse_gxgp_api_response( + response, + source_code="ZcyAnnouncement1", + source_name="采购公告", + crawled_at=crawled_at, + category_id=66485, + ) + assert len(results) == 1 + assert results[0]["title"] == "测试采购公告" + assert results[0]["source_code"] == "ZcyAnnouncement1" + assert "content_hash" in results[0] + + +def test_parse_gxgp_empty_response(): + response = make_api_response([]) + crawled_at = datetime(2026, 5, 9, 10, 0, 0) + results = parse_gxgp_api_response( + response, "ZcyAnnouncement1", "采购公告", crawled_at, 66485 + ) + assert len(results) == 0 + + +def test_parse_gxgp_missing_title(): + response = make_api_response([ + {"title": "", "publishDate": 1746720000000, "purchaseName": "x", "articleId": 1} + ]) + crawled_at = datetime(2026, 5, 9, 10, 0, 0) + results = parse_gxgp_api_response( + response, "ZcyAnnouncement1", "采购公告", crawled_at, 66485 + ) + assert len(results) == 0 + + +def test_extract_pagination(): + response = make_api_response([]) + pagination = extract_pagination(response) + assert pagination["total"] == 0 + assert pagination["page_no"] == 1 + assert pagination["has_next"] is False + + +def test_parse_dahuagov_html(): + html = """ + + + + """ + crawled_at = datetime(2026, 5, 9, 10, 0, 0) + results = parse_dahuagov_html(html, crawled_at) + assert len(results) == 2 + assert results[0]["title"] == "大化县某项目采购公告" + assert results[0]["source_code"] == "dahuagov" + assert results[0]["source_name"] == "大化县政府网采购公告" + assert results[0]["purchase_name"] == "大化瑶族自治县" + + +def test_parse_dahuagov_html_no_list(): + html = "" + crawled_at = datetime(2026, 5, 9, 10, 0, 0) + results = parse_dahuagov_html(html, crawled_at) + assert len(results) == 0 +``` + +- [ ] **Step 2: 运行测试确认失败** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_crawler/test_parsers.py -v +# Expected: FAIL — ModuleNotFoundError or ImportError +``` + +- [ ] **Step 3: 创建 app/crawler/parsers.py** + +```python +import hashlib +from datetime import datetime +from typing import Any, Dict, List, Optional +from urllib.parse import urljoin +from bs4 import BeautifulSoup + + +def parse_gxgp_api_response( + response_data: Dict[str, Any], + source_code: str, + source_name: str, + crawled_at: datetime, + category_id: int, +) -> List[Dict[str, Any]]: + if not response_data.get("success"): + return [] + data = response_data.get("result", {}).get("data", {}) + records = data.get("data", []) + if not records: + return [] + + results = [] + for record in records: + title = str(record.get("title", "")).strip() + if not title: + continue + + timestamp = record.get("publishDate") + if not timestamp: + continue + try: + publish_date = datetime.fromtimestamp(int(timestamp) / 1000) + except (ValueError, TypeError): + continue + + purchase_name = str(record.get("purchaseName", "")).strip() + article_id = record.get("articleId") + if not article_id: + continue + + content_url = f"https://zfcg.gxzf.gov.cn/site/detail?parentId={category_id}&articleId={article_id}" + + announce = { + "title": title, + "publish_date": publish_date, + "purchase_name": purchase_name, + "content_url": content_url, + "source_code": source_code, + "source_name": source_name, + "announcement_type": "purchase", + "crawl_mode": "auto", + "is_new": True, + "is_today": publish_date.date() == datetime.now().date(), + } + announce["content_hash"] = _generate_hash(announce) + results.append(announce) + + return results + + +def extract_pagination(response_data: Dict[str, Any]) -> Dict[str, Any]: + data = response_data.get("result", {}).get("data", {}) + return { + "total": data.get("total", 0), + "page_no": data.get("pageNo", 1), + "page_size": data.get("pageSize", 100), + "pages": data.get("pages", 0), + "empty": data.get("empty", True), + "has_next": data.get("hasNext", False), + "has_previous": data.get("hasPrevious", False), + } + + +def parse_dahuagov_html(html: str, crawled_at: datetime) -> List[Dict[str, Any]]: + soup = BeautifulSoup(html, "html.parser") + lists = soup.find_all("ul", class_="more-list") + if not lists: + return [] + + results = [] + base_url = "http://www.gxdh.gov.cn" + base_path = "/xxgk/zdlyxxgk/ggzypzly/zfcgly/cggg/" + + for ul in lists: + for li in ul.find_all("li"): + date_span = li.find("span") + if not date_span: + continue + date_text = date_span.get_text(strip=True) + try: + publish_date = datetime.strptime(date_text, "%Y-%m-%d") + except ValueError: + continue + + link_tag = li.find("a") + if not link_tag: + continue + title = link_tag.get("title", "") or link_tag.get_text(strip=True) + href = link_tag.get("href", "") + if not title or not href: + continue + + if href.startswith("./") or href.startswith("../"): + content_url = urljoin(base_url + base_path, href) + elif href.startswith("/"): + content_url = base_url + href + elif href.startswith("http"): + content_url = href + else: + content_url = urljoin(base_url + base_path, href) + + announce = { + "title": title, + "publish_date": publish_date, + "purchase_name": "大化瑶族自治县", + "content_url": content_url, + "source_code": "dahuagov", + "source_name": "大化县政府网采购公告", + "announcement_type": "purchase", + "crawl_mode": "auto", + "is_new": True, + "is_today": publish_date.date() == datetime.now().date(), + } + announce["content_hash"] = _generate_hash(announce) + results.append(announce) + + return results + + +def _generate_hash(ann: Dict[str, Any]) -> str: + content = ( + f"{ann['title']}|{ann['publish_date'].strftime('%Y-%m-%d')}" + f"|{ann['purchase_name']}|{ann['content_url']}|{ann['source_code']}" + ) + return hashlib.sha256(content.encode("utf-8")).hexdigest() +``` + +- [ ] **Step 4: 运行测试验证通过** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_crawler/test_parsers.py -v +# Expected: 6 passed +``` + +- [ ] **Step 5: 运行全部测试** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/ -v +# Expected: 12 passed +``` + +- [ ] **Step 6: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/crawler/parsers.py tests/test_crawler/test_parsers.py +git commit -m "feat: 添加数据解析器模块 + 测试" +``` + +--- + +### Task 5: GXGP Spider（广西政府采购网） + +**Files:** +- Create: `app/crawler/gxgp_spider.py` +- Test: `tests/test_crawler/test_gxgp_spider.py` + +- [ ] **Step 1: 创建测试 tests/test_crawler/test_gxgp_spider.py** + +```python +import pytest +from datetime import datetime +from unittest.mock import AsyncMock, patch, MagicMock +from app.crawler.gxgp_spider import GXGPSpider +from app.crawler.base import PipelineConfig + + +@pytest.mark.asyncio +async def test_gxgp_spider_attributes(): + spider = GXGPSpider() + assert spider.name == "gxgp" + assert spider.source_code == "gxgp" + assert spider.source_name == "广西政府采购网" + + +@pytest.mark.asyncio +async def test_gxgp_spider_pipeline_config(): + spider = GXGPSpider() + config = spider.get_pipeline_config() + assert isinstance(config, PipelineConfig) + assert config.filter_enabled is True + assert config.notify_mode == "filtered" + assert config.mark_sent is False + + +@pytest.mark.asyncio +async def test_gxgp_spider_crawl_empty(): + spider = GXGPSpider() + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "success": True, + "result": {"data": {"data": [], "total": 0, "pageNo": 1, "pageSize": 100, + "pages": 0, "empty": True, "hasNext": False, "hasPrevious": False}}, + } + + with patch("httpx.AsyncClient.post", AsyncMock(return_value=mock_response)): + result = await spider.crawl() + assert result.total_count == 0 + assert len(result.announcements) == 0 + assert result.success is True + + +@pytest.mark.asyncio +async def test_gxgp_spider_crawl_with_data(): + spider = GXGPSpider() + + mock_data = { + "success": True, + "result": { + "data": { + "data": [{ + "title": "测试采购公告", + "publishDate": 1746720000000, + "purchaseName": "测试单位", + "articleId": 12345, + }], + "total": 1, "pageNo": 1, "pageSize": 100, + "pages": 1, "empty": False, "hasNext": False, "hasPrevious": False, + } + }, + } + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = mock_data + + with patch("httpx.AsyncClient.post", AsyncMock(return_value=mock_response)): + result = await spider.crawl(sources=["ZcyAnnouncement1"]) + assert result.total_count >= 0 + assert result.success is True +``` + +- [ ] **Step 2: 运行测试确认失败** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_crawler/test_gxgp_spider.py -v +# Expected: FAIL — ImportError +``` + +- [ ] **Step 3: 创建 app/crawler/gxgp_spider.py** + +```python +import json +import random +import time +from datetime import datetime +from typing import List, Optional +import httpx +from app.config import settings +from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig +from app.crawler.parsers import parse_gxgp_api_response, extract_pagination + + +class GXGPSpider(BaseSpider): + name = "gxgp" + source_code = "gxgp" + source_name = "广西政府采购网" + + def __init__(self): + self.base_url = settings.crawler_base_url + self.announcement_api = f"{self.base_url}/portal/category" + + def get_pipeline_config(self) -> PipelineConfig: + return PipelineConfig( + filter_enabled=True, + keywords=list(settings.crawler_keywords), + dedup_enabled=True, + notify_mode="filtered", + mark_sent=False, + ) + + async def crawl(self, sources: Optional[List[str]] = None, + max_pages: Optional[int] = None) -> CrawlResult: + if max_pages is None: + max_pages = settings.crawler_max_pages + if sources is None: + source_map = json.loads(settings.announcement_sources) + sources = list(source_map.keys()) + + start_time = datetime.now() + all_announcements = [] + error_messages = [] + + async with httpx.AsyncClient(timeout=settings.crawler_timeout) as client: + for source_code in sources: + source_info = json.loads(settings.announcement_sources).get(source_code) + if not source_info: + continue + + category_id = source_info["category_id"] + source_name = source_info["name"] + + for page_no in range(1, max_pages + 1): + if page_no > 1: + await self._delay() + + try: + data = await self._fetch_page( + client, source_code, category_id, page_no + ) + if data is None: + break + + records = parse_gxgp_api_response( + data, source_code, source_name, + start_time, category_id + ) + if not records: + break + + all_announcements.extend(records) + + pagination = extract_pagination(data) + if not pagination["has_next"]: + break + except Exception as e: + error_messages.append(f"{source_code} page {page_no}: {e}") + break + + duration = (datetime.now() - start_time).total_seconds() + return CrawlResult( + source_code=self.source_code, + source_name=self.source_name, + total_count=len(all_announcements), + new_count=len(all_announcements), + announcements=all_announcements, + error_message="; ".join(error_messages) if error_messages else None, + crawled_at=start_time, + duration=duration, + ) + + async def _fetch_page(self, client: httpx.AsyncClient, source_code: str, + category_id: int, page_no: int) -> Optional[dict]: + payload = { + "keyword": "", + "publishDateBegin": "", + "publishDateEnd": "", + "pageNo": page_no, + "pageSize": settings.crawler_page_size, + "categoryCode": source_code, + "_t": int(time.time() * 1000), + } + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "Content-Type": "application/json;charset=UTF-8", + "Origin": self.base_url, + "Referer": f"{self.base_url}/site/category?parentId={category_id}&childrenCode={source_code}", + } + response = await client.post( + self.announcement_api, json=payload, headers=headers + ) + if response.status_code != 200: + return None + return response.json() + + async def _delay(self): + import asyncio + delay = random.uniform(1.0, 3.0) + await asyncio.sleep(delay) +``` + +- [ ] **Step 4: 运行测试验证通过** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_crawler/test_gxgp_spider.py -v +# Expected: 3 passed +``` + +- [ ] **Step 5: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/crawler/gxgp_spider.py tests/test_crawler/test_gxgp_spider.py +git commit -m "feat: 添加 GXGP Spider（广西政府采购网爬虫）+ 测试" +``` + +--- + +### Task 6: Dahuagov Spider（大化县政府网） + +**Files:** +- Create: `app/crawler/dahuagov_spider.py` +- Test: `tests/test_crawler/test_dahuagov_spider.py` + +- [ ] **Step 1: 创建测试 tests/test_crawler/test_dahuagov_spider.py** + +```python +import pytest +from unittest.mock import AsyncMock, patch, MagicMock +from app.crawler.dahuagov_spider import DahuagovSpider +from app.crawler.base import PipelineConfig + + +@pytest.mark.asyncio +async def test_dahuagov_spider_attributes(): + spider = DahuagovSpider() + assert spider.name == "dahuagov" + assert spider.source_code == "dahuagov" + assert spider.source_name == "大化县政府网采购公告" + + +@pytest.mark.asyncio +async def test_dahuagov_spider_pipeline_config(): + spider = DahuagovSpider() + config = spider.get_pipeline_config() + assert isinstance(config, PipelineConfig) + assert config.filter_enabled is False + assert config.notify_mode == "all" + assert config.mark_sent is True + + +@pytest.mark.asyncio +async def test_dahuagov_spider_crawl(): + spider = DahuagovSpider() + + html = """ + + + + """ + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = html + + with patch("httpx.AsyncClient.get", AsyncMock(return_value=mock_response)): + result = await spider.crawl() + assert result.total_count == 2 + assert len(result.announcements) == 2 + assert result.success is True + + +@pytest.mark.asyncio +async def test_dahuagov_spider_crawl_empty(): + spider = DahuagovSpider() + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = "" + + with patch("httpx.AsyncClient.get", AsyncMock(return_value=mock_response)): + result = await spider.crawl() + assert result.total_count == 0 + assert result.success is True +``` + +- [ ] **Step 2: 运行测试确认失败** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_crawler/test_dahuagov_spider.py -v +# Expected: FAIL — ImportError +``` + +- [ ] **Step 3: 创建 app/crawler/dahuagov_spider.py** + +```python +import asyncio +import random +from datetime import datetime +import httpx +from app.config import settings +from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig +from app.crawler.parsers import parse_dahuagov_html + + +class DahuagovSpider(BaseSpider): + name = "dahuagov" + source_code = "dahuagov" + source_name = "大化县政府网采购公告" + + BASE_URL = "http://www.gxdh.gov.cn" + ANNOUNCEMENT_PATH = "/xxgk/zdlyxxgk/ggzypzly/zfcgly/cggg/" + + def get_pipeline_config(self) -> PipelineConfig: + return PipelineConfig( + filter_enabled=False, + keywords=[], + dedup_enabled=True, + notify_mode="all", + mark_sent=True, + ) + + async def crawl(self) -> CrawlResult: + start_time = datetime.now() + url = self.BASE_URL + self.ANNOUNCEMENT_PATH + + async with httpx.AsyncClient(timeout=settings.crawler_timeout) as client: + await self._delay() + try: + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "Accept": "text/html,application/xhtml+xml", + "Accept-Language": "zh-CN,zh;q=0.9", + "Referer": self.BASE_URL, + } + response = await client.get(url, headers=headers) + if response.status_code != 200: + return CrawlResult( + source_code=self.source_code, + source_name=self.source_name, + error_message=f"HTTP {response.status_code}", + crawled_at=start_time, + ) + html = response.text + except Exception as e: + return CrawlResult( + source_code=self.source_code, + source_name=self.source_name, + error_message=str(e), + crawled_at=start_time, + ) + + announcements = parse_dahuagov_html(html, start_time) + duration = (datetime.now() - start_time).total_seconds() + + return CrawlResult( + source_code=self.source_code, + source_name=self.source_name, + total_count=len(announcements), + new_count=len(announcements), + announcements=announcements, + crawled_at=start_time, + duration=duration, + ) + + async def _delay(self): + delay = random.uniform(1.0, 3.0) + await asyncio.sleep(delay) +``` + +- [ ] **Step 4: 运行测试验证通过** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_crawler/test_dahuagov_spider.py -v +# Expected: 4 passed +``` + +- [ ] **Step 5: 运行全部测试** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/ -v +# Expected: 19 passed +``` + +- [ ] **Step 6: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/crawler/dahuagov_spider.py tests/test_crawler/test_dahuagov_spider.py +git commit -m "feat: 添加 Dahuagov Spider（大化县政府网爬虫）+ 测试" +``` + +--- + +### Task 7: CrawlService（爬取编排器） + +**Files:** +- Create: `app/services/__init__.py` +- Create: `app/services/crawl_service.py` +- Test: `tests/test_services/__init__.py` +- Test: `tests/test_services/test_crawl_service.py` + +- [ ] **Step 1: 创建测试 tests/test_services/test_crawl_service.py** + +```python +import pytest +from datetime import datetime +from unittest.mock import AsyncMock, patch +from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig +from app.services.crawl_service import CrawlService + + +class MockGXGPSpider(BaseSpider): + name = "mock_gxgp" + source_code = "mock_gxgp" + source_name = "Mock GXGP" + + async def crawl(self, **kwargs): + return CrawlResult( + source_code=self.source_code, + source_name=self.source_name, + total_count=10, + new_count=5, + announcements=[ + { + "title": f"公告{i}", + "publish_date": datetime(2026, 5, 9), + "purchase_name": "单位", + "content_url": f"https://x.com/{i}", + "source_code": "ZcyAnnouncement1", + "source_name": "采购公告", + "announcement_type": "purchase", + "crawl_mode": "auto", + "is_new": True, + "is_today": True, + "content_hash": f"hash{i}", + } + for i in range(10) + ], + ) + + def get_pipeline_config(self): + return PipelineConfig( + filter_enabled=True, + keywords=["大化"], + notify_mode="filtered", + ) + + +class MockDahuagovSpider(BaseSpider): + name = "mock_dahuagov" + source_code = "mock_dahuagov" + source_name = "Mock Dahuagov" + + async def crawl(self, **kwargs): + return CrawlResult( + source_code=self.source_code, + source_name=self.source_name, + total_count=3, + new_count=3, + announcements=[ + { + "title": f"大化公告{j}", + "publish_date": datetime(2026, 5, 9), + "purchase_name": "大化瑶族自治县", + "content_url": f"https://dh.com/{j}", + "source_code": "dahuagov", + "source_name": "大化县政府网采购公告", + "announcement_type": "purchase", + "crawl_mode": "auto", + "is_new": True, + "is_today": True, + "content_hash": f"dh_hash{j}", + } + for j in range(3) + ], + ) + + def get_pipeline_config(self): + return PipelineConfig( + filter_enabled=False, + notify_mode="all", + mark_sent=True, + ) + + +@pytest.mark.asyncio +async def test_crawl_service_registers_spiders(): + service = CrawlService() + service.register(MockGXGPSpider()) + service.register(MockDahuagovSpider()) + assert len(service.spiders) == 2 + + +@pytest.mark.asyncio +async def test_crawl_service_run_all(): + service = CrawlService() + service.register(MockGXGPSpider()) + service.register(MockDahuagovSpider()) + results = await service.run_all() + assert len(results) == 2 + assert results[0].total_count == 10 + assert results[1].total_count == 3 + assert all(r.success for r in results) + + +@pytest.mark.asyncio +async def test_crawl_service_run_specific(): + service = CrawlService() + service.register(MockGXGPSpider()) + service.register(MockDahuagovSpider()) + results = await service.run_spider("mock_dahuagov") + assert len(results) == 1 + assert results[0].source_code == "mock_dahuagov" +``` + +- [ ] **Step 2: 运行测试确认失败** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_services/test_crawl_service.py -v +# Expected: FAIL — ImportError +``` + +- [ ] **Step 3: 创建 app/services/__init__.py** (空文件) + +- [ ] **Step 4: 创建 app/services/crawl_service.py** + +```python +from typing import Dict, List +from app.crawler.base import BaseSpider, CrawlResult + + +class CrawlService: + def __init__(self): + self.spiders: Dict[str, BaseSpider] = {} + + def register(self, spider: BaseSpider): + self.spiders[spider.name] = spider + + async def run_all(self) -> List[CrawlResult]: + results = [] + for name, spider in self.spiders.items(): + result = await spider.crawl() + results.append(result) + return results + + async def run_spider(self, name: str, **kwargs) -> List[CrawlResult]: + spider = self.spiders.get(name) + if spider is None: + return [CrawlResult( + source_code=name, source_name=name, + error_message=f"Spider not found: {name}" + )] + result = await spider.crawl(**kwargs) + return [result] + + def get_spider_names(self) -> List[str]: + return list(self.spiders.keys()) + + def get_pipeline_config(self, name: str): + spider = self.spiders.get(name) + if spider: + return spider.get_pipeline_config() + return None +``` + +- [ ] **Step 5: 运行测试验证通过** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_services/test_crawl_service.py -v +# Expected: 3 passed +``` + +- [ ] **Step 6: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/services/ tests/test_services/ +git commit -m "feat: 添加 CrawlService 爬取编排器 + 测试" +``` + +--- + +### Task 8: PostCrawlPipeline 统一管道 + +**Files:** +- Create: `app/services/pipeline.py` +- Create: `app/services/filter_service.py` +- Test: `tests/test_services/test_pipeline.py` +- Test: `tests/test_services/test_filter_service.py` + +- [ ] **Step 1: 创建过滤测试 tests/test_services/test_filter_service.py** + +```python +from datetime import datetime +from app.services.filter_service import filter_by_keywords, filter_by_date + + +def test_filter_by_keywords_match(): + announcements = [ + {"title": "大化县采购公告", "purchase_name": "大化县财政局", + "source_code": "test", "source_name": "test"}, + {"title": "南宁市采购公告", "purchase_name": "南宁市财政局", + "source_code": "test", "source_name": "test"}, + ] + result = filter_by_keywords(announcements, ["大化"]) + assert len(result) == 1 + assert result[0]["title"] == "大化县采购公告" + + +def test_filter_by_keywords_no_keywords(): + announcements = [ + {"title": "大化县采购公告", "purchase_name": "x", + "source_code": "test", "source_name": "test"}, + ] + result = filter_by_keywords(announcements, []) + assert len(result) == 1 + + +def test_filter_by_date_range(): + today = datetime(2026, 5, 9) + announcements = [ + {"title": "t1", "publish_date": datetime(2026, 5, 9), + "source_code": "test", "source_name": "test"}, + {"title": "t2", "publish_date": datetime(2026, 5, 1), + "source_code": "test", "source_name": "test"}, + {"title": "t3", "publish_date": datetime(2026, 4, 30), + "source_code": "test", "source_name": "test"}, + ] + result = filter_by_date(announcements, start_date=today, end_date=today) + assert len(result) == 1 + assert result[0]["title"] == "t1" +``` + +- [ ] **Step 2: 创建管道测试 tests/test_services/test_pipeline.py** + +```python +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from datetime import datetime +from app.services.pipeline import PostCrawlPipeline +from app.crawler.base import CrawlResult, PipelineConfig + + +@pytest.mark.asyncio +async def test_pipeline_filtered_mode(): + config = PipelineConfig( + filter_enabled=True, + keywords=["大化"], + dedup_enabled=True, + notify_mode="filtered", + mark_sent=False, + ) + result = CrawlResult( + source_code="test", source_name="test", + total_count=3, new_count=3, + announcements=[ + { + "title": "大化县公告", "publish_date": datetime(2026, 5, 9), + "purchase_name": "大化县", "content_url": "https://1.com", + "source_code": "test", "source_name": "测试", + "announcement_type": "purchase", "crawl_mode": "auto", + "is_new": True, "is_today": True, + "content_hash": "abc123", + }, + { + "title": "南宁市公告", "publish_date": datetime(2026, 5, 9), + "purchase_name": "南宁市", "content_url": "https://2.com", + "source_code": "test", "source_name": "测试", + "announcement_type": "purchase", "crawl_mode": "auto", + "is_new": True, "is_today": True, + "content_hash": "def456", + }, + ], + ) + + mock_db = AsyncMock() + mock_notify = AsyncMock() + + pipeline = PostCrawlPipeline(db_session=mock_db, notification_service=mock_notify) + + with patch.object(pipeline, "_save_to_db", AsyncMock(return_value=2)): + with patch.object(pipeline, "_send_notifications", AsyncMock(return_value=1)): + pipe_result = await pipeline.process( + result.announcements, config + ) + assert pipe_result.stored == 2 + assert pipe_result.filtered == 1 # 3 total, 2 pass keyword, 1 filtered + assert pipe_result.notified == 1 + + +@pytest.mark.asyncio +async def test_pipeline_all_mode(): + config = PipelineConfig( + filter_enabled=False, + keywords=[], + dedup_enabled=True, + notify_mode="all", + mark_sent=True, + ) + result = CrawlResult( + source_code="dahuagov", source_name="大化县政府网", + total_count=2, new_count=2, + announcements=[ + { + "title": f"公告{i}", "publish_date": datetime(2026, 5, 9), + "purchase_name": "大化县", "content_url": f"https://x.com/{i}", + "source_code": "dahuagov", "source_name": "大化县政府网采购公告", + "announcement_type": "purchase", "crawl_mode": "auto", + "is_new": True, "is_today": True, + "content_hash": f"hash{i}", + } + for i in range(2) + ], + ) + + mock_db = AsyncMock() + mock_notify = AsyncMock() + pipeline = PostCrawlPipeline(db_session=mock_db, notification_service=mock_notify) + + with patch.object(pipeline, "_save_to_db", AsyncMock(return_value=2)): + with patch.object(pipeline, "_send_notifications", AsyncMock(return_value=2)): + with patch.object(pipeline, "_mark_sent", AsyncMock(return_value=2)): + pipe_result = await pipeline.process( + result.announcements, config + ) + assert pipe_result.stored == 2 + assert pipe_result.filtered == 0 + assert pipe_result.notified == 2 + pipeline._mark_sent.assert_awaited_once() +``` + +- [ ] **Step 3: 运行测试确认失败** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_services/test_filter_service.py tests/test_services/test_pipeline.py -v +# Expected: FAIL — ModuleNotFoundError +``` + +- [ ] **Step 4: 创建 app/services/filter_service.py** + +```python +from datetime import date +from typing import Any, Dict, List, Optional + + +def filter_by_keywords(announcements: List[Dict[str, Any]], + keywords: List[str]) -> List[Dict[str, Any]]: + if not keywords: + return announcements + + filtered = [] + for ann in announcements: + search_text = f"{ann.get('title', '')} {ann.get('purchase_name', '')}" + if any(kw in search_text for kw in keywords): + ann["keyword_matched"] = True + filtered.append(ann) + else: + ann["keyword_matched"] = False + filtered.append(ann) # Still include, just mark not matched + + return filtered + + +def filter_by_date(announcements: List[Dict[str, Any]], + start_date: Optional[date] = None, + end_date: Optional[date] = None) -> List[Dict[str, Any]]: + if not start_date and not end_date: + return announcements + + filtered = [] + for ann in announcements: + pub_date = ann.get("publish_date") + if not pub_date: + continue + if isinstance(pub_date, date): + pub_date = pub_date + else: + pub_date = pub_date.date() if hasattr(pub_date, "date") else pub_date + + if start_date and pub_date < start_date: + continue + if end_date and pub_date > end_date: + continue + filtered.append(ann) + + return filtered + + +def dedup_by_hash(announcements: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + seen = set() + result = [] + for ann in announcements: + h = ann.get("content_hash") + if h and h not in seen: + seen.add(h) + result.append(ann) + return result +``` + +- [ ] **Step 5: 创建 app/services/pipeline.py** + +```python +from typing import Any, Dict, List +from app.crawler.base import PipelineConfig, PipelineResult +from app.services.filter_service import filter_by_keywords, dedup_by_hash + + +class PostCrawlPipeline: + def __init__(self, db_session, notification_service): + self.db = db_session + self.notify = notification_service + + async def process(self, announcements: List[Dict[str, Any]], + config: PipelineConfig) -> PipelineResult: + result = PipelineResult() + + if not announcements: + return result + + # 1. 去重 + if config.dedup_enabled: + announcements = dedup_by_hash(announcements) + + # 2. 存储到数据库 + stored = await self._save_to_db(announcements) + result.stored = stored + + to_notify = announcements + + # 3. 筛选 + if config.filter_enabled and config.keywords: + before = len(to_notify) + to_notify = [a for a in to_notify + if self._match_keywords(a, config.keywords)] + result.filtered = before - len(to_notify) + + # 4. 推送 + if config.notify_mode == "all": + result.notified = await self._send_notifications(to_notify) + elif config.notify_mode == "filtered": + if config.filter_enabled and config.keywords: + result.notified = await self._send_notifications(to_notify) + elif not config.filter_enabled: + result.notified = await self._send_notifications(to_notify) + + # 5. 标记已发送 + if config.mark_sent and result.notified > 0: + await self._mark_sent(to_notify) + + return result + + async def _save_to_db(self, announcements: List[Dict[str, Any]]) -> int: + from sqlalchemy.dialects.postgresql import insert + from app.models.announcement import Announcement + + if not announcements: + return 0 + + values = [{ + "title": a["title"], + "publish_date": a["publish_date"], + "purchase_name": a.get("purchase_name", ""), + "content_url": a.get("content_url", ""), + "source_code": a["source_code"], + "source_name": a["source_name"], + "announcement_type": a.get("announcement_type", "purchase"), + "content_hash": a["content_hash"], + "crawl_mode": a.get("crawl_mode", "auto"), + "is_new": a.get("is_new", True), + "is_sent": False, + "keyword_matched": a.get("keyword_matched", False), + } for a in announcements] + + stmt = insert(Announcement).values(values) + stmt = stmt.on_conflict_do_nothing(index_elements=["content_hash"]) + + result_proxy = await self.db.execute(stmt) + await self.db.commit() + return result_proxy.rowcount or len(values) + + async def _send_notifications(self, announcements: List[Dict[str, Any]]) -> int: + return await self.notify.send(announcements) + + async def _mark_sent(self, announcements: List[Dict[str, Any]]) -> int: + from app.models.announcement import Announcement + from sqlalchemy import update + + hashes = [a["content_hash"] for a in announcements if a.get("content_hash")] + if not hashes: + return 0 + + stmt = ( + update(Announcement) + .where(Announcement.content_hash.in_(hashes)) + .values(is_sent=True) + ) + result = await self.db.execute(stmt) + await self.db.commit() + return result.rowcount + + @staticmethod + def _match_keywords(announcement: Dict[str, Any], keywords: List[str]) -> bool: + text = f"{announcement.get('title', '')} {announcement.get('purchase_name', '')}" + return any(kw in text for kw in keywords) +``` + +- [ ] **Step 6: 运行测试验证通过** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_services/ -v +# Expected: all passed +``` + +- [ ] **Step 7: 运行全部测试** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/ -v +# Expected: all passed +``` + +- [ ] **Step 8: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/services/pipeline.py app/services/filter_service.py tests/test_services/ +git commit -m "feat: 添加 PostCrawlPipeline 统一管道 + 筛选服务 + 测试" +``` + +--- + +### Task 9: 企业微信模块 + +**Files:** +- Create: `app/wechat/__init__.py` +- Create: `app/wechat/crypto.py` — 复制现有 WXBizMsgCrypt +- Create: `app/wechat/client.py` — 企业微信 API 客户端 +- Create: `app/wechat/handler.py` — 消息处理器 + +- [ ] **Step 1: 创建 app/wechat/__init__.py** (空文件) + +- [ ] **Step 2: 创建 app/wechat/crypto.py** + +从现有 `gx_gp_monitor/wechat/WXBizMsgCrypt.py` 和 `gx_gp_monitor/wechat/ierror.py` 复制并整合： + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +cp gx_gp_monitor/wechat/WXBizMsgCrypt.py app/wechat/crypto.py +cp gx_gp_monitor/wechat/ierror.py app/wechat/ierror.py +``` + +- [ ] **Step 3: 创建 app/wechat/client.py** + +```python +import time +import httpx +from typing import Optional +from app.config import settings + + +class WeChatClient: + def __init__(self): + self._access_token: Optional[str] = None + self._token_expires_at: float = 0 + + async def _get_access_token(self) -> Optional[str]: + now = time.time() + if self._access_token and now < self._token_expires_at: + return self._access_token + + url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" + params = { + "corpid": settings.wechat_corp_id, + "corpsecret": settings.wechat_secret, + } + async with httpx.AsyncClient(timeout=30) as client: + response = await client.get(url, params=params) + data = response.json() + if data.get("errcode") == 0: + self._access_token = data["access_token"] + self._token_expires_at = now + data.get("expires_in", 7200) - 300 + return self._access_token + return None + + async def send_text(self, content: str, to_user: str = "@all") -> bool: + return await self._send_message("text", {"content": content}, to_user) + + async def send_markdown(self, content: str, to_user: str = "@all") -> bool: + return await self._send_message("markdown", {"content": content}, to_user) + + async def send_textcard(self, title: str, description: str, url: str, + to_user: str = "@all", btn_txt: str = "查看详情") -> bool: + return await self._send_message("textcard", { + "title": title, + "description": description, + "url": url, + "btntxt": btn_txt, + }, to_user) + + async def _send_message(self, msgtype: str, msg_data: dict, + to_user: str = "@all") -> bool: + token = await self._get_access_token() + if not token: + return False + + url = "https://qyapi.weixin.qq.com/cgi-bin/message/send" + params = {"access_token": token} + body = { + "touser": to_user, + "msgtype": msgtype, + "agentid": int(settings.wechat_agent_id), + msgtype: msg_data, + } + + async with httpx.AsyncClient(timeout=30) as client: + response = await client.post(url, params=params, json=body) + data = response.json() + return data.get("errcode") == 0 +``` + +- [ ] **Step 4: 创建 app/wechat/handler.py** + +```python +import xml.etree.cElementTree as ET +from typing import Optional +from app.wechat.crypto import WXBizMsgCrypt +from app.config import settings + + +class WeChatMessageHandler: + def __init__(self): + self.wxcpt = WXBizMsgCrypt( + sToken=settings.wechat_token, + sEncodingAESKey=settings.wechat_encoding_aes_key, + sReceiveId=settings.wechat_corp_id, + ) + + def verify_url(self, msg_signature: str, timestamp: str, + nonce: str, echostr: str) -> Optional[str]: + ret, sEchoStr = self.wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr) + if ret == 0: + return sEchoStr.decode("utf-8") if isinstance(sEchoStr, bytes) else sEchoStr + return None + + def decrypt_message(self, post_data: str, msg_signature: str, + timestamp: str, nonce: str) -> Optional[ET.Element]: + ret, xml_content = self.wxcpt.DecryptMsg( + post_data, msg_signature, timestamp, nonce + ) + if ret != 0: + return None + return ET.fromstring(xml_content) + + def encrypt_response(self, response_xml: str, nonce: str, + timestamp: str) -> Optional[str]: + ret, encrypted = self.wxcpt.EncryptMsg(response_xml, nonce, timestamp) + if ret == 0: + return encrypted + return None + + def handle_event(self, event: str, event_key: Optional[str], + from_user: str) -> Optional[str]: + return None + + def handle_text(self, content: str, from_user: str) -> Optional[str]: + return None +``` + +- [ ] **Step 5: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/wechat/ +git commit -m "feat: 迁移企业微信模块（crypto + client + handler）" +``` + +--- + +### Task 10: Notification Service + +**Files:** +- Create: `app/services/notification_service.py` +- Test: `tests/test_services/test_notification_service.py` + +- [ ] **Step 1: 创建测试 tests/test_services/test_notification_service.py** + +```python +import pytest +from unittest.mock import AsyncMock, patch +from app.services.notification_service import NotificationService + + +@pytest.mark.asyncio +async def test_notification_send(): + svc = NotificationService() + announcements = [ + { + "title": "测试公告", + "publish_date": None, + "purchase_name": "测试单位", + "content_url": "https://x.com/1", + "source_code": "test", + "source_name": "测试来源", + "announcement_type": "purchase", + } + ] + + with patch.object(svc.client, "send_textcard", AsyncMock(return_value=True)): + count = await svc.send(announcements) + assert count == 1 + + +@pytest.mark.asyncio +async def test_notification_send_empty(): + svc = NotificationService() + count = await svc.send([]) + assert count == 0 + + +@pytest.mark.asyncio +async def test_notification_send_disabled(): + svc = NotificationService() + announcements = [{"title": "test"}] + + with patch.object(svc.client, "send_textcard", AsyncMock(return_value=False)): + count = await svc.send(announcements) + assert count == 0 +``` + +- [ ] **Step 2: 运行测试确认失败** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_services/test_notification_service.py -v +# Expected: FAIL +``` + +- [ ] **Step 3: 创建 app/services/notification_service.py** + +```python +from datetime import datetime +from typing import Any, Dict, List +from app.config import settings +from app.wechat.client import WeChatClient + + +class NotificationService: + def __init__(self): + self.client = WeChatClient() + + async def send(self, announcements: List[Dict[str, Any]]) -> int: + if not settings.wechat_enabled: + return 0 + if not announcements: + return 0 + + sent = 0 + for ann in announcements: + try: + title = ann.get("title", "") + if len(title) > 128: + title = title[:125] + "..." + + purchase_name = ann.get("purchase_name", "") + if len(purchase_name) > 25: + purchase_name = purchase_name[:22] + "..." + + pub_date = ann.get("publish_date") + time_str = pub_date.strftime("%Y-%m-%d %H:%M") if pub_date else "时间未知" + + source_name = ann.get("source_name", "") + + description = ( + f'
' + f'{source_name} | {purchase_name} | {time_str}' + f'
' + ) + + url = ann.get("content_url", "") + + if await self.client.send_textcard(title, description, url): + sent += 1 + except Exception: + continue + + return sent +``` + +- [ ] **Step 4: 运行测试验证通过** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_services/test_notification_service.py -v +# Expected: 3 passed +``` + +- [ ] **Step 5: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/services/notification_service.py tests/test_services/test_notification_service.py +git commit -m "feat: 添加 NotificationService 通知服务 + 测试" +``` + +--- + +### Task 11: API 路由 + +**Files:** +- Create: `app/api/__init__.py` +- Create: `app/api/deps.py` +- Create: `app/api/router.py` +- Create: `app/api/announcements.py` +- Create: `app/api/crawl.py` +- Create: `app/api/wechat.py` +- Modify: `app/main.py` — 注册路由 +- Test: `tests/test_api/__init__.py` +- Test: `tests/test_api/test_health.py` +- Test: `tests/test_api/test_announcements.py` +- Test: `tests/test_api/test_crawl.py` + +- [ ] **Step 1: 创建 app/api/__init__.py** (空文件) + +- [ ] **Step 2: 创建 app/api/deps.py** + +```python +from sqlalchemy.ext.asyncio import AsyncSession +from app.main import async_session +from app.services.crawl_service import CrawlService +from app.crawler.gxgp_spider import GXGPSpider +from app.crawler.dahuagov_spider import DahuagovSpider + + +async def get_db() -> AsyncSession: + async with async_session() as session: + yield session + + +_crawl_service: CrawlService | None = None + + +def get_crawl_service() -> CrawlService: + global _crawl_service + if _crawl_service is None: + _crawl_service = CrawlService() + _crawl_service.register(GXGPSpider()) + _crawl_service.register(DahuagovSpider()) + return _crawl_service +``` + +- [ ] **Step 3: 创建 app/api/router.py** + +```python +from fastapi import APIRouter +from app.api import announcements, crawl, wechat + +api_router = APIRouter(prefix="/api/v1") +api_router.include_router(announcements.router, tags=["announcements"]) +api_router.include_router(crawl.router, tags=["crawl"]) +api_router.include_router(wechat.router, tags=["wechat"]) +``` + +- [ ] **Step 4: 创建 app/api/announcements.py** + +```python +from datetime import datetime +from typing import Optional +from fastapi import APIRouter, Depends, Query, HTTPException +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, func, text +from app.api.deps import get_db +from app.models.announcement import Announcement +from app.models.schemas import AnnouncementResponse, AnnouncementListResponse + +router = APIRouter() + + +@router.get("/announcements", response_model=AnnouncementListResponse) +async def list_announcements( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + source_code: Optional[str] = None, + keyword: Optional[str] = None, + start_date: Optional[str] = None, + end_date: Optional[str] = None, + crawl_mode: Optional[str] = None, + db: AsyncSession = Depends(get_db), +): + conditions = [] + if source_code: + conditions.append(Announcement.source_code == source_code) + if crawl_mode: + conditions.append(Announcement.crawl_mode == crawl_mode) + if start_date: + conditions.append(Announcement.publish_date >= start_date) + if end_date: + conditions.append(Announcement.publish_date <= end_date) + if keyword: + conditions.append( + Announcement.title.ilike(f"%{keyword}%") + ) + + base_query = select(Announcement) + if conditions: + base_query = base_query.where(*conditions) + + count_query = select(func.count()).select_from(base_query.subquery()) + total_result = await db.execute(count_query) + total = total_result.scalar() or 0 + + items_query = base_query.order_by(Announcement.publish_date.desc()) \ + .offset((page - 1) * page_size).limit(page_size) + items_result = await db.execute(items_query) + items = items_result.scalars().all() + + return AnnouncementListResponse( + total=total, + page=page, + page_size=page_size, + items=[AnnouncementResponse.model_validate(item) for item in items], + ) + + +@router.get("/announcements/{announcement_id}", response_model=AnnouncementResponse) +async def get_announcement(announcement_id: int, db: AsyncSession = Depends(get_db)): + result = await db.execute( + select(Announcement).where(Announcement.id == announcement_id) + ) + item = result.scalar_one_or_none() + if item is None: + raise HTTPException(status_code=404, detail="公告不存在") + return AnnouncementResponse.model_validate(item) + + +@router.get("/announcements/today", response_model=AnnouncementListResponse) +async def get_today_announcements(db: AsyncSession = Depends(get_db)): + today = datetime.now().date() + result = await db.execute( + select(Announcement).where( + func.date(Announcement.publish_date) == today + ).order_by(Announcement.publish_date.desc()) + ) + items = result.scalars().all() + return AnnouncementListResponse( + total=len(items), page=1, page_size=len(items), + items=[AnnouncementResponse.model_validate(item) for item in items], + ) + + +@router.get("/announcements/stats") +async def get_stats(db: AsyncSession = Depends(get_db)): + total = await db.execute(select(func.count()).select_from(Announcement)) + today_count = await db.execute( + select(func.count()).where( + func.date(Announcement.publish_date) == func.current_date() + ).select_from(Announcement) + ) + new_count = await db.execute( + select(func.count()).where(Announcement.is_new == True) + .select_from(Announcement) + ) + unsent = await db.execute( + select(func.count()).where( + Announcement.is_sent == False, Announcement.is_new == True + ).select_from(Announcement) + ) + return { + "total": total.scalar() or 0, + "today": today_count.scalar() or 0, + "new": new_count.scalar() or 0, + "unsent": unsent.scalar() or 0, + } +``` + +- [ ] **Step 5: 创建 app/api/crawl.py** + +```python +from fastapi import APIRouter, Depends +from app.api.deps import get_crawl_service +from app.models.schemas import CrawlTriggerRequest + +router = APIRouter() + + +@router.post("/crawl/trigger") +async def trigger_crawl(request: CrawlTriggerRequest): + from app.services.pipeline import PostCrawlPipeline + from app.services.notification_service import NotificationService + from app.api.deps import get_db + import asyncio + + service = get_crawl_service() + names = service.get_spider_names() + + all_results = [] + for name in names: + results = await service.run_spider(name) + all_results.extend(results) + + return { + "spiders_run": names, + "total_announcements": sum(r.total_count for r in all_results), + "errors": [r.error_message for r in all_results if not r.success], + } + + +@router.get("/crawl/status") +async def crawl_status(): + service = get_crawl_service() + return { + "spiders": service.get_spider_names(), + "running": False, + } + + +@router.get("/crawl/sources") +async def crawl_sources(): + import json + from app.config import settings + sources = json.loads(settings.announcement_sources) + return { + "sources": [ + {"code": code, "name": info["name"], "type": info["type"]} + for code, info in sources.items() + ] + } +``` + +- [ ] **Step 6: 创建 app/api/wechat.py** + +```python +from fastapi import APIRouter, Request, Response +from app.wechat.handler import WeChatMessageHandler + +router = APIRouter() + +_handler: WeChatMessageHandler | None = None + + +def get_handler() -> WeChatMessageHandler: + global _handler + if _handler is None: + _handler = WeChatMessageHandler() + return _handler + + +@router.get("/wechat/callback") +async def wechat_verify(request: Request): + handler = get_handler() + params = request.query_params + echostr = handler.verify_url( + params.get("msg_signature", ""), + params.get("timestamp", ""), + params.get("nonce", ""), + params.get("echostr", ""), + ) + if echostr: + return Response(content=echostr, media_type="text/plain") + return Response(content="verification failed", status_code=403) + + +@router.post("/wechat/callback") +async def wechat_callback(request: Request): + handler = get_handler() + params = request.query_params + post_data = await request.body() + post_text = post_data.decode("utf-8") + + xml_tree = handler.decrypt_message( + post_text, + params.get("msg_signature", ""), + params.get("timestamp", ""), + params.get("nonce", ""), + ) + if xml_tree is None: + return Response(content="decrypt failed", status_code=403) + + msg_type = xml_tree.find("MsgType") + msg_type = msg_type.text if msg_type is not None else "unknown" + + if msg_type == "event": + event = xml_tree.find("Event") + event_key = xml_tree.find("EventKey") + from_user = xml_tree.find("FromUserName") + handler.handle_event( + event.text if event is not None else "", + event_key.text if event_key is not None else None, + from_user.text if from_user is not None else "", + ) + elif msg_type == "text": + content = xml_tree.find("Content") + from_user = xml_tree.find("FromUserName") + handler.handle_text( + content.text if content is not None else "", + from_user.text if from_user is not None else "", + ) + + return Response(content="success") +``` + +- [ ] **Step 7: 创建 app/api/scheduler.py** (scheduler endpoints 在 spec 4.3) + +```python +from fastapi import APIRouter +from app.scheduler.jobs import scheduler +from app.models.schemas import JobResponse + +router = APIRouter(prefix="/scheduler", tags=["scheduler"]) + + +@router.get("/jobs") +async def list_jobs(): + jobs = [] + for job in scheduler.get_jobs(): + jobs.append(JobResponse( + id=job.id, + name=job.name, + next_run_time=str(job.next_run_time) if job.next_run_time else None, + )) + return {"jobs": jobs} + + +@router.post("/pause/{job_id}") +async def pause_job(job_id: str): + scheduler.pause_job(job_id) + return {"status": "paused", "job_id": job_id} + + +@router.post("/resume/{job_id}") +async def resume_job(job_id: str): + scheduler.resume_job(job_id) + return {"status": "resumed", "job_id": job_id} +``` + +- [ ] **Step 8: 修改 app/api/router.py 注册 scheduler 路由** + +在 `app/api/router.py` 中添加: + +```python +from app.api import scheduler as scheduler_module +api_router.include_router(scheduler_module.router) +``` + +- [ ] **Step 9: 修改 app/main.py 注册路由** + +在 app/main.py 的 `app = FastAPI(...)` 之后添加： + +```python +from app.api.router import api_router + +app.include_router(api_router) +``` + +- [ ] **Step 10: 创建测试 tests/test_api/__init__.py** (空文件) + +- [ ] **Step 11: 创建测试 tests/test_api/test_health.py** + +```python +import pytest +from httpx import AsyncClient, ASGITransport +from app.main import app + + +@pytest.mark.asyncio +async def test_health(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.get("/health") + assert response.status_code == 200 + assert response.json() == {"status": "ok"} +``` + +- [ ] **Step 12: 运行 API 测试** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/test_api/test_health.py -v +# Expected: 1 passed +``` + +- [ ] **Step 13: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/api/ app/main.py tests/test_api/ +git commit -m "feat: 添加 API 路由（公告/爬取/微信回调/健康检查）+ 测试" +``` + +--- + +### Task 12: APScheduler 定时任务 + +**Files:** +- Create: `app/scheduler/__init__.py` +- Create: `app/scheduler/jobs.py` +- Modify: `app/main.py` — lifespan 中启动调度器 + +- [ ] **Step 1: 创建 app/scheduler/__init__.py** (空文件) + +- [ ] **Step 2: 创建 app/scheduler/jobs.py** + +```python +import logging +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from app.config import settings +from app.api.deps import get_crawl_service + +logger = logging.getLogger(__name__) +scheduler = AsyncIOScheduler() + + +async def scheduled_crawl(): + """定时爬取任务：运行所有 Spider""" + logger.info("开始定时爬取任务") + service = get_crawl_service() + names = service.get_spider_names() + for name in names: + try: + results = await service.run_spider(name) + for r in results: + if not r.success: + logger.error(f"Spider {name} 失败: {r.error_message}") + else: + logger.info(f"Spider {name} 完成: {r.total_count} 条") + except Exception as e: + logger.error(f"Spider {name} 异常: {e}") + logger.info("定时爬取任务完成") + + +def start_scheduler(): + if not settings.scheduler_enabled: + return + scheduler.add_job( + scheduled_crawl, + "cron", + hour="8,14,18", + minute="0", + id="scheduled_crawl", + name="定时爬取", + timezone="Asia/Shanghai", + ) + scheduler.start() + logger.info("APScheduler 已启动 (8:00, 14:00, 18:00)") + + +def shutdown_scheduler(): + if scheduler.running: + scheduler.shutdown(wait=False) + logger.info("APScheduler 已停止") +``` + +- [ ] **Step 3: 修改 app/main.py 的 lifespan** + +```python +@asynccontextmanager +async def lifespan(app: FastAPI): + import logging + logging.basicConfig( + level=getattr(logging, settings.log_level), + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + ) + from app.scheduler.jobs import start_scheduler, shutdown_scheduler + start_scheduler() + yield + shutdown_scheduler() + await engine.dispose() +``` + +- [ ] **Step 4: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add app/scheduler/ app/main.py +git commit -m "feat: 添加 APScheduler 定时爬取任务" +``` + +--- + +### Task 13: Docker 化 + +**Files:** +- Create: `docker/Dockerfile` +- Create: `docker/docker-compose.yml` + +- [ ] **Step 1: 创建 docker/Dockerfile** + +```dockerfile +FROM python:3.12-slim + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends \ + libpq-dev && \ + rm -rf /var/lib/apt/lists/* + +COPY pyproject.toml . +RUN pip install --no-cache-dir -e ".[dev]" + +COPY . . + +EXPOSE 8000 + +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] +``` + +- [ ] **Step 2: 创建 docker/docker-compose.yml** + +```yaml +services: + app: + build: + context: .. + dockerfile: docker/Dockerfile + ports: + - "8000:8000" + env_file: + - ../.env + volumes: + - ../logs:/app/logs + restart: unless-stopped +``` + +- [ ] **Step 3: 构建验证** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +docker build -f docker/Dockerfile -t gx-gp-notify . +# Expected: Successfully built +``` + +- [ ] **Step 4: 提交** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add docker/ +git commit -m "feat: 添加 Docker 部署配置" +``` + +--- + +### Task 14: 运行 ruff 代码检查 + +- [ ] **Step 1: 安装 ruff 并运行** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +pip install ruff +ruff check app/ tests/ +# Fix any issues +ruff check --fix app/ tests/ +``` + +- [ ] **Step 2: 提交修复** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add -A +git commit -m "chore: ruff 代码检查与修复" +``` + +--- + +### Task 15: 清理旧文件并最终验证 + +- [ ] **Step 1: 运行全部测试最终确认** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +python -m pytest tests/ -v +# Expected: all passed +``` + +- [ ] **Step 2: 验证服务启动** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +uvicorn app.main:app --host 0.0.0.0 --port 8000 & +sleep 2 +curl http://localhost:8000/health +curl http://localhost:8000/docs # 需要 debug=true +kill %1 +``` + +- [ ] **Step 3: 提交最终状态** + +```bash +cd /home/v6ole/PythonProject/GX-gp-notify +git add -A +git commit -m "chore: 清理旧文件，最终验证通过" +```