```
```
# epr_fr_weee_register — 法国WEEE和运动休闲注册项目设计开发文档
---
## 1. 项目概述
### 1.1 项目定位
创建新业务项目 `epr_fr_weee_register`（法国WEEE和运动休闲注册自动化），复用 `apps/_example/` 脚手架模板，面向 **Ecologic** 门户（`https://producteur.ecologic-extranet.com/`）的网页注册流程。
**文件来源**：SaaS 端 PHP 脚本 `epr_weee_query_api.php` 自动生成并上传到 OSS（腾讯云 COS）的两个文件：
- `Mandat`（授权书 POA）— ZIP 格式，内含 PDF
- `WEEE_Sports_Leisure`（注册表格）— CSV UTF-8 格式
**技术栈**：
- **FastAPI** — 作为服务框架
- **Playwright** — 浏览器自动化引擎
- **SQLAlchemy 2.0** — ORM
- **APScheduler** — 任务调度
- **SQL Server** — 数据库（源库 `vat_db`，目标库 `rpa_test`）
### 1.2 业务流程概览
整个项目分为 **5 个独立的 CLI 脚本流程**，按时间顺序串行依赖：
| 流程 | 命令 | 职责 | 触发时机 | 状态流转 |
| ---------- | ----------------------- | ------------------------------------------------------------ | ------------------------- | ----------------------------------------------- |
| **流程一** | `run-data-poller` | 从 SaaS 库查询数据 → 清洗 → 写入 `epr_fr_weee_register` 表（PushType=301+305） | 持续轮询 | SaaS `status=1` → 写入 `status=0`，记录 `push_type` |
| **流程二** | `run-submission-poller` | 从 `epr_fr_weee_register` 取数据 → Playwright 自动注册 → 更新状态（仅处理 `push_type=301`） | 持续轮询 | 本地 `status=0` → status=1/2（成功）/-1（失败） |
| **流程三** | `run-signing-poller` | IMAP 监控签字邮件 → Playwright 自动签字 → 下载已签文件 | 持续轮询（仅处理邀请/提醒邮件，排除已完成通知） | `status=2` → `status=3`（已签字） |
| **流程四** | `run-uin-poller` | Playwright 登录 Ecologic 门户获取注册号（UIN）→ 下载证书 | 每天 8:00 / 17:30（签字满 5 天后才查） | `status=3` → `status=4`（已下号） |
| **流程五** | `run-submission-overdue-notify` | 检查 `status=2` 且 `submitted_at` 超过 5 天仍未签字的记录，发送企微报警 | 每天 10:00 | 仅通知，不改状态 |
| **流程六** | `run-bill-poller` | 查询已下号（`status=4`）且未解析账单的订单 → 门户下载账单 PDF → 解析数据写入机构账单表 | 每天 8:30 / 16:00 | `status=4` 不变，`bill_status=0` → `1`（已解析） |
**时间线示意：**
```
流程一 (Data) ─── 持续运行（每 N 分钟轮询源库）
│
▼
流程二 (Submit) ─── 持续运行（每 N 分钟处理待提交记录）
│ status → 2 (成功)
│
▼ ⏳ 等待 1-2 天（Ecologic 审核 + 发送签字邮件）
│
流程三 (Sign) ─── 持续运行（IMAP 监控，排除 tout le monde a signé / venez de signer）
│ status → 3 (已签字)
│
├─ 正常 → 流程四 (UIN) ─── 每天 8:00 / 17:30（签字满 5 天后才查下号）
│ status → 4 (已下号,终态)
│
├─ 提交超期报警 ── status=2 且 submitted_at 超 5 天
│ │
│ ▼
│ 流程五 (Overdue Notify) ─── 每天 10:00 企微报警
│
└─ 下号超期报警 ── status=3 且 signed_at 超 15 天
│
▼
流程四 (UIN) 内置检查 ─── 每天 8:00 / 17:30 各报警 1 次
```
**流程一：数据抓取与入库**（PushType = 301 + 305）
```
SaaS (vat_db) → 8 表 LEFT JOIN 查询 (ei.PushType IN ('301','305')) → validateData() 校验 → getFilePaths() 获取 OSS URL → insertOrUpdateRegistrationInfo() 写入 epr_fr_weee_register（含 push_type）→ update SaaS status=2
```
**流程二：自动化提交（Ecologic Import de comptes）**（仅 PushType = 301）
```
epr_fr_weee_register (status=0) → Playwright 打开 Chrome → 登录 Ecologic → 导航到 Import de comptes → 上传 CSV（WEEE_Sports_Leisure）→ 上传 ZIP（Mandat）→ 提交 → 截图留存 → 更新状态
```
---
## 2. 系统架构
### 2.1 总体架构图
```
┌─────────────────────────────────────────────────────────────────────┐
│ SaaS 系统 (SQL Server: vat_db) │
│ │
│ EPRBusinessRecord ────┐ │
│ EPRRegInfo ─┼─── LEFT JOIN (8 表) │
│ Base_Customer_Company─┼───────▶ epr_weee_query_api.php │
│ GeneralTemplateEPR ─┘ 生成 Mandat + CSV │
│ Country │
│ ServiceItems │
│ InseeApiClient ─── 法国公司获取 NAF Code │
└─────────────────────────────────────────────────────────────────────┘
│
│ epr_weee_query_api.php 生成文件 → OSS 上传
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 腾讯云 COS (OSS) │
│ │
│ oss://vat_factory/fr_declar/ │
│ ├── EPR_Ecologic_{NameEng}.zip ← Mandat (授权书 PDF ZIP) │
│ └── EPR_Ecologic_{NameEng}.csv ← WEEE_Sports_Leisure (CSV) │
└─────────────────────────────────────────────────────────────────────┘
│
│ API 返回 OSS URL → 流程一写入
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 本地业务库 (SQL Server: rpa_test) │
│ │
│ epr_fr_weee_register 表 │
│ ┌─────────────────────────────────────┐ │
│ │ id PK │ │
│ │ tid VARCHAR(64) │ │
│ │ status INT(0/1/2/-1) │ │
│ │ retry_count INT │ │
│ │ Mandat VARCHAR(512) │ ◀── 授权书 ZIP │
│ │ WEEE_Sports_Leisure VARCHAR(512) │ ◀── 注册表格 CSV│
│ │ Company_Name_en VARCHAR(60) │ │
│ │ ... 其余 30+ 业务字段 │ │
│ │ (追踪字段) screenshot_urls │ ◀── 流程二写入 │
│ └─────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
│
│ 流程二: run-submission-poller
▼
┌─────────────────────────────────────────────────────────────────────┐
│ Playwright 浏览器自动化 │
│ │
│ ┌─────────┐ ┌──────────────────────────────────────┐ │
│ │ Login │──│ Import de comptes 页面 │ │
│ │ Page │ │ 1. 上传 CSV (WEEE_Sports_Leisure) │ │
│ └─────────┘ │ └─ 检查 #response-csv 错误 │ │
│ │ └─ 有错误 → 通知 bu.F_Mobile │ │
│ │ 2. Dépôt des documents 上传 ZIP │ │
│ │ └─ 检查 #response-poa 错误 │ │
│ │ └─ 有错误 → 通知 bu.F_Mobile │ │
│ │ 3. 确认提交 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 提交前/后截图 → COS 存储 │
│ 企业微信通知 │
└─────────────────────────────────────────────────────────────────────┘
│ │ │
│ 流程二 │ 流程三 │ 流程四
▼ ▼ ▼
┌──────────────┐ ┌───────────────────┐ ┌───────────────────┐
│ Ecologic │ │ 企业微信邮箱 │ │ Ecologic 门户 │
│ 提交注册 │ │ info@seamew.de │ │ 查看 UIN + 证书 │
│ (CSV+ZIP) │ │ 监控签字邮件 │ │ 下载证书 PDF │
└──────────────┘ └───────────────────┘ └───────────────────┘
```
### 2.2 技术架构分层
```
apps/epr_fr_weee_register/
├── cli.py # Typer CLI 入口（4 个命令）
├── config.py # EcologicSettings (继承 CommonSettings)
├── models.py # SQLAlchemy ORM: EcologicRegistration
├── data_poller.py # 流程一: EcologicDataPoller
├── submission_poller.py # 流程二: EcologicSubmissionPoller
├── signing_poller.py # 流程三: EcologicSigningPoller (新增)
├── uin_poller.py # 流程四: EcologicUinPoller (新增)
├── service.py # 业务服务层
├── results.py # 数据契约 (SubmissionResult / SigningResult / UinResult)
├── pages/
│ ├── login_page.py # EcologicLoginPage
│ ├── import_mandats_page.py # EcologicImportMandatsPage (流程二)
│ ├── signing_page.py # EcologicSigningPage (新增 - 流程三)
│ └── uin_page.py # EcologicUinPage (新增 - 流程四)
├── mail/
│ └── signing_mail_handler.py # 签字邮件匹配与通知 (新增)
├── repository/
│ ├── source_repo.py # 源库操作
│ └── target_repo.py # 目标库操作
└── migrations/ # Alembic 迁移
```
---
### 2.3 状态机扩展
原有状态值：
| 状态值 | 常量 | 含义 |
| ------ | ------------------- | ------------------ |
| 0 | PENDING | 待提交 |
| 1 | PROCESSING | 处理中（认领态） |
| 2 | SUCCESS | 提交成功 |
| -1 | FAILED | 提交失败（可重试） |
| 6 | MANUAL_INTERVENTION | 需人工介入 |
| 7 | PERMANENT_FAILURE | 永久失败 |
| 8 | MAX_RETRIES_REACHED | 重试次数耗尽 |
**新增状态值（签字 + 下号）：**
| 状态值 | 常量 | 含义 | 触发条件 |
| ------ | ----------- | ------------------ | ----------------------- |
| 3 | SIGNED | 已签字 | 流程三完成 Yousign 签字 |
| 4 | UIN_ISSUED | 已下号 | 流程四获取 UIN + 证书 |
| -2 | SIGN_FAILED | 签字失败（可重试） | 流程三签字异常 |
| -3 | UIN_FAILED | 下号失败（可重试） | 流程四下号异常 |
**状态流转图：**
```
┌─────────────┐
│ 0 │
│ PENDING │
└──────┬──────┘
│ 流程二: 提交
▼
┌─────────────┐ 失败 ┌────────────┐
│ 1 │──────────▶│ -1 │
│ PROCESSING │ │ FAILED │
└──────┬──────┘ └─────┬──────┘
│ 成功 │ 重试 < max
▼ ▼
┌─────────────┐ ┌────────────┐
│ 2 │ │ 0 │
│ SUCCESS │ │ PENDING │
└──────┬──────┘ └────────────┘
│ 流程三: 签字 (1-2天后)
▼
┌─────────────┐ 失败 ┌────────────┐
│ 3 │──────────▶│ -2 │
│ SIGNED │ │ SIGN_FAILED│
└──────┬──────┘ └─────┬──────┘
│ 成功 │ 重试
▼ │
┌─────────────┐ │
│ 4 │ │
│ UIN_ISSUED │ │
│ (终态) │ │
└─────────────┘ │
▲ │
│ 流程四: 下号 (1-2周后) │
│ │
│ ┌──────────┘
│ ▼
│ ┌────────────┐
│ │ -3 │
└─────│ UIN_FAILED │
└────────────┘
```
##
## 3. 流程一：数据抓取与入库（run-data-poller）
### 3.1 入口命令
```bash
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-data-poller --once
```
### 3.2 调用链路
```
cli.run_data_poller()
│
├─ 1. 加载配置 (EcologicSettings)
├─ 2. 初始化日志
├─ 3. LifecycleManager 注册信号
├─ 4. 构建数据库连接池
│
├─ 5. EcologicDataPoller.__init__()
│
├─ 6. PollerScheduler 调度
│ └─ interval = settings.data_poll_interval_seconds
│
└─ 7. poller.run_once() ← 执行一轮抓取
│
├─ fetch_qualifying_records()
│ └─ 源库 8 表 LEFT JOIN SELECT
│
├─ foreach record:
│ ├─ should_stop()?
│ ├─ process_record() ← 数据清洗 + 校验
│ └─ claim_and_write()
│ ├─ INSERT epr_fr_weee_register 或 UPDATE 重置
│ └─ update SaaS status=2
│
└─ bound_logger.info(...)
```
### 3.3 源库查询（8 表 JOIN）
**SQL 查询逻辑**（位于 `repository/source_query.py`）：
```sql
SELECT TOP 10
br.ID AS RegisterID,
br.BusinessSerialNumber AS BusinessSerialNumber,
br.BusinessCode AS BusinessCode,
br.ServiceYear AS ServiceYear,
ei.ID AS EPRRegInfoID,
br.VATNumber AS VATNumber,
-- 公司信息
bc.NameEng AS NameEng,
bc.NameCN AS NameCN,
bc.RegNumber AS RegNumber,
bc.CompanyAddressLine1En AS CompanyAddressLine1En,
bc.CompanyAddressLine2En AS CompanyAddressLine2En,
bc.CompanyAddressPostcode AS CompanyAddressPostcode,
bc.CityEngName AS CityEngName,
bc.Country AS CompanyCountry,
bc.RegisteredCapital AS RegisteredCapital,
bc.LegalPersonFullNamePinYin AS LegalPersonFullNamePinYin,
-- 国家信息
c.IsEUMember AS IsEUMember,
c.CountryTwoCode AS CountryTwoCode,
-- 服务项名（用于判断 WEEE / 运动休闲）
si.ServiceItemName AS ServiceItemName,
bu.F_Mobile,
ei.PushType AS PushType
FROM EPRBusinessRecord br
LEFT JOIN EPRRegInfo ei ON br.ID = ei.EPRBusinessRecordId
LEFT JOIN GeneralTemplateEPR gt ON gt.ID = br.ID
LEFT JOIN Base_Customer_Company bc ON bc.ID = br.CompanyId
LEFT JOIN Country c ON c.CountryName = bc.Country
LEFT JOIN ServiceItems si ON br.ServiceItemId = si.ID
LEFT JOIN Base_User AS bu ON br.BusinessCounselorId = bu.F_UserId
LEFT JOIN SupplierInformation sp ON br.OfficialFeeRecycleMerchant = sp.ID
WHERE br.Country = 'FR'
AND (ei.PushType = '301' or ei.PushType = '305') -- 301=WEEE注册 305=添加合同
AND sp.ServiceItemName='ECOLOGIC' -- 法国 WEEE Ecologic
AND (si.ServiceItemName='WEEE注册' or si.ServiceItemName='运动户外注册')
AND ei.PushTaxBureauStatus = 1 -- 1 = 待处理
ORDER BY br.ID ASC
```
### 3.4 文件 URL 获取（关联附件表）
`Mandat` 和 `WEEE_Sports_Leisure` 两个文件通过 **关联附件表 `Base_AnnexesFile`** 获取文件路径，再进行路径替换得到 OSS URL。
**SQL 关联逻辑：**
```sql
-- 授权书（Mandat / POA）：匹配文件名含 mandat / poa / vollmacht
LEFT JOIN Base_AnnexesFile pa_annex ON pa_annex.InfoId = ei.ID
AND (LOWER(pa_annex.F_FileName) LIKE '%mandat%'
OR LOWER(pa_annex.F_FileName) LIKE '%poa%'
OR LOWER(pa_annex.F_FileName) LIKE '%vollmacht%')
-- 注册表格（WEEE_Sports_Leisure）：匹配文件名含 weee / csv / ecologic 等
LEFT JOIN Base_AnnexesFile mb_annex ON mb_annex.InfoId = ei.ID
AND (LOWER(mb_annex.F_FileName) LIKE '%weee%'
OR LOWER(mb_annex.F_FileName) LIKE '%csv%'
OR LOWER(mb_annex.F_FileName) LIKE '%ecologic%'
OR LOWER(mb_annex.F_FileName) LIKE '%sports%')
```
**查询字段：**
```sql
SELECT
pa_annex.F_FilePath AS MandatFilePath,
mb_annex.F_FilePath AS WEEESportsFilePath
```
**路径替换逻辑：**
```python
# MAIN_SITE_PATH → FILE_SITE_PATH
# G:/fileAnnexes/xxx → https://file.usaeu.com/xxx
# 或从 OSS URL 直接获取（如果已上传 OSS）
```
| 字段 | 说明 | 获取方式 |
| --------------------- | --------------------- | -------------------------------------- |
| `Mandat` | 授权书（POA） | `pa_annex.F_FilePath` → 路径替换后 URL |
| `WEEE_Sports_Leisure` | WEEE/运动休闲注册表格 | `mb_annex.F_FilePath` → 路径替换后 URL |
### 3.5 数据校验规则（validateData）
| 校验项 | 规则 |
| ----------------- | ------------------------------------------------ |
| **14 个必填字段** | 非空 trim() |
| 公司地址 | `CompanyAddressLine1En` 至少一个非空 |
| 地址长度 | `CompanyAddressLine1En` ≤ 50 字符（文档C列限制） |
| 注册号 | `CTRegNumber` 非空 |
| 公司名称长度 | 无硬性截断 |
| 法人邮箱 | 格式校验 (FILTER_VALIDATE_EMAIL) |
| 法人电话 | 法国格式：纯数字（空格由 Excel 自定义格式处理） |
| 法人拼音姓名 | 含空格分隔（名 姓格式） |
**必填字段清单**：
`BusinessSerialNumber`, `RegisterID`, `EPRRegInfoID`, `NameEng`, `LegalPersonFullNamePinYin`, `LegalPersonPhone`, `LegalPersonEmail`, `CompanyAddressPostcode`, `CityEngName`, `CompanyAddressLine1En`, `CTRegNumber`, `CountryTwoCode`
### 3.6 数据映射表（字段映射）
| epr_fr_weee_register 字段 | 类型 | 源表字段 | 转换规则 |
| ------------------------- | ------------ | ------------------------------ | -------------------------------------------------- |
| `tid` | VARCHAR(64) | `br.BusinessSerialNumber` | 直接映射 |
| `VATBusinessID` | VARCHAR(64) | `br.ID`（RegisterID） | 直接映射 |
| `EPRRegInfoID` | VARCHAR(64) | `ei.ID` | 直接映射 |
| `VAT_or_company_number` | VARCHAR(64) | `ei.CTRegNumber` | SIRET 号（法国公司由 INSEE API 获取） |
| `Company_Name_en` | VARCHAR(256) | `bc.NameEng` | 直接映射（不截断） |
| `Company_Name_cn` | VARCHAR(256) | `bc.NameCN` | 直接映射 |
| `StreetNrBox` | VARCHAR(256) | `bc.CompanyAddressLine1En` | 直接映射 |
| `Zipcode` | VARCHAR(16) | `bc.CompanyAddressPostcode` | 直接映射 |
| `City` | VARCHAR(128) | `bc.CityEngName` | 直接映射 |
| `Country` | VARCHAR(64) | `c.CountryName_en` | 直接映射 |
| `CountryTwoCode` | VARCHAR(8) | `c.CountryTwoCode` | 直接映射 |
| `Phone_number` | VARCHAR(32) | `bc.LegalPersonPhone` | 法国格式：纯数字（`preg_replace('/[^0-9]/', '')`） |
| `VAT_nr` | VARCHAR(64) | `br.VATNumber` | 欧盟公司填 VAT，其他为空 |
| `Legal_LastName` | VARCHAR(64) | `bc.LegalPersonFullNamePinYin` | `parseLegalPersonName()` 拆分姓 |
| `Legal_FirstName` | VARCHAR(128) | `bc.LegalPersonFullNamePinYin` | `parseLegalPersonName()` 拆分名 |
| `Legal_Salutation` | VARCHAR(8) | `bc.LegalPersonGender` | 1→Mr, 0/其他→Mrs |
| `Legal_Email` | VARCHAR(256) | `bc.LegalPersonEmail` | 直接映射 |
| `Sector` | VARCHAR(256) | `gt.ProductsRange_En` | 直接映射 |
| `NAFCode` | VARCHAR(16) | `ei.NAFCode` | 法国 NAF 编码（仅法国公司） |
| `legal_form` | VARCHAR(32) | INSEE API 返回 | 公司类型（默认 AUTRE） |
| `RegisteredCapital` | VARCHAR(64) | `bc.RegisteredCapital` | 去除逗号 |
| `ServiceYear` | VARCHAR(8) | `br.ServiceYear` | 服务年份，决定日期格式 01/01/{year} |
| `ContractType` | VARCHAR(32) | `si.ServiceItemName` | WEEE→'EEE Ménager', 运动休闲→'ASL' |
| `Mandat` | VARCHAR(512) | `pa_annex.F_FilePath` | 路径替换后 → 授权书 ZIP 的 OSS URL |
| `WEEE_Sports_Leisure` | VARCHAR(512) | `mb_annex.F_FilePath` | 路径替换后 → 注册表格 CSV 的 OSS URL |
| `status` | INT | - | 固定 `0`（待推送） |
| `msg` | VARCHAR(512) | - | 固定 `''` |
| `push_type` | VARCHAR(16) | `ei.PushType` | SaaS PushType 值：301=WEEE注册，305=添加合同 |
### 3.7 法人姓名解析函数
```python
def parseLegalPersonName(full_name_pinyin: str) -> tuple[str, str]:
"""
拆分法人拼音姓名
"ZHANG SAN LI" → ("ZHANGSAN", "LI")
"WANG" → ("", "WANG")
注意：CSV 生成中法人姓氏取第一个单词（如 "Zhang"），
法人名取后面的单词合并（如 "Sanli"）
"""
parts = full_name_pinyin.strip().split()
if len(parts) == 0:
return ("", "")
if len(parts) == 1:
return ("", parts[0])
return (" ".join(parts[:-1]), parts[-1])
```
### 3.8 幂等性处理
```python
existing = SELECT * FROM epr_fr_weee_register WHERE tid = ?
if existing:
if existing.status == 2: # 已完成
skip, update SaaS status = -1
else:
UPDATE epr_fr_weee_register SET ... # 更新记录
else:
INSERT INTO epr_fr_weee_register ...
```
---
## 4. 流程二：自动化提交（run-submission-poller）
> **注意：** 流程二仅处理 `push_type = '301'`（WEEE注册）的订单。`push_type = '305'`（添加合同）的订单不会被此流程处理。
### 4.1 入口命令
```bash
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-submission-poller --once
```
### 4.2 调用链路
```
cli.run_submission_poller()
│
├─ 1. 加载配置 (EcologicSettings)
├─ 2. 初始化日志
├─ 3. LifecycleManager 注册信号
├─ 4. 构建数据库连接池
│
├─ 5. EcologicSubmissionPoller 初始化
│
└─ 6. poller.run_once()
│
├─ select_submittable_records()
│ └─ status IN (0, -1) ORDER BY id LIMIT 1
│
├─ claim_record()
│ └─ UPDATE status=1 WHERE id=? AND status IN (0, -1)
│
├─ idempotency_check()
│ └─ 基础幂等校验
│
├─ submit() ← 异步 Playwright 自动化
│ ├─ 浏览器启动 (BrowserManager)
│ ├─ 登录 Ecologic (EcologicLoginPage)
│ ├─ 导入页面 (EcologicImportMandatsPage)
│ │ ├─ 下载 CSV (OSS) → 上传到 Ecologic
│ │ ├─ 检查 #response-csv 校验错误
│ │ ├─ 下载 ZIP (OSS) → 上传到 Ecologic
│ │ ├─ 检查 #response-poa 校验错误
│ │ └─ 点击 Confirm sending of documents
│ └─ 失败截图 + 企业微信通知
│
└─ apply_result()
├─ 更新本地 status (2/-1/6/7)
├─ 回写 SaaS EPRRegInfo PushTaxBureauStatus
└─ 企业微信通知
### 4.3 提交页面 HTML 结构（Ecologic Import de comptes）
```
### 4.3 提交状态机
| 状态值 | 枚举 | 含义 | 流转触发 |
| ------ | --------------------- | ------------ | ------------------------- |
| `0` | `PENDING` | 待提交 | 流程一写入初始状态 |
| `1` | `PROCESSING` | 处理中 | 认领记录后更新（防并发） |
| `2` | `SUCCESS` | 提交成功 | Ecologic 确认提交成功 |
| `-1` | `FAILED` | 失败 | 可重试（retry_count < 3） |
| `6` | `MANUAL_INTERVENTION` | 需要人工介入 | 非可重试错误 |
### 4.4 Ecologic 门户操作完整流程（Playwright 精确选择器）
#### 登录流程 (Login)
```python
# ===== 0. 登录 Ecologic =====
browser = playwright.chromium.launch(headless=False)
context = browser.new_context()
page = context.new_page()
# Step 0a: 打开 Ecologic 登录页
page.goto("https://producteur.ecologic-extranet.com/")
# Step 0b: 填写邮箱
page.get_by_role("textbox", name="E-mail").click()
page.get_by_role("textbox", name="E-mail").fill("info@seamew.de")
# Step 0c: 填写密码
page.get_by_role("textbox", name="Mot de passe").click()
page.get_by_role("textbox", name="Mot de passe").fill("Meiouwang1332!")
# Step 0d: 点击登录按钮
page.get_by_role("button", name="Se connecter").click()
# Step 0e: 登录后等待页面加载
page.wait_for_load_state("networkidle")
```
#### 导入账户流程 (Import de comptes)
```python
# ===== 1. 导航到 Import de comptes =====
# Step 1a: 点击 "Mes producteurs"
# 注意: 该链接有 tabindex="0"，详情页中 "Back" 链接也含 compte-manager，用此区分
page.locator("a[tabindex='0'][href*='compte-manager']").click()
# Step 1b: 等待页面加载 (Ecologic 有长轮询，不能用 networkidle)
page.wait_for_url("**/compte-manager**", timeout=30000)
page.wait_for_timeout(2000)
# Step 1c: 点击 "Import de comptes"
page.get_by_role("link", name="Import de comptes").click()
page.wait_for_url("https://producteur.ecologic-extranet.com/import-mandats")
# ===== 2. 上传 CSV 文件（WEEE_Sports_Leisure）=====
# Step 2a: 点击 Choose File 按钮选择 CSV 文件
page.get_by_role("button", name="Choose File").click()
# Step 2b: 设置文件输入（CSV 文件路径）
csv_file_path = "/tmp/downloads/EPR_Ecologic_{CompanyName}.csv"
page.get_by_role("button", name="Choose File").set_input_files(csv_file_path)
# Step 2c: 等待上传完成后，检查 CSV 校验错误
page.wait_for_timeout(2000)
# Step 2d: 检查 #response-csv 是否有校验报错
csv_errors = extract_csv_validation_errors(page)
if csv_errors:
# 记录错误信息并发送企业微信通知
record.msg = csv_errors
send_wework_notification(
phone=record.bu_F_Mobile,
message=f"【法国WEEE Ecologic CSV校验失败】\n"
f"流水号: {record.tid}\n"
f"公司名称: {record.Company_Name_en}\n"
f"错误详情: {csv_errors}"
)
return SubmissionResult(success=False, message=csv_errors)
# ===== 3. 上传授权书 ZIP 文件（Mandat）=====
# Step 3a: 找到 "Dépôt des documents" 区域
page.get_by_text("- Dépôt des documents").click()
# Step 3b: 上传 ZIP 文件
page.get_by_role("button", name="Choose File").click()
zip_file_path = "/tmp/downloads/EPR_Ecologic_{CompanyName}.zip"
page.get_by_role("button", name="Choose File").set_input_files(zip_file_path)
# Step 3c: 等待上传完成后，检查 ZIP 校验错误
page.wait_for_timeout(2000)
# Step 3d: 检查 #response-poa 是否有校验报错
poa_errors = extract_poa_validation_errors(page)
if poa_errors:
record.msg = poa_errors
send_wework_notification(
phone=record.bu_F_Mobile,
message=f"【法国WEEE Ecologic 授权书校验失败】\n"
f"流水号: {record.tid}\n"
f"公司名称: {record.Company_Name_en}\n"
f"错误详情: {poa_errors}"
)
return SubmissionResult(success=False, message=poa_errors)
```
#### CSV 上传校验错误检测
上传 CSV 后，Ecologic 会校验文件格式。如有错误，页面会在 `#response-csv` 区域显示错误详情。
**HTML 结构示例：**
```html
There is 1 line(s) in error
```
**错误提取函数：**
```python
def extract_csv_validation_errors(page) -> str | None:
"""
提取 CSV 上传后的校验错误信息。
返回 None 表示无错误，返回字符串表示有错误。
"""
# 检查 #response-csv 中是否有 .alert-danger
csv_response = page.locator("#response-csv")
if csv_response.count() == 0:
return None
danger_box = csv_response.locator(".alert-danger")
if danger_box.count() == 0:
return None
# 提取错误行数概览
summary = danger_box.locator("div").first.text_content().strip() # "There is 1 line(s) in error"
# 提取表格中每条错误信息
errors = []
rows = danger_box.locator("table tbody tr")
for i in range(rows.count()):
row = rows.nth(i)
line_no = row.locator("td").nth(0).text_content().strip()
col_name = row.locator("td").nth(1).text_content().strip()
msg = row.locator("td").nth(2).text_content().strip()
errors.append(f"行{line_no} 列{col_name}: {msg}")
return f"{summary}\n" + "\n".join(errors)
```
**企业微信通知函数：**
```python
def send_wework_notification(phone: str, message: str):
"""
通过企业微信发送通知消息给指定手机号。
@param phone: bu.F_Mobile 字段值，如 '13800138000'
@param message: 通知内容
"""
# 调用企业微信 API 发送消息
# 具体实现根据项目的企业微信通知模块调整
wework_api.send_text(phone=phone, content=message)
```
#### POA 上传校验错误检测
上传 ZIP（授权书）后，Ecologic 会校验文件。如有错误，页面会在 `#response-poa` 区域显示错误详情。
**HTML 结构示例：**
```html
There is 1 line(s) in error
```
**错误提取函数：**
```python
def extract_poa_validation_errors(page) -> str | None:
"""
提取 ZIP（授权书）上传后的校验错误信息。
返回 None 表示无错误，返回字符串表示有错误。
"""
# 检查 #response-poa 中是否有 .alert-danger
poa_response = page.locator("#response-poa")
if poa_response.count() == 0:
return None
danger_box = poa_response.locator(".alert-danger")
if danger_box.count() == 0:
return None
# 提取错误行数概览
summary = danger_box.locator("div").first.text_content().strip()
# 提取表格中每条错误信息
errors = []
rows = danger_box.locator("table tbody tr")
for i in range(rows.count()):
row = rows.nth(i)
line_no = row.locator("td").nth(0).text_content().strip()
col_name = row.locator("td").nth(1).text_content().strip()
msg = row.locator("td").nth(2).text_content().strip()
errors.append(f"行{line_no} 列{col_name}: {msg}")
return f"{summary}\n" + "\n".join(errors)
```
#### 提交确认
页面最后一步的 HTML 结构：
```html
Confirmation
# ===== 4. 确认提交 =====
# Step 4a: 点击确认提交按钮
page.get_by_role("button", name="Confirm sending of documents").click()
# Step 4b: 等待页面响应
page.wait_for_load_state("networkidle")
# Step 4c: 截图留存
page.screenshot(path=f"/tmp/screenshots/{record.tid}_confirmation.png")
# Step 4d: 检查提交结果
# 成功：可能跳转到产品列表页或显示成功提示
# 失败：页面可能显示错误提示框 .alert-danger
if page.locator(".alert-danger").count() > 0:
error_msg = page.locator(".alert-danger").text_content()
raise Exception(f"Ecologic 提交失败: {error_msg}")
# Step 4e: 提取成功信息
success_msg = page.locator(".alert-success").text_content() if page.locator(".alert-success").count() > 0 else "Submitted"
```
### 4.5 完整 Playwright 自动化脚本
```python
def run(playwright: Playwright) -> None:
browser = playwright.chromium.launch(headless=False)
context = browser.new_context()
page = context.new_page()
# 登录
page.goto("https://producteur.ecologic-extranet.com/")
page.get_by_role("textbox", name="E-mail").click()
page.get_by_role("textbox", name="E-mail").fill("info@seamew.de")
page.get_by_role("textbox", name="Mot de passe").click()
page.get_by_role("textbox", name="Mot de passe").fill("Meiouwang1332!")
page.get_by_role("button", name="Se connecter").click()
# 导航到 Import de comptes
page.locator("a[tabindex='0'][href*='compte-manager']").click()
page.wait_for_url("**/compte-manager**", timeout=30000)
page.get_by_role("link", name="Import de comptes").click()
page.goto("https://producteur.ecologic-extranet.com/import-mandats")
# 上传 CSV
page.get_by_role("button", name="Choose File").click()
page.get_by_role("button", name="Choose File").set_input_files("EPR_Ecologic_{Name}.csv")
# 上传 ZIP (Dépôt des documents)
page.get_by_text("- Dépôt des documents").click()
page.get_by_role("button", name="Choose File").click()
page.get_by_role("button", name="Choose File").set_input_files("EPR_Ecologic_{Name}.zip")
# 确认提交
page.get_by_role("button", name="Confirm sending of documents").click()
page.wait_for_load_state("networkidle")
context.close()
browser.close()
```
### 4.6 文件下载机制
#### 下载目录路径
| 配置来源 | 路径 |
|---------|------|
| `EcologicSettings.download_dir` 默认值 | `~/.autobot/downloads/ecologic/` |
| `automation` 用户主目录（`/etc/passwd`） | `/opt/autobot` |
| **实际下载目录** | **`/opt/autobot/.autobot/downloads/ecologic/`** |
> 目录由 `os.makedirs(exist_ok=True)` 首次下载时自动创建。
#### 下载前清空目录（Step 0）
[import_mandats_page.py](file:///d:/phpstudy_pro/WWW/fastapi-playwright-automation/apps/epr_fr_weee_register/pages/import_mandats_page.py) 的 `submit_registration` 方法在每次提交流程开头先清空下载目录：
```python
# Step 0: 清空下载目录，确保每次提交流程都下载最新文件
if os.path.isdir(self._download_dir):
shutil.rmtree(self._download_dir)
```
> 确保没有旧文件残留，后续 `_download_file` 会通过 `os.makedirs(exist_ok=True)` 自动重建目录。
#### 下载逻辑
`_download_file` 方法：
- **每次强制重新下载**，不检查文件是否已存在
- 同名文件直接 `open(local_path, "wb")` 覆盖
- 超时时间使用 `SUBMISSION_TIMEOUT_SECONDS`（默认 300 秒）
```python
async def _download_file(self, url: str, filename: str) -> str:
os.makedirs(self._download_dir, exist_ok=True)
local_path = os.path.join(self._download_dir, filename)
async with httpx.AsyncClient(timeout=download_timeout, follow_redirects=True) as client:
response = await client.get(url)
response.raise_for_status()
with open(local_path, "wb") as f:
f.write(response.content)
return local_path
```
#### 文件命名规则
| 文件类型 | 命名格式 | 示例 |
|---------|---------|------|
| CSV（注册表格） | `{流水号}_WEEE.csv` | `POEPR20260716000061_WEEE.csv` |
| ZIP（授权书） | `{流水号}_mandat.zip` | `POEPR20260716000061_mandat.zip` |
> **注意**：每次提交前自动清空下载目录，OSS 文件有更新时无需任何手动操作。
#### 文件名来源
文件名由 OSS URL 的 basename 决定（[import_mandats_page.py](file:///d:/phpstudy_pro/WWW/fastapi-playwright-automation/apps/epr_fr_weee_register/pages/import_mandats_page.py) 第 156 / 196 行）：
```python
# CSV 文件名
csv_filename = os.path.basename(csv_url.split("?")[0]) or "EPR_Ecologic.csv"
# ZIP 文件名
zip_filename = os.path.basename(zip_url.split("?")[0]) or "EPR_Ecologic.zip"
```
因此实际文件名取决于 OSS 上的原始文件名（如 `EPR_Ecologic_{公司英文名}.csv`），而非固定格式。
#### 服务器端 .env 实际配置
服务器上 `/opt/autobot/.env` 中的实际值可能与代码默认值不同：
| 配置项 | 代码默认值 | 服务器实际值 |
|--------|-----------|-------------|
| `SUBMISSION_TIMEOUT_SECONDS` | 300 | **120** |
| `APP_ENV` | - | `prod` |
| `HEADLESS` | - | `true` |
> **重要**：服务器 `SUBMISSION_TIMEOUT_SECONDS=120`，比代码默认值 300 少，大文件下载可能超时。如需调整，修改 `/opt/autobot/.env` 后重启服务。
#### 检查是否已有下载文件
```bash
# 在服务器上检查
sudo ls -lt /opt/autobot/.autobot/downloads/ecologic/
# 如果目录不存在（首次下载前），手动创建可确认路径正确
sudo -u automation mkdir -p /opt/autobot/.autobot/downloads/ecologic/
```
> 目录仅在首次执行 `_download_file` 时由 `os.makedirs(exist_ok=True)` 自动创建，因此新部署或清理后该目录可能不存在。
### 4.7 运行时生成的其他文件
#### 错误截图
当提交流程失败时，[submission_poller.py](file:///d:/phpstudy_pro/WWW/fastapi-playwright-automation/apps/epr_fr_weee_register/submission_poller.py) 的 `_take_error_screenshot` 方法会在 `/tmp/` 下生成错误截图：
| 文件类型 | 路径 | 命名规则 |
|---------|------|---------|
| 提交失败截图 | `/tmp/ecologic_error_{流水号}.png` | `ecologic_error_POEPR20260716000083.png` |
```python
# submission_poller.py _take_error_screenshot
path = os.path.join(tempfile.gettempdir(), f"ecologic_error_{serial}.png")
await page.screenshot(path=path, full_page=True)
```
#### Playwright 浏览器运行时文件
Playwright 运行期间会在 `/tmp/` 下生成临时文件，正常退出后会自动清理：
| 文件/目录 | 说明 |
|-----------|------|
| `/tmp/playwright-artifacts-*` | 浏览器运行临时工件（如下载文件） |
| `/tmp/playwright_chromiumdev_profile-*` | Chromium 开发者工具 profile |
> 如果进程异常退出（`kill -9` 等），这些文件可能残留。定期清理 `/tmp/playwright-*` 可释放磁盘空间。
#### 进程检查命令
```bash
# 查看流程二服务进程
ps aux | grep -i "run-submission-poller" | grep -v grep
# 示例输出：
# automat+ 582920 1.7 1.6 518332 129836 ? Ssl 15:25 0:48 \
# /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-submission-poller
```
#### 手动清理下载缓存
```bash
# 清理所有下载的临时文件
sudo rm -rf /opt/autobot/.autobot/downloads/ecologic/*
# 清理错误截图
sudo rm -f /tmp/ecologic_error_*.png
# 清理 Playwright 残留文件
sudo rm -rf /tmp/playwright-artifacts-* /tmp/playwright_chromiumdev_profile-*
```
---
## 5. 预设
---
## 6. 配置说明
### 6.1 配置类（EcologicSettings）
```python
from common.settings import CommonSettings
class EcologicSettings(CommonSettings):
# 数据库
source_db_url: str # SaaS 源库 (vat_db)
target_db_url: str # 本地目标库 (rpa_test)
# 轮询间隔
data_poll_interval_seconds: int = 3600 # 流程一 数据抓取间隔（默认 1 小时）
submission_poll_interval_seconds: int = 120 # 流程二 提交间隔（默认 2 分钟）
# Playwright
headless: bool = False
slow_mo_ms: int = 200
# Ecologic 登录
ecologic_email: str = "info@seamew.de"
ecologic_password: str = "Meiouwang1332!"
ecologic_login_url: str = "https://producteur.ecologic-extranet.com/"
ecologic_import_url: str = "https://producteur.ecologic-extranet.com/import-mandats"
# 文件下载
download_dir: str = "/tmp/downloads/ecologic/"
# 重试
max_retry_count: int = 3
# 调试
debug_mode: bool = False
```
---
## 7. CLI 命令设计
### 7.1 cli.py 结构
```python
import typer
from apscheduler.schedulers.background import BackgroundScheduler
from playwright.async_api import async_playwright
from common.lifecycle.manager import LifecycleManager
from common.db.repository import build_db_session
from .config import get_settings
from .data_poller import EcologicDataPoller
from .submission_poller import EcologicSubmissionPoller
app = typer.Typer(help="法国WEEE和运动休闲注册自动化 CLI")
settings = get_settings()
@app.command()
def run_data_poller(
once: bool = typer.Option(False, "--once", help="只执行一轮然后退出"),
):
"""从 SaaS 库抓取数据并写入 epr_fr_weee_register 表"""
lifecycle = LifecycleManager()
source_session = build_db_session(settings.get_source_db_url())
target_session = build_db_session(settings.get_target_db_url())
poller = EcologicDataPoller(settings, lifecycle, source_session, target_session)
if once:
poller.run_once()
else:
scheduler = BackgroundScheduler()
scheduler.add_job(
poller.run_once,
"interval",
seconds=settings.data_poll_interval_seconds,
id="ecologic_data_poller",
coalesce=True,
max_instances=1,
)
scheduler.start()
lifecycle.wait_for_stop()
scheduler.shutdown()
@app.command()
def run_submission_poller(
once: bool = typer.Option(False, "--once", help="只执行一轮然后退出"),
max_retry: int = typer.Option(3, help="最大重试次数"),
):
"""从 epr_fr_weee_register 取数据，Playwright 自动注册到 Ecologic"""
lifecycle = LifecycleManager()
db_session = build_db_session(settings.get_target_db_url())
async def run_with_browser():
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=not settings.debug_mode,
slow_mo=settings.slow_mo_ms,
)
poller = EcologicSubmissionPoller(
settings, lifecycle, db_session, browser
)
poller.run_once()
import asyncio
if once:
asyncio.run(run_with_browser())
else:
scheduler = BackgroundScheduler()
scheduler.add_job(
lambda: asyncio.run(run_with_browser()),
"interval",
seconds=settings.submission_poll_interval_seconds,
id="ecologic_submission_poller",
coalesce=True,
max_instances=1,
)
scheduler.start()
lifecycle.wait_for_stop()
scheduler.shutdown()
if __name__ == "__main__":
app()
```
---
## 8. 关键组件接口定义
### 8.1 EcologicDataPoller 接口
```python
class EcologicDataPoller(DataPollerBase):
def __init__(self, settings, lifecycle, source_session, target_session):
self.source_session = source_session
self.target_session = target_session
def fetch_qualifying_records(self) -> list[dict]:
"""执行 8 表 JOIN 查询（Country=FR, PushType=304）"""
pass
def process_record(self, record: dict) -> dict:
"""数据清洗与校验"""
# 1. parseLegalPersonName 拆分姓和名
# 2. 电话清洗（法国格式：纯数字）
# 3. 地址长度校验（≤50 字符）
# 4. 必填字段校验
# 5. 法国公司 NAF Code 校验
pass
def claim_and_write(self, record: dict, processed: dict) -> None:
"""写入目标库 + 回写源库状态"""
# 1. INSERT or UPDATE epr_fr_weee_register
# 2. UPDATE EPRRegInfo SET PushTaxBureauStatus = 2
pass
```
### 8.2 EcologicSubmissionPoller 接口
```python
class EcologicSubmissionPoller(SubmissionPollerBase):
def __init__(self, settings, lifecycle, db_session, browser):
self.db_session = db_session
self.browser = browser
def select_submittable_records(self) -> list[EcologicRegistration]:
"""选取待提交记录"""
# status IN (0, 1, -1) AND retry_count < max_retry
pass
async def submit_record(self, record: EcologicRegistration) -> SubmissionResult:
"""单条记录的完整提交流程"""
# 1. 下载 CSV 和 ZIP 文件
# 2. 登录 Ecologic
# 3. 导航到 Import de comptes
# 4. 上传 CSV 文件
# 5. 检查 #response-csv 校验错误（如有错误 → 通知 bu.F_Mobile → 返回 FAILED）
# 6. 上传 ZIP 文件（Dépôt des documents）
# 7. 检查 #response-poa 校验错误（如有错误 → 通知 bu.F_Mobile → 返回 FAILED）
# 8. 确认提交
# 9. 截图上传 COS
# 10. 提取提交结果
pass
def apply_submission_result(self, record_id: int, result: SubmissionResult) -> None:
"""更新数据库状态"""
pass
```
---
## 9. 错误处理与通知机制
### 9.1 企业微信通知场景
| 通知类型 | 触发条件 | 接收人 |
| ----------------------- | ---------------------------------------- | ------------- |
| **数据异常** | 数据校验失败（必填字段缺失、格式不对等） | 数据维护组 |
| **CSV 校验失败** | Ecologic 返回 CSV 格式校验错误 | `bu.F_Mobile` |
| **POA 校验失败** | Ecologic 返回 ZIP 授权书校验错误 | `bu.F_Mobile` |
| **提交成功** | Ecologic 提交成功 | 全体 |
| **提交失败 - 人工介入** | retry_count >= 3 或 non-retryable 错误 | 运营组 |
### 9.2 通知内容模板
```markdown
【法国WEEE和运动休闲注册 - 数据异常通知】
流水号: POEPR20260407000161
公司名称: Example Company Ltd.
异常原因: 公司英文地址超过50个字符限制
时间: 2026-07-11 14:30:00
【法国WEEE Ecologic CSV校验失败】
流水号: POEPR20260407000161
公司名称: Example Company Ltd.
错误详情: There is 1 line(s) in error
行2 列M: Le champ Phone doit être au format texte, avec un espace entre les chiffres (exemple 01 02 03 04 05)
【法国WEEE Ecologic 授权书校验失败】
流水号: POEPR20260407000161
公司名称: Example Company Ltd.
错误详情: There is 1 line(s) in error
行2 列AR: document.file_not_found
【法国WEEE和运动休闲注册 - 提交成功通知】
流水号: POEPR20260407000161
公司名称: Example Company Ltd.
Ecologic 提交状态: 成功
提交时间: 2026-07-11 14:35:00
```
---
## 10. 部署与运维
### 10.1 生产部署
```bash
# 安装依赖
pip install -r requirements.txt
# 执行数据库迁移
python -m apps.epr_fr_weee_register.cli db upgrade
# === systemd 服务（推荐：崩溃自动拉起 + 开机自启）===
# 流程一：数据抓取轮询
sudo systemctl enable --now autobot-data-poller@epr_fr_weee_register
# 流程二：自动注册轮询
sudo systemctl enable --now autobot-submission-poller@epr_fr_weee_register
# 流程三：签字邮件监控 + Yousign 自动签字
sudo systemctl enable --now autobot-signing-poller@epr_fr_weee_register
# 流程四：下号检查 + 证书下载 (每天 8:00 / 17:30)
sudo systemctl enable --now autobot-uin-poller@epr_fr_weee_register
```
**systemd 服务模板文件**（位于 `deploy/` 目录）：
| 服务模板 | ExecStart |
|----------|-----------|
| `autobot-data-poller@.service` | `python -m apps.%i.cli run-data-poller` |
| `autobot-submission-poller@.service` | `python -m apps.%i.cli run-submission-poller` |
| `autobot-signing-poller@.service` | `python -m apps.%i.cli run-signing-poller` |
| `autobot-uin-poller@.service` | `python -m apps.%i.cli run-uin-poller` |
所有服务均以 `User=automation` 运行，`Restart=always`，`TimeoutStopSec=180`。
**UIN Poller 调度特殊处理**：`PollerScheduler` 新增 `schedule_cron()` 方法支持 CronTrigger（原有只有 IntervalTrigger），下号流程使用两个 cron job 分别触发 8:00 和 17:30。
### 10.2 日志路径
```
logs/epr_fr_weee_register/
├── epr_fr_weee_register_data_poller_20260711.log # 数据抓取日志
├── epr_fr_weee_register_submission_20260711.log # 自动注册日志
└── error_20260711.log # 错误日志
```
### 10.3 手动触发
```bash
# 单轮数据抓取
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-data-poller --once
# 单轮自动注册
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-submission-poller --once
# 单轮签字邮件检查
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-signing-poller --once
# 单轮下号检查
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-uin-poller --once
```
---
## 11. 关键风险点与注意事项
### 11.1 文件 URL 获取
`Mandat` 和 `WEEE_Sports_Leisure` 的文件路径通过关联附件表 `Base_AnnexesFile` 获取（按文件名关键词匹配），再进行路径替换（`MAIN_SITE_PATH` → `FILE_SITE_PATH`）得到可访问的 OSS URL。
**注意事项：**
- `pa_annex` 匹配文件名关键词：`mandat` / `poa` / `vollmacht`
- `mb_annex` 匹配文件名关键词：`weee` / `csv` / `ecologic` / `sports`
- 如果文件名不符合上述关键词规则，会导致 JOIN 不到数据，需要更新关键词列表
### 11.2 CSV 格式要求
CSV 文件严格按照「法国申请weee或运动休闲表格（格式转换说明）.xlsx」文档要求：
- 分号分隔
- UTF-8 编码（BOM）
- 第 1 行：`sep=;`
- 第 2 行：48 个表头
- 第 3 行：48 个数据
- 电话号码：纯数字（Excel 格式 `0#" "##" "##" "##" "##` 处理显示）
- 地址：≤50 字符，不截断（超出报错）
- 包含逗号的字段自动用双引号包裹
### 11.3 Ecologic 网站变更风险
- 选择器变更 → Page Objects 需要同步更新
- 登录页面改版 → LoginPage 需要适配
- 导入流程变化 → ImportMandatsPage 需要扩展
### 11.4 并发控制
- `coalesce=True` + `max_instances=1` 防止任务叠加
- `status=1` 认领机制防止多进程重复处理同一条记录
### 11.5 数据库连接池
两个数据库连接（SaaS + Local）需要独立配置连接池大小，避免连接泄漏。
---
## 12. 项目文件结构
```
apps/epr_fr_weee_register/
├── cli.py # FastAPI CLI 入口（5 个流程命令）
├── config.py # EcologicSettings 配置
├── models.py # SQLAlchemy ORM 模型
├── data_poller.py # 流程一：数据抓取入库
├── submission_poller.py # 流程二：Playwright 自动注册
├── signing_poller.py # 流程三：Yousign 签字监控
├── signing_mail_handler.py # 流程三：邮件匹配与解析
├── uin_poller.py # 流程四：下号检查
├── submission_overdue_notify.py # 流程五：提交超期通知
├── service.py # 业务服务层（状态机、SaaS 回写等）
├── results.py # 数据契约定义
│
├── pages/
│ ├── __init__.py
│ ├── login_page.py # 登录页 Page Object (Ecologic)
│ ├── import_mandats_page.py # Import de comptes Page Object
│ ├── signing_page.py # Yousign 签字 Page Object
│ └── uin_page.py # 下号查询 Page Object
│
├── repository/
│ ├── __init__.py
│ ├── source_repo.py # 源库操作 (vat_db)
│ ├── target_repo.py # 目标库操作 (rpa_test)
│ └── source_query.sql # 8 表 JOIN 查询语句
│
├── migrations/
│ ├── env.py
│ ├── script.py.mako
│ └── versions/
│ └── 001_init_ecologic_schema.py # ALTER TABLE 迁移
│
└── .env.example # 环境变量模板
```
---
## 附录 A：数据库表 DDL
```sql
CREATE TABLE epr_fr_weee_register (
id INT IDENTITY(1,1) PRIMARY KEY,
tid VARCHAR(64) NOT NULL, -- BusinessSerialNumber
status INT NOT NULL DEFAULT 0, -- 0=PENDING, 1=PROCESSING, 2=SUCCESS, -1=FAILED, 6=MANUAL
retry_count INT NOT NULL DEFAULT 0,
msg VARCHAR(512) DEFAULT '',
-- 源数据关联
VATBusinessID VARCHAR(64), -- RegisterID
EPRRegInfoID VARCHAR(64), -- EPRRegInfo.ID
-- 公司基本信息
Company_Name_en VARCHAR(256),
Company_Name_cn VARCHAR(256),
StreetNrBox VARCHAR(256), -- 公司地址
Zipcode VARCHAR(16), -- 邮编
City VARCHAR(128), -- 城市
Country VARCHAR(64), -- 国家
CountryTwoCode VARCHAR(8), -- 国家二字码
-- 注册信息
VAT_or_company_number VARCHAR(64), -- SIRET 号
VAT_nr VARCHAR(64), -- VAT 税号（欧盟）
NAFCode VARCHAR(16), -- NAF 编码（法国）
legal_form VARCHAR(32), -- 公司类型
RegisteredCapital VARCHAR(64), -- 注册资本
ServiceYear VARCHAR(8), -- 服务年份
ContractType VARCHAR(32), -- EEE Ménager / ASL
-- 法人信息
Legal_LastName VARCHAR(64),
Legal_FirstName VARCHAR(128),
Legal_Salutation VARCHAR(8), -- Mr / Mrs
Legal_Email VARCHAR(256),
Phone_number VARCHAR(32),
-- 业务信息
Sector VARCHAR(256), -- 产品范围
-- 文件 URL
Mandat VARCHAR(512), -- 授权书 ZIP (OSS URL)
WEEE_Sports_Leisure VARCHAR(512), -- 注册表格 CSV (OSS URL)
-- 追踪字段
screenshot_urls VARCHAR(2048), -- 截图 URL（JSON 数组）
created_at DATETIME DEFAULT GETDATE(),
updated_at DATETIME DEFAULT GETDATE()
);
CREATE INDEX idx_epr_fr_weee_status ON epr_fr_weee_register(status);
CREATE INDEX idx_epr_fr_weee_tid ON epr_fr_weee_register(tid);
```
## 附录 B：API 响应格式
```json
{
"code": 200,
"msg": "success",
"data": [{
"Mandat": "http://vat-1259285998.cos.ap-guangzhou.myqcloud.com/vat_factory/fr_declar/EPR_Ecologic_Songzishifuhuajiancaidiangetigongshanghu.zip",
"WEEE_Sports_Leisure": "http://vat-1259285998.cos.ap-guangzhou.myqcloud.com/vat_factory/fr_declar/EPR_Ecologic_Songzishifuhuajiancaidiangetigongshanghu.csv"
}]
}
```
| 字段 | 说明 | 格式 |
| --------------------- | --------------------- | ----------- |
| `Mandat` | 授权书（POA） | ZIP (含PDF) |
| `WEEE_Sports_Leisure` | WEEE/运动休闲注册表格 | CSV UTF-8 |
## 13. 流程三：签字邮件监控 + Yousign 自动签字（run-signing-poller）
### 13.1 业务背景
流程二（提交注册）成功后，Ecologic 会在 **1-2 天** 内发送签字邮件到 `info@seamew.de`。邮件来自 `notifications@yousign.app`，要求法人在 Yousign 平台完成电子签字。签字完成后才算注册正式生效。
### 13.2 入口命令
```bash
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-signing-poller --once
```
### 13.3 邮件匹配规则
签字邮件发送到企业微信邮箱 `info@seamew.de`，存放在 IMAP 文件夹 `法国WEEE/Ecologic/` 下。
**邮件规则：**
| 属性 | 规则 |
| -------------------- | ------------------------------------------------------------ |
| 来源 | `notifications@yousign.app` |
| 文件夹 | `法国WEEE/Ecologic/` |
| WEEE 签字邀请邮件标题 | `Ecologic France vous invite à signer Document - Contrat C{合同编号} - EEE Ménager sur Yousign` |
| WEEE 签字提醒邮件标题 | `Rappel : vous n'avez pas signé Document - Contrat C{合同编号} - EEE Ménager` |
| WEEE Youtrust 邀请 | `Vous avez été invité(e) à signer Document - Contrat C{合同编号} - EEE Ménager par Ecologic France sur Youtrust` |
| 运动休闲签字邮件标题 | `Ecologic France vous invite à signer Document - Contrat C{合同编号} - ASL` |
| **排除邮件** | `"Document - Contrat CXXXXX" : tout le monde a signé`（全体已签通知，不处理） |
| **排除邮件** | `Vous venez de signer Document - Contrat CXXXXX`（本人刚签完通知，不处理） |
| 验证码邮件标题 | `{验证码} est votre code de securite pour Document - Contrat C{合同编号} - EEE Menager`（或 ASL） |
**合同编号与公司匹配步骤：**
1. 从邮件标题提取合同编号（`CXXXXX` 格式）
2. **仅当 Yousign 链接有效时**（`_check_yousign_link` 返回 `ready` 或 `already_signed`）
3. Playwright 登录 Ecologic 门户 → 点击「My producers」
4. DataTables 搜索框输入 C{合同编号} 搜索
5. 提取搜索结果第一行：Member code（列0）、Company name（列1）、Registration number（列2）
6. **先验证 Member code 非空** → 空则搜索失败，报错：`"签字邮件C{xxxx}未在门户网站匹配到合同编码"`
7. 用营业执照号（Registration number）精确匹配 DB `vat_or_company_number` 字段
8. 回退：用公司名匹配 DB `company_name` 字段
9. DB 匹配失败 → 企微通知：`"签字邮件C{xxxx}未在数据库中匹配到对应单号"`
### 13.4 签字邮件标题解析
```python
import re
# 签字邮件标题正则
_SIGNING_SUBJECT_PATTERNS = {
"weee": re.compile(
r"Document - Contrat (C\d+) - EEE M.nager",
re.IGNORECASE
),
"asl": re.compile(
r"Document - Contrat (C\d+) - ASL",
re.IGNORECASE
),
}
```
签字邀请邮件筛选正则：
```python
# 仅用于判断邮件是否为签字邀请（不分组）
_SIGNING_INVITE_SUBJECT_RE = re.compile(
r"Document - Contrat (C\d+) - (EEE M.nager|ASL)",
re.IGNORECASE
)
# 排除已完成的邮件通知
_SIGNING_EXCLUDE_RE = re.compile(
r"tout le monde a signé|venez de signer",
re.IGNORECASE,
)
# 在 extract_signing_link() 中先匹配 _SIGNING_INVITE_SUBJECT_RE，
# 再排除 _SIGNING_EXCLUDE_RE，确保只处理签字邀请和提醒邮件
# 验证码邮件标题正则
_VERIFY_CODE_PATTERNS = {
"weee": re.compile(
r"(\d{6}) est votre code de s.curit. pour Document - Contrat (C\d+) - EEE M.nager",
re.IGNORECASE
),
"asl": re.compile(
r"(\d{6}) est votre code de s.curit. pour Document - Contrat (C\d+) - ASL",
re.IGNORECASE
),
}
def parse_signing_subject(subject: str) -> tuple[str | None, str | None]:
"""解析签字邮件标题，提取合同类型和合同编号。
:returns: (contract_type, contract_code)
contract_type: "weee" / "asl" / None
contract_code: "C45507" / None
"""
for ctype, pattern in _SIGNING_SUBJECT_PATTERNS.items():
match = pattern.search(subject)
if match:
return ctype, match.group(1)
return None, None
def parse_verify_code_subject(subject: str) -> tuple[str | None, str | None, str | None]:
"""解析验证码邮件标题，提取验证码、合同编号和合同类型。
:returns: (code, contract_code, contract_type)
"""
for ctype, pattern in _VERIFY_CODE_PATTERNS.items():
match = pattern.search(subject)
if match:
return match.group(1), match.group(2), ctype
return None, None, None
```
### 13.5 签字流程调用链路
```
cli.run_signing_poller()
│
├─ 1. 加载配置 (EcologicSettings)
├─ 2. 初始化 IMAP 客户端 (ImapToolsClient)
├─ 3. LifecycleManager 注册信号
│
└─ 4. EcologicSigningPoller.run_once()
│
├─ 4a. IMAP 拉取未读邮件
│ └─ folder = "Ecologic/Ecomaison验证码"
│ └─ limit = 50
│
├─ 4b. 快速检测 Yousign 链接状态 (_check_yousign_link)
│ ├─ "Déverrouiller l'accès" → 链接过期，跳过
│ ├─ "Signature finalisée" → 已签完字
│ ├─ "Not available" → 链接失效，跳过
│ └─ "Commencer" 按钮可见 → 有效链接，继续
│
├─ 4c. 门户搜索 + DB 匹配 (_find_record_by_contract)
│ └─ 仅当 Yousign 状态为 ready / already_signed 时执行
│ └─ Playwright 登录 Ecologic → My producers
│ └─ DataTables 搜索 C{合同编号} → 提取 Member code + Company name + Reg number
│ └─ Member code 非空 → 搜索成功；空 → 报错: "未在门户网站匹配到合同编码"
│ └─ Reg number 匹配 DB vat_or_company_number → 公司名匹配
│ └─ DB 无匹配 → 企微通知: "签字邮件C{xxxx}未在数据库中匹配到对应单号"
│
├─ 4d. 执行 Yousign 签字 (仅当 y_status=ready + DB 匹配成功)
│ └─ 见 13.6 详细步骤
│ └─ 已签完字 (y_status=already_signed) → 直接更新 DB
│
└─ 4e. apply_result()
├─ status=3 (SIGNED) 或 status=-2 (SIGN_FAILED)
├─ contract_code / ecologic_member_code
└─ 企业微信通知
```
### 13.6 Yousign 签字 Playwright 自动化流程
#### Phase A: Detect Yousign page status (_check_yousign_link)
Open the link and wait for React to render before checking state:
```python
# 1. Open the Yousign signing link from IMAP
await page.goto(signing_link, wait_until="domcontentloaded", timeout=120000)
# 2. Wait for React to render (Yousign is a React SPA)
await page.wait_for_function(
"() => document.querySelectorAll('button').length > 0 || document.querySelectorAll('h1').length > 0",
timeout=30000
)
# 3. Handle anti-phishing page (optional)
anti_phishing = page.locator("button:has-text('Continue')")
if await anti_phishing.count() > 0:
await anti_phishing.click()
# 4. Check page status (by priority)
- Deverrouiller l'accès
-> "unlock_required" (link expired -> skip)
- Signature finalisée
-> "already_signed" (already signed -> portal search + update DB)
- h1: "Sorry, this content is not available" -> "not_available" (link broken -> skip)
- button: "Commencer" -> "ready" (valid link -> portal search + DB match + execute signing)
Portal search + DB match only triggered for "ready" or "already_signed" statuses.
```
#### 阶段 B：门户搜索 + DB 匹配（_find_record_by_contract → _search_contract_on_portal）
仅当 Yousign 链接有效时才执行，避免无效链接触发昂贵的门户搜索：
```
1. 从邮件标题提取 C{合同编号}
2. Playwright 登录 Ecologic → My producers → DataTables 搜索合同编号
3. 提取同行: Member code (AXXXX) + Company name + Registration number (营业执照号)
4. 营业执照号 匹配 DB vat_or_company_number 字段
5. 回退: 公司名 匹配 DB company_name 字段
6. 无匹配 → 企微通知: "签字邮件C{xxxx}未在数据库中匹配到对应单号"
```
#### Phase C: Yousign signing execution (signing_page.py execute_signing)
Signing flow (Yousign new version, no ACCEDER button). **所有超时时间已调整为原始值的 3 倍**以适配服务器慢网络环境（签字按钮 20s 等待除外）。
```python
# 0. Open link -> wait for React render
await page.goto(signing_link, wait_until="domcontentloaded", timeout=360000)
await page.wait_for_function("() => document.querySelectorAll('button').length > 0", timeout=90000)
# 1. Click Commencer (start)
await page.locator("button:has-text('Commencer')").click()
await page.wait_for_timeout(6000)
# 2. Wait for document page to load (can take up to 360s)
try:
await page.wait_for_function("() => document.querySelectorAll('button').length > 1", timeout=360000)
except: ...
# 3. Scroll to bottom with mouse wheel until Continuer is enabled
for i in range(300):
await page.mouse.wheel(0, 500)
await page.wait_for_timeout(800)
btn = page.locator("button:has-text('Continuer')")
if await btn.count() > 0:
disabled = await btn.get_attribute("disabled")
if disabled is None:
break
await page.locator("button:has-text('Continuer')").click()
await page.wait_for_timeout(9000)
# 4. Verification code input
code_input = page.locator("input[autocomplete='one-time-code']")
if await code_input.count() > 0:
verify_code = await wait_for_verify_code(imap, folder)
await code_input.click()
await code_input.fill(verify_code)
await page.wait_for_timeout(3000)
# 等待签字按钮出现 (Yousign 本地处理验证码，固定 20s)
await page.wait_for_timeout(20000)
# 5. Cliquer pour signer (final) — 按钮特征: data-monitoring="sign-flow-sign-slider"
sign_btn = page.locator("button[data-monitoring='sign-flow-sign-slider']")
await sign_btn.wait_for(state="visible", timeout=180000)
await sign_btn.click()
await page.wait_for_timeout(15000)
# 6. Verify Signature finalisee
success = page.locator("h1:has-text('Signature finalis\u00e9e')")
await success.wait_for(state="visible", timeout=180000)
```
**Key changes from old Yousign:**
- Removed ACCEDER AU DOCUMENT step (not in new Yousign UI)
- Added `wait_for_function` to wait for React SPA rendering
- Document scrolling uses `page.mouse.wheel()` instead of `window.scrollTo()`
- Continuer button is disabled until document is fully scrolled
- Scroll loop increased to `range(300)` — Yousign 文档页加载较慢
- **服务器慢网络适配**：所有超时时间翻 3 倍（goto 360s, 按钮等待 180s 等），签字按钮本地等待保持 20s
- **DataTable AJAX 加载循环重试**：门户搜索中 My producers 点击后，循环等待最多 10 次 × 60s 直到 DataTable 数据加载完成（适配服务器慢网络）
- 验证码提取：直接从邮件标题取前 6 位数字 + 匹配合同编号，无需正文回退
- **签字按钮选择器**：优先使用 `data-monitoring="sign-flow-sign-slider"` 属性，回退到文本匹配
### 13.7 签字结果数据结构
```python
@dataclass
class SigningResult:
"""签字流程结果。"""
status: SubmissionStatusEnum # SUCCESS / FAILED / MANUAL_INTERVENTION
contract_type: str | None # "weee" / "asl"
contract_code: str | None # "C45507"
error_class: str | None # "retryable" / "non-retryable"
error_message: str | None
evidence_path: str | None # 截图路径（成功后页面截图）
```
### 13.8 签字状态机与企微通知
```
status=2 (SUBMIT_SUCCESS)
│
├─ 收到签字邮件 → 执行签字
│ ├─ 成功 → status=3 (SIGNED)
│ │ signed_doc_url = OSS URL
│ │ 企业微信通知: "Yousign 签字完成" (含 @bu_f_mobile + @15817402851)
│ │ 回写 SaaS EPRRegInfo.Remarks = "已经签字成功"
│ │
│ └─ 失败 → status=-2 (SIGN_FAILED)
│ retry_count++
│ 企业微信通知: "签字失败" (含 @bu_f_mobile + @15817402851)
│ └─ retry_count < max → 下一轮重试 (最多重试 3 次)
│
└─ 超过 N 天未收到签字邮件
└─ 企业微信通知: "等待签字超时, 请人工跟进"
```
### 13.9 邮件去重机制
```python
def _build_mail_fingerprint(mail: FetchedMail) -> str:
"""构建邮件指纹用于去重。"""
import hashlib
raw = f"{mail.message_id}|{mail.subject}|{mail.from_addr}"
return hashlib.md5(raw.encode(), usedforsecurity=False).hexdigest()
```
### 13.10 签字流程核心代码
**重要说明：为什么不通过 Playwright 登录邮箱获取签字链接？**
自动化时使用 **IMAP 协议** 替代 Playwright 登录邮箱：
| 对比维度 | Playwright 登录邮箱 | IMAP 拉取邮件 |
| -------- | ----------------------------------- | ---------------------- |
| 速度 | 慢（加载 iframe、导航文件夹） | 快（直接 API 查询） |
| 稳定性 | 差（邮箱 UI 经常改版、iframe 跨域） | 稳（RFC 标准协议不变） |
| 资源 | 占用浏览器实例 | 无浏览器开销 |
因此签字流程分为两个模块：
1. `signing_mail_handler.py` — **IMAP** 拉取未读邮件，匹配签字邮件，提取链接 + 验证码
2. `signing_page.py` — **Playwright** 打开 Yousign 完成签字
3. `signing_poller.py` — 编排层，协调 Yousign 检测 → 门户搜索 → DB 匹配 → 签字执行
**核心方法：**
| 方法 | 文件 | 职责 |
|------|------|------|
| `extract_signing_link()` | signing_mail_handler.py | 从 HTML 提取 Yousign 链接 |
| `wait_for_verify_code()` | signing_mail_handler.py | IMAP 轮询获取验证码 |
| `_check_yousign_link()` | signing_poller.py | 打开 Yousign，返回页面状态 |
| `_search_contract_on_portal()` | signing_poller.py | 门户 DataTables 搜索合同 |
| `_find_record_by_contract()` | signing_poller.py | 营业执照号 + 公司名匹配 DB |
| `execute_signing()` | signing_page.py | Commencer → 滚动 → Continuer → 验证码 → 签字 → Signature finalisée |
**Yousign 页面关键选择器：**
| 步骤 | 选择器 |
| -------------------- | --------------------------------------------- |
| 反钓鱼继续访问 | `button:has-text("继续访问")` |
| 开始按钮 | `button:has-text("Commencer")` |
| 浏览文档后继续 | `button:has-text("Continuer")` |
| 签字按钮 | `button:has-text("Cliquer pour signer")` |
| 备用签字按钮 | `Signer / Valider / Sign / Validate` |
| 验证码输入框 | `input[autocomplete='one-time-code']` |
| 已签完字 | `h1:has-text("Signature finalisée")` |
| 链接过期需解锁 | `h1:has-text("Déverrouiller l'accès")` |
| 链接彻底失效 | `h1:has-text("Sorry, this content is not available")` |
### 13.11 异常处理
| 异常场景 | 处理策略 |
| ---------------------- | ------------------------------------- |
| IMAP 连接失败 | 记录错误, 下轮重试 |
| Yousign 链接已失效 | `_check_yousign_link` 返回 not_available → 标记已读跳过 |
| Yousign 链接过期 | `_check_yousign_link` 返回 unlock_required → 标记已读跳过（等待重发） |
| 门户搜索无结果 | 企微通知: "签字邮件C{xxxx}未在门户网站匹配到合同编码" |
| DB 匹配无对应记录 | 企微通知: "签字邮件C{xxxx}未在数据库中匹配到对应单号" |
| Yousign 页面加载超时 | 截图留存, status=-2, 下轮重试 |
| 验证码超时未收到 | 截图留存, status=-2, 下轮重试 |
| 签字确认页未出现 | 截图留存, status=-2, 下轮重试 |
---
## 14. 流程四：下号检查 + 证书下载（run-uin-poller）
### 14.1 业务背景
签字完成后，Ecologic 需要 **1-2 周** 审批时间，审批通过后会在 Ecologic 门户生成注册号（UIN）和证书。
**触发条件**：签字完成 **5 天后** 才开始查询下号（`signed_at + 5天 <= 当前时间`），避免过早查询浪费资源。
**调度频率**：每天 **8:00** 和 **17:30** 各运行一次（CronTrigger `hour="8" minute="0"` + `hour="17" minute="30"`）。
### 14.2 入口命令
```bash
sudo -u automation /opt/autobot/.venv/bin/python -m apps.epr_fr_weee_register.cli run-uin-poller --once
```
### 14.3 下号验证流程
#### 步骤 A：登录 + 进入 My producers + 搜索公司
```
1. 登录 Ecologic 门户
https://producteur.ecologic-extranet.com/
邮箱: info@seamew.de / 密码: Meiouwang1332!
2. 点击 "Mes producteurs" 进入 DataTables 搜索页面
3. 等待 networkidle + 2s 确保 DataTables 初始化完成
4. 在搜索框 (input[type='search']) 输入营业执照号或公司名
5. 按 Enter 触发搜索，等待 networkidle + 2s
6. 查看搜索结果列表
```
**搜索结果字段解读：**
| 列 | 字段 | 说明 |
| ---- | ------------------- | -------------------------------- |
| 1 | Member code | Ecologic 会员编号 (如 A6967) |
| 2 | Company name | 公司英文名 |
| 3 | Registration number | 营业执照号 / SIRET |
| 4 | **UIN Number** | 注册号（唯一识别号），下号后出现 |
| 5(末) | 操作按钮 | 蓝色图标，进入详情 |
> 匹配时除了检查 UIN 列，还会提取同行全部列（member_code / company_name / vat_number）与 DB 做双重校验（营业执照号 + 公司名）。
#### 步骤 B：进入详情页 + 确认信息
```
7. 点击最后一列 (td_count-1) 的详情/编辑按钮进入详情
8. 出现发票信息确认页面，点击 "Record" 确认
```
#### 步骤 C：下载证书（路由拦截方式）
```
9. 点击 "Contracts & Membership Certificates" 标签
10. 通过 Sector 字段区分合同类型:
- "EEE Menager" → WEEE 证书
- "ASL" → 运动休闲证书
11. 拦截 attestation 链接: page.route("**/generate-attestation/**", capture_pdf)
→ 直接捕获 PDF 响应字节，不触发浏览器下载
12. 去掉 target="_blank" 避免跳转新标签页，点击 attestation 链接
13. 保存证书 PDF 到 tmp/ 目录，命名: 公司中文名(优先)-法国WEEE证书.pdf
14. 上传证书到 COS（路径: oss_cer_path，默认 "epr/fr/weee/cer/"）
15. 提取 UIN Number 并更新数据库（uin_number 字段，同时同步存 contract_code）
```
### 14.4 调用链路
```
cli.run_uin_poller()
│
├─ 1. 加载配置 (EcologicSettings)
├─ 2. LifecycleManager 注册信号
│
└─ 3. EcologicUinPoller.run_once()
│ (override: 开头通知 + 超期报警 + 无记录跳过 + 整个 run 只登录一次)
│
├─ _send_start_notify(): 发一条 "现在开始查询下号" @15817402851
│
├─ _check_overdue_alerts(): 扫描所有签字满15天仍未下号的记录，逐条发报警通知
│
├─ 3a. 选取已签字记录（一次返回全部，不再 LIMIT 1）
│ └─ status=3 AND uin_number IS NULL
│ └─ signed_at IS NOT NULL AND signed_at + 5天 <= 当前时间
│ └─ uin_retry_count < uin_max_retry_limit（默认7）
│ └─ ORDER BY signed_at ASC（每次运行把全部符合条件的记录都循环查一遍）
│ └─ 无待处理记录时直接结束（避免空跑登录门户）
│
├─ 3b. 建立共享会话（仅一次）: BrowserManager → new_context → new_page
│ ├─ EcologicLoginPage.login(): 登录一次
│ ├─ 点击 "My producers" → wait_for_url("**/compte-manager**", wait_until="commit")
│ │ （不等 load 事件，避免被 cdn.rawgit.com 卡死，详见 14.13）
│ └─ DataTable AJAX 加载循环重试(10×60s)
│ 登录/进搜索页失败 → 本轮跳过（不触发逐条 pipeline 错误通知）
│
└─ 3c. 基类 run_once 循环（复用共享会话页面，一条 run 查完所有记录）:
├─ 跳过本轮已处理过的记录、继续处理未处理记录，直到本批全部处理完
│ （单轮内每条记录至多处理一次）
├─ claim_record: 原子认领为 status=1(processing)
└─ submit(record):
├─ _ensure_search_page(): 不在搜索页则回到 compte-manager
│ （上一条成功时页面停在详情/证书页；失败则尝试恢复会话，
│ 恢复不了则抛给基类回退 SIGNED 并结束本轮）
├─ EcologicUinPage.fetch_uin_and_certificate()
│ ├─ 搜索营业执照号 → 双重校验匹配
│ ├─ 检查 UIN 列是否有值
│ ├─ 有 UIN → 详情页 → 确认 Record
│ ├─ Contracts & Membership Certificates
│ ├─ 路由拦截 attestation → 下载证书 PDF
│ └─ 返回 UinResult
│
├─ 上传证书到 COS (oss_cer_path)
│
└─ apply_result():
├─ SUCCESS → status=4, uin_number, certificate_url, 企微通知
├─ PENDING → 回退 status=3 (不发通知)
├─ FAILED → 仅当签字超15天才发企微报警
└─ SaaS 回写（仅 SUCCESS 时执行）
```
### 14.5 下号检查 Page Object
```python
class EcologicUinPage(BasePage):
"""Ecologic 下号证书页面 Page Object。"""
_SEARCH_INPUT = "input[type='search']"
_SEARCH_RESULT_ROWS = "table tbody tr"
_RECORD_BUTTON = "button:has-text('Record')"
_CONTRACTS_TAB = "text=Contracts & Membership Certificates"
_DETAIL_BTN_SELECTOR = "a, button"
async def fetch_uin_and_certificate(self, record) -> UinResult:
"""
流程:
1. 已在调用方点击 "My producers" → 进入 DataTables 搜索页面
2. 搜索公司 → 检查 UIN 列
3. 如有 UIN → 进入详情 → 确认发票 → 下载证书
"""
page = self.page
# 注意: 调用方已在 submit() 中完成 "My producers" 点击，
# 此处直接搜索公司，不再重复导航
# Step 1: 搜索公司
search_keyword = record.vat_or_company_number or record.company_name or ""
await page.locator(self._SEARCH_INPUT).fill(search_keyword)
await page.keyboard.press("Enter")
await page.wait_for_load_state("networkidle")
await page.wait_for_timeout(2000)
# Step 2: 查找搜索结果中的 UIN (第4列)
rows = page.locator(self._SEARCH_RESULT_ROWS)
row_count = await rows.count()
uin_number = None
for i in range(row_count):
row = rows.nth(i)
tds = row.locator("td")
td_count = await tds.count()
if td_count < 5:
continue
# 提取各列
member_code = (await tds.nth(0).text_content() or "").strip()
company_name = (await tds.nth(1).text_content() or "").strip()
vat_number = (await tds.nth(2).text_content() or "").strip()
uin_cell = tds.nth(3) # UIN Number 在第4列
# 用营业执照号或公司名匹配记录
record_vat = (record.vat_or_company_number or "").strip()
record_company = (record.company_name or "").strip()
if vat_number != record_vat and company_name.lower() != record_company.lower():
continue
uin_text = (await uin_cell.text_content() or "").strip()
if uin_text and uin_text != "-":
uin_number = uin_text
# 点击最后一列的详情/编辑按钮
detail_btn = tds.nth(td_count - 1).locator(self._DETAIL_BTN_SELECTOR).first
await detail_btn.click()
await page.wait_for_load_state("load")
await page.wait_for_timeout(2000)
break
if not uin_number:
return UinResult(
status=S.PENDING,
error_message="UIN 尚未生成, 等待下号",
)
# Step 3: 确认发票信息 → 点击 Record
await page.wait_for_load_state("load")
await page.locator(self._RECORD_BUTTON).click()
await page.wait_for_timeout(1000)
# Step 4: 切换到 Contracts & Membership Certificates 标签
await page.locator(self._CONTRACTS_TAB).click()
await page.wait_for_timeout(2000)
# Step 5: 根据 Sector 匹配并下载证书（路由拦截方式）
sector = getattr(record, "sector", "") or ""
contract_type = self._resolve_contract_type(sector)
cert_row = page.locator(f"tr:has(td:text('{contract_type}'))")
attestation_link = cert_row.locator(
"a[href*='attestation']"
).first
pdf_data = None
async def _capture_pdf(route):
nonlocal pdf_data
response = await route.fetch()
pdf_data = await response.body()
await route.fulfill(response=response)
await page.route("**/generate-attestation/**", _capture_pdf)
# 去掉 target="_blank" 避免跳转到新标签页
await attestation_link.evaluate("el => el.removeAttribute('target')")
await attestation_link.click()
await page.wait_for_load_state("load")
await page.wait_for_timeout(2000)
serial = getattr(record, "business_serial_number", "unknown")
company = getattr(record, "company_name_cn", None) or getattr(record, "company_name", "unknown") or "unknown"
cert_label = "法国WEEE证书" if contract_type == "EEE Ménager" else "法国运动证书"
filename = f"{company}-{cert_label}.pdf"
local_path = os.path.join(self._download_dir, filename)
if pdf_data:
with open(local_path, "wb") as f:
f.write(pdf_data)
else:
raise RuntimeError("未捕获到证书 PDF 数据")
return UinResult(
status=S.SUCCESS,
uin_number=uin_number,
certificate_local_path=local_path,
contract_type=contract_type,
)
def _resolve_contract_type(self, sector: str) -> str:
"""根据 Sector 字段解析合同类型。"""
sector_lower = sector.lower()
if "weee" in sector_lower or "eee" in sector_lower or "menager" in sector_lower:
return "EEE Ménager"
if "asl" in sector_lower or "sport" in sector_lower or "leisure" in sector_lower:
return "ASL"
return "EEE Ménager"
```
### 14.6 下号结果数据结构
```python
@dataclass
class UinResult:
"""下号流程结果。"""
status: SubmissionStatusEnum # SUCCESS / FAILED
uin_number: str | None # Ecologic 注册号 UIN
certificate_local_path: str | None # 证书本地路径
certificate_oss_url: str | None # 证书 OSS URL
contract_type: str | None # "EEE Menager" / "ASL"
error_class: str | None
error_message: str | None
evidence_path: str | None # 截图路径
```
### 14.7 下号状态机与企微通知
```
status=3 (SIGNED)
│
├─ 签字满 5 天后 → 每天 8:00 / 17:30 检查 Ecologic 门户
│ ├─ 每轮开始: 发 "现在开始查询下号" @15817402851
│ ├─ 扫描: 签字满15天仍未下号 → 逐条发送超期报警 @bu_f_mobile + @15817402851
│ │
│ ├─ UIN 已生成 → status=4 (UIN_ISSUED)
│ │ uin_number = 提取值
│ │ certificate_url = OSS URL
│ │ 企业微信通知: "已下号: UIN=XXX"
│ │
│ └─ UIN 未生成 → 回退 status=3 (PENDING)，不发通知
│
└─ 下号失败 (超时/异常)
├─ 签字 < 15 天 → 不发通知，继续重试
└─ 签字 ≥ 15 天 → 企业微信报警: "下号失败(签字已超15天)" @bu_f_mobile + @15817402851
```
### 14.8 同一公司多合同处理
如果一条记录同时注册了 WEEE 和运动休闲，在 `Contracts & Membership Certificates` 标签下会出现两行。
> **注意**：当前 `EcologicUinPage.fetch_uin_and_certificate()` 仅处理单个合同类型（根据 Sector 字段解析），如需处理同一公司同时有 WEEE + ASL 的场景，需在调用层循环处理多个 contract_type。
### 14.9 Ecologic 门户页面结构
```
/compte-manager ← 点击 "Mes producteurs" 后跳转
├── input[type='search'] ← DataTables 搜索框
├── table tbody tr*N ← 搜索结果
│ ├── td:nth(0) = Member code
│ ├── td:nth(1) = Company name
│ ├── td:nth(2) = Registration number
│ ├── td:nth(3) = UIN Number
│ └── td:nth(末) = 详情/编辑按钮
/consulter-compte/{id}/detail
├── 发票确认表单
│ └── button:has-text('Record') ← 确认按钮
├── Tab 1-3: 其他信息
└── Tab 4: Contracts & Membership Certificates
├── tr:has(td:text('EEE Ménager')) ← WEEE 证书行
│ └── a[href*='attestation'] ← 证书下载链接（路由拦截）
└── tr:has(td:text('ASL')) ← 运动休闲证书行
└── a[href*='attestation'] ← 证书下载链接（路由拦截）
```
### 14.10 异常处理
| 异常场景 | 处理策略 |
| ----------------- | ----------------------------- |
| Ecologic 登录失败 | status=-3, 下轮重试 |
| 搜索结果无匹配 | 公司可能尚未录入, 等待下轮 |
| UIN 列为空 | 返回 S.PENDING，回退到 SIGNED，等待下轮 |
| 证书下载失败 | 截图留存, status=-3, 下轮重试 |
| Record 确认报错 | 截图留存, status=-3, 下轮重试 |
| COS 上传失败 | 记录 warning，不影响主流程 |
| 下号超时 (N 周) | 企业微信通知人工跟进 |
| 流水线异常崩溃 | cleanup_after_error 释放认领，回退 SIGNED |
### 14.11 与流程三的衔接关系
```
流程二 status=2 (SUBMIT_SUCCESS)
│
▼ 1-2 天 (等待 Ecologic 发送签字邮件)
流程三 status=3 (SIGNED)
│
▼ 1-2 周 (等待 Ecologic 审批出号)
流程四 status=4 (UIN_ISSUED, 终态)
│
├─ 步骤 A: 更新 epr_fr_weee_register 表
│ └─ uin_number, certificate_url, uin_issued_at, status=4
│
├─ 步骤 B: 更新 SaaS EPRRegInfo 表
│ ├─ PushTaxBureauStatus = 4
│ └─ RegBackNumber = Ecologic UIN 注册号
│
└─ 步骤 C: 调用 SaaS SaveBusinessForFRWEEE 接口
└─ POST 保存业务记录 (id, eprCode, certificateUrl)
详见 14.12 节
```
流程四选取记录条件：`status=3 AND uin_number IS NULL AND signed_at IS NOT NULL AND signed_at + 5天 <= 当前时间 AND uin_retry_count < 7`，按 `signed_at` 升序处理，**一次返回全部符合条件的记录**。基类 `run_once` 循环逐条处理：跳过本轮已处理过的记录继续取未处理的，直到本批全部处理完（单轮内每条记录至多处理一次；`--once` 与服务调度共用同一逻辑）。
**调度方式**：使用 `PollerScheduler.schedule_cron()` 以 CronTrigger 每天 8:00 和 17:30 触发（两个独立的 cron job：`epr-fr-weee-uin-08` 和 `epr-fr-weee-uin-17`）。
**企微通知规则**：
- 每轮开始时发一条 "现在开始查询下号" @15817402851
- 每轮开头扫描所有签字满 15 天未下号记录，逐条发超期报警
- 下号成功 → 正常通知（含下号失败时的 SaaS 回写失败信息）
- 下号 PENDING（还没下号）→ 不发通知
- 下号 FAILED 但签字 < 15 天 → 不发通知
- 下号 FAILED 且签字 ≥ 15 天 → 发超期报警通知
- 所有通知均 @bu_f_mobile + @15817402851
### 14.12 SaaS 回写与 SaveBusinessForFRWEEE 接口调用
下号成功（status=4）后，需要执行三步 SaaS 侧操作：
| 步骤 | 操作 | 说明 |
|------|------|------|
| 1 | UPDATE `EPRRegInfo` | `PushTaxBureauStatus` = 4, `RegBackNumber` = Ecologic UIN 注册号 |
| 2 | POST `SaveBusinessForFRWEEE` | 调用 SaaS 接口保存业务记录 |
#### SaveBusinessForFRWEEE 接口
**请求方式和实例：**
```
POST http://localhost:20472/EPRBusiness/EPRBusinessRecord/SaveBusinessForFRWEEE
Content-Type: application/json
X-Requested-With: XMLHttpRequest
实例:
curl --location --request POST 'https://backend.usaeu.com/EPRBusiness/EPRBusinessRecord/SaveBusinessForFRWEEE?id=20a4749b-83a9-4118-bfe9-91efa5215b60&eprCode=FR491888_05NRUQ&certificateUrl=https://vat-1259285998.cos.ap-guangzhou.myqcloud.com/epr/fr/weee/cer/深圳市红圣网络科技有限公司-法国WEEE证书.pdf' \
--header 'Pragma: no-cache' \
--header 'x-requested-with: XMLHttpRequest' \
--header 'Cookie: hncjpms_Mark=ec065669-cb39-4704-a597-80b35e139bad; __RequestVerificationToken=tS3ho4pqc5ESI3K7udHQw84wuHbDyefwrrLFk15xQfJTQ15uqUq5_SpB18AQhLBE0rSMMciJJT3SIQ5jnX21tBmZThnw3zHaJCef3qb8Pjw1; ASP.NET_SessionId=yoci2y1h3yeo3ccke04xkzoo' \
--header 'Content-Type: application/json' \
--data-raw '{}'
```
**请求体：**
| 参数 | 类型 | 说明 |
|------|------|------|
| `id` | string | 单据 ID（EPRBusinessRecord.ID） |
| `eprCode` | string | EPR 码（Ecologic UIN 注册号） |
| `certificateUrl` | string | 证书 OSS URL |
**Python 调用参考实现**（参照 `save_business_de_package` 模式）：
```python
def save_business_fr_weee(record_id, epr_code, certificate_url):
"""调用 SaveBusinessForFRWEEE 接口，返回 (success, info, data)"""
url = "http://localhost:20472/EPRBusiness/EPRBusinessRecord/SaveBusinessForFRWEEE"
headers = {
"Pragma": "no-cache",
"x-requested-with": "XMLHttpRequest",
"Content-Type": "application/json"
}
payload = {
"id": record_id,
"eprCode": epr_code,
"certificateUrl": certificate_url
}
try:
response = requests.post(
url,
headers=headers,
data=json.dumps(payload),
timeout=30
)
print(f"[HTTP] 状态码: {response.status_code}")
result = response.json()
print(f"[HTTP] 返回结果: {json.dumps(result, ensure_ascii=False, indent=2)}")
code = result.get("code")
info = result.get("info", "未知响应")
data = result.get("data", {})
if code == 200:
return True, info, data
else:
return False, info, data
except requests.exceptions.Timeout:
msg = "请求超时，接口未在30秒内响应"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
except requests.exceptions.ConnectionError:
msg = "连接失败，请检查网络或接口地址是否正确"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
except json.JSONDecodeError:
msg = f"接口返回内容无法解析为JSON，原始内容: {response.text[:200]}"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
except Exception as e:
msg = f"请求异常: {str(e)}"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
```
**完整的下号成功 SaaS 回写流程：**
```python
# 仅当 UIN 获取成功（status=SUCCESS）时执行 SaaS 回写
if result.status == S.SUCCESS:
uin_number = result.uin_number
certificate_oss_url = result.certificate_oss_url
# 步骤 B: 更新 SaaS EPRRegInfo 表
update_saas_eprreginfo(
epr_reg_info_id=record.EPRRegInfoID,
push_tax_bureau_status=4,
reg_back_number=uin_number
)
# 步骤 C: 调用 SaveBusinessForFRWEEE
success, info, data = save_business_fr_weee(
record_id=record.VATBusinessID,
epr_code=uin_number,
certificate_url=certificate_oss_url
)
if not success:
# SaaS 接口调用失败，记录 warning 但不影响本地状态
logger.warning(f"SaaS SaveBusinessForFRWEEE 调用失败: {info}")
```
> **注意**：`SaveBusinessForFRWEEE` 接口调用失败不阻塞流程，仅记录 warning 日志。因为本地 `epr_fr_weee_register` 表已正确更新为 status=4（终态），SaaS 侧可通过重试或人工补录处理。
#### 完整编排函数参考（参照 `save_business_de_package` 模式）
```python
import requests
import json
import pymssql
from datetime import datetime
def get_current_datetime_formatted():
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
def update_db_status(tid, status, msg='',
db_host='localhost', db_user='sa',
db_password='password', db_database='vat', db_port=1433):
"""更新数据库状态，记录 SaaS 接口调用结果"""
connection = None
try:
connection = pymssql.connect(
server=db_host,
user=db_user,
password=db_password,
database=db_database,
port=db_port,
charset='utf8',
as_dict=True
)
cursor = connection.cursor()
update_sql = """
UPDATE epr_fr_weee_register
SET status_saas_url = %s, saas_url_msg = %s
WHERE tid = %s
"""
update_params = (status, msg, tid)
cursor.execute(update_sql, update_params)
rows_affected = cursor.rowcount
connection.commit()
if rows_affected > 0:
print(f"[DB] 成功更新记录 tid={tid}, status={status}")
return True
else:
print(f"[DB] 未找到记录 tid={tid}")
return False
except pymssql.Error as e:
print(f"[DB ERROR] 数据库操作失败: {e}")
if connection:
connection.rollback()
return False
except Exception as e:
print(f"[DB ERROR] 处理异常: {e}")
if connection:
connection.rollback()
return False
finally:
if connection:
cursor.close()
connection.close()
def save_business_fr_weee(record_id, register_num, osspath):
"""调用 SaveBusinessForFRWEEE 接口，返回 (success, info, data)"""
url = "http://localhost:20472/EPRBusiness/EPRBusinessRecord/SaveBusinessForFRWEEE"
headers = {
"Pragma": "no-cache",
"x-requested-with": "XMLHttpRequest",
"Cookie": "hncjpms_Mark=ec065669-cb39-4704-a597-80b35e139bad; __RequestVerificationToken=tS3ho4pqc5ESI3K7udHQw84wuHbDyefwrrLFk15xQfJTQ15uqUq5_SpB18AQhLBE0rSMMciJJT3SIQ5jnX21tBmZThnw3zHaJCef3qb8Pjw1; ASP.NET_SessionId=yoci2y1h3yeo3ccke04xkzoo",
"Content-Type": "application/json"
}
payload = {
"id": record_id,
"eprCode": register_num,
"certificateUrl": osspath
}
try:
response = requests.post(
url,
headers=headers,
data=json.dumps(payload),
timeout=30
)
print(f"[HTTP] 状态码: {response.status_code}")
result = response.json()
print(f"[HTTP] 返回结果: {json.dumps(result, ensure_ascii=False, indent=2)}")
code = result.get("code")
info = result.get("info", "未知响应")
data = result.get("data", {})
if code == 200:
return True, info, data
else:
return False, info, data
except requests.exceptions.Timeout:
msg = "请求超时，接口未在30秒内响应"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
except requests.exceptions.ConnectionError:
msg = "连接失败，请检查网络或接口地址是否正确"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
except json.JSONDecodeError:
msg = f"接口返回内容无法解析为JSON，原始内容: {response.text[:200]}"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
except Exception as e:
msg = f"请求异常: {str(e)}"
print(f"[HTTP ERROR] {msg}")
return False, msg, {}
def run(tid, record_id, register_num, osspath='',
db_host='localhost', db_user='sa',
db_password='password', db_database='vat', db_port=1433):
"""编排入口：调用 SaaS 接口 → 更新本地状态"""
print(f"\n{'='*50}")
print(f"[START] {get_current_datetime_formatted()} 开始处理 tid={tid}")
print(f"[PARAM] record_id={record_id}, register_num={register_num}")
print(f"[PARAM] osspath={osspath}")
success, info, data = save_business_fr_weee(record_id, register_num, osspath)
db_kwargs = dict(
db_host=db_host, db_user=db_user,
db_password=db_password, db_database=db_database, db_port=db_port
)
if success:
print(f"[SUCCESS] 接口调用成功: {info}")
update_db_status(tid, status=2, msg='', **db_kwargs)
else:
print(f"[FAIL] 接口调用失败: {info}")
update_db_status(tid, status=-1, msg=info, **db_kwargs)
print(f"[END] {get_current_datetime_formatted()} 处理完成")
print(f"{'='*50}\n")
# # ========== 调用入口 ==========
# run(
# tid = "", # BusinessSerialNumber
# record_id = "", # EPRBusinessRecord.ID
# register_num = "", # Ecologic UIN 注册号
# osspath = "", # 证书 OSS URL
# db_host = "localhost",
# db_user = "sa",
# db_password = "password",
# db_database = "vat",
# db_port = 1433
# )
```
### 14.13 已知问题与修复：Ecologic 页面引用已停运的 RawGit CDN
**现象**：服务器上 `run-uin-poller --once` / 服务调度（08:00 / 17:30）全部报 `TimeoutError: Timeout 90000ms exceeded`，最终企微通知「下号流程异常: pipeline 错误，已释放认领回退至 SIGNED」。
**根因**：Ecologic 的 `/compte-manager` 页面引用了两个**同步 `