# 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: 清理旧文件，最终验证通过" ```